Skip to content
126 changes: 119 additions & 7 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ type BatchItem struct {
func main() {
// create a batcher
batcher := batcher.New[*BatchItem](
batcher.WithBatchSize[*BatchItem](100), // will batch each 100 items.
batcher.WithBatchInterval[*BatchItem](1*time.Second), // or each second.
batcher.WithBatchSize[*BatchItem](100), // flush at 100 items,
batcher.WithBatchInterval[*BatchItem](10*time.Millisecond), // or when the oldest item is 10ms old.
// then run this processor with each batch
batcher.WithProcessor(func(items []*BatchItem) error {
fmt.Printf("processing batch with %d items...\n", len(items))
Expand Down Expand Up @@ -83,8 +83,8 @@ To create a batcher you can use the `New` function:

```go
batcher := batcher.New[*BatchItem](
batcher.WithBatchSize[*BatchItem](100), // will batch each 100 items.
batcher.WithBatchInterval[*BatchItem](1*time.Second), // or each second.
batcher.WithBatchSize[*BatchItem](100), // flush at 100 items,
batcher.WithBatchInterval[*BatchItem](10*time.Millisecond), // or when the oldest item is 10ms old.
// then run this processor with each batch
batcher.WithProcessor(func(items []*BatchItem) error {
fmt.Printf("processing batch with %d items...\n", len(items))
Expand Down Expand Up @@ -344,9 +344,88 @@ fmt.Printf("batcher has %d items\n", batcher.Len())

## Available Options to configure batcher

- `WithBatchSize[*BatchItem](size int)`: sets the batch size.
- `WithBatchInterval[*BatchItem](interval time.Duration)`: sets the batch interval.
- `WithProcessor(func(items []*BatchItem) error)`: sets the processor function.
| Option | Default | Purpose |
| --- | --- | --- |
| `WithProcessor(fn)` | no-op | The function called with each batch. |
| `WithBatchSize[T](n)` | 1000 | Flush as soon as a batch holds `n` items. |
| `WithBatchInterval[T](d)` | 1s | Maximum age of a partial batch. 1s is conservative; see [Choosing a batch interval](#choosing-a-batch-interval). |
| `WithMaxQueueSize[T](n)` | unbounded | Bound queued items to get back-pressure instead of unbounded growth. |
| `WithConcurrency[T](n)` | 1 | Process `n` batches at once. Requires `WithoutOrderedProcessing`. |
| `WithoutOrderedProcessing[T]()` | off | Acknowledges that `n > 1` gives up cross-batch ordering and processor mutual exclusion. |
| `WithCloseGrace[T](d)` | 30s | How long `Close` waits before reporting an incomplete drain. |
| `WithErrorBufferSize[T](n)` | 1024 | How many diagnostics `Errors()` buffers before dropping newer ones. |
| `WithSkipAutoStart[T]()` | off | Do not start processing in `New`; call `Start` yourself. |

## Choosing a batch interval

The interval is the **maximum age of a partial batch**, not a periodic flush tick:
the timer starts when the first item enters an empty batch. Sparse traffic therefore
waits the full interval per item, and a request crossing several batching services
pays up to one interval per hop.

Expected batch size is approximately **min(arrival rate x interval, BatchSize)**.
The arrival-rate term is measured, not estimated: at 10,000 items/s with a 10ms
interval it predicts 100 items per batch, and the measured mean is ~101.

`BatchSize` is the binding constraint whenever it fills first, and the interval then
never fires. With the default `BatchSize` of 1,000, a service at 50,000 items/s
reaches 1,000 items after about 20ms, so configuring a 100ms interval changes nothing
— which is exactly what the 50,000/s row below shows.

Use the formula to pick an interval from the coalescing you actually need:

| Arrival rate | 10ms | 100ms | 1s (default) |
| --- | --- | --- | --- |
| 1,000/s | ~11 items/batch, ~94 calls/s, p99 ~12ms | ~100 items/batch, ~10 calls/s, p99 ~100ms | ~1000 items/batch, ~1 call/s, p99 ~981ms |
| 10,000/s | ~101 items/batch, ~99 calls/s, p99 ~10ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms | ~1000 items/batch, ~10 calls/s, p99 ~99ms |
| 50,000/s | ~500 items/batch, ~100 calls/s, p99 ~10ms | ~1000 items/batch (capped by `BatchSize`), ~50 calls/s, p99 ~20ms | ~1000 items/batch, ~50 calls/s, p99 ~20ms |

**The default is 1s, and it is deliberately conservative.** It favours coalescing and
a low downstream call rate over latency, which is the safe direction for a shared
library: a caller who needs lower latency can measure it and set one option, whereas a
default that quietly multiplied everyone's downstream call rate would not show up
until something else saturated.

**Set 10ms when latency matters.** At sparse rates it is the difference between p99
~981ms and ~12ms while still coalescing ~11 items per batch. Note the cost in the
table: at 1,000/s it moves ~1 call/s to ~94 calls/s. Above ~10k/s, `BatchSize` closes
batches before either timer fires and the choice stops mattering.

```go
b := batcher.New(
batcher.WithProcessor(processor),
batcher.WithBatchInterval[Item](10*time.Millisecond),
)
```

The full matrix, and why 10ms is recommended rather than made the default, are in
[`docs/improvements/default-window.md`](./docs/improvements/default-window.md).

**Shrinking the interval is not overload protection.** Measured with a saturated
downstream, every interval from 10ms to 1s accepted the same offered load while
completing the same amount, with the queue absorbing the difference. Use
`WithMaxQueueSize` and `Enqueue` for that, or upstream flow control.

**A slow processor bounds the effective interval.** At the default `Concurrency` of 1,
one batch must finish before the next starts, so the effective interval is
`max(BatchInterval, processor duration)`. A window below the processor duration buys
nothing and can measure worse, because queueing dominates once the processor is the
bottleneck.

Measured with a 50ms processor at 10k items/s on darwin/arm64 (Apple M4 Pro, Go
1.26.5, `GOMAXPROCS=12`):

| Configured interval | Effective behaviour | p50 latency |
| --- | --- | --- |
| 5ms | bounded by the 50ms processor | ~120ms |
| 100ms | bounded by the interval | ~100ms |

`TestReproducesInlineSlowProcessorInversion` asserts the ordering — the smaller
window is slower — and not these millisecond values, which are host-specific. If your
processor is slow, raise `WithConcurrency` (with `WithoutOrderedProcessing`) rather
than lowering the interval: `TestConcurrencyRemovesSlowProcessorCoupling` pins that
8 workers reduce p50 against 1 worker in the same scenario, because workers add the
capacity a shorter window cannot.

## FX Integration

Expand Down Expand Up @@ -431,6 +510,39 @@ func main() {
}
```

### Configuring the Fx batcher beyond size and interval

`ProvideBatcherInFX` takes batch size and interval positionally, which covers most
applications. For anything else — bounded queues, worker concurrency, close grace,
diagnostics buffer — use `ProvideBatcherInFXWithOptions`, which accepts the same
options as `New`:

```go
batcher.ProvideBatcherInFXWithOptions[*BatchItem](
func(processor *Processor) batcher.Processor[*BatchItem] {
return processor.Process
},
batcher.WithBatchSize[*BatchItem](500),
batcher.WithBatchInterval[*BatchItem](10*time.Millisecond),
batcher.WithMaxQueueSize[*BatchItem](10_000),
batcher.WithConcurrency[*BatchItem](4),
batcher.WithoutOrderedProcessing[*BatchItem](),
)
```

Do not pass `WithProcessor` here: the processor comes from the injected factory, and
supplying it as an option would bypass dependency injection.

Both variants forward the Fx stop-hook context to `Shutdown`, so your application's
shutdown deadline governs the drain:

| Case | Result |
| --- | --- |
| Normal drain | `nil` |
| Stop context expires | `*ShutdownIncompleteError` as an Fx lifecycle error; the drain continues in the background |
| Batcher holds queued work but never started | Drains anyway |
| Repeated stop | Idempotent |

The key part of the FX integration is the processor function passed to `ProvideBatcherInFX`. This function is an FX resolver that:

1. Takes any dependencies you need (like the `Processor` struct) as parameters
Expand Down
Loading
Loading