Batching — Junior Level¶

Table of Contents¶

1. Introduction
2. Prerequisites
3. Glossary
4. Core Concepts
5. Real-World Analogies
6. Mental Models
7. Pros and Cons
8. Use Cases
9. Code Examples
10. Coding Patterns
11. Clean Code
12. Product Use
13. Error Handling
14. Security Considerations
15. Performance Tips
16. Best Practices
17. Edge Cases and Pitfalls
18. Common Mistakes
19. Common Misconceptions
20. Tricky Points
21. Test
22. Tricky Questions
23. Cheat Sheet
24. Self-Assessment Checklist
25. Summary
26. What You Can Build
27. Further Reading
28. Related Topics
29. Diagrams and Visual Aids

Introduction¶

Focus: "What is a batcher? Why is it faster than calling the sink for every item? How do I write one in 60 lines of Go without losing data?"

A batcher is a small piece of code that collects items as they arrive and hands them off to a downstream sink in groups instead of one at a time. The "downstream sink" might be a database, a message broker, an HTTP endpoint, a log file — anything where the cost of one call is mostly fixed regardless of how many items the call carries.

Think of it like a delivery driver. Driving the truck across town costs time and fuel whether you carry one parcel or fifty. So you fill the truck before you set out. Batching is exactly that trick, applied to function calls.

The two questions every batcher must answer:

1. When is the batch full enough to send? This is the size trigger — flush when N items have accumulated.
2. When have we waited too long? This is the time trigger — flush whatever we have if the oldest item is older than D.

A batcher with only a size trigger can hold an item forever if traffic suddenly dies. A batcher with only a time trigger never amortises the per-call cost across many items under high load. Real production batchers always have both.

In Go, a batcher is one of the most natural things you can build with channels and select. The entire pattern fits on the back of an envelope:

for {
    select {
    case item := <-input:
        buf = append(buf, item)
        if len(buf) >= maxSize {
            flush(buf); buf = buf[:0]
        }
    case <-ticker.C:
        if len(buf) > 0 {
            flush(buf); buf = buf[:0]
        }
    }
}

That nine-line loop is the heart of every batcher you will ever see in production. Everything else — graceful shutdown, retry policy, observability, adaptive sizing, NUMA-aware accumulators — is added on top of it. You will recognise this shape in golang/x/time/rate, in the Kafka producer client, in Prometheus' remote-write batch, in OpenTelemetry's batch processor, in Vector's sink. Internalise it once and you will see it everywhere.

After reading this file you will be able to:

• Explain to a colleague what a batcher does and why it is faster.
• Write a working batcher with both triggers from scratch, without referring to notes.
• Read time.Ticker and select code well enough to predict batcher behaviour on a whiteboard.
• Recognise and fix the most common first-time bugs: dropped flushes on shutdown, shared-buffer aliasing, blocked producers, time-trigger races, only-one-trigger bugs.
• Decide whether batching is appropriate for a given downstream sink based on the per-call versus per-item cost model.

You do not yet need to know about graceful shutdown contracts, partial flush guarantees, observability metrics, retry strategies, adaptive sizing, or the internals of Kafka and Postgres. Those come at the middle, senior, and professional levels.

A note on naming. In conversation Go engineers use "batcher", "buffer", "aggregator", "collector", and "accumulator" interchangeably. They all mean the same thing for our purposes. Where the distinction matters — for instance, "aggregator" is sometimes reserved for batchers that combine items, like summing counters, rather than just grouping them — we will say so explicitly. This file uses "batcher" throughout.

Prerequisites¶

• Required: a working Go toolchain, version 1.18 or newer. Run go version to check. All examples in this file compile on 1.20 and above.
• Required: comfort writing and running a main function, importing packages, and starting a goroutine with the go keyword.
• Required: comfort with channels: make(chan T, n), send ch <- x, receive x := <-ch, and close(ch).
• Required: comfort with select statements, including the default clause.
• Required: knowledge that time.NewTicker(d) returns a *time.Ticker whose C field is a channel that delivers a tick value every d.
• Recommended: awareness that sync.WaitGroup exists and is used to wait for goroutines to finish.
• Recommended: some prior exposure to database/sql or any HTTP client library. We do not use them in this file — we use a tiny fake Sink — but knowing they exist will help the motivation land.

If you can read this snippet and predict its output:

ticker := time.NewTicker(100 * time.Millisecond)
for i := 0; i < 3; i++ {
    <-ticker.C
    fmt.Println(i)
}
ticker.Stop()

you have every prerequisite this file assumes.

The output is "0", "1", "2" spaced 100 ms apart, with the program taking about 300 ms total. If that is news to you, read the time.Ticker junior file first and come back.

Glossary¶

Core Concepts¶

A batcher is a one-goroutine loop¶

The heart of every batcher is a single goroutine running a select over two channels and a ticker:

for {
    select {
    case item := <-input:
        buf = append(buf, item)
        if len(buf) >= maxSize {
            flush(buf); buf = buf[:0]
        }
    case <-ticker.C:
        if len(buf) > 0 {
            flush(buf); buf = buf[:0]
        }
    }
}

That is essentially the whole pattern. Everything else — shutdown, retries, metrics, adaptive sizing — is added on top of this loop. Internalise this shape; you will see it again and again.

Note three properties of this loop:

1. Single ownership of the buffer. Only one goroutine ever touches buf. No mutex is needed; the channel does the synchronisation.
2. Two flush paths. Inside the case item branch we flush on size; inside the case ticker.C branch we flush on time. The two branches share the flush helper.
3. buf = buf[:0] after each flush. This empties the slice but keeps the underlying array. The next batch will reuse that memory.

Producers send, the consumer batches¶

The batcher exposes one method, Add(item), which is just a channel send. Many goroutines can call Add simultaneously; the channel serialises them. Only the one batcher goroutine touches the buffer, so the buffer itself needs no lock.

This is a classic example of using Go's "share by communicating, do not communicate by sharing" model. Synchronisation is implicit in the channel.

producer1  --\
producer2  ---*--[input channel]--> [consumer goroutine + buffer] --> sink
producer3  --/

The producers never see the buffer. The consumer never sees the producers as anything other than items appearing on the channel. The decoupling is what makes the design easy to reason about.

Size and time are independent triggers¶

The size trigger fires when the buffer is full. The time trigger fires when too much time has passed. They are independent — either can fire first. If you only implement one, you have a bug waiting:

• Size only: a burst of 10 items at 09:00, then silence. Those 10 items wait until 17:00 when the next item arrives. Audit logs that are supposed to be flushed every minute sit in RAM for the entire workday. On crash they are gone. This is not theoretical — every shop has had this incident at least once.
• Time only: 100 000 items per second arriving. The size never fills, so the batcher flushes every 100 ms by ticker — meaning a batch of 10 000 items every tick. Now you are amortising, but only by accident; if traffic doubles, your downstream sees 20 000-item batches and falls over. Worse, your batches are now too big and your downstream latency spikes.

The right answer is "flush at size N or after D, whichever comes first." Every production batcher uses both.

The buffer is just a slice¶

For a junior-level batcher, buffer []Item is enough. Reset it with buffer = buffer[:0] after flush. This reuses the underlying array and is allocation-free for steady-state operation. You do not need ring buffers, lock-free accumulators, or any of the structures professional.md will discuss.

buf := make([]Item, 0, maxSize)
// ... use buf, append, flush ...
buf = buf[:0]  // empty, but cap(buf) is still maxSize

If you accidentally write buf = buf[:0] before flushing instead of after, you lose the whole batch. We will see that bug in find-bug.md.

"Flush" is a synchronous call¶

In the simplest batcher, flush(buf) is a blocking call. If the database write takes 200 ms, the batcher goroutine is busy for 200 ms and cannot accept new items during that window. That is fine for a starter implementation — the producers will block in their channel send until the batcher returns. We will see how to overlap flush with accumulation in middle.md.

The synchronous flush gives you a property worth noting: at most one batch is in flight at a time. That simplifies error handling enormously. If the flush fails, you know exactly which items are affected — the ones in the buffer at the moment you started the flush. With a concurrent flush you have to manage that mapping yourself.

The sink is an interface¶

Always pass the sink as an interface, not as a concrete type:

type Sink interface {
    Write(batch []Item) error
}

This lets you swap a fake sink in tests, a logging sink in development, and a Postgres sink in production without recompiling the batcher. It also forces you to think about the contract: what does it mean for Write to return an error? Does it mean "none of the items landed", "some of them landed", "all of them landed but the ack was lost"? The sink interface is where those questions get pinned down. We will revisit this when we cover error handling.

Defensive copy before handing the buffer to the sink¶

If your sink stores the slice it receives, you have a problem: the batcher reuses the buffer for the next batch, and now the sink sees overwritten data. In the junior implementation we make a defensive copy before passing the buffer to the sink, so the sink can hold the slice as long as it likes.

batch := make([]Item, len(buf))
copy(batch, buf)
sink.Write(batch)
buf = buf[:0]

The copy costs O(N) memory writes per flush, which is usually negligible compared to the network round-trip the flush triggers. If profiling shows the copy as a hotspot you can move to a double-buffer scheme covered in senior.md.

Real-World Analogies¶

The delivery truck¶

A parcel costs more or less the same to deliver whether the truck is full or empty: the driver's hourly wage, the fuel for the route, the toll. So the dispatcher waits until the truck is full or until the oldest parcel is in danger of missing its promised delivery date, then sends the truck out. Size trigger and time trigger together. Without the time trigger, a parcel addressed to a small village in the mountains might wait six months for the truck to fill up; without the size trigger, the truck makes the trip half empty every day. Real logistics companies use both.

The bus¶

A city bus leaves the terminal when full or when its scheduled departure time arrives, whichever comes first. The departure time is the time trigger, the seat count is the size trigger. Once again, both are necessary: rush hour fills the bus before the time trigger; midnight Tuesday hits the time trigger long before the bus is full.

The dishwasher¶

Nobody runs the dishwasher with two dishes in it. You wait until it is full. But if you have guests coming, you run it at the time threshold even if it is half full. "Guests coming" is the time trigger; "tray full" is the size trigger.

The bakery oven¶

Bakers do not bake one croissant. They fill a tray. The tray is the size trigger; "we open at 7 AM" is the time trigger. At 6:30 AM, even if the tray is only 60% full, it has to go in the oven. By 6:55 the croissants are ready.

The shipping manifest¶

Customs paperwork is the same whether a container holds one item or a thousand. So freight forwarders accumulate items until they have a container's worth and ship them as one entry. The "per-call cost" is the customs paperwork; the "per-item cost" is the warehouse handling. Batching reduces the first; nothing reduces the second.

The shape of the problem repeats everywhere: per-unit fixed cost amortised across many units, with a cap on how long any single unit may wait.

Mental Models¶

"Cost per call, cost per item"¶

Every downstream sink has two cost components:

• Per-call cost (C_call): connection lookup, query planning, network round-trip, transaction overhead. Roughly constant regardless of payload size up to some limit. Examples: 5 ms for a Postgres INSERT round-trip, 2 ms for a Kafka produce request, 50 ms for an HTTP _bulk call to Elastic.
• Per-item cost (C_item): proportional to how many items are in the payload. Examples: the time Postgres spends parsing each row, the bytes Kafka must serialise, the JSON overhead of each Elastic document.

Without batching, sending N items costs N * (C_call + C_item). With batching, it costs C_call + N * C_item. The savings — (N-1) * C_call — is exactly what batching buys you. If C_call is small compared to C_item, batching is not worth it. If C_call is huge (network round-trip to a database 5 ms away), batching can be a 100x throughput win.

Example numbers:

• Postgres INSERT, local network: C_call ≈ 1 ms, C_item ≈ 50 µs. Batch of 100 takes 1 ms + 5 ms = 6 ms, versus 100 unbatched calls at 100 * 1.05 ms = 105 ms. 17x speedup.
• Kafka produce, same datacentre: C_call ≈ 2 ms, C_item ≈ 1 µs. Batch of 1000 takes 2 ms + 1 ms = 3 ms, versus 1000 unbatched calls at 1000 * 2.001 ms = 2001 ms. 667x speedup.
• In-process function call: C_call ≈ 50 ns, C_item ≈ 50 ns. Batch of 100 saves 99 * 50 ns = 5 µs. Not worth it.

Always start with this back-of-envelope. If (N-1) * C_call is not at least an order of magnitude more than your batcher's overhead, you do not need a batcher.

"Latency budget"¶

Every operation in your service has a latency budget. If a user-facing API must respond in 100 ms, and your batcher adds 50 ms of waiting (the time trigger), you have spent half your budget. So the time trigger is not free — it is exactly the worst-case extra latency you are paying for amortisation. Pick it deliberately, document it, and make sure everyone downstream knows.

A common mistake is to copy linger.ms = 100 from a tutorial without realising that you are committing to a 100 ms latency floor. If your SLA is 50 ms p99, you have already failed the SLA before any other code runs.

"Worst case is the only case that matters"¶

The size trigger fires on the lucky day when traffic is high. The time trigger fires on the slow day. The slow day is the one that determines correctness — if your time trigger is 1 hour and your service handles audit logs, an item that arrives at 14:59:59 might not reach the database until 16:00. Whether that is acceptable is a business decision, but it must be a decision, not an accident.

This generalises: when you design a batcher, draw the worst-case timeline. "What does the timeline look like when the size trigger never fires?" If the answer is "it is fine", proceed. If the answer is "we lose data on crash", redesign.

"Two clocks"¶

A batcher has two notions of time:

1. Wall clock: what time.Now() says.
2. Batch clock: how long the oldest item in the current buffer has been waiting.

The time trigger is about the batch clock, not the wall clock. A time.Ticker is wall-clock-based and is an approximation: if the buffer was just flushed and a new item arrives 1 ms before the next tick, the next flush happens 1 ms later instead of D later. That is usually fine. If you need a strict batch clock, use time.Timer reset on the first arrival of each batch. We will see both in this file.

"Block, drop, or grow"¶

When a producer's Add arrives and the input channel is full, exactly three things can happen:

1. Block: the producer waits until there is space. Default behaviour of an unbuffered or full bounded channel.
2. Drop: the item is discarded. The producer never blocks, but data is lost.
3. Grow: the buffer expands. The producer never blocks and no data is lost, but memory can grow unboundedly.

The junior batcher uses block, because it is the safest default and exposes the problem to upstream code. The other two are policy choices documented in 01-backpressure. There is no "right" answer; there are only consequences.

Pros and Cons¶

Pros¶

• Throughput: orders of magnitude higher than per-item calls in many setups (see the cost-per-call model above).
• Fewer round-trips: lower network load, fewer connection slots used, fewer TCP packets, less TLS handshake amortisation work.
• Lower downstream load: one big INSERT is much cheaper for Postgres than 1000 small ones (one WAL flush, one transaction, one parser pass, one query-plan lookup).
• Better resource utilisation: connection pools, prepared statements, and HTTP keep-alive all benefit. Fewer connections means less memory at every layer.
• Smoother CPU usage: bursty per-item calls produce CPU spikes; batched calls smooth them.
• Cheaper retries: one retry for a batch of 100 items is one retry; 100 individual retries with backoff can mean 100x the wall time.

Cons¶

• Added latency: every item waits up to MaxBatchDelay. This is the fundamental cost.
• All-or-nothing failure: if a batch fails, all items in it fail. You need policy for retry, split, or dead-letter.
• Memory: pending items sit in RAM. Under backpressure this can grow unless bounded.
• Shutdown complexity: items in the buffer must be flushed before the process exits, or they are lost.
• Ordering subtleties: items are not necessarily processed in the order they arrived if the batcher uses multiple flush goroutines or if retries reorder things.
• Hard to reason about correlation: a batch of mixed-tenant items may fail because of one bad row, taking the rest down with it.
• Observability blind spots: per-item latency includes the wait-in-buffer time, which is invisible if the batcher is not instrumented.

Use Cases¶

Excellent fits¶

• Database INSERTs: 500 rows per multi-row INSERT instead of 500 individual ones. Postgres INSERT INTO t (...) VALUES (...), (...), ... or COPY FROM. MySQL multi-row INSERT. SQLite WAL mode.
• Kafka producer: this is what Kafka's linger.ms and batch.size configure internally. You can layer your own batcher on top for application-level grouping.
• HTTP bulk APIs: Elastic _bulk, BigQuery insertAll, Splunk HTTP Event Collector, Datadog /api/v1/series, Prometheus remote write, OpenTelemetry OTLP.
• Log shipping: collect log lines, ship every N or every D. The pattern behind Fluentd, Vector, Filebeat.
• Metrics aggregation: collect counter increments, aggregate, push as one statsd packet. The pattern behind statsd clients, the OpenTelemetry SDK, and Prometheus' own scrape buffer.
• Email or notification fan-out: batch up "you have N new messages" digests instead of one email per event.
• CDC sinks: change-data-capture pipelines almost always batch downstream writes.
• Audit logs: batch flush to an append-only store like S3 or a journaling DB.

Poor fits¶

• Strictly per-request semantics: when each item needs an individual synchronous response to the caller before the caller can move on. A login endpoint should not batch.
• Very low traffic: if you average 1 item per minute, the time trigger fires before the size trigger ever does, and you have only paid the latency cost without buying any throughput.
• Idempotency cost is high: if retrying a batch means having to deduplicate downstream, and the dedup is expensive, the math may not favour batching.
• Heterogeneous items: if items cannot easily be combined (different shards, different tenants, different schemas), batching forces routing logic that may cost more than it saves.

Code Examples¶

A minimal in-memory sink¶

We start with a fake sink so we can focus on the batcher itself.

package main

import (
    "sync"
)

type Sink struct {
    mu      sync.Mutex
    flushed [][]int
}

func (s *Sink) Write(batch []int) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    cp := make([]int, len(batch))
    copy(cp, batch)
    s.flushed = append(s.flushed, cp)
    return nil
}

func (s *Sink) Count() int {
    s.mu.Lock()
    defer s.mu.Unlock()
    return len(s.flushed)
}

func (s *Sink) Total() int {
    s.mu.Lock()
    defer s.mu.Unlock()
    n := 0
    for _, b := range s.flushed {
        n += len(b)
    }
    return n
}

We make a defensive copy inside Write because, even though the batcher also makes a copy, real sinks may hold the slice across goroutines and we want to be explicit. In production code, document who copies; do not rely on a copy at both ends.

A minimal batcher with both triggers¶

package main

import "time"

type Batcher struct {
    in       chan int
    sink     *Sink
    maxSize  int
    maxDelay time.Duration
    done     chan struct{}
}

func NewBatcher(sink *Sink, maxSize int, maxDelay time.Duration) *Batcher {
    b := &Batcher{
        in:       make(chan int, 1024),
        sink:     sink,
        maxSize:  maxSize,
        maxDelay: maxDelay,
        done:     make(chan struct{}),
    }
    go b.run()
    return b
}

func (b *Batcher) Add(item int) { b.in <- item }

func (b *Batcher) flush(buf []int) {
    batch := make([]int, len(buf))
    copy(batch, buf)
    _ = b.sink.Write(batch)
}

func (b *Batcher) run() {
    defer close(b.done)
    buf := make([]int, 0, b.maxSize)
    ticker := time.NewTicker(b.maxDelay)
    defer ticker.Stop()
    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                if len(buf) > 0 {
                    b.flush(buf)
                }
                return
            }
            buf = append(buf, item)
            if len(buf) >= b.maxSize {
                b.flush(buf)
                buf = buf[:0]
            }
        case <-ticker.C:
            if len(buf) > 0 {
                b.flush(buf)
                buf = buf[:0]
            }
        }
    }
}

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

That is roughly 45 lines and handles both triggers, a basic shutdown, and an in-flight flush on close.

Driving it¶

package main

import (
    "fmt"
    "time"
)

func main() {
    sink := &Sink{}
    b := NewBatcher(sink, 5, 100*time.Millisecond)

    // Send 12 items, then wait for the time trigger, then close.
    for i := 0; i < 12; i++ {
        b.Add(i)
    }
    time.Sleep(250 * time.Millisecond)
    b.Close()

    fmt.Println("batches flushed:", sink.Count())
    fmt.Println("items written:", sink.Total())
}

You should see two size-triggered flushes (of 5 each) and one time- or close-triggered flush of the remaining 2. batches = 3, items = 12. Run it a few times — the exact split between time-trigger and close-trigger depends on the race between the 250 ms sleep and the 100 ms ticker, but the total is always 12.

A simple HTTP example¶

To make the value tangible, here is the same shape against a fake HTTP endpoint that simulates network latency:

package main

import (
    "fmt"
    "time"
)

type HTTPSink struct {
    latency time.Duration
    calls   int
}

func (h *HTTPSink) Write(batch []int) error {
    time.Sleep(h.latency)
    h.calls++
    return nil
}

func unbatched(n int, sink *HTTPSink) time.Duration {
    start := time.Now()
    for i := 0; i < n; i++ {
        _ = sink.Write([]int{i})
    }
    return time.Since(start)
}

func batched(n int, sink *HTTPSink, size int) time.Duration {
    start := time.Now()
    buf := make([]int, 0, size)
    for i := 0; i < n; i++ {
        buf = append(buf, i)
        if len(buf) == size {
            _ = sink.Write(buf)
            buf = buf[:0]
        }
    }
    if len(buf) > 0 {
        _ = sink.Write(buf)
    }
    return time.Since(start)
}

func main() {
    sink := &HTTPSink{latency: 5 * time.Millisecond}
    nVal := 1000

    fmt.Println("unbatched:", unbatched(nVal, sink))
    sink = &HTTPSink{latency: 5 * time.Millisecond}
    fmt.Println("batched 100:", batched(nVal, sink, 100))
    sink = &HTTPSink{latency: 5 * time.Millisecond}
    fmt.Println("batched 1000:", batched(nVal, sink, 1000))
}

Run this. You will see something like:

unbatched: 5.0s
batched 100: 50ms
batched 1000: 5ms

A thousand-fold speedup is not unusual for round-trip-dominated workloads. The numbers do not always look this dramatic — local databases with sub-millisecond round trips and CPU-bound items see modest gains — but the shape of the win is universal.

Wiring up a goroutine-safe Add with WaitGroup¶

In a real service, many goroutines call Add concurrently. The channel handles that automatically, but you also need to wait for them all to finish enqueueing before you call Close. A WaitGroup works:

func main() {
    sink := &Sink{}
    b := NewBatcher(sink, 100, 50*time.Millisecond)

    var wg sync.WaitGroup
    for w := 0; w < 8; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for i := 0; i < 1000; i++ {
                b.Add(id*1000 + i)
            }
        }(w)
    }
    wg.Wait()
    b.Close()
    fmt.Println("items written:", sink.Total()) // 8000
}

Note the order: producers finish (wg.Wait), then we Close the batcher. If you Close first, the producers will panic on send to a closed channel.

Using time.Timer instead of time.Ticker¶

The ticker fires on a fixed schedule, regardless of when the last flush happened. If you want the time trigger to be measured from "first item arrived after the last flush", use a time.Timer that you reset on each flush:

func (b *Batcher) runTimer() {
    defer close(b.done)
    buf := make([]int, 0, b.maxSize)
    var timer *time.Timer
    var timerC <-chan time.Time
    armTimer := func() {
        if timer == nil {
            timer = time.NewTimer(b.maxDelay)
            timerC = timer.C
        }
    }
    disarmTimer := func() {
        if timer != nil {
            if !timer.Stop() {
                select { case <-timer.C: default: }
            }
            timer = nil
            timerC = nil
        }
    }
    for {
        select {
        case item, ok := <-b.in:
            if !ok {
                if len(buf) > 0 {
                    b.flush(buf)
                }
                disarmTimer()
                return
            }
            buf = append(buf, item)
            if len(buf) == 1 {
                armTimer() // first item: start the clock
            }
            if len(buf) >= b.maxSize {
                b.flush(buf)
                buf = buf[:0]
                disarmTimer()
            }
        case <-timerC:
            if len(buf) > 0 {
                b.flush(buf)
                buf = buf[:0]
            }
            disarmTimer()
        }
    }
}

This version waits exactly maxDelay from the first item, no more, no less. It is slightly more complex than the ticker version but has stronger latency guarantees per batch — which is what you usually want in production.

Most production batchers use the timer style. The ticker style is easier to write and good enough for many cases.

Coding Patterns¶

"Reset by reslice, not by realloc"¶

buf = buf[:0]

This keeps the underlying array, so the next batch reuses the same memory. buf = make([]int, 0, cap) would allocate a fresh array every time and pressure the GC.

If your items are pointers (or contain pointers), you may also want to nil out the entries before reslicing, so the GC can reclaim what the items point to:

for i := range buf {
    buf[i] = nil
}
buf = buf[:0]

For value items (ints, small structs without pointers), this is unnecessary. For pointer items in a long-running batcher, forgetting the nil-out can pin large object graphs and look like a memory leak.

"One goroutine owns the buffer"¶

The buffer is touched only inside run(). No mutex is needed. This is the idiom; if you find yourself locking the buffer, you have probably structured the batcher wrong.

The corollary: do not store the buffer on the struct. Make it a local variable inside run(). That way the type system enforces single ownership.

"Send on a channel is your only public API"¶

Add is a channel send. That gives you backpressure for free: when the channel is full, callers block. You do not have to write any extra code for that.

func (b *Batcher) Add(item Item) { b.in <- item }

If you want a non-blocking send, write a separate method and name it explicitly:

func (b *Batcher) TryAdd(item Item) bool {
    select {
    case b.in <- item:
        return true
    default:
        return false
    }
}

Never silently change Add from blocking to dropping. That is a guaranteed production incident.

"Centralised close"¶

Only the orchestrator — the code that constructed the batcher — calls Close. Producers never close the channel. The batcher's Close method is the only call site that touches close(b.in). This single rule prevents a whole class of "send on closed channel" panics.

"Flush in one place"¶

Resist the urge to write flush inline in both the size and time branches. Factor it into a method. That way:

• A future addition (metrics, tracing, retry) goes in one place.
• The reset of buf = buf[:0] lives next to the flush call, so they cannot drift apart.
• Code review sees one flush path, not two slightly different ones.

func (b *Batcher) flushAndReset(buf *[]int, reason string) {
    if len(*buf) == 0 {
        return
    }
    batch := make([]int, len(*buf))
    copy(batch, *buf)
    _ = b.sink.Write(batch)
    *buf = (*buf)[:0]
    // future: emit metric tagged with reason
}

The pointer-to-slice argument is a small wart but lets the helper reset the caller's slice. Alternatively, return the new slice — *buf = b.flushAndReset(*buf, "size") — same idea.

"The flush function takes a context.Context"¶

A production-grade flush takes a context so the caller can cancel a slow downstream call. We omit it from the junior examples to keep the focus on the pattern, but in middle.md you will see:

type Sink interface {
    Write(ctx context.Context, batch []Item) error
}

Start adding the ctx parameter now even if you do not use it. Adding it later forces an API break.

Clean Code¶

• Name the trigger reason. When you log or emit metrics, say reason="size" or reason="time" or reason="shutdown". Future-you will thank present-you when debugging a "why is the batch size always exactly 47?" incident.
• Make maxSize and maxDelay explicit configuration with units, not magic numbers. MaxBatchSize int and MaxBatchDelay time.Duration not N int and D time.Duration.
• Use a struct, not free functions. A batcher is naturally stateful; pretending otherwise leads to global variables.
• Keep the flush function pluggable — pass a Sink interface, not a concrete *sql.DB. This makes the batcher unit-testable.
• Document the policy: in the package doc-comment, write down the blocking semantics of Add, the latency floor introduced by MaxBatchDelay, and the shutdown contract.
• Use named return values sparingly inside the batcher — the consumer goroutine is hot code and named returns can confuse the reader about what is assigned where.
• Lift error handling out of the run loop. The run loop should call b.handleFlushError(err) and stay focused on the orchestration. Putting retry logic inline turns the loop into spaghetti.

Product Use¶

A real product team uses batchers in places like:

• The audit log subsystem: 1000 user actions per second arrive as gRPC calls. The handler enqueues, the batcher flushes 500 rows per INSERT to Postgres. Saves both the DB CPU (one parse instead of 500) and the WAL fsyncs (one instead of 500).
• The notification fan-out: a "user followed you" event arrives; the batcher accumulates events per recipient for 5 seconds and sends one push notification with "5 new followers". This is also a combining batcher; we will discuss that in middle.md.
• The metrics SDK: every counter increment is enqueued; the batcher emits one statsd packet per 100 ms with all increments aggregated.
• The analytics pipeline: every page view emits an event; the batcher flushes 5 MB of events to S3 every 30 seconds or 100 000 events, whichever first.
• The webhook delivery: each customer gets at most one webhook per minute regardless of how many events they generate, by batching downstream notifications.

Without batching, each of these subsystems would either fall over under load or require 10x the infrastructure.

A product question that often comes up: "Can we batch user-facing writes too?" Usually no — the user is waiting for the response. But some teams have made it work by writing the user-facing record synchronously to a fast store (Redis, an in-memory queue) and batching the slow store write (Postgres) asynchronously. The user sees their action persisted; the durable store catches up within seconds. This is the "write-behind" pattern, covered in the caching subsection.

Error Handling¶

The junior batcher above does _ = b.sink.Write(batch). That is terrible — it silently drops errors. At the junior level we are mostly concerned with mechanics, but you should at least know the policy options for when flush fails:

• Log and continue — fine for best-effort metrics and logs. The batch is dropped but operation continues.
• Retry with backoff — for everything that matters. We cover the retry algorithm in middle.md.
• Dead-letter — push to a separate queue for human or automated review.
• Crash — if the whole batch failing means the system is in an unrecoverable state. Crash early and let the orchestrator restart.

For a junior implementation, at least log:

func (b *Batcher) flush(buf []int) {
    batch := make([]int, len(buf))
    copy(batch, buf)
    if err := b.sink.Write(batch); err != nil {
        log.Printf("batcher: flush failed (%d items): %v", len(batch), err)
    }
}

This is still wrong for production (the items are gone), but at least an operator sees the failure.

Partial vs total failure¶

Some sinks return "failed for items X, Y, Z" (Elastic _bulk does, Kafka's per-record-callback API does). Others give you a single error per call (database/sql.Exec does). The shape of the sink's failure model determines the shape of your error handling. We will return to this in middle.md.

Panic in the flush¶

If the sink's Write panics, your batcher goroutine dies and the channel keeps filling until producers block forever. Recover at the goroutine boundary:

func (b *Batcher) run() {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("batcher: panic in run loop: %v", r)
        }
        close(b.done)
    }()
    // ... rest of run
}

This is the bare minimum. A real implementation would also restart the goroutine or signal the supervisor. Covered in senior.md.

Security Considerations¶

Batching is not directly a security topic, but a few things are worth flagging:

• Mixed-tenant batches: if your batch combines items from many tenants, a failure attributable to one tenant's data can take down processing for the rest. Some teams batch per tenant to localise blast radius.
• Sensitive data in memory: items sit in RAM longer with a batcher than without. Zero them out on flush if the items contain secrets.
• Resource exhaustion: an unbounded input channel is a DoS vector — an attacker who can flood your endpoint can drive the batcher's memory unbounded. Bounded channels with explicit backpressure prevent this.
• Log injection: if you log "flushed batch of 500 items: [...]" with item contents, an attacker controlling item content can inject log lines. Standard logging hygiene applies.
• Replay during partial flush on crash: if a process crashes mid-flush and is restarted from a checkpoint, items can be replayed. The downstream sink must be idempotent or de-duplicating.

We treat these in more depth in the security review file under senior.md and the api-security-checklist skill.

Performance Tips¶

• Pre-allocate the buffer with make([]T, 0, maxSize) so the first batch does not grow the slice.
• Make Add non-allocating — pass values, not pointers, when items are small. A chan int64 does not allocate per send; a chan *Event may or may not, depending on escape analysis.
• Buffer the input channel (make(chan T, 1024)) so producers do not block on every send.
• Pick maxSize based on the downstream's sweet spot. Postgres likes 100–1000 rows per INSERT; Kafka likes ~16 KB batches; Elastic _bulk likes 5–15 MB; statsd likes one UDP packet.
• Avoid interface{} items if you can. They allocate (boxing) on every Add.
• If your items are large, consider passing pointers but documenting the ownership: "after calling Add, the batcher owns the pointer; do not modify the pointee."
• Use a sync.Pool to reuse batch slices across flushes if profiling shows allocation pressure. This is overkill for most cases.
• Measure before tuning. Add a metric for batch size before changing maxSize. A common mistake is to triple maxSize and discover the size trigger was almost never firing — the time trigger was driving everything.

Best Practices¶

• Always have both triggers. Never just size, never just time.
• Always make maxDelay configurable per environment. Test, staging, and production often want different values.
• Always pass a context.Context into the flush function so it can be cancelled. Even if you do not use it today, start with it.
• Never block the producer indefinitely. If your input channel can fill, decide now whether to drop, block, or grow — and document the decision.
• Document the latency budget your time trigger consumes. "This batcher adds up to 50 ms of latency to every item" should be in the package doc.
• Always flush on shutdown. A batcher that loses the last batch is a batcher that loses data every deploy.
• Emit metrics: batch size, flush latency, flush reason, queue depth, dropped items (if you drop). These four numbers are the difference between a fragile batcher and a maintainable one.
• Write a test for the size trigger, a test for the time trigger, and a test for empty close. These three tests catch 80% of regressions.
• Pin the order of trigger checks. Always check size after appending, time after waking on tick. If you reverse them you can introduce subtle bugs.
• Always copy the buffer before handing it to the sink. The cost is O(N) writes; the bug is silent and catastrophic.

Edge Cases and Pitfalls¶

The "stuck-at-N-minus-1" bug¶

If your maxSize is 100 and you receive 99 items, then no more, those 99 sit there forever — unless you have the time trigger. This is the canonical reason to always have both triggers.

The "flush during flush" question¶

What happens if flush is slow and items keep arriving? In our junior implementation, items pile up in the input channel until it fills, then producers block. That is correct behaviour — backpressure — but it does mean a slow flush stalls producers. The middle.md file discusses how to overlap flush with accumulation.

The "ticker drifts" trap¶

time.Ticker fires on a fixed schedule, not "D after the last flush". If a flush took 200 ms and the ticker is 100 ms, the next tick may already be queued when the flush finishes. For a junior batcher this is fine. For a precise time trigger, use time.Timer reset on each flush. We saw both above.

The "missed tick" trap¶

time.Ticker's channel has buffer 1. If your consumer is busy in a flush, multiple ticks can queue up. When select finally reaches the <-ticker.C branch, only one of those ticks is consumed; the rest are dropped. This is documented behaviour, not a bug, but it can confuse newcomers.

Closing the channel from a producer¶

If a producer closes the input channel, the batcher exits — but other producers panic on their next send. Only the owner closes the channel, and the owner is the orchestrator, not any producer. The batcher's Close method enforces this.

maxDelay == 0¶

time.NewTicker(0) panics. Always validate the configuration:

if maxDelay <= 0 {
    panic("batcher: maxDelay must be positive")
}
if maxSize <= 0 {
    panic("batcher: maxSize must be positive")
}

Or, better, return an error from NewBatcher.

maxSize == 1¶

The batcher flushes on every Add. It is a degenerate "batcher" that is just a goroutine indirection. It costs more than calling sink.Write directly because of the channel hop. Almost never what you want.

A panic in the sink¶

If sink.Write panics, our defer close(b.done) still runs but the channel stays open. Subsequent Add calls block forever. Recover inside flush (or run the whole loop under recover, depending on policy).

Shutdown ordering¶

If you call b.Close() from the same goroutine that is still calling b.Add(...), you will hang. The producer is sending on the channel, the batcher is waiting for the channel to close, but the channel only closes when the producer stops sending and the orchestrator calls Close. This is a special case of "do not deadlock by ordering". The fix is to ensure all producers have finished — via WaitGroup — before Close.

Loss on crash¶

A batcher's items are durable only when they reach the sink. If the process crashes between Add and flush, those items are gone. If your sink is itself unreliable (Kafka with no acks), they can be gone even after flush. Treat "in-buffer" items as not-yet-durable and design the upstream accordingly (idempotent retries, at-least-once delivery, dead-letter queue for re-processing).

Common Mistakes¶

1. Only one trigger. "I have a size trigger, that is enough." It is not; low traffic exposes the bug.
2. Forgetting to flush on close. The buffer has pending items; the program exits; they are gone. Always flush on shutdown.
3. Sharing the buffer with the sink. The sink keeps a reference and reads from it later; the batcher reuses the slice; the sink sees garbage. Always copy.
4. Unbounded input channel. make(chan T) (or make(chan T, 1_000_000)) hides backpressure until OOM.
5. Producer calls close(b.in). Now another producer's Add panics. Always centralise closes.
6. No metric on batch size. When throughput regresses, you have no way to know whether the size trigger is firing or the time trigger.
7. No metric on flush latency. The downstream gets slower; your producers slow down; you have no insight into where the time is going.
8. select over only one channel — losing the trigger structure. If you find yourself writing for x := range b.in you have given up on the time trigger.
9. Flushing inside a mutex. A common refactor: someone adds a mutex around the buffer "to be safe", then flush is called with the mutex held, and the entire system blocks on the downstream call. Single-ownership avoids this entirely.
10. Forgetting ticker.Stop(). A goroutine leak that masquerades as a memory leak.
11. Re-using buf after handing to the sink. Same as #3 stated differently: if the sink stores the slice and you buf = buf[:0] and append, the sink sees garbage.
12. Add returning before the item is enqueued. A select with default that drops silently looks like Add succeeded. Rename it to TryAdd if you want that semantics.

Common Misconceptions¶

• "Batching adds latency, so it is bad for user-facing APIs." Only true if the API waits for the batch to flush. Audit logs, metrics, and notifications can be async; the API returns immediately.
• "Bigger batches are always better." False. There is a knee in the curve. Above some size, downstream latency increases faster than throughput, and the per-batch failure cost grows.
• "A batcher needs a mutex around the buffer." Not if exactly one goroutine touches it.
• "A batcher with linger.ms = 0 is not a batcher." False. Even with no waiting, items arriving in the same scheduling window are still batched. The size trigger still fires; the time trigger just fires immediately on the next tick.
• "Channels are slow, so the batcher overhead is high." Channels cost on the order of 50–100 ns per send-receive pair on modern hardware. Compared to a 1 ms Postgres round-trip, the channel overhead is in the noise.
• "Two batchers in series are equivalent to one." No. The intermediate stage adds latency without adding amortisation if the second batcher's size trigger never fires. Resist this design.
• "If the sink supports streaming, batching is pointless." Streaming reduces per-call cost, not per-item cost. Batching can still help by amortising context switches and giving the kernel larger send-msg chunks.

Tricky Points¶

• The size trigger should fire when len(buf) >= maxSize, after the append, not before. If you check before, you either flush a batch of maxSize - 1 and lose the new item, or grow past max.
• The time trigger should not flush an empty buffer. Wasted call, wasted log line, wasted metric event.
• select is randomised among ready cases. If both input and ticker.C are ready, the choice is unpredictable. This usually does not matter, but if you ever debug a "missing tick", that is why.
• close(b.in) is what tells the loop "no more items". A done channel is what the outside uses to wait for the loop to exit. Two channels, two directions, two different responsibilities.
• The compiler does not warn you about _ = sink.Write(...). Silent error discards are valid Go. Be deliberate.
• time.Ticker.C has a buffer of 1. Two ticks queued look like one to the consumer.
• time.Timer requires drain-before-reset. The pattern is if !t.Stop() { <-t.C }; t.Reset(d). Forgetting the drain gives you a spurious tick. Go 1.23 relaxed this for some cases.
• Buffers that grow above maxSize are valid Go but indicate a bug. cap(buf) may be larger than maxSize after a single oversized batch; the next buf = buf[:0] keeps the larger capacity, so memory stays high. Consider rebuilding the buffer if it grew beyond expected size.
• On Go 1.22+ the per-iteration loop-variable scoping eliminates the classic capture bug, but old habits and old code still apply.

Test¶

A test for the size trigger:

func TestSizeTrigger(t *testing.T) {
    sink := &Sink{}
    b := NewBatcher(sink, 3, time.Hour) // huge delay so only size can fire
    for i := 0; i < 6; i++ {
        b.Add(i)
    }
    b.Close()
    if got := sink.Count(); got != 2 {
        t.Fatalf("expected 2 size-triggered batches, got %d", got)
    }
}

A test for the time trigger:

func TestTimeTrigger(t *testing.T) {
    sink := &Sink{}
    b := NewBatcher(sink, 1000, 50*time.Millisecond)
    b.Add(1)
    b.Add(2)
    time.Sleep(150 * time.Millisecond)
    b.Close()
    if got := sink.Count(); got < 1 {
        t.Fatalf("expected at least one time-triggered flush, got %d", got)
    }
}

A test for empty close:

func TestEmptyClose(t *testing.T) {
    sink := &Sink{}
    b := NewBatcher(sink, 10, time.Hour)
    b.Close()
    if got := sink.Count(); got != 0 {
        t.Fatalf("expected 0 flushes on empty close, got %d", got)
    }
}

A test for partial close:

func TestPartialClose(t *testing.T) {
    sink := &Sink{}
    b := NewBatcher(sink, 100, time.Hour)
    for i := 0; i < 7; i++ {
        b.Add(i)
    }
    b.Close()
    if got := sink.Count(); got != 1 {
        t.Fatalf("expected 1 close-triggered flush, got %d", got)
    }
    if got := sink.Total(); got != 7 {
        t.Fatalf("expected 7 items, got %d", got)
    }
}

A test for concurrent producers:

func TestConcurrentProducers(t *testing.T) {
    sink := &Sink{}
    b := NewBatcher(sink, 50, 10*time.Millisecond)
    var wg sync.WaitGroup
    for w := 0; w < 10; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := 0; i < 100; i++ {
                b.Add(i)
            }
        }()
    }
    wg.Wait()
    b.Close()
    if got := sink.Total(); got != 1000 {
        t.Fatalf("expected 1000 items, got %d", got)
    }
}

Run all four with go test -race. Race-free, deterministic, fast — exactly what a unit test should be.

Tricky Questions¶

Q: What does the batcher do if maxSize is 1? A: It flushes on every Add — a degenerate "batcher" that is just a goroutine indirection. It costs more than calling sink.Write directly because of the channel hop. Real batchers have maxSize >= 10 to make the cost worthwhile.

Q: Can two flushes overlap? A: In the junior implementation, no — the run loop is single-threaded. In production batchers we often overlap flush with accumulation; that is middle.md.

Q: What if the producer is faster than the batcher? A: The input channel fills, the next Add blocks. That is backpressure. Whether to convert it to drop-on-floor or queue-grow is a policy decision documented in 01-backpressure.

Q: Why a done channel? A: Because close(b.in) returns immediately; we need a way for Close() to wait until the loop has actually flushed and exited. The done channel closes from inside the loop on exit.

Q: What happens if I call Close twice? A: close(b.in) panics on the second call. Add a sync.Once if your callers might do that:

type Batcher struct {
    closeOnce sync.Once
    // ...
}
func (b *Batcher) Close() {
    b.closeOnce.Do(func() {
        close(b.in)
    })
    <-b.done
}

Q: What happens if the ticker fires while a flush is in progress? A: In our implementation, the run loop is busy in the case item branch executing flush, so the case <-ticker.C cannot be selected. The tick queues (buffer 1) and is consumed on the next iteration. The buffer is empty at that point, so the time-triggered branch sees len(buf) == 0 and skips. No bug.

Q: Can flush and Add race on buf? A: No. flush is called only from inside run, and run is single-threaded. The only way to race on buf is to expose it; do not.

Q: How do I make sure Close is always called? A: Either: - Use defer b.Close() in main. - Wire it to a signal.Notify for SIGTERM/SIGINT. - Let the orchestrator (lifecycle library, fx, wire, etc.) call it.

Forgetting to call Close means the last partial batch is lost. Critical.

Q: If I have 10 batchers, do I need 10 ticker goroutines? A: Each time.NewTicker registers with the runtime timer heap; it does not spawn a goroutine. So you can have thousands of tickers cheaply. The runtime fires them all from a small set of internal helpers.

Q: Can the batcher flush less than maxSize if maxSize is reached during shutdown? A: Yes. On Close, whatever is in the buffer flushes regardless of size. That is the point of the close-triggered flush.

Q: What happens if I Add after Close? A: Send on a closed channel panics. Document this; callers must coordinate.

Cheat Sheet¶

// Triggers
size:  flush when len(buf) >= MaxBatchSize  (after append)
time:  flush when ticker fires AND len(buf) > 0
close: flush remaining items on Close

// Buffer
make([]T, 0, MaxBatchSize); reset by buf = buf[:0]

// Concurrency
one goroutine owns the buffer; producers send on channel

// Shutdown
close(input) -> loop drains remaining -> close(done)
caller calls Close() -> blocks on <-done

// Tests
TestSizeTrigger:     huge delay, exact size, expect N/maxSize batches
TestTimeTrigger:     huge size, small delay, sleep > delay, expect >=1 batch
TestEmptyClose:      close immediately, expect 0 batches
TestPartialClose:    items < maxSize, close, expect 1 batch with remainder
TestConcurrent:      many producers, sum of items = expected total

// Metrics to emit
batch_size_items      (histogram)
flush_duration_ms     (histogram)
flush_reason          (counter, labeled: size|time|shutdown)
queue_depth           (gauge)
dropped_items         (counter, only if dropping)

// Pitfalls
- one trigger only: bug
- unbounded channel: bug
- producer-side close: bug
- shared buffer with sink: bug
- forget Close: bug (last batch lost)
- ticker.Stop missing: leak

Self-Assessment Checklist¶

• I can write a batcher with size and time triggers from memory.
• I know why both triggers are necessary, with a worked example for each failure mode.
• I can explain why the buffer needs no mutex.
• I know what buf = buf[:0] does and why.
• I can write a test that proves the size trigger fires.
• I can write a test that proves the time trigger fires.
• I can write a test that proves the close trigger fires for a partial buffer.
• I can explain what happens on Close if the buffer is non-empty.
• I know the per-call versus per-item cost model and can decide whether batching helps for a given sink.
• I can identify a "stuck-at-N-minus-1" bug in someone else's code.
• I know how to avoid the "shared buffer with sink" bug.
• I can convert a time.Ticker-based time trigger into a time.Timer-based one and explain when each is preferable.
• I can explain "block, drop, or grow" and articulate which the junior batcher chooses and why.
• I know the four metrics every batcher should emit.
• I can list at least five common mistakes and explain how to fix each.

Summary¶

A batcher is one goroutine, one buffer, one channel, one ticker. It collects items until size or time says "ship it", then hands the batch to a sink. The reason every Go service in production has one hidden somewhere is that it turns 50 000 small calls per second into 500 medium-sized ones — an order of magnitude less load on every downstream system, at the cost of a few tens of milliseconds of latency per item.

The hard parts at the junior level are:

• Picking both triggers (size and time), not just one.
• Flushing on shutdown so the last batch is not lost.
• Copying the buffer before handing it to the sink.
• Centralising channel close in the orchestrator, not in producers.
• Using buf = buf[:0] to avoid per-flush allocation.

You now know enough to write a working batcher and reason about its behaviour under both burst and idle traffic. The middle level adds production-grade shutdown, error handling, retries, observability, and integration with real sinks. The senior level adds adaptive sizing, latency budgeting, and architectural composition with backpressure and worker pools. The professional level dives into the implementation of Kafka's producer client and Postgres' COPY protocol so you can tune them with full understanding.

What You Can Build¶

• A standalone batcher library that wraps any sink (database, HTTP, Kafka stub) and exposes Add, Close, and a metric stream.
• A buffered audit-log writer for a small web service that flushes 100 rows per INSERT or every second.
• A metrics aggregator that emits one statsd packet per second instead of one per increment.
• A "save-game" buffer in a game server that flushes player state every 100 writes or every 10 seconds.
• A simple log shipper that buffers stdout lines and posts them to a remote endpoint every 5 seconds.
• A notification debouncer that combines "5 new followers" into one push instead of five.

Further Reading¶

• The Go blog: Go Concurrency Patterns (Pike, 2012) — channel patterns including pipelines that prefigure batchers.
• The Go blog: Advanced Go Concurrency Patterns (Cox, 2013) — select and timer patterns you will use here.
• Kafka documentation: linger.ms and batch.size — the canonical real-world batcher, and the configuration knobs that everyone has touched at least once.
• The database/sql documentation on multi-row INSERT and prepared statements.
• The pgx documentation on CopyFrom — Postgres' fastest bulk insert path.
• Elasticsearch documentation on the _bulk endpoint — the canonical HTTP batch API.
• The OpenTelemetry Go SDK source: sdktrace/batch_span_processor.go — a production-grade batcher in ~300 lines.
• The next file: middle.md — production-ready shutdown, partial flush, retries, and real-world integration.

Related Topics¶

• 01-backpressure — what to do when producers outrun the batcher.
• 02-dynamic-worker-scaling — how many goroutines should consume in parallel.
• 04-graceful-shutdown — the broader contract that close-on-SIGTERM participates in.
• 05-drain-pattern — what to do with the in-flight queue when shutting down.
• 16-time-based-concurrency/01-ticker — the prerequisite ticker mechanics.
• 16-time-based-concurrency/02-afterfunc — time.AfterFunc as an alternative time trigger.
• 06-errgroup-x-sync/01-errgroup — the standard tool for waiting on producer goroutines.
• 04-context-package — for cancelling slow flushes.
• 02-channels/06-closing-channels — the rules around close that govern who can close b.in.
• 17-goroutine-pools-3rd-party/01-ants — a pool that often sits in front of a batcher.

Step-by-Step: Build a Batcher From Nothing¶

This section walks through writing a batcher one step at a time so you can build the intuition by motion, not by reading. Open an editor, create a new file batcher.go, and follow along.

Step 1: The Sink Interface¶

Start with the smallest possible interface:

package main

type Sink interface {
    Write(batch []int) error
}

That is it. The batcher's only contract with the world is "I will call Write with a slice of items, and I will respect the error it returns." Anything more elaborate — context, partial results, metrics — comes later.

Step 2: The No-Op Batcher¶

Now write the simplest "batcher" possible. It is not really a batcher; it just forwards every item.

type NoBatcher struct{ sink Sink }

func (n *NoBatcher) Add(item int) error {
    return n.sink.Write([]int{item})
}

Run it against the Sink from the Postgres-like example with callLatency = 5 ms. 1000 items take 5 seconds. The slowest possible "batcher" — useful as a baseline.

Step 3: Size-Only Batcher¶

Add a slice buffer and a size trigger.

type SizeBatcher struct {
    sink    Sink
    maxSize int
    buf     []int
}

func (s *SizeBatcher) Add(item int) error {
    s.buf = append(s.buf, item)
    if len(s.buf) >= s.maxSize {
        return s.flush()
    }
    return nil
}

func (s *SizeBatcher) flush() error {
    if len(s.buf) == 0 {
        return nil
    }
    err := s.sink.Write(s.buf)
    s.buf = s.buf[:0]
    return err
}

func (s *SizeBatcher) Close() error { return s.flush() }

This works but is single-threaded — concurrent callers will race on buf. Run it under -race to see the report. We will fix that in step 5.

It also has the "stuck-at-N-minus-1" bug: if the producer sends 99 items and stops, the batcher never flushes. The fix is the time trigger, coming in step 4.

Step 4: Adding the Time Trigger¶

The time trigger requires a goroutine, because no producer call is going to wake us up at the deadline. We need a separate consumer.

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

func New(sink Sink, maxSize int, maxDelay time.Duration) *TimedBatcher {
    t := &TimedBatcher{
        sink: sink, maxSize: maxSize, maxDelay: maxDelay,
        in:   make(chan int, 1024),
        done: make(chan struct{}),
    }
    go t.run()
    return t
}

func (t *TimedBatcher) Add(item int) { t.in <- item }

func (t *TimedBatcher) run() {
    defer close(t.done)
    buf := make([]int, 0, t.maxSize)
    ticker := time.NewTicker(t.maxDelay)
    defer ticker.Stop()
    for {
        select {
        case item, ok := <-t.in:
            if !ok {
                if len(buf) > 0 {
                    _ = t.sink.Write(buf)
                }
                return
            }
            buf = append(buf, item)
            if len(buf) >= t.maxSize {
                _ = t.sink.Write(buf)
                buf = buf[:0]
            }
        case <-ticker.C:
            if len(buf) > 0 {
                _ = t.sink.Write(buf)
                buf = buf[:0]
            }
        }
    }
}

func (t *TimedBatcher) Close() {
    close(t.in)
    <-t.done
}

This is the canonical junior batcher. It has both triggers, supports many concurrent producers, flushes on close. Run it under -race — clean.

Step 5: Defensive Copy¶

Replace t.sink.Write(buf) with the copy idiom:

batch := make([]int, len(buf))
copy(batch, buf)
_ = t.sink.Write(batch)

Now the sink can hold the slice as long as it likes; the next buf = buf[:0]; append(buf, ...) will not corrupt it.

Step 6: Error Logging¶

Replace _ = t.sink.Write(batch) with:

if err := t.sink.Write(batch); err != nil {
    log.Printf("batcher: flush of %d items failed: %v", len(batch), err)
}

Silent error drops are a junior anti-pattern that becomes a senior incident.

Step 7: Validation¶

Add input validation to New:

func New(sink Sink, maxSize int, maxDelay time.Duration) (*TimedBatcher, error) {
    if sink == nil {
        return nil, errors.New("batcher: sink is nil")
    }
    if maxSize <= 0 {
        return nil, errors.New("batcher: maxSize must be positive")
    }
    if maxDelay <= 0 {
        return nil, errors.New("batcher: maxDelay must be positive")
    }
    // ... rest
}

time.NewTicker(0) panics, so the maxDelay check is not just hygiene.

Step 8: Idempotent Close¶

type TimedBatcher struct {
    // ...
    closeOnce sync.Once
}

func (t *TimedBatcher) Close() {
    t.closeOnce.Do(func() {
        close(t.in)
    })
    <-t.done
}

Now Close is safe to call any number of times. Repeated calls are no-ops; the second caller just waits for done.

Step 9: Reason-Tagged Flushes¶

flush := func(reason string) {
    if len(buf) == 0 {
        return
    }
    log.Printf("batcher: flush reason=%s size=%d", reason, len(buf))
    batch := make([]int, len(buf))
    copy(batch, buf)
    if err := t.sink.Write(batch); err != nil {
        log.Printf("batcher: flush failed: %v", err)
    }
    buf = buf[:0]
}

When we add metrics in middle.md, this single change is what makes "what fraction of flushes are time-triggered?" trivially queryable.

Step 10: Stop the Ticker¶

The defer ticker.Stop() is already there. Verify it. If you forget it, the goroutine leaks every time you create a batcher and discard it without Close.

Step 11: Write the Three Canonical Tests¶

We have them above. Run them. They should all pass.

Step 12: Profile and Tune¶

Run a benchmark:

func BenchmarkBatcher(b *testing.B) {
    sink := &Sink{}
    bat := New(sink, 100, 10*time.Millisecond)
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        bat.Add(i)
    }
    bat.Close()
}

Compare against the NoBatcher baseline. Compare with maxSize of 10, 100, 1000. Plot the curve. You should see the throughput-vs-size knee that the diagram earlier sketched.

That is the whole junior journey: 12 steps, ~50 lines of code, and a working production-quality batcher.

Worked Example: A Postgres-Like Sink¶

Let us extend the in-memory sink to simulate a realistic database. We will add per-call latency, per-item cost, and an occasional failure, so we can see how the batcher behaves under realistic conditions.

package main

import (
    "errors"
    "math/rand"
    "sync"
    "time"
)

type DBSink struct {
    callLatency time.Duration
    perItemCost time.Duration
    failureRate float64 // 0..1
    mu          sync.Mutex
    written     int
    calls       int
    failures    int
}

func (d *DBSink) Write(batch []int) error {
    time.Sleep(d.callLatency + time.Duration(len(batch))*d.perItemCost)
    d.mu.Lock()
    defer d.mu.Unlock()
    d.calls++
    if rand.Float64() < d.failureRate {
        d.failures++
        return errors.New("simulated db error")
    }
    d.written += len(batch)
    return nil
}

Drive it under three configurations:

func bench(name string, sink interface{ Write([]int) error }, n int) {
    t := time.Now()
    for i := 0; i < n; i++ {
        _ = sink.Write([]int{i})
    }
    elapsed := time.Since(t)
    fmt.Printf("%s: %v for %d items\n", name, elapsed, n)
}

Run it. With callLatency = 5 ms, perItemCost = 10 µs, you see:

• Per-item (unbatched): about 5 s for 1000 items.
• Batched 100 per call: about 60 ms for 1000 items.
• Batched 1000 per call: about 15 ms for 1000 items.

The batched-1000 case shows the per-item cost dominating; the per-call cost has been almost completely amortised. Going from 100 to 1000 only saves another 45 ms; going from 1 to 100 saved 4.94 seconds. That is the curve in the diagram earlier.

Worked Example: A Notifications Combiner¶

Some batchers combine items rather than just grouping them. A "you have N new messages" digest is the canonical example: instead of flushing N separate notifications, you flush one notification per recipient with a count.

type Combiner struct {
    in       chan string // recipient
    maxDelay time.Duration
    done     chan struct{}
    sink     func(recipient string, count int)
}

func (c *Combiner) run() {
    defer close(c.done)
    counts := map[string]int{}
    ticker := time.NewTicker(c.maxDelay)
    defer ticker.Stop()
    flush := func() {
        for r, n := range counts {
            c.sink(r, n)
        }
        counts = map[string]int{}
    }
    for {
        select {
        case r, ok := <-c.in:
            if !ok {
                flush()
                return
            }
            counts[r]++
        case <-ticker.C:
            if len(counts) > 0 {
                flush()
            }
        }
    }
}

Now sending 100 "user followed" events for the same recipient within the window produces one push: "you have 100 new followers". The size trigger is gone here because the value of batching is in combining, not just grouping — there is no hard cap on the map size, only the time window.

This kind of batcher is everywhere: rate-limited alerts, debounced UI updates, "compact" log lines that fold repeated entries. It shares the structure of the basic batcher but exchanges the slice buffer for a map and the size trigger for nothing or for a cap on map size.

Worked Example: Per-Tenant Sub-Batches¶

A real system often has multiple downstreams or multiple shards. A batcher that mixes items from many tenants into one INSERT loses tenant isolation — a single bad row from tenant A can fail the whole batch including tenant B's rows.

The fix is to keep one sub-buffer per tenant and flush them independently:

type ShardedBatcher struct {
    in       chan Item
    maxSize  int
    maxDelay time.Duration
    sinks    map[string]Sink // one per tenant
    done     chan struct{}
}

type Item struct {
    Tenant string
    Body   []byte
}

func (s *ShardedBatcher) run() {
    defer close(s.done)
    bufs := map[string][]Item{}
    ticker := time.NewTicker(s.maxDelay)
    defer ticker.Stop()
    flush := func(tenant string) {
        if len(bufs[tenant]) == 0 {
            return
        }
        _ = s.sinks[tenant].Write(bufs[tenant])
        bufs[tenant] = bufs[tenant][:0]
    }
    flushAll := func() {
        for t := range bufs {
            flush(t)
        }
    }
    for {
        select {
        case item, ok := <-s.in:
            if !ok {
                flushAll()
                return
            }
            bufs[item.Tenant] = append(bufs[item.Tenant], item)
            if len(bufs[item.Tenant]) >= s.maxSize {
                flush(item.Tenant)
            }
        case <-ticker.C:
            flushAll()
        }
    }
}

This is the same shape as the basic batcher but with one buffer per shard key. Production data pipelines extend this pattern with bounded per-tenant memory, fairness quotas, and per-tenant time triggers (so a quiet tenant does not have its tiny batch starve next to a noisy one). We cover the fairness aspect in senior.md.

Worked Example: An HTTP Bulk Endpoint Client¶

The classic non-database batcher is an HTTP bulk endpoint. Here is a stub of an Elastic _bulk-style client:

type BulkClient struct {
    url      string
    client   *http.Client
    in       chan json.RawMessage
    maxSize  int
    maxBytes int
    maxDelay time.Duration
    done     chan struct{}
}

func (b *BulkClient) run() {
    defer close(b.done)
    var buf bytes.Buffer
    var count int
    ticker := time.NewTicker(b.maxDelay)
    defer ticker.Stop()
    flush := func() {
        if count == 0 {
            return
        }
        req, _ := http.NewRequest("POST", b.url, bytes.NewReader(buf.Bytes()))
        req.Header.Set("Content-Type", "application/x-ndjson")
        resp, err := b.client.Do(req)
        if err == nil {
            resp.Body.Close()
        }
        buf.Reset()
        count = 0
    }
    for {
        select {
        case doc, ok := <-b.in:
            if !ok {
                flush()
                return
            }
            buf.Write(doc)
            buf.WriteByte('\n')
            count++
            if count >= b.maxSize || buf.Len() >= b.maxBytes {
                flush()
            }
        case <-ticker.C:
            flush()
        }
    }
}