Imported from marquesfelip/agents-and-skills (skills/backpressure-patterns/SKILL.md). Install upstream with npx skills add marquesfelip/agents-and-skills --skill backpressure-patterns. Copyright stays with the author.

Backpressure Patterns Specialist

When to Use

Invoke this skill when you need to:

• Prevent a fast producer from overwhelming a slow consumer
• Design bounded queues and buffer constraints
• Implement backpressure in Go channels, message queues, or HTTP pipelines
• Detect and alert on queue depth and consumer lag
• Throttle producers based on downstream capacity signals
• Stabilize a pipeline that collapses under burst load

Step 1 — Understand the Backpressure Model

Backpressure means the consumer signals its capacity upstream, and the producer slows down or stops rather than overwhelming the consumer.

Without backpressure — the failure cascade:

Producer (10,000 events/s)
  → Unbounded queue grows → OOM
  → Consumer can only process 1,000 events/s → queue depth: 9,000/s growth
  → Memory exhausted in ~2 minutes
  → System crash

With backpressure — stable operation:

Producer (10,000 events/s)
  → Bounded queue (capacity: 500 items)
  → Consumer processes at 1,000 events/s
  → Queue full → Producer blocks OR drops OR slows down
  → Producer throttled to ~1,000 events/s
  → System stable at consumer capacity

Backpressure propagation strategies:

Checklist:

• Backpressure strategy chosen before implementation — not left as "we'll see"
• Consumer capacity measured at realistic data volumes — establishes the throughput ceiling
• Producer behavior under full queue defined explicitly — block, drop, or throttle

Step 2 — Bounded Channels and Goroutine Pools (Go)

Unbounded goroutine creation is the most common Go backpressure failure:

// ❌ Unbounded — every incoming message spawns a goroutine; OOM on burst
for msg := range messages {
    go processMessage(msg) // goroutine count = queue depth → memory exhaustion
}

// ✅ Bounded worker pool — fixed concurrency, backpressure via channel
func NewWorkerPool(workers int) (chan<- Message, func()) {
    jobs := make(chan Message, workers*2) // bounded buffer

    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for msg := range jobs {
                processMessage(msg)
            }
        }()
    }

    shutdown := func() {
        close(jobs)
        wg.Wait()
    }
    return jobs, shutdown
}

// Usage: send blocks when pool is saturated (backpressure to caller)
jobs <- msg

Semaphore-based bounded concurrency:

sem := make(chan struct{}, 20) // max 20 concurrent workers

for _, item := range items {
    item := item
    sem <- struct{}{}           // acquire — blocks if 20 already running
    go func() {
        defer func() { <-sem }() // release
        process(item)
    }()
}
// Drain remaining work
for i := 0; i < cap(sem); i++ { sem <- struct{}{} }

errgroup with bounded concurrency:

import "golang.org/x/sync/errgroup"
import "golang.org/x/sync/semaphore"

g, ctx := errgroup.WithContext(ctx)
sem := semaphore.NewWeighted(20)

for _, item := range items {
    item := item
    if err := sem.Acquire(ctx, 1); err != nil {
        break // context cancelled
    }
    g.Go(func() error {
        defer sem.Release(1)
        return process(ctx, item)
    })
}
if err := g.Wait(); err != nil {
    return err
}

Checklist:

• No unbounded goroutine creation in loops — worker pool or semaphore always bounds concurrency
• Channel buffers sized based on consumer throughput × acceptable latency (Little's Law)
• select with a default case used for non-blocking sends — explicit drop, not silent block
• Worker pool size configurable — tunable without code changes (env var or config)

Step 3 — Message Queue Backpressure

Kafka — consumer lag as the backpressure signal:

// Consumer: process messages at a controlled rate
func runConsumer(ctx context.Context, reader *kafka.Reader, workers int) {
    pool, shutdown := NewWorkerPool(workers)
    defer shutdown()

    for {
        msg, err := reader.ReadMessage(ctx) // blocks when no messages
        if err != nil {
            break
        }

        select {
        case pool <- msg:     // send to worker pool
        default:              // pool buffer full — backpressure from workers
            // Option A: Block until space available (strongest backpressure)
            pool <- msg
            // Option B: Pause consumption briefly
            // time.Sleep(10 * time.Millisecond)
        }
    }
}

Kafka partition assignment — control consumer parallelism:

reader := kafka.NewReader(kafka.ReaderConfig{
    Brokers:        []string{"kafka:9092"},
    Topic:          "orders",
    GroupID:        "order-processor",
    MaxBytes:       10e6,           // 10MB max batch
    CommitInterval: time.Second,    // async commit every 1s
    // Match partition count to worker pool size for even distribution
})

RabbitMQ — prefetch count as backpressure:

// Consumer prefetch: only buffer N unacknowledged messages per consumer
ch.Qos(
    20,     // prefetch count — consumer never holds more than 20 unacked messages
    0,      // prefetch size (not used)
    false,  // apply per-consumer (not per-channel)
)
// Without prefetch: consumer tries to load the entire queue into memory

Consumer lag monitoring (Kafka):

# Check lag for all consumer groups
kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --all-groups

# Alert when lag grows — means consumers are falling behind

Prometheus metric for lag:

- alert: KafkaConsumerLagHigh
  expr: kafka_consumer_group_lag{topic="orders"} > 10000
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Orders Kafka consumer lag > 10,000 — consumer is falling behind"
    runbook: "https://runbooks.internal/kafka/consumer-lag"

Checklist:

• Kafka prefetch / RabbitMQ QoS set — consumer does not buffer the entire queue in memory
• Consumer worker pool size matches partition count — all partitions consumed in parallel
• Consumer lag monitored and alerted — lag growth is the primary backpressure health signal
• Manual lag offset reset procedure documented — recovery path when lag is enormous

Step 4 — Pipeline Backpressure (Stage-to-Stage)

For multi-stage processing pipelines, each stage must propagate backpressure to the previous stage.

Ingest → Parse → Validate → Enrich → Persist
  ↑         ↑        ↑         ↑
  backpressure propagates upstream when downstream is slow

Go pipeline with bounded channels:

func buildPipeline(ctx context.Context, input <-chan RawEvent) <-chan error {
    // Each channel is bounded — backpressure propagates stage to stage
    parsed  := make(chan ParsedEvent, 100)
    valid   := make(chan ValidEvent, 100)
    enriched := make(chan EnrichedEvent, 50)

    go parseStage(ctx, input, parsed)
    go validateStage(ctx, parsed, valid)
    go enrichStage(ctx, valid, enriched)

    return persistStage(ctx, enriched)
}

func parseStage(ctx context.Context, in <-chan RawEvent, out chan<- ParsedEvent) {
    defer close(out)
    for raw := range in {
        parsed, err := parse(raw)
        if err != nil {
            continue // or send to dead letter channel
        }
        select {
        case out <- parsed: // blocks if next stage is full — backpressure
        case <-ctx.Done():
            return
        }
    }
}

Monitoring pipeline stage health:

var pipelineChannelDepth = promauto.NewGaugeVec(prometheus.GaugeOpts{
    Namespace: "pipeline", Name: "channel_depth",
    Help: "Current depth of each pipeline stage channel.",
}, []string{"stage"})

// Report channel depths periodically
go func() {
    ticker := time.NewTicker(5 * time.Second)
    for range ticker.C {
        pipelineChannelDepth.WithLabelValues("parse").Set(float64(len(parsed)))
        pipelineChannelDepth.WithLabelValues("validate").Set(float64(len(valid)))
        pipelineChannelDepth.WithLabelValues("enrich").Set(float64(len(enriched)))
    }
}()

Checklist:

• All inter-stage channels are bounded — no make(chan T) with unbounded buffer in pipelines
• Slowest stage identified and its throughput used to size upstream buffers
• Pipeline stage channel depths instrumented — visible in monitoring before backup occurs
• Dead letter channel or error sink defined — malformed records don't stall the pipeline

Step 5 — HTTP and gRPC Flow Control

HTTP/2 — built-in flow control: HTTP/2 has per-stream and per-connection flow control windows. When the client is slow, the server automatically stops writing.

// Go HTTP server — HTTP/2 enabled automatically with TLS
srv := &http.Server{
    Addr:    ":8443",
    Handler: mux,
    // HTTP/2 configured via TLSConfig
}
// For streaming responses, check client disconnection to stop producing
func (h *Handler) StreamLargeResponse(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming not supported", 500)
        return
    }

    for _, chunk := range generateChunks(r.Context()) {
        select {
        case <-r.Context().Done(): // client disconnected — stop producing
            return
        default:
        }
        fmt.Fprintf(w, "%s\n", chunk)
        flusher.Flush()
    }
}

gRPC streaming — flow control via Send blocking:

// Server streaming — gRPC buffers and blocks Send when client is slow
func (s *Service) StreamOrders(req *pb.StreamRequest, stream pb.OrderService_StreamOrdersServer) error {
    for {
        order, err := s.repo.Next(stream.Context())
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
        // Send blocks when the client's receive buffer is full — natural backpressure
        if err := stream.Send(order); err != nil {
            return err // client disconnected or buffer permanently full
        }
    }
}

Checklist:

• HTTP streaming handlers check r.Context().Done() — stop producing when client disconnects
• gRPC server streaming uses Send return value — stop on error (client buffer full / disconnected)
• Large response bodies streamed via http.Flusher — not buffered entirely in server memory
• HTTP/2 enabled on all services — inherits built-in flow control; outperforms HTTP/1.1 for streaming

Step 6 — Producer Throttling Based on Consumer Signals

When the consumer cannot signal backpressure directly (e.g., fire-and-forget producers), use explicit throttling.

Token bucket producer throttling:

import "golang.org/x/time/rate"

// Allow 100 events/s with a burst of 50
limiter := rate.NewLimiter(rate.Limit(100), 50)

for _, event := range events {
    // Block until a token is available
    if err := limiter.Wait(ctx); err != nil {
        return err // context cancelled
    }
    producer.Send(event)
}

Adaptive throttling — slow down when consumer lag grows:

type AdaptiveProducer struct {
    limiter     *rate.Limiter
    lagMonitor  func() int64
    baseRate    float64
    currentRate float64
}

func (p *AdaptiveProducer) adjustRate() {
    lag := p.lagMonitor()
    switch {
    case lag > 100000:
        p.currentRate = p.baseRate * 0.1  // 10% of base rate
    case lag > 10000:
        p.currentRate = p.baseRate * 0.5  // 50%
    case lag < 1000:
        p.currentRate = min(p.currentRate*1.1, p.baseRate) // recover 10% per cycle
    }
    p.limiter.SetLimit(rate.Limit(p.currentRate))
}

Checklist:

• Producer rate is bounded — no fire-and-forget without a rate limit
• Adaptive throttling adjusts producer rate based on consumer lag — not static
• Rate limiter state shared across all producer goroutines — not per-goroutine (defeats the purpose)
• Rate limit values configurable — adjustable without code deployment

Output Report

Critical

• Unbounded goroutine creation in a consumer loop — OOM crash under moderate burst load
• No consumer prefetch limit (RabbitMQ/Kafka) — consumer buffers the entire queue; OOM during replay
• Pipeline inter-stage channels unbounded — fast stage fills memory of downstream slow stage

High

• No backpressure from consumers to producers — fast producer + slow consumer → queue grows unboundedly
• Worker pool size not tuned to consumer throughput — pool too small wastes available capacity; too large causes downstream overload
• Consumer lag not monitored — lag growth detected only when system is already in distress

Medium

• Drop strategy used without metrics — events silently lost with no visibility into loss rate
• Pipeline channel depth not instrumented — bottleneck stage invisible until the pipeline backs up
• Rate limiter not shared across goroutines — each goroutine has its own limit; effective rate = N × limit

Low

• Consumer lag alert threshold too high (> 1M) — meaningful backlog before alert fires
• Worker pool size hardcoded — capacity cannot be adjusted without a code deployment
• HTTP streaming handler does not check client disconnect — continues generating data after client leaves

Passed

• Bounded worker pool used for all consumer loops; no unbounded goroutine creation
• Kafka prefetch / RabbitMQ QoS configured; consumer lag monitored and alerted
• Pipeline inter-stage channels bounded; slowest stage throughput measured and used to size buffers
• Producer rate limiter applied; adaptive throttling adjusts rate based on consumer lag signal
• HTTP streaming checks context cancellation; gRPC Send errors handled to stop on slow/disconnected clients