Client Library and Advanced Networking

Learning Objectives

• Master connection pooling for high-performance clients
• Implement intelligent retry logic with exponential backoff
• Build circuit breakers to prevent cascading failures
• Use pipelining to achieve 10x throughput improvements
• Handle network errors and timeouts gracefully
• Create production-ready client libraries

Lesson 8.1: Client Implementation

Connection Management

A client needs to manage connections to the database server efficiently. Let's start with basic connection handling and evolve to production-grade pooling.

Basic Connection

// Simple client without pooling
type Client struct {
    conn net.Conn
    reader *bufio.Reader
    writer *bufio.Writer
}

func NewClient(addr string) (*Client, error) {
    conn, err := net.Dial("tcp", addr)
    if err != nil {
        return nil, err
    }
    
    return &Client{
        conn: conn,
        reader: bufio.NewReader(conn),
        writer: bufio.NewWriter(conn),
    }, nil
}

func (c *Client) Get(key string) (string, error) {
    // Send GET command
    cmd := fmt.Sprintf("*2\r\n$3\r\nGET\r\n$%d\r\n%s\r\n", len(key), key)
    if _, err := c.writer.WriteString(cmd); err != nil {
        return "", err
    }
    if err := c.writer.Flush(); err != nil {
        return "", err
    }
    
    // Parse response
    marker, _ := c.reader.ReadByte()
    
    if marker == '-' {
        // Error response
        line, _ := bufio.NewReader(c.reader).ReadString('\n')
        return "", fmt.Errorf(line)
    }
    
    if marker == '$' {
        // Bulk string
        line, _ := bufio.NewReader(c.reader).ReadString('\n')
        length, _ := strconv.Atoi(strings.TrimSpace(line))
        
        if length == -1 {
            return "", nil  // Null
        }
        
        data := make([]byte, length)
        c.reader.Read(data)
        c.reader.ReadBytes('\n')  // Read trailing \r\n
        
        return string(data), nil
    }
    
    return "", fmt.Errorf("unexpected response type: %c", marker)
}

func (c *Client) Close() error {
    return c.conn.Close()
}

Problem: One connection per client means:

• Limited concurrency (connection blocks on read)
• New connection overhead for each operation
• No connection reuse

Connection Pooling

Connection Pool: Maintain multiple reusable connections

// Pool of connections to single server
type ConnPool struct {
    addr        string
    conns       chan net.Conn
    maxConns    int
    idleTimeout time.Duration
    closed      bool
    mu          sync.Mutex
}

func NewConnPool(addr string, maxConns int) *ConnPool {
    return &ConnPool{
        addr:        addr,
        conns:       make(chan net.Conn, maxConns),
        maxConns:    maxConns,
        idleTimeout: 5 * time.Minute,
    }
}

// Get retrieves connection from pool or creates new
func (cp *ConnPool) Get(ctx context.Context) (net.Conn, error) {
    select {
    case conn := <-cp.conns:
        // Connection available from pool
        
        // Test if still alive (optional)
        if err := cp.testConn(conn); err == nil {
            return conn, nil
        }
        conn.Close()
        // Fall through to create new
        
    case <-ctx.Done():
        return nil, ctx.Err()
    default:
    }
    
    // Create new connection
    conn, err := net.Dial("tcp", cp.addr)
    if err != nil {
        return nil, err
    }
    
    return conn, nil
}

// Put returns connection to pool
func (cp *ConnPool) Put(conn net.Conn) error {
    cp.mu.Lock()
    if cp.closed {
        cp.mu.Unlock()
        return conn.Close()
    }
    cp.mu.Unlock()
    
    select {
    case cp.conns <- conn:
        return nil
    default:
        // Pool full, close connection
        return conn.Close()
    }
}

// testConn verifies connection is still alive
func (cp *ConnPool) testConn(conn net.Conn) error {
    // Set short timeout for test
    conn.SetDeadline(time.Now().Add(100 * time.Millisecond))
    defer conn.SetDeadline(time.Time)
    
    // Send PING (simple test)
    fmt.Fprintf(conn, "*1\r\n$4\r\nPING\r\n")
    
    // Read response
    reader := bufio.NewReader(conn)
    resp, err := reader.ReadString('\n')
    return err
}

Pooled Client

type PooledClient struct {
    pool    *ConnPool
    timeout time.Duration
}

func NewPooledClient(addr string, maxConns int) *PooledClient {
    return &PooledClient{
        pool:    NewConnPool(addr, maxConns),
        timeout: 5 * time.Second,
    }
}

func (pc *PooledClient) Get(key string) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), pc.timeout)
    defer cancel()
    
    conn, err := pc.pool.Get(ctx)
    if err != nil {
        return "", err
    }
    defer pc.pool.Put(conn)
    
    // Send command and read response
    return pc.sendCommand(conn, "GET", key)
}

func (pc *PooledClient) Close() error {
    return pc.pool.Close()
}

Retry Logic and Exponential Backoff

Problem: Transient failures (temporary network issues)

Attempt 1: Connection refused
Attempt 2 (wait 10ms): Connection refused
Attempt 3 (wait 100ms): Success!

// RetryConfig configures retry behavior
type RetryConfig struct {
    MaxAttempts int
    InitialWait time.Duration
    MaxWait     time.Duration
}

// Retry executes operation with exponential backoff
func Retry(cfg RetryConfig, fn func() error) error {
    var lastErr error
    wait := cfg.InitialWait
    
    for attempt := 0; attempt < cfg.MaxAttempts; attempt++ {
        if attempt > 0 {
            time.Sleep(wait)
            // Exponential backoff: wait = wait * 2, capped at MaxWait
            if wait < cfg.MaxWait {
                wait *= 2
                if wait > cfg.MaxWait {
                    wait = cfg.MaxWait
                }
            }
        }
        
        err := fn()
        if err == nil {
            return nil
        }
        
        // Check if error is retryable
        if !isRetryable(err) {
            return err
        }
        
        lastErr = err
    }
    
    return fmt.Errorf("max retries exceeded: %w", lastErr)
}

// isRetryable determines if error should trigger retry
func isRetryable(err error) bool {
    if err == nil {
        return false
    }
    
    // Network errors are retryable
    if ne, ok := err.(net.Error); ok {
        return ne.Temporary() || ne.Timeout()
    }
    
    // Check error message
    errStr := err.Error()
    retryable := []string{
        "connection refused",
        "connection reset",
        "i/o timeout",
        "temporary failure",
    }
    
    for _, msg := range retryable {
        if strings.Contains(strings.ToLower(errStr), msg) {
            return true
        }
    }
    
    return false
}

Circuit Breaker Pattern

Problem: Retry loop can overwhelm failing server

Server crashes:
  Client retries → Sends traffic to dead server
  More clients retry → Thundering herd
  Server recovers but overloaded

Solution: Circuit Breaker - Stops sending requests when failures exceed threshold

// CircuitBreaker prevents cascading failures
type CircuitBreaker struct {
    maxFailures  int
    resetTimeout time.Duration
    
    failures     int
    lastFailTime time.Time
    state        CircuitState
    mu           sync.RWMutex
}

type CircuitState int

const (
    CircuitClosed CircuitState = iota  // Normal operation
    CircuitOpen                        // Too many failures
    CircuitHalfOpen                    // Testing recovery
)

func NewCircuitBreaker(maxFailures int, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        maxFailures:  maxFailures,
        resetTimeout: resetTimeout,
        state:        CircuitClosed,
    }
}

// Do executes operation, managing circuit state
func (cb *CircuitBreaker) Do(fn func() error) error {
    cb.mu.Lock()
    
    // Check if should reset from Open to HalfOpen
    if cb.state == CircuitOpen {
        if time.Since(cb.lastFailTime) > cb.resetTimeout {
            cb.state = CircuitHalfOpen
            cb.failures = 0
        } else {
            cb.mu.Unlock()
            return ErrCircuitOpen
        }
    }
    
    cb.mu.Unlock()
    
    // Execute operation
    err := fn()
    
    cb.mu.Lock()
    defer cb.mu.Unlock()
    
    if err != nil {
        cb.failures++
        cb.lastFailTime = time.Now()
        
        if cb.failures >= cb.maxFailures {
            cb.state = CircuitOpen
            return fmt.Errorf("circuit open after %d failures: %w", cb.failures, err)
        }
        
        return err
    }
    
    // Success: reset circuit
    cb.failures = 0
    if cb.state == CircuitHalfOpen {
        cb.state = CircuitClosed
    }
    
    return nil
}

Lesson 8.2: Advanced Networking Features

Pipelining for Throughput

Problem: Without pipelining:

Client sends command 1  →
                        ← Server sends response 1
Client sends command 2  →
                        ← Server sends response 2
Client sends command 3  →
                        ← Server sends response 3

Round-trip latency: 3 × 10ms = 30ms

Solution: Pipelining sends multiple commands before waiting for responses

Client sends commands 1,2,3 →
                             ← Server sends responses 1,2,3

Latency: 1 × 10ms = 10ms (3x faster!)

// Pipeline sends multiple commands without waiting
type Pipeline struct {
    conn    net.Conn
    writer  *bufio.Writer
    reader  *bufio.Reader
    commands []*Command
}

type Command struct {
    Name string
    Args []string
    Response chan interface
    Error chan error
}

func NewPipeline(conn net.Conn) *Pipeline {
    return &Pipeline{
        conn:     conn,
        writer:   bufio.NewWriter(conn),
        reader:   bufio.NewReader(conn),
        commands: make([]*Command, 0),
    }
}

// Add queues command for execution
func (p *Pipeline) Add(name string, args ...string) <-chan interface {
    cmd := &Command{
        Name:     name,
        Args:     args,
        Response: make(chan interface, 1),
        Error:    make(chan error, 1),
    }
    
    p.commands = append(p.commands, cmd)
    return cmd.Response
}

// Execute sends all queued commands and reads responses
func (p *Pipeline) Execute(ctx context.Context) error {
    // Send all commands
    for _, cmd := range p.commands {
        req := p.formatCommand(cmd)
        if _, err := p.writer.WriteString(req); err != nil {
            return err
        }
    }
    
    if err := p.writer.Flush(); err != nil {
        return err
    }
    
    // Read all responses
    for _, cmd := range p.commands {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }
        
        resp, err := p.readResponse()
        if err != nil {
            cmd.Error <- err
        } else {
            cmd.Response <- resp
        }
    }
    
    p.commands = p.commands[:0]  // Clear for next pipeline
    return nil
}

Multiplexing Connections

Problem: Multiple goroutines on same connection conflicts

// BAD: Race condition
conn := shared connection
go func() {
    conn.Write(command1)
    conn.Read(response1)
}()
go func() {
    conn.Write(command2)
    conn.Read(response2)
}()
// Commands/responses may interleave!

Solution: Multiplex with request ID matching

// Multiplexer allows concurrent operations on single connection
type Multiplexer struct {
    conn        net.Conn
    requestID   uint64
    responses   map[uint64]chan interface
    mu          sync.Mutex
}

// Request with unique ID
type MuxRequest struct {
    ID    uint64
    Name  string
    Args  []string
}

// Send sends request and returns channel for response
func (m *Multiplexer) Send(name string, args ...string) (<-chan interface, error) {
    m.mu.Lock()
    
    requestID := m.requestID
    m.requestID++
    
    respCh := make(chan interface, 1)
    m.responses[requestID] = respCh
    
    m.mu.Unlock()
    
    // Send request with ID
    req := fmt.Sprintf("%d %s %s\r\n", requestID, name, strings.Join(args, " "))
    _, err := m.conn.Write([]byte(req))
    if err != nil {
        m.mu.Lock()
        delete(m.responses, requestID)
        m.mu.Unlock()
        return nil, err
    }
    
    return respCh, nil
}

Request Timeouts

// Timeout ensures requests don't hang forever
func (pc *PooledClient) GetWithTimeout(key string, timeout time.Duration) (string, error) {
    ctx, cancel := context.WithTimeout(context.Background(), timeout)
    defer cancel()
    
    resultCh := make(chan string, 1)
    errCh := make(chan error, 1)
    
    go func() {
        val, err := pc.Get(key)
        if err != nil {
            errCh <- err
        } else {
            resultCh <- val
        }
    }()
    
    select {
    case result := <-resultCh:
        return result, nil
    case err := <-errCh:
        return "", err
    case <-ctx.Done():
        return "", fmt.Errorf("timeout exceeded")
    }
}

Lesson 8.3: Error Handling

Network Error Types

import "errors"
import "net"

// Classify network errors
func classifyError(err error) string {
    if err == nil {
        return "none"
    }
    
    if errors.Is(err, io.EOF) {
        return "connection_closed"
    }
    
    if errors.Is(err, context.DeadlineExceeded) {
        return "timeout"
    }
    
    if ne, ok := err.(net.Error); ok {
        if ne.Timeout() {
            return "timeout"
        }
        if ne.Temporary() {
            return "temporary"
        }
        return "permanent"
    }
    
    return "unknown"
}

// Common network errors
var (
    ErrConnectionRefused = fmt.Errorf("connection refused")
    ErrConnectionReset   = fmt.Errorf("connection reset")
    ErrTimeout           = fmt.Errorf("operation timeout")
    ErrCircuitOpen       = fmt.Errorf("circuit breaker open")
)

Partial Failure Handling

// Multi-key operation with partial failures
func (pc *PooledClient) MGet(keys ...string) (map[string]string, error) {
    results := make(map[string]string)
    var lastErr error
    
    for _, key := range keys {
        val, err := pc.Get(key)
        if err != nil {
            lastErr = err
            continue
        }
        results[key] = val
    }
    
    // Return partial results even if some failed
    return results, lastErr
}

Client-Side Retries

// Transient vs permanent errors
func isTransient(err error) bool {
    if err == nil {
        return false
    }
    
    if ne, ok := err.(net.Error); ok {
        return ne.Temporary() || ne.Timeout()
    }
    
    // Check for specific errors
    errMsg := err.Error()
    transient := []string{
        "ECONNREFUSED",
        "ETIMEDOUT",
        "ECONNRESET",
        "temporary failure",
    }
    
    for _, msg := range transient {
        if strings.Contains(errMsg, msg) {
            return true
        }
    }
    
    return false
}

// Retry with backoff
func (pc *PooledClient) GetWithRetry(key string, maxRetries int) (string, error) {
    var lastErr error
    
    for attempt := 0; attempt < maxRetries; attempt++ {
        if attempt > 0 {
            backoff := time.Duration(math.Pow(2, float64(attempt-1))) * 100 * time.Millisecond
            time.Sleep(backoff)
        }
        
        val, err := pc.Get(key)
        if err == nil {
            return val, nil
        }
        
        if !isTransient(err) {
            return "", err  // Permanent error, don't retry
        }
        
        lastErr = err
    }
    
    return "", fmt.Errorf("max retries exceeded: %w", lastErr)
}

Lab 8.1: Build Production-Ready Client Library

Objective

Implement a feature-complete, production-grade client library with pooling, retries, and advanced features.

Requirements

• Connection Pooling: Configurable pool size, connection reuse, idle cleanup
• Retry Logic: Exponential backoff, configurable max attempts, transient error detection
• Circuit Breaker: Prevent cascading failures, automatic recovery, state tracking
• Pipelining: Batch multiple commands, single flush to server, collect all responses
• Error Handling: Proper error types, timeout handling, partial failure support
• Testing: Test with real server, test error scenarios, test concurrent operations

Starter Code

package client

import (
    "bufio"
    "context"
    "net"
    "sync"
    "time"
)

// Client is a production-ready database client
type Client struct {
    // TODO: Add connection pool, circuit breaker, retry config
}

// NewClient creates a new client
// TODO: Implement with sensible defaults
func NewClient(addr string, opts ...Option) (*Client, error) {
    return nil, nil
}

// Get retrieves a value with automatic retry and timeout
// TODO: Implement with connection pooling
func (c *Client) Get(key string) (string, error) {
    return "", nil
}

// Set stores a value
// TODO: Implement
func (c *Client) Set(key, value string) error {
    return nil
}

// Delete removes a key
// TODO: Implement
func (c *Client) Delete(key string) error {
    return nil
}

// Pipeline returns a new pipeline for batching
// TODO: Implement
func (c *Client) Pipeline() *Pipeline {
    return nil
}

// Close gracefully closes the client
// TODO: Implement
func (c *Client) Close() error {
    return nil
}

// Option configures client behavior
type Option func(*Client)

// WithPoolSize sets connection pool size
func WithPoolSize(size int) Option {
    return func(c *Client) {
        // TODO: Implement
    }
}

// WithTimeout sets operation timeout
func WithTimeout(t time.Duration) Option {
    return func(c *Client) {
        // TODO: Implement
    }
}

// WithRetry configures retry behavior
func WithRetry(maxAttempts int, initialWait time.Duration) Option {
    return func(c *Client) {
        // TODO: Implement
    }
}

Test Template

func TestClientPooling(t *testing.T) {
    // Start test server
    server := startTestServer()
    defer server.Stop()
    
    // Create client with pool size 5
    client := NewClient(server.Addr(), WithPoolSize(5))
    defer client.Close()
    
    // Concurrent operations
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(idx int) {
            defer wg.Done()
            key := fmt.Sprintf("key-%d", idx)
            client.Set(key, "value")
            val, _ := client.Get(key)
            assert.Equal(t, "value", val)
        }(i)
    }
    
    wg.Wait()
}

func TestClientRetry(t *testing.T) {
    // Use flaky server that fails then succeeds
    server := startFlakyServer(failCount: 2)
    defer server.Stop()
    
    client := NewClient(server.Addr(), WithRetry(5, 10*time.Millisecond))
    
    // Should succeed after retries
    val, err := client.Get("key")
    assert.NoError(t, err)
}

func BenchmarkClientThroughput(b *testing.B) {
    server := startTestServer()
    defer server.Stop()
    
    client := NewClient(server.Addr())
    defer client.Close()
    
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        client.Set("key", "value")
    }
}

Acceptance Criteria

✅ All tests pass
✅ Connection pooling working
✅ Retries succeed with flaky servers
✅ Circuit breaker prevents cascading failures
✅ Pipelining achieves 10x throughput improvement
✅ 100K+ operations/sec on single client
✅ > 85% code coverage
✅ No goroutine leaks

Summary: Week 8 Complete

By completing Week 8, you've learned and implemented:

1. Connection Management

• Simple connections
• Connection pooling with reuse
• Idle connection cleanup
• Connection testing

2. Retry Logic

• Exponential backoff
• Transient error detection
• Max retry limits
• Backoff strategies

3. Circuit Breaker Pattern

• Prevents cascading failures
• State management
• Automatic recovery
• Failure threshold tracking

4. Advanced Features

• Pipelining: 10x throughput
• Multiplexing: Concurrent ops
• Timeouts: Prevent hanging
• Keep-alive: Detect dead conns

Metrics Achieved:

✅ 100K+ ops/sec with pooling
✅ 10x throughput with pipelining
✅ Sub-100ms failover with circuit breaker
✅ <1% error rate with retries
✅ Support for 1000+ concurrent clients

Module 4 Complete: Full Client-Server System

Congratulations! You now have a complete client-server system:

✅ Week 7: TCP Server

• RESP protocol implementation
• 10,000+ concurrent connections
• Command execution framework
• Graceful shutdown

✅ Week 8: Client Library

• Connection pooling (5-10x throughput)
• Intelligent retries (99%+ success rate)
• Circuit breaking (prevents cascades)
• Pipelining (10x throughput)

Ready for Module 5?

Next we'll implement advanced features including transactions with MVCC, replication, and high availability.

Continue to Module 5: Advanced Features →

Course Navigation

All Lessons