Go Concurrency Web

Go Concurrency for Web Applications

Quick Reference

Core Rules

1. Goroutines are cheap but not free — each goroutine consumes ~2-8 KB of stack. Unbounded spawning under load leads to OOM. 2. Always have a shutdown path — every goroutine you start must have a way to exit. Use context.Context, channel closing, or sync.WaitGroup. 3. Prefer channels for communication — use channels to coordinate work between goroutines and signal completion. 4. Use mutexes for state protection — when goroutines share mutable state, protect it with sync.Mutex, sync.RWMutex, or sync/atomic. 5. Never spawn raw goroutines in HTTP handlers — use worker pools, errgroup, or other bounded concurrency primitives.

Gates (check before merge or review)

Use these sequenced checks for objective pass/fail; do not replace them with “I verified mentally.”

1. Race detector

• Run go test -race ./... on packages that changed concurrent code, or go build -race for binaries under test.
• Pass: exit code 0. If you report “no races,” attach or cite CI output / saved terminal transcript—do not assert cleanliness without that artifact.

2. Bounded background work from HTTP

• Inspect handlers and middleware that start work beyond the request goroutine.
• Pass: every such path uses a bounded primitive (worker pool, buffered channel with documented capacity, errgroup with an explicit concurrency cap)—not unbounded go per incoming request.

3. Graceful teardown

• For processes that start long-lived goroutines, trace from shutdown signal (or test defer) to Wait() / channel close / context cancel for each goroutine family.
• Pass: you can point to the call chain or a test that proves shutdown completes without hang (no orphan goroutines).

Worker Pool Pattern

Use worker pools for background tasks dispatched from HTTP handlers. This bounds concurrency and provides graceful shutdown.

// Worker pool for background tasks (e.g., sending emails)
type WorkerPool struct {
    jobs   chan Job
    wg     sync.WaitGroup
    logger *slog.Logger
}

type Job struct {
    ID      string
    Execute func(ctx context.Context) error
}

func NewWorkerPool(numWorkers int, queueSize int, logger *slog.Logger) *WorkerPool {
    wp := &WorkerPool{
        jobs:   make(chan Job, queueSize),
        logger: logger,
    }

    for i := 0; i < numWorkers; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }

    return wp
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        wp.logger.Info("processing job", "worker", id, "job_id", job.ID)
        if err := job.Execute(context.Background()); err != nil {
            wp.logger.Error("job failed", "worker", id, "job_id", job.ID, "err", err)
        }
    }
}

func (wp *WorkerPool) Submit(job Job) {
    wp.jobs <- job
}

func (wp *WorkerPool) Shutdown() {
    close(wp.jobs)
    wp.wg.Wait()
}

Usage in HTTP Handler

func (s *Server) handleCreateUser(w http.ResponseWriter, r *http.Request) {
    user, err := s.userService.Create(r.Context(), decodeUser(r))
    if err != nil {
        handleError(w, r, err)
        return
    }

    // Dispatch background task — never spawn raw goroutines in handlers
    s.workers.Submit(Job{
        ID: "welcome-email-" + user.ID,
        Execute: func(ctx context.Context) error {
            return s.emailService.SendWelcome(ctx, user)
        },
    })

    writeJSON(w, http.StatusCreated, user)
}

See references/worker-pools.md for sizing guidance, backpressure, error handling, retry patterns, and errgroup as a simpler alternative.

Rate Limiting

Use golang.org/x/time/rate for token bucket rate limiting. Apply as middleware for global limits or per-IP/per-user limits.

Key points:

• Global rate limiting protects overall service capacity
• Per-IP rate limiting prevents individual clients from monopolizing resources
• Always return 429 Too Many Requests with a Retry-After header

See references/rate-limiting.md for middleware implementation, per-IP limiting, stale limiter cleanup, and API key-based limiting.

Race Detection

Run the race detector in development and CI:

go test -race ./...
go build -race -o myserver ./cmd/server

The race detector catches concurrent reads and writes to shared memory. It does not catch logical races (e.g., TOCTOU bugs) or deadlocks.

See references/race-detection.md for common web handler races, fixing strategies, and CI integration.

Handler Safety

Every incoming HTTP request runs in its own goroutine. Any shared mutable state on the server struct is a potential data race.

// BAD — shared state without protection
type Server struct {
    requestCount int // data race!
}

func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
    s.requestCount++ // concurrent writes = race condition
}

// GOOD — use atomic or mutex
type Server struct {
    requestCount atomic.Int64
}

func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
    s.requestCount.Add(1)
}

// GOOD — use mutex for complex state
type Server struct {
    mu    sync.RWMutex
    cache map[string]*CachedItem
}

func (s *Server) handleGetCached(w http.ResponseWriter, r *http.Request) {
    s.mu.RLock()
    item, ok := s.cache[r.PathValue("key")]
    s.mu.RUnlock()
    // ...
}

Rules for Handler Safety

• Request-scoped data is safe — r.Context(), request body, URL params are isolated per request.
• Server struct fields are shared — any field on *Server accessed by handlers needs synchronization.
• Database connections are safe — *sql.DB manages its own connection pool with internal locking.
• Maps are not safe — use sync.Map or protect with a mutex.
• Slices are not safe — concurrent append or read/write requires a mutex.

Anti-Patterns

Unbounded goroutine spawning

// BAD — no limit on concurrent goroutines
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    go func() {
        // What if 10,000 requests arrive at once?
        s.processWebhook(r.Context(), decodeWebhook(r))
    }()
    w.WriteHeader(http.StatusAccepted)
}

// GOOD — use a worker pool
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    webhook := decodeWebhook(r)
    s.workers.Submit(Job{
        ID:      "webhook-" + webhook.ID,
        Execute: func(ctx context.Context) error {
            return s.processWebhook(ctx, webhook)
        },
    })
    w.WriteHeader(http.StatusAccepted)
}

Forgetting to propagate context

// BAD — loses cancellation signal
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
    results, err := s.search(context.Background(), r.URL.Query().Get("q"))
    // ...
}

// GOOD — use request context
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
    results, err := s.search(r.Context(), r.URL.Query().Get("q"))
    // ...
}

Goroutine leak from missing channel receiver

// BAD — goroutine blocks forever if nobody reads the channel
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
    ch := make(chan *Response)
    go func() {
        resp, _ := http.Get(url) // blocks forever if ctx cancels
        ch <- resp               // stuck here if nobody reads
    }()
    select {
    case resp := <-ch:
        return resp, nil
    case <-ctx.Done():
        return nil, ctx.Err() // goroutine leaked!
    }
}

// GOOD — use buffered channel so goroutine can exit
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
    ch := make(chan *Response, 1) // buffered — goroutine can always send
    go func() {
        resp, _ := http.Get(url)
        ch <- resp
    }()
    select {
    case resp := <-ch:
        return resp, nil
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

Using time.Sleep for coordination

// BAD — sleeping to wait for goroutines
go doWork()
time.Sleep(5 * time.Second) // hoping it finishes

// GOOD — use sync primitives
var wg sync.WaitGroup
wg.Add(1)
go func() {
    defer wg.Done()
    doWork()
}()
wg.Wait()

Race Detection

Running the Race Detector

The Go race detector instruments memory accesses at compile time and detects concurrent unsynchronized access at runtime.

# Run tests with race detection
go test -race ./...

# Build a binary with race detection (for integration testing)
go build -race -o myserver ./cmd/server

# Run a specific test with race detection
go test -race -run TestHandlerConcurrency ./internal/server/

Always run `-race` in CI. Race conditions are intermittent; the race detector increases the chance of catching them.

What the Race Detector Catches

The race detector detects data races: two goroutines access the same memory location concurrently, and at least one access is a write.

It catches:

• Concurrent map reads and writes
• Concurrent struct field modifications
• Concurrent slice access
• Concurrent variable increments

It does not catch:

• Logical races (TOCTOU) — checking a condition and acting on it non-atomically
• Deadlocks — goroutines waiting on each other forever
• Starvation — a goroutine never gets scheduled
• Race conditions that don't execute — it only detects races that actually occur during the test run

Common Race Conditions in Web Handlers

Race 1: Shared Map Without Lock

// BAD — concurrent map write causes panic
var cache = map[string]string{}

func handler(w http.ResponseWriter, r *http.Request) {
    cache[r.URL.Path] = "value" // concurrent map write!
}

// Fix: use sync.Map or mutex
var cache sync.Map

func handler(w http.ResponseWriter, r *http.Request) {
    cache.Store(r.URL.Path, "value") // safe
}

Note: Concurrent map writes in Go cause a runtime panic, not just incorrect data.

Race 2: Incrementing a Counter

// BAD — data race on counter
var count int

func handler(w http.ResponseWriter, r *http.Request) {
    count++ // data race!
}

// Fix: use atomic
var count atomic.Int64

func handler(w http.ResponseWriter, r *http.Request) {
    count.Add(1) // safe
}

Race 3: Slice Append

// BAD — concurrent append is not safe
type Server struct {
    events []Event
}

func (s *Server) handleEvent(w http.ResponseWriter, r *http.Request) {
    event := decodeEvent(r)
    s.events = append(s.events, event) // data race!
}

// Fix: protect with mutex
type Server struct {
    mu     sync.Mutex
    events []Event
}

func (s *Server) handleEvent(w http.ResponseWriter, r *http.Request) {
    event := decodeEvent(r)
    s.mu.Lock()
    s.events = append(s.events, event)
    s.mu.Unlock()
}

Race 4: Lazy Initialization

// BAD — multiple goroutines may initialize simultaneously
type Server struct {
    client *http.Client
}

func (s *Server) getClient() *http.Client {
    if s.client == nil {        // race: read
        s.client = &http.Client{  // race: write
            Timeout: 10 * time.Second,
        }
    }
    return s.client
}

// Fix: use sync.Once
type Server struct {
    clientOnce sync.Once
    client     *http.Client
}

func (s *Server) getClient() *http.Client {
    s.clientOnce.Do(func() {
        s.client = &http.Client{
            Timeout: 10 * time.Second,
        }
    })
    return s.client
}

Race 5: Read-Modify-Write on Struct Field

// BAD — read and write are separate operations
type Server struct {
    healthy bool
}

func (s *Server) healthCheck(w http.ResponseWriter, r *http.Request) {
    if s.healthy {  // race: read
        writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
    }
}

func (s *Server) setHealthy(healthy bool) {
    s.healthy = healthy  // race: write
}

// Fix: use atomic.Bool
type Server struct {
    healthy atomic.Bool
}

func (s *Server) healthCheck(w http.ResponseWriter, r *http.Request) {
    if s.healthy.Load() {
        writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
    }
}

func (s *Server) setHealthy(healthy bool) {
    s.healthy.Store(healthy)
}

Fixing Races: When to Use What

sync.Mutex vs sync.RWMutex

Use sync.RWMutex when reads significantly outnumber writes:

type Cache struct {
    mu   sync.RWMutex
    data map[string]*Entry
}

// Multiple goroutines can read concurrently
func (c *Cache) Get(key string) (*Entry, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    entry, ok := c.data[key]
    return entry, ok
}

// Only one goroutine can write at a time
func (c *Cache) Set(key string, entry *Entry) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = entry
}

If reads and writes are roughly equal, use sync.Mutex (simpler, less overhead from lock upgrades).

sync.Map vs Mutex-Protected Map

sync.Map is optimized for two patterns: 1. Write-once, read-many (append-only maps) 2. Multiple goroutines read/write disjoint key sets

For everything else, a mutex-protected map is usually faster and provides type safety.

Testing for Race Conditions

Write Concurrent Tests

func TestHandlerConcurrency(t *testing.T) {
    srv := NewServer()
    ts := httptest.NewServer(srv)
    defer ts.Close()

    // Hammer the endpoint from multiple goroutines
    var wg sync.WaitGroup
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            resp, err := http.Get(ts.URL + "/api/data")
            if err != nil {
                t.Errorf("request failed: %v", err)
                return
            }
            resp.Body.Close()
        }()
    }
    wg.Wait()
}

Test Specific Race Scenarios

func TestCacheRace(t *testing.T) {
    cache := NewCache()

    var wg sync.WaitGroup

    // Concurrent writes
    for i := 0; i < 50; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            cache.Set(fmt.Sprintf("key-%d", i), &Entry{Value: i})
        }(i)
    }

    // Concurrent reads
    for i := 0; i < 50; i++ {
        wg.Add(1)
        go func(i int) {
            defer wg.Done()
            cache.Get(fmt.Sprintf("key-%d", i))
        }(i)
    }

    wg.Wait()
}

CI Integration

GitHub Actions

name: Test
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with:
          go-version: '1.22'

      - name: Test with race detector
        run: go test -race -count=1 ./...

      - name: Build with race detector (integration)
        run: go build -race -o ./bin/server ./cmd/server

Key Flags

• -race enables the race detector
• -count=1 disables test caching (ensures fresh run for race detection)
• The race detector adds ~2-10x runtime overhead and ~5-10x memory overhead
• Acceptable for CI and testing; do not ship race-enabled binaries to production

Race Detector Environment Variables

# Customize race detector behavior
GORACE="log_path=/tmp/race.log" go test -race ./...

# Halt on first race (useful in CI)
GORACE="halt_on_error=1" go test -race ./...

# Increase history size for complex programs
GORACE="history_size=7" go test -race ./...

Rate Limiting

Token Bucket Algorithm

golang.org/x/time/rate implements a token bucket rate limiter:

• A bucket holds up to burst tokens
• Tokens are added at a rate of rps (requests per second)
• Each request consumes one token
• If no tokens are available, the request is rejected (or waits)

Example: rate.NewLimiter(10, 20) allows 10 requests/second sustained with bursts up to 20.

Global Rate Limiting

Protect the entire service from being overwhelmed:

// Rate limit middleware using x/time/rate
func RateLimit(rps float64, burst int) func(http.Handler) http.Handler {
    limiter := rate.NewLimiter(rate.Limit(rps), burst)

    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            if !limiter.Allow() {
                w.Header().Set("Retry-After", "1")
                http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
                return
            }
            next.ServeHTTP(w, r)
        })
    }
}

// Usage
mux := http.NewServeMux()
mux.HandleFunc("GET /api/data", s.handleGetData)
handler := RateLimit(100, 200)(mux) // 100 rps, burst of 200

Per-IP Rate Limiting

Prevent individual clients from monopolizing resources:

// Per-IP rate limiting
type IPRateLimiter struct {
    mu       sync.RWMutex
    limiters map[string]*rate.Limiter
    rps      float64
    burst    int
}

func NewIPRateLimiter(rps float64, burst int) *IPRateLimiter {
    return &IPRateLimiter{
        limiters: make(map[string]*rate.Limiter),
        rps:      rps,
        burst:    burst,
    }
}

func (l *IPRateLimiter) GetLimiter(ip string) *rate.Limiter {
    l.mu.RLock()
    limiter, exists := l.limiters[ip]
    l.mu.RUnlock()

    if exists {
        return limiter
    }

    l.mu.Lock()
    defer l.mu.Unlock()

    // Double-check after acquiring write lock
    if limiter, exists = l.limiters[ip]; exists {
        return limiter
    }

    limiter = rate.NewLimiter(rate.Limit(l.rps), l.burst)
    l.limiters[ip] = limiter
    return limiter
}

Per-IP Middleware

func PerIPRateLimit(rps float64, burst int, trustProxy bool) func(http.Handler) http.Handler {
    limiter := NewIPRateLimiter(rps, burst)

    // Start cleanup goroutine
    go limiter.cleanup(5 * time.Minute)

    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ip := extractIP(r, trustProxy)
            if !limiter.GetLimiter(ip).Allow() {
                w.Header().Set("Retry-After", "1")
                http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
                return
            }
            next.ServeHTTP(w, r)
        })
    }
}

// extractIP returns the client IP address from the request.
// WARNING: X-Forwarded-For and X-Real-IP headers can be spoofed by clients.
// Only trust these headers when behind a known reverse proxy that strips/overwrites them.
func extractIP(r *http.Request, trustProxy bool) string {
    if trustProxy {
        // Check X-Forwarded-For for proxied requests
        if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
            // Take the first IP (client IP)
            if idx := strings.Index(xff, ","); idx != -1 {
                return strings.TrimSpace(xff[:idx])
            }
            return strings.TrimSpace(xff)
        }
        // Check X-Real-IP
        if xri := r.Header.Get("X-Real-IP"); xri != "" {
            return xri
        }
    }
    // Fall back to RemoteAddr
    host, _, err := net.SplitHostPort(r.RemoteAddr)
    if err != nil {
        return r.RemoteAddr
    }
    return host
}

Cleaning Up Stale Limiters

Without cleanup, the limiter map grows indefinitely. Remove entries that haven't been used recently:

type trackedLimiter struct {
    limiter  *rate.Limiter
    lastSeen time.Time
}

type IPRateLimiter struct {
    mu       sync.RWMutex
    limiters map[string]*trackedLimiter
    rps      float64
    burst    int
}

func (l *IPRateLimiter) GetLimiter(ip string) *rate.Limiter {
    l.mu.RLock()
    tracked, exists := l.limiters[ip]
    l.mu.RUnlock()

    if exists {
        // Update last seen time under write lock
        l.mu.Lock()
        tracked.lastSeen = time.Now()
        l.mu.Unlock()
        return tracked.limiter
    }

    l.mu.Lock()
    defer l.mu.Unlock()

    // Double-check
    if tracked, exists = l.limiters[ip]; exists {
        tracked.lastSeen = time.Now()
        return tracked.limiter
    }

    limiter := rate.NewLimiter(rate.Limit(l.rps), l.burst)
    l.limiters[ip] = &trackedLimiter{
        limiter:  limiter,
        lastSeen: time.Now(),
    }
    return limiter
}

func (l *IPRateLimiter) cleanup(maxAge time.Duration) {
    ticker := time.NewTicker(maxAge)
    defer ticker.Stop()

    for range ticker.C {
        l.mu.Lock()
        cutoff := time.Now().Add(-maxAge)
        for ip, tracked := range l.limiters {
            if tracked.lastSeen.Before(cutoff) {
                delete(l.limiters, ip)
            }
        }
        l.mu.Unlock()
    }
}

Rate Limiting by API Key or User

For authenticated endpoints, rate limit by user identity instead of IP:

type KeyRateLimiter struct {
    mu       sync.RWMutex
    limiters map[string]*trackedLimiter
    tiers    map[string]Tier // API key -> tier
}

type Tier struct {
    RPS   float64
    Burst int
}

var defaultTiers = map[string]Tier{
    "free":       {RPS: 10, Burst: 20},
    "pro":        {RPS: 100, Burst: 200},
    "enterprise": {RPS: 1000, Burst: 2000},
}

func (l *KeyRateLimiter) GetLimiter(apiKey string) *rate.Limiter {
    l.mu.RLock()
    tracked, exists := l.limiters[apiKey]
    l.mu.RUnlock()

    if exists {
        l.mu.Lock()
        tracked.lastSeen = time.Now()
        l.mu.Unlock()
        return tracked.limiter
    }

    l.mu.Lock()
    defer l.mu.Unlock()

    if tracked, exists = l.limiters[apiKey]; exists {
        tracked.lastSeen = time.Now()
        return tracked.limiter
    }

    tier, ok := l.tiers[apiKey]
    if !ok {
        tier = defaultTiers["free"]
    }

    limiter := rate.NewLimiter(rate.Limit(tier.RPS), tier.Burst)
    l.limiters[apiKey] = &trackedLimiter{
        limiter:  limiter,
        lastSeen: time.Now(),
    }
    return limiter
}

API Key Middleware

func APIKeyRateLimit(keyLimiter *KeyRateLimiter) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            apiKey := r.Header.Get("X-API-Key")
            if apiKey == "" {
                http.Error(w, "missing API key", http.StatusUnauthorized)
                return
            }

            limiter := keyLimiter.GetLimiter(apiKey)
            if !limiter.Allow() {
                w.Header().Set("Retry-After", "1")
                w.Header().Set("X-RateLimit-Limit", fmt.Sprintf("%.0f", float64(limiter.Limit())))
                w.Header().Set("X-RateLimit-Remaining", "0")
                http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
                return
            }

            next.ServeHTTP(w, r)
        })
    }
}

Returning Proper 429 Responses

Always include informative headers when rejecting rate-limited requests:

func rateLimitResponse(w http.ResponseWriter, limiter *rate.Limiter) {
    reservation := limiter.Reserve()
    if !reservation.OK() {
        http.Error(w, "rate limit exceeded", http.StatusTooManyRequests)
        return
    }

    delay := reservation.Delay()
    reservation.Cancel() // We're rejecting, not waiting

    w.Header().Set("Retry-After", fmt.Sprintf("%.0f", delay.Seconds()+1))
    w.Header().Set("X-RateLimit-Limit", fmt.Sprintf("%.0f", float64(limiter.Limit())))
    w.Header().Set("X-RateLimit-Remaining", "0")

    w.WriteHeader(http.StatusTooManyRequests)
    json.NewEncoder(w).Encode(map[string]any{
        "error":       "rate limit exceeded",
        "retry_after": delay.Seconds(),
    })
}

Combining Rate Limiters

Layer global and per-IP limits for defense in depth:

mux := http.NewServeMux()
mux.HandleFunc("GET /api/data", s.handleGetData)

// Apply per-IP first (inner), then global (outer)
handler := RateLimit(1000, 2000)(          // global: 1000 rps
    PerIPRateLimit(10, 20)(                // per-IP: 10 rps
        mux,
    ),
)

This ensures no single IP can use more than 10 rps, while the service overall caps at 1000 rps.

Worker Pools

Why Worker Pools

Worker pools prevent goroutine leaks and OOM by bounding concurrency. Without a pool, every incoming request that spawns a goroutine can create unbounded parallelism:

• 10,000 requests/second = 10,000 goroutines = ~20 MB minimum (stacks start at ~2 KB, grow as needed)
• Each goroutine may hold open database connections, file descriptors, or network sockets
• The Go scheduler slows down with millions of goroutines

A worker pool with N workers guarantees at most N concurrent background tasks, regardless of request volume.

Sizing Workers

CPU-Bound Tasks

For tasks that primarily consume CPU (compression, hashing, image processing):

• Workers = `runtime.NumCPU()` or slightly more
• More workers than CPUs adds context-switching overhead with no throughput gain

I/O-Bound Tasks

For tasks that wait on external services (HTTP calls, database queries, email sending):

• Workers = 10x to 100x the number of CPUs is common
• The bottleneck is the external service, not CPU
• Tune based on the external service's capacity and latency

General Guidance

// CPU-bound: match CPU count
pool := NewWorkerPool(runtime.NumCPU(), 1000, logger)

// I/O-bound: more workers, they spend most time waiting
pool := NewWorkerPool(50, 5000, logger)

Queue Sizing and Backpressure

The buffered channel acts as a queue. Its size determines backpressure behavior:

// Small queue = fast backpressure signal
jobs: make(chan Job, 10)

// Large queue = absorbs bursts but uses more memory
jobs: make(chan Job, 10000)

Handling a Full Queue

When the queue is full, Submit blocks. For HTTP handlers, blocking is usually unacceptable. Use a non-blocking submit:

func (wp *WorkerPool) TrySubmit(job Job) bool {
    select {
    case wp.jobs <- job:
        return true
    default:
        wp.logger.Warn("worker pool full, dropping job", "job_id", job.ID)
        return false
    }
}

// In handler
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    webhook := decodeWebhook(r)
    if !s.workers.TrySubmit(Job{
        ID:      "webhook-" + webhook.ID,
        Execute: func(ctx context.Context) error {
            return s.processWebhook(ctx, webhook)
        },
    }) {
        http.Error(w, "server busy, try again later", http.StatusServiceUnavailable)
        return
    }
    w.WriteHeader(http.StatusAccepted)
}

Graceful Shutdown Integration

Worker pools must integrate with your server's shutdown sequence. Process in-flight jobs before exiting:

func main() {
    logger := slog.Default()
    pool := NewWorkerPool(10, 1000, logger)

    srv := &http.Server{
        Addr:    ":8080",
        Handler: newRouter(pool),
    }

    // Start server
    go func() {
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            logger.Error("server error", "err", err)
        }
    }()

    // Wait for interrupt
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit

    logger.Info("shutting down server...")

    // 1. Stop accepting new requests
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    srv.Shutdown(ctx)

    // 2. Drain the worker pool (finish in-flight jobs)
    logger.Info("draining worker pool...")
    pool.Shutdown()

    logger.Info("shutdown complete")
}

Error Handling and Retry Patterns

Logging Errors

At minimum, log all job failures:

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        start := time.Now()
        if err := job.Execute(context.Background()); err != nil {
            wp.logger.Error("job failed",
                "worker", id,
                "job_id", job.ID,
                "duration", time.Since(start),
                "err", err,
            )
        } else {
            wp.logger.Info("job completed",
                "worker", id,
                "job_id", job.ID,
                "duration", time.Since(start),
            )
        }
    }
}

Retry with Backoff

For transient failures, wrap jobs with retry logic:

type RetryJob struct {
    Job
    MaxRetries int
    Backoff    time.Duration
}

func (wp *WorkerPool) workerWithRetry(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        retryJob, hasRetry := job.(RetryJob) // type assertion if using interface
        maxRetries := 1
        backoff := time.Second

        if hasRetry {
            maxRetries = retryJob.MaxRetries
            backoff = retryJob.Backoff
        }

        var lastErr error
        for attempt := 0; attempt < maxRetries; attempt++ {
            if attempt > 0 {
                time.Sleep(backoff * time.Duration(attempt))
            }
            if err := job.Execute(context.Background()); err != nil {
                lastErr = err
                wp.logger.Warn("job attempt failed",
                    "worker", id,
                    "job_id", job.ID,
                    "attempt", attempt+1,
                    "err", err,
                )
                continue
            }
            lastErr = nil
            break
        }

        if lastErr != nil {
            wp.logger.Error("job exhausted retries",
                "worker", id,
                "job_id", job.ID,
                "err", lastErr,
            )
        }
    }
}

Context Propagation

Workers should use their own context, not the request context. The request context is cancelled when the HTTP response is sent, which happens before the background job runs:

// BAD — request context cancels when response is written
s.workers.Submit(Job{
    ID: "send-email",
    Execute: func(ctx context.Context) error {
        return s.email.Send(r.Context(), user) // r.Context() is already cancelled!
    },
})

// GOOD — use background context with timeout
s.workers.Submit(Job{
    ID: "send-email",
    Execute: func(ctx context.Context) error {
        ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
        defer cancel()
        return s.email.Send(ctx, user)
    },
})

If you need to pass values from the request context (like trace IDs), extract them before submitting:

traceID := traceIDFromContext(r.Context())
userID := userIDFromContext(r.Context())

s.workers.Submit(Job{
    ID: "audit-log-" + userID,
    Execute: func(ctx context.Context) error {
        ctx = withTraceID(ctx, traceID)
        return s.audit.Log(ctx, userID, action)
    },
})

errgroup as a Simpler Alternative

For fan-out/fan-in within a single request (parallel API calls, batch processing), golang.org/x/sync/errgroup is simpler than a worker pool:

// errgroup for bounded parallel work
func (s *Server) handleBatchProcess(w http.ResponseWriter, r *http.Request) {
    items := decodeItems(r)

    g, ctx := errgroup.WithContext(r.Context())
    g.SetLimit(10) // max 10 concurrent goroutines

    results := make([]*Result, len(items))
    for i, item := range items {
        i, item := i, item
        g.Go(func() error {
            result, err := s.processItem(ctx, item)
            if err != nil {
                return fmt.Errorf("processing item %d: %w", i, err)
            }
            results[i] = result
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        handleError(w, r, err)
        return
    }

    writeJSON(w, http.StatusOK, results)
}

When to Use errgroup vs Worker Pool

errgroup Key Points

• g.SetLimit(n) bounds concurrency (available since Go 1.20)
• Context cancellation propagates automatically — if one goroutine returns an error, the context is cancelled for all others
• g.Wait() blocks until all goroutines complete and returns the first error
• Safe to write to results[i] from goroutine i without a mutex because each goroutine writes to a distinct index