Batching — Middle Level¶

Table of Contents¶

1. Introduction
2. Recap and What This File Adds
3. The Production Shutdown Contract
4. Partial Flush Semantics
5. Retries with Exponential Backoff
6. Observability: The Four Metrics
7. Concurrency: Double-Buffer Pattern
8. Integrating with database/sql
9. Integrating with pgx and CopyFrom
10. Integrating with Kafka Producers
11. Integrating with HTTP Bulk Endpoints
12. Per-Tenant Sub-Batchers
13. Backpressure Composition
14. Testing in CI
15. Operational Runbook
16. Common Pitfalls
17. Self-Assessment
18. Summary

Introduction¶

The junior file gave you a working batcher: one goroutine, two triggers, a Close() that drains. This file takes that batcher and makes it production-grade. The differences are not about the core algorithm — that is the same nine lines of select. The differences are about everything around the core: shutdown contracts, error handling, retries, observability, integration with real sinks, composition with backpressure, testing in CI, runbooks.

After this file you will be able to:

• Write a graceful-shutdown contract that bounds drain time and never silently loses data.
• Choose between log-and-drop, retry, dead-letter, and crash policies for failed flushes — and implement each.
• Wire a batcher into database/sql multi-row INSERT, pgx.CopyFrom, and a Kafka producer with confidence.
• Emit the four metrics every batcher needs and read their values to detect saturation, slowdown, and loss.
• Run flushes concurrently with accumulation using the double-buffer pattern.
• Shard a batcher across tenants and bound per-tenant memory.
• Write deterministic CI tests for shutdown, retry, and partial flush.

You should already have read junior.md. If "size trigger", "time trigger", "defensive copy", and "reason-tagged flush" do not ring bells, go back.

Recap and What This File Adds¶

The junior batcher's shape:

for {
    select {
    case item, ok := <-b.in:
        if !ok { flush("shutdown"); return }
        buf = append(buf, item)
        if len(buf) >= b.maxSize { flush("size") }
    case <-ticker.C:
        flush("time")
    }
}

What is missing for production:

1. Bounded shutdown: Close() can hang forever if the sink is slow. We need a deadline.
2. Retry policy: _ = sink.Write(...) silently drops errors. We need retries and dead-letters.
3. Observability: no metrics. We cannot tell whether the batcher is healthy.
4. Concurrent flush: a slow flush stalls accumulation. We can double-buffer.
5. Real sinks: the in-memory Sink is fine for examples; production batchers integrate with database/sql, pgx, Kafka, HTTP.
6. Multi-tenancy: one batch failing one tenant's row fails the whole batch. We can shard.
7. Backpressure integration: producers blocking is one option, but you may need to propagate the signal upstream.
8. Testing: timing-based tests are flaky. We need deterministic patterns.
9. Runbooks: when something goes wrong, what do you check?

This file works through each.

The Production Shutdown Contract¶

A junior Close() looks like:

func (b *Batcher) Close() {
    close(b.in)
    <-b.done
}

It is correct for normal shutdown, but it has no time bound. If the sink hangs, Close() hangs. If the orchestrator does not call it, the buffer is lost.

A production Close() needs:

• A deadline so it cannot hang forever.
• A bounded flush queue so the drain itself is bounded.
• An error return so the orchestrator knows whether the drain succeeded.
• An idempotent contract so calling it twice is safe.

Shutdown With Deadline¶

func (b *Batcher) Shutdown(ctx context.Context) error {
    b.closeOnce.Do(func() { close(b.in) })
    select {
    case <-b.done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

The caller passes a context with a deadline. If the drain takes longer than the deadline, Shutdown returns context.DeadlineExceeded and the items left in the buffer are not durable. The caller has to decide: log and exit (data lost), wait longer, or kill the process.

A typical orchestrator looks like:

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := batcher.Shutdown(ctx); err != nil {
    log.Printf("shutdown timed out: %v (buffer not durable)", err)
}

Telling the Run Loop About the Deadline¶

The Shutdown context can also propagate into the run loop, so the loop knows to flush immediately rather than waiting for the next tick:

type Batcher struct {
    // ...
    shutdownReq chan struct{}
}

func (b *Batcher) Shutdown(ctx context.Context) error {
    b.closeOnce.Do(func() {
        close(b.shutdownReq)
        close(b.in)
    })
    select {
    case <-b.done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

In the run loop:

select {
case item, ok := <-b.in:
    // ...
case <-ticker.C:
    flush("time")
case <-b.shutdownReq:
    flush("shutdown_request")
    // continue draining b.in until it is closed
}

Now an immediate flush happens when shutdown is requested, even if the timer has not fired. The remaining items continue to drain through the <-b.in case.

What "Drain" Means Precisely¶

There are three reasonable definitions of "drain":

1. All buffered items are flushed: the buffer is empty when Shutdown returns.
2. All in-flight Add calls have completed: producers' calls have returned, and their items are in the buffer or flushed.
3. All items that will ever be added have been flushed: practically impossible without a fence from upstream.

Production batchers implement #1 (with the deadline cap) and rely on the orchestrator to enforce #2 by stopping upstreams before calling Shutdown. The HTTP server Shutdown(ctx) does this: it stops accepting new connections and waits for in-flight ones to finish. Then you call the batcher's Shutdown.

The dance:

// 1. Stop accepting new work.
srv.Shutdown(ctx)

// 2. Wait for in-flight producers (often done by Shutdown's wait-for-handlers).

// 3. Drain the batcher.
batcher.Shutdown(ctx)

// 4. Exit.

If you reverse 1 and 3, the batcher's Shutdown returns but new items arrive after, and the channel is closed, and producers panic. Don't do that.

Partial Flush Semantics¶

A real production case: the deadline strikes during a flush. What is the state?

• Items already accepted by the sink: durable.
• Items in the batcher's in-flight batch: in unknown state. The sink call might have started, might have partially landed, might not have started.
• Items still in the channel: lost (the channel is closed, the loop exited).
• Items still in the buffer slice but not yet handed to the sink: lost.

What you want to log: a tally per state. "On shutdown timeout: 1234 items in unknown state, 56 items lost." That number is what your post-mortem references.

To produce this tally, the batcher must track:

• enqueued (total items received from Add).
• flushed_ok (items the sink accepted with no error).
• flushed_fail (items the sink rejected).
• dropped_on_shutdown (items in the buffer when the deadline struck).

The difference enqueued - flushed_ok - flushed_fail - dropped_on_shutdown should be zero. If it is not, you have a bug somewhere.

A Partial-Flush-Aware Batcher¶

type Batcher struct {
    // ...
    metrics struct {
        enqueued           atomic.Int64
        flushedOK          atomic.Int64
        flushedFail        atomic.Int64
        droppedOnShutdown  atomic.Int64
    }
}

func (b *Batcher) Add(item Item) error {
    select {
    case b.in <- item:
        b.metrics.enqueued.Add(1)
        return nil
    case <-b.closeCh:
        return ErrClosed
    }
}

func (b *Batcher) flush(batch []Item, reason string) {
    err := b.sink.Write(context.Background(), batch)
    if err == nil {
        b.metrics.flushedOK.Add(int64(len(batch)))
    } else {
        b.metrics.flushedFail.Add(int64(len(batch)))
        log.Printf("batcher: flush failed reason=%s items=%d err=%v", reason, len(batch), err)
    }
}

func (b *Batcher) run() {
    defer close(b.done)
    buf := make([]Item, 0, b.maxSize)
    ticker := time.NewTicker(b.maxDelay)
    defer ticker.Stop()
    flushBuf := func(reason string) {
        if len(buf) == 0 {
            return
        }
        batch := make([]Item, len(buf))
        copy(batch, buf)
        buf = buf[:0]
        b.flush(batch, reason)
    }
    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                flushBuf("shutdown")
                b.metrics.droppedOnShutdown.Add(int64(len(buf)))
                return
            }
            buf = append(buf, item)
            if len(buf) >= b.maxSize {
                flushBuf("size")
            }
        case <-ticker.C:
            flushBuf("time")
        }
    }
}

This batcher always returns a clean tally. The post-mortem can answer "where did the missing items go?" precisely.

Retries with Exponential Backoff¶

The sink returns an error. Now what?

Decision Tree¶

1. Permanent error (bad payload, schema mismatch, auth failure): retrying does not help. Send to dead-letter or log-and-drop.
2. Transient error (network, timeout, downstream restart): retry with backoff.
3. Throttle response (downstream says "slow down"): retry with longer backoff; also feed back into upstream rate limiter.
4. Partial success (some items accepted, some rejected): retry only the rejected ones, if the sink tells you which.

You can usually tell which by inspecting the error type or HTTP status:

• 5xx, network error, timeout: transient.
• 429: throttle.
• 4xx (other than 429): permanent.
• 200 with per-item rejections: partial.

The Retry Wrapper¶

A common pattern is to wrap the sink with a retry layer:

type RetryingSink struct {
    inner     Sink
    maxTries  int
    baseDelay time.Duration
    classify  func(error) Retryability
}

type Retryability int

const (
    Permanent Retryability = iota
    Transient
    Throttle
)

func (r *RetryingSink) Write(ctx context.Context, batch []Item) error {
    var lastErr error
    delay := r.baseDelay
    for try := 0; try < r.maxTries; try++ {
        err := r.inner.Write(ctx, batch)
        if err == nil {
            return nil
        }
        lastErr = err
        kind := r.classify(err)
        if kind == Permanent {
            return err
        }
        // Add jitter to avoid synchronised retry storms.
        sleepFor := delay + time.Duration(rand.Int63n(int64(delay)))
        select {
        case <-time.After(sleepFor):
        case <-ctx.Done():
            return ctx.Err()
        }
        delay *= 2
        if kind == Throttle {
            // Be extra conservative.
            delay *= 2
        }
    }
    return lastErr
}

The retry layer is a sink wrapping a sink. The batcher does not know about retries; it just sees a sink that takes longer to return. This is the decorator pattern, and it composes cleanly with rate limiting, circuit breakers, and metrics.

When to Move Retries Out of the Hot Path¶

If retries can take seconds, you do not want them blocking the run loop. Move the retry to a separate goroutine:

type Batcher struct {
    // ...
    retryQueue chan []Item
}

// In run:
flushBuf := func(reason string) {
    if len(buf) == 0 {
        return
    }
    batch := make([]Item, len(buf))
    copy(batch, buf)
    buf = buf[:0]
    select {
    case b.retryQueue <- batch:
    default:
        // Retry queue full; drop or block depending on policy.
    }
}

// Separate goroutine:
func (b *Batcher) retryWorker() {
    for batch := range b.retryQueue {
        for try := 0; try < maxTries; try++ {
            if err := b.sink.Write(ctx, batch); err == nil {
                break
            }
            // backoff...
        }
    }
}

Now the run loop is decoupled from sink latency. Trade-off: a slow sink causes the retry queue to fill, and you must decide what happens when it fills (the same "block, drop, or grow" question as the input channel).

Dead-Letter Queue¶

For permanent errors, you usually want a dead-letter queue (DLQ): a separate sink (often a Kafka topic, S3 bucket, or local disk file) where failed batches go for human or automated review.

type DLQSink struct {
    primary Sink
    dlq     Sink
}

func (d *DLQSink) Write(ctx context.Context, batch []Item) error {
    err := d.primary.Write(ctx, batch)
    if err == nil {
        return nil
    }
    if !isTransient(err) {
        _ = d.dlq.Write(ctx, batch)
        return nil // we are not blocking the upstream; we have stored
    }
    return err
}

The DLQ is a critical part of the production design. Without it, permanent errors either get retried forever (clogging the retry queue) or are silently dropped.

Observability: The Four Metrics¶

Every batcher must emit at least these four metrics. Without them you have a black box.

1. Batch Size (Histogram)¶

batcher_batch_size_items{name="audit"}

A histogram of how many items were in each flushed batch. Buckets like 1, 5, 10, 50, 100, 500, 1000, 5000, +Inf.

What it tells you:

• p99 near MaxBatchSize: size trigger is the main driver; downstream is keeping up.
• p99 well below MaxBatchSize: time trigger dominates; traffic is low or MaxBatchSize is too high.
• p50 unstable: traffic is bursty; consider adaptive sizing.

2. Flush Latency (Histogram)¶

batcher_flush_duration_seconds{name="audit"}

A histogram of how long each sink.Write call took. Buckets in seconds: 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, +Inf.

What it tells you:

• p50 stable, p99 spiky: occasional slow downstream calls; investigate the long tail.
• p50 rising: downstream is getting slower over time; consider scaling downstream.
• p99 > maxDelay: the time trigger may fire while a flush is in progress; expect backpressure on producers.

3. Flush Reason (Counter)¶

batcher_flush_total{name="audit", reason="size"} batcher_flush_total{name="audit", reason="time"} batcher_flush_total{name="audit", reason="shutdown"}

A counter for each flush, labelled by reason. Increment on every flush.

What it tells you:

• All size: high traffic; throughput-saturated.
• All time: low traffic; latency-dominant.
• Mix: healthy operating zone.
• Many shutdown: lots of restarts; investigate process churn.

4. Queue Depth (Gauge)¶

batcher_queue_depth{name="audit"}

The current number of items in the input channel (len(b.in)).

What it tells you:

• Near zero: consumer keeps up.
• Near cap: backpressure on producers.
• At cap: producers blocked.

How to Wire Them Up With Prometheus¶

import "github.com/prometheus/client_golang/prometheus"

var (
    batchSize = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name: "batcher_batch_size_items",
        Help: "Number of items in each flushed batch.",
        Buckets: []float64{1, 5, 10, 50, 100, 500, 1000, 5000, 10000},
    }, []string{"name"})

    flushDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
        Name: "batcher_flush_duration_seconds",
        Help: "Duration of each flush call.",
        Buckets: prometheus.ExponentialBuckets(0.001, 2, 12),
    }, []string{"name", "result"})

    flushTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "batcher_flush_total",
        Help: "Total flushes, labelled by reason.",
    }, []string{"name", "reason"})

    queueDepth = prometheus.NewGaugeVec(prometheus.GaugeOpts{
        Name: "batcher_queue_depth",
        Help: "Current input channel depth.",
    }, []string{"name"})
)

func init() {
    prometheus.MustRegister(batchSize, flushDuration, flushTotal, queueDepth)
}

In the flush:

flushBuf := func(reason string) {
    if len(buf) == 0 {
        return
    }
    batch := make([]Item, len(buf))
    copy(batch, buf)
    buf = buf[:0]
    flushTotal.WithLabelValues(b.name, reason).Inc()
    batchSize.WithLabelValues(b.name).Observe(float64(len(batch)))
    start := time.Now()
    err := b.sink.Write(context.Background(), batch)
    result := "ok"
    if err != nil {
        result = "error"
    }
    flushDuration.WithLabelValues(b.name, result).Observe(time.Since(start).Seconds())
}

And update queue depth periodically (from the run loop on every iteration, or from a separate goroutine):

queueDepth.WithLabelValues(b.name).Set(float64(len(b.in)))

Optional fifth metric: batcher_dropped_total{name, reason} if your batcher drops on overflow. Always emit if you drop.

Reading the Metrics¶

A healthy batcher has:

• batch_size_items{name} p50 between 0.1x and 0.5x of MaxBatchSize.
• flush_duration_seconds{name} p99 below maxDelay.
• flush_total{name, reason} showing both size and time, neither at 100%.
• queue_depth{name} p99 below 50% of channel capacity.

Any one of these going out of range is worth an alert.

Concurrency: Double-Buffer Pattern¶

In the junior batcher, the flush is synchronous. While the run loop is in sink.Write, new items pile up in the input channel. If the sink is slow and traffic is high, producers block.

The double-buffer pattern allows the run loop to start a new batch while the previous one is being flushed:

type Batcher struct {
    in       chan Item
    sink     Sink
    maxSize  int
    maxDelay time.Duration
    flushReq chan []Item // batches handed off to the flush worker
    done     chan struct{}
}

func (b *Batcher) flushWorker() {
    for batch := range b.flushReq {
        _ = b.sink.Write(context.Background(), batch)
    }
}

func (b *Batcher) run() {
    defer close(b.done)
    buf := make([]Item, 0, b.maxSize)
    ticker := time.NewTicker(b.maxDelay)
    defer ticker.Stop()
    handoff := func(reason string) {
        if len(buf) == 0 {
            return
        }
        batch := make([]Item, len(buf))
        copy(batch, buf)
        buf = buf[:0]
        b.flushReq <- batch
    }
    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                handoff("shutdown")
                close(b.flushReq)
                return
            }
            buf = append(buf, item)
            if len(buf) >= b.maxSize {
                handoff("size")
            }
        case <-ticker.C:
            handoff("time")
        }
    }
}

The flush worker (or a small pool of them) drains flushReq independently. The run loop hands off the batch and immediately starts a new one.

Trade-offs:

• Pro: higher throughput when the sink is slow.
• Pro: latency is bounded by maxDelay + handoff_latency, regardless of sink latency.
• Con: the flushReq channel can fill, applying backpressure differently. Decide on its cap and policy.
• Con: ordering is less obvious; multiple batches may be in flight.
• Con: more memory; two batches' worth of items can be in transit.

Sizing the Flush Queue¶

If your sink has tail latency of 1 s and your size-trigger rate is 100 per second, you need flushReq cap of at least 100 to avoid blocking the run loop during a 1 s tail event. Pick a number, observe len(flushReq), tune.

Multiple Flush Workers¶

If your sink is partitioned (multiple Postgres replicas, multiple Kafka brokers), you can have multiple flush workers:

for i := 0; i < numWorkers; i++ {
    go b.flushWorker()
}

All workers drain the same flushReq channel. Each takes the next available batch. Now you can saturate a multi-broker downstream.

Caveat: ordering across workers is undefined. If your sink requires per-key ordering (Kafka with a key), use one worker per key or partition.

Integrating with database/sql¶

Multi-Row INSERT¶

The classic Postgres batch insert:

INSERT INTO events (user_id, action, ts) VALUES
  ($1, $2, $3),
  ($4, $5, $6),
  ($7, $8, $9),
  ...

In Go:

type SQLSink struct {
    db *sql.DB
}

func (s *SQLSink) Write(ctx context.Context, batch []Event) error {
    if len(batch) == 0 {
        return nil
    }
    args := make([]any, 0, len(batch)*3)
    placeholders := make([]string, 0, len(batch))
    for i, e := range batch {
        base := i * 3
        placeholders = append(placeholders,
            fmt.Sprintf("($%d, $%d, $%d)", base+1, base+2, base+3))
        args = append(args, e.UserID, e.Action, e.Timestamp)
    }
    query := "INSERT INTO events (user_id, action, ts) VALUES " +
        strings.Join(placeholders, ",")
    _, err := s.db.ExecContext(ctx, query, args...)
    return err
}

Pitfalls¶

• Parameter limit: PostgreSQL has a 65535 parameter limit per query. If batch_size * cols_per_row > 65535, the query is rejected. For 3-column rows, that is 21845 rows max. Stay well below — 1000 is typical.
• Query plan cache thrash: if len(batch) varies wildly, every batch is a "new" query as far as the plan cache is concerned. Either prepared statements per size bucket, or always pad to a fixed size.
• Transaction overhead: by default, each ExecContext is its own transaction. Group into one transaction only if you need atomicity across batches.
• MySQL has a different max_allowed_packet limit: 64 MB by default. Big batches can hit it.

Prepared Statements¶

For repeated batches of the same size:

type SQLSink struct {
    db   *sql.DB
    stmt *sql.Stmt // prepared for a specific batch size
    size int
}

func NewSQLSink(db *sql.DB, batchSize int) (*SQLSink, error) {
    placeholders := make([]string, batchSize)
    for i := 0; i < batchSize; i++ {
        base := i * 3
        placeholders[i] = fmt.Sprintf("($%d, $%d, $%d)", base+1, base+2, base+3)
    }
    query := "INSERT INTO events (user_id, action, ts) VALUES " +
        strings.Join(placeholders, ",")
    stmt, err := db.Prepare(query)
    if err != nil {
        return nil, err
    }
    return &SQLSink{db: db, stmt: stmt, size: batchSize}, nil
}

func (s *SQLSink) Write(ctx context.Context, batch []Event) error {
    if len(batch) != s.size {
        // Fall back to a non-prepared query for partial batches.
        return s.writeUnprepared(ctx, batch)
    }
    args := make([]any, 0, len(batch)*3)
    for _, e := range batch {
        args = append(args, e.UserID, e.Action, e.Timestamp)
    }
    _, err := s.stmt.ExecContext(ctx, args...)
    return err
}

The prepared path is faster (no parse), the unprepared path handles partial batches (time-triggered or close-triggered).

Idempotency: ON CONFLICT¶

If retries can fan out, add ON CONFLICT:

INSERT INTO events (id, user_id, action, ts) VALUES ...
ON CONFLICT (id) DO NOTHING

This requires items to have a unique ID, which is good practice anyway. With it, you can retry the same batch and get the same result.

Integrating with pgx and CopyFrom¶

For Postgres, the fastest bulk-insert path is COPY FROM, not INSERT. pgx.CopyFrom exposes this:

import "github.com/jackc/pgx/v5"

type PGXCopySink struct {
    conn *pgx.Conn
}

func (p *PGXCopySink) Write(ctx context.Context, batch []Event) error {
    if len(batch) == 0 {
        return nil
    }
    rows := make([][]any, len(batch))
    for i, e := range batch {
        rows[i] = []any{e.UserID, e.Action, e.Timestamp}
    }
    _, err := p.conn.CopyFrom(ctx,
        pgx.Identifier{"events"},
        []string{"user_id", "action", "ts"},
        pgx.CopyFromRows(rows))
    return err
}

COPY FROM is dramatically faster than multi-row INSERT for batches of 1000+. It uses Postgres' streaming binary protocol, bypassing parser overhead and reducing per-row cost.

When to use INSERT vs COPY¶

• INSERT VALUES (..), (..), ...: small batches (10–500). Better for ON CONFLICT and RETURNING.
• COPY FROM: large batches (500+). Faster but limited — no ON CONFLICT (without a temp table dance).
• Temp table + INSERT ... SELECT: best of both — copy into a temp table, then INSERT SELECT with ON CONFLICT. Use for large idempotent batches.

Pool Awareness¶

*pgx.Conn is single-connection. For a batcher, use *pgxpool.Pool and call Acquire(ctx) before CopyFrom:

func (p *PGXCopySink) Write(ctx context.Context, batch []Event) error {
    conn, err := p.pool.Acquire(ctx)
    if err != nil {
        return err
    }
    defer conn.Release()
    _, err = conn.CopyFrom(ctx, ...)
    return err
}

The pool handles concurrent flushes (one per connection). Set pool size to at least the number of concurrent flush workers.

Integrating with Kafka Producers¶

Kafka producers (franz-go, confluent-kafka-go, IBM sarama) all internally batch. So you have a choice: rely on the producer's internal batching, or layer your own on top.

Just Use the Producer's Batcher¶

If you use franz-go, you can Produce(ctx, record, callback) per item and let franz-go batch internally. Set ProducerLinger(d) and ProducerBatchMaxBytes(n):

client, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.DefaultProduceTopic("events"),
    kgo.ProducerLinger(5*time.Millisecond),
    kgo.ProducerBatchMaxBytes(1024*1024), // 1 MB
)

for _, e := range events {
    rec := &kgo.Record{Value: serialise(e)}
    client.Produce(ctx, rec, func(rec *kgo.Record, err error) {
        if err != nil {
            log.Printf("kafka produce failed: %v", err)
        }
    })
}

This is clean — no application-level batcher needed.

When to Add an App-Level Batcher¶

Add your own batcher when:

• You need to combine items before producing (e.g., dedupe, aggregate).
• You want per-tenant ordering or partitioning that the producer does not handle.
• You need a shared latency budget across multiple sinks (one batcher feeding Kafka, Postgres, and S3 with the same window).
• You want stronger observability than the producer client provides.

Sink Adapter for Kafka¶

type KafkaSink struct {
    client *kgo.Client
    topic  string
}

func (k *KafkaSink) Write(ctx context.Context, batch []Event) error {
    records := make([]*kgo.Record, len(batch))
    for i, e := range batch {
        records[i] = &kgo.Record{
            Topic: k.topic,
            Key:   []byte(e.Key),
            Value: serialise(e),
        }
    }
    results := k.client.ProduceSync(ctx, records...)
    return results.FirstErr()
}

ProduceSync blocks until all records have been acked. This integrates cleanly with the batcher: one call per batch, one error per batch.

Per-Key Ordering¶

If items must arrive in order per key, you have a problem: Kafka delivers in order per partition, and the partition is determined by the key hash. If your batcher hands records to the producer in some order, they will end up in different partitions (different orders).

Solutions:

• One batcher per partition key (high overhead).
• Send all records in one batch with a single ProduceSync; per-partition records within a batch are sent in batch order.
• Use MaxBufferedRecords(1) (no in-flight) for strictly ordered producers — but this destroys throughput.

For most use cases, "ordered within batch, batches in submission order" is what you get and is enough.

Integrating with HTTP Bulk Endpoints¶

Elastic _bulk, BigQuery insertAll, Datadog /api/v1/series — all HTTP endpoints that accept many records per call.

type HTTPSink struct {
    client *http.Client
    url    string
    auth   string
}

func (h *HTTPSink) Write(ctx context.Context, batch []Event) error {
    body, err := encodeNDJSON(batch)
    if err != nil {
        return err
    }
    req, err := http.NewRequestWithContext(ctx, "POST", h.url, bytes.NewReader(body))
    if err != nil {
        return err
    }
    req.Header.Set("Content-Type", "application/x-ndjson")
    req.Header.Set("Authorization", h.auth)
    resp, err := h.client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()
    if resp.StatusCode >= 400 {
        b, _ := io.ReadAll(resp.Body)
        return fmt.Errorf("http %d: %s", resp.StatusCode, string(b))
    }
    return parseBulkResponse(resp.Body)
}

Considerations¶

• Body size: bulk APIs usually have a hard limit (5–15 MB). Track body size in addition to count.
• gzip: most bulk APIs accept gzip. Compress before sending; saves bandwidth.
• Per-item errors: bulk APIs often return per-item status (_bulk returns an array). Parse it and route per-item failures to retry or dead-letter.
• Idempotency keys: include a per-record _id for idempotent retries.

Per-Item Result Parsing¶

type BulkItemResult struct {
    Index struct {
        Status int    `json:"status"`
        Error  string `json:"error"`
    } `json:"index"`
}

type BulkResponse struct {
    Items []BulkItemResult `json:"items"`
}

func parseBulkResponse(body io.Reader) error {
    var r BulkResponse
    if err := json.NewDecoder(body).Decode(&r); err != nil {
        return err
    }
    for i, item := range r.Items {
        if item.Index.Status >= 400 {
            log.Printf("item %d failed: %s", i, item.Index.Error)
        }
    }
    return nil
}

For each failed item, decide: retry (if transient), dead-letter (if permanent), or log (if best-effort).

Per-Tenant Sub-Batchers¶

A single shared batcher mixes items from many tenants. A failure caused by one tenant's bad payload fails the whole batch. The fix is to keep one buffer per tenant:

type TenantBatcher struct {
    in       chan Item
    maxSize  int
    maxDelay time.Duration
    sink     Sink
    done     chan struct{}
}

func (t *TenantBatcher) run() {
    defer close(t.done)
    bufs := make(map[string][]Item)
    timers := make(map[string]*time.Timer)
    flush := func(tenant string, reason string) {
        b := bufs[tenant]
        if len(b) == 0 {
            return
        }
        batch := make([]Item, len(b))
        copy(batch, b)
        bufs[tenant] = b[:0]
        if tm, ok := timers[tenant]; ok {
            tm.Stop()
            delete(timers, tenant)
        }
        _ = t.sink.Write(context.Background(), batch)
    }
    flushAll := func() {
        for tenant := range bufs {
            flush(tenant, "shutdown")
        }
    }
    for {
        select {
        case item, ok := <-t.in:
            if !ok {
                flushAll()
                return
            }
            bufs[item.Tenant] = append(bufs[item.Tenant], item)
            if len(bufs[item.Tenant]) == 1 {
                tt := item.Tenant
                timers[tt] = time.AfterFunc(t.maxDelay, func() {
                    // BUG: cannot directly call flush from another goroutine
                })
            }
            if len(bufs[item.Tenant]) >= t.maxSize {
                flush(item.Tenant, "size")
            }
        }
    }
}

The buggy version above shows the trap: time.AfterFunc runs the callback in a different goroutine, but our bufs map is owned by the run loop. We cannot safely call flush from the callback.

The fix is to signal the run loop:

case tenant := <-t.timeTrigger:
    flush(tenant, "time")

And the timer callback sends to t.timeTrigger:

timers[tt] = time.AfterFunc(t.maxDelay, func() {
    t.timeTrigger <- tt
})

Now all map operations happen in the run loop, one goroutine, no race.

Memory Bounds¶

Per-tenant batching multiplies the worst-case memory: numTenants * maxSize. For 1000 tenants and a 1000-row batch, that is 1 million items in RAM. Decide a global cap and either:

• Reject new items when total > cap.
• Force-flush the oldest tenant when total > cap.
• Cap per tenant at a smaller value.

This is the area of "fairness in multi-tenant systems", covered more in senior.md.

Backpressure Composition¶

A batcher's input channel is the first backpressure boundary. When it fills, producers block. But that may not be the right behaviour for your service.

Producer-side: choose your policy¶

// Block.
b.Add(item)

// Drop.
select {
case b.in <- item:
default:
    drops.Inc()
}

// Drop oldest.
select {
case b.in <- item:
default:
    select {
    case <-b.in: // discard one
    default:
    }
    b.in <- item
    dropsOldest.Inc()
}

// Block with deadline.
ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
defer cancel()
select {
case b.in <- item:
case <-ctx.Done():
    return ctx.Err()
}

Each policy is correct for some workload, wrong for others. Document which you chose.

Upstream signalling¶

If you cannot block producers (an HTTP handler must respond within an SLA), return an HTTP 503 when the batcher is saturated:

func handler(w http.ResponseWriter, r *http.Request) {
    item := parse(r)
    select {
    case batcher.in <- item:
        w.WriteHeader(202)
    case <-time.After(50 * time.Millisecond):
        w.WriteHeader(503)
        w.Header().Set("Retry-After", "5")
    }
}

This lets the client retry, often with exponential backoff, instead of the request piling up server-side. See 01-backpressure for the full pattern.

Testing in CI¶

Time-based tests are flaky. Here are deterministic patterns.

Inject a Fake Clock¶

type Clock interface {
    Now() time.Time
    After(d time.Duration) <-chan time.Time
    NewTicker(d time.Duration) *time.Ticker // or a Ticker interface
}

type RealClock struct{}
func (RealClock) Now() time.Time { return time.Now() }
// ...

type FakeClock struct {
    mu   sync.Mutex
    now  time.Time
    chs  []chan time.Time
}
// ...

func (b *Batcher) run() {
    ticker := b.clock.NewTicker(b.maxDelay)
    // ...
}

In tests, advance the fake clock and assert flushes:

func TestTimeTriggerDeterministic(t *testing.T) {
    clock := NewFakeClock(time.Now())
    sink := NewFakeSink()
    b := NewBatcher(sink, 100, 50*time.Millisecond, clock)

    b.Add(Item{})
    b.Add(Item{})
    if sink.Count() != 0 {
        t.Fatalf("expected 0 flushes before tick")
    }
    clock.Advance(50 * time.Millisecond)
    waitFor(t, func() bool { return sink.Count() == 1 })
}

waitFor is a polling helper with a generous timeout (1 s) and a short interval (1 ms). It is the bridge between "test wants determinism" and "fake clock advances out of band of the run loop".

Libraries: clockwork, github.com/benbjohnson/clock. Either works.

Test the Shutdown Contract¶

func TestShutdownFlushes(t *testing.T) {
    sink := NewFakeSink()
    b := NewBatcher(sink, 100, time.Hour, clock)
    for i := 0; i < 5; i++ {
        b.Add(Item{ID: i})
    }
    if err := b.Shutdown(context.Background()); err != nil {
        t.Fatal(err)
    }
    if sink.Total() != 5 {
        t.Fatalf("expected 5 items, got %d", sink.Total())
    }
}

Test Shutdown Timeout¶

func TestShutdownTimeout(t *testing.T) {
    slowSink := &SlowSink{delay: 1 * time.Second}
    b := NewBatcher(slowSink, 100, time.Hour, clock)
    for i := 0; i < 100; i++ {
        b.Add(Item{ID: i})
    }
    ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
    defer cancel()
    err := b.Shutdown(ctx)
    if !errors.Is(err, context.DeadlineExceeded) {
        t.Fatalf("expected deadline exceeded, got %v", err)
    }
}

Test Retry on Transient¶

func TestRetryOnTransient(t *testing.T) {
    flaky := &FlakySink{failures: 2}
    retry := &RetryingSink{inner: flaky, maxTries: 5, baseDelay: time.Millisecond}
    b := NewBatcher(retry, 5, time.Hour, clock)
    for i := 0; i < 5; i++ {
        b.Add(Item{ID: i})
    }
    b.Shutdown(context.Background())
    if flaky.successes != 1 {
        t.Fatalf("expected 1 successful write after retries")
    }
    if flaky.attempts != 3 {
        t.Fatalf("expected 3 attempts (2 fail + 1 success), got %d", flaky.attempts)
    }
}

Run with -race¶

Every batcher test must run with -race. The race detector catches the subtle bugs that timing tests miss.

Operational Runbook¶

When a batcher misbehaves in production, the first place to look is the four metrics.

"Throughput is low"¶

1. Check flush_total{reason}. All time? Increase MaxBatchSize or accept low throughput.
2. Check batch_size_items. p50 near MaxBatchSize? Sink is the bottleneck; scale sink.
3. Check flush_duration. p99 high? Sink is slow; investigate sink-side.

"Latency is high"¶

1. Check flush_duration p99. Above MaxBatchDelay? Sink is slow; reduce MaxBatchSize to flush smaller batches faster.
2. Check queue_depth. Near cap? Backpressure; producers blocked.

"Items are missing after deploy"¶

1. Was Shutdown(ctx) called? Check shutdown handler.
2. Did Shutdown return an error? Check logs for "deadline exceeded".
3. Were there in-flight retries? Check retry queue depth at SIGTERM.

"Memory is rising"¶

1. Check queue_depth. Bounded by channel cap; if rising, channel cap is the issue.
2. Check per-tenant memory (if sharded). One tenant dominating?
3. Check retry queue. If retries can grow unboundedly, that is the leak.

"Sink is returning errors"¶

1. Check error classification. Are transient errors being treated as permanent?
2. Check retry budget. Are retries exhausting too quickly?
3. Check dead-letter queue. Is it accumulating? Drain it.

The runbook is the connective tissue between metrics and action. Without it, "the dashboard is on fire" produces panic; with it, you have a checklist.

More on Multi-Sink Batchers¶

Sometimes the same batch must be written to multiple sinks: a primary (Postgres) and a stream (Kafka) so consumers can subscribe to changes. Options:

Option 1: Sequential Multi-Sink¶

type MultiSink struct {
    sinks []Sink
}

func (m *MultiSink) Write(ctx context.Context, batch []Event) error {
    for _, s := range m.sinks {
        if err := s.Write(ctx, batch); err != nil {
            return err
        }
    }
    return nil
}

Simple, but each sink waits for the previous. Latency adds up.

Option 2: Parallel Multi-Sink with errgroup¶

func (m *MultiSink) Write(ctx context.Context, batch []Event) error {
    g, gctx := errgroup.WithContext(ctx)
    for _, s := range m.sinks {
        s := s
        g.Go(func() error {
            return s.Write(gctx, batch)
        })
    }
    return g.Wait()
}

Concurrent flushes; the slowest sink dominates latency. errgroup cancels remaining if any fails.

Option 3: Best-Effort Fan-Out¶

func (m *MultiSink) Write(ctx context.Context, batch []Event) error {
    var firstErr error
    var mu sync.Mutex
    var wg sync.WaitGroup
    for _, s := range m.sinks {
        wg.Add(1)
        go func(s Sink) {
            defer wg.Done()
            if err := s.Write(ctx, batch); err != nil {
                mu.Lock()
                if firstErr == nil {
                    firstErr = err
                }
                mu.Unlock()
            }
        }(s)
    }
    wg.Wait()
    return firstErr
}

All sinks attempted regardless of others' failures. Trade-off: a slow sink does not abort the others, but you may lose data on a failing sink without knowing which.

Two-Phase Commit?¶

Tempting to think about transactional fan-out. In practice, distributed transactions across heterogeneous sinks (Postgres + Kafka + S3) are impractical. The standard pattern is outbox: write to Postgres in one transaction with a marker; a separate CDC pipeline streams to Kafka. The batcher just writes to Postgres; the propagation is decoupled.

More on the Pipeline Architecture¶

Refining the pipeline:

type Batcher struct {
    in        chan Item
    flushReq  chan []Item
    sink      Sink
    maxSize   int
    maxDelay  time.Duration
    flushers  int
    done      chan struct{}
    wg        sync.WaitGroup
}

func New(cfg Config) *Batcher {
    b := &Batcher{
        in:       make(chan Item, cfg.QueueDepth),
        flushReq: make(chan []Item, cfg.FlushQueueDepth),
        sink:     cfg.Sink,
        maxSize:  cfg.MaxBatchSize,
        maxDelay: cfg.MaxBatchDelay,
        flushers: cfg.Flushers,
        done:     make(chan struct{}),
    }
    for i := 0; i < cfg.Flushers; i++ {
        b.wg.Add(1)
        go b.flushWorker()
    }
    go b.run()
    return b
}

func (b *Batcher) flushWorker() {
    defer b.wg.Done()
    for batch := range b.flushReq {
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        _ = b.sink.Write(ctx, batch)
        cancel()
    }
}

func (b *Batcher) run() {
    defer func() {
        close(b.flushReq)
        b.wg.Wait()
        close(b.done)
    }()
    buf := make([]Item, 0, b.maxSize)
    ticker := time.NewTicker(b.maxDelay)
    defer ticker.Stop()
    handoff := func(reason string) {
        if len(buf) == 0 {
            return
        }
        batch := make([]Item, len(buf))
        copy(batch, buf)
        buf = buf[:0]
        b.flushReq <- batch
    }
    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                handoff("shutdown")
                return
            }
            buf = append(buf, item)
            if len(buf) >= b.maxSize {
                handoff("size")
            }
        case <-ticker.C:
            handoff("time")
        }
    }
}