Keeping Write Paths Fast with Batching and Aggregation in Go

Fast request paths often become slow for an unglamorous reason: every event triggers its own write. One log record, one analytics event, or one view increment may be cheap alone. At sustained traffic, database round trips, lock contention, queue coordination, and allocation pressure turn that work into latency and backend load.

gbuffer is a small Go library built around one goal: accept typed work cheaply now, then flush it efficiently later.

More project details: iam.foxie.vip.

Problem: Per-Event Writes Do Not Scale Cleanly

Consider a video view counter:

request -> increment video_views in database -> response

This preserves a simple mental model, but makes storage part of every request. For logs and audit events, writing each item separately also increases network calls and transaction overhead. For counters, repeated updates to one key produce redundant work:

video-42 + 1
video-42 + 1
video-42 + 1

Writing those three updates independently costs three storage operations. The useful result is often only this:

video-42 + 3

The fix is not "put everything in a queue." Different workloads need different shapes, bounded memory, explicit shutdown, and predictable scheduling.

Solution: Buffer Before Writing

gbuffer has two typed buffering paths:

Both paths flush to a Sinker[T]:

type Sinker[T any] func(context.Context, T) error

The sink owns persistence. The buffer owns when and how pending values become one payload.

Batch Independent Records

Use a batcher for values that must all be preserved but can be written together.

writeLogs := gbuffer.Sinker[[]Log](func(ctx context.Context, logs []Log) error {
    return repo.InsertLogs(ctx, logs)
})

logs, err := gbuffer.NewBatcher(
    writeLogs,
    gbuffer.WithBatchSize[Log](500),
    gbuffer.WithBatchFlushInterval[Log](time.Second),
)
if err != nil {
    return err
}

return logs.Add(ctx, log)

Add appends to pending memory. Flush happens at configured size or interval. One sink call can persist many records.

Aggregate Repeated Updates

Use an aggregator when updates to same key can be combined before persistence.

views, err := gbuffer.NewAggregator(
    writeViews,
    gbuffer.SumInt64,
    gbuffer.WithEventThreshold[string, int64](1000),
    gbuffer.WithKeyThreshold[string, int64](500),
)
if err != nil {
    return err
}

return views.Add(ctx, videoID, int64(1))

Before flush, repeated values merge in memory:

video-42 + 1
video-42 + 1
video-99 + 1

becomes

map[video-42:2 video-99:1]

This removes redundant writes while keeping merge logic explicit through a Combiner[V].

Share Scheduling, Keep Types

One application may buffer logs, orders, and view counters. Creating separate worker pools for every type fragments concurrency limits and makes priority hard to control.

gbuffer wraps each typed flush payload in a non-generic Job. Typed buffers can then share one priority pool:

pool := gbuffer.NewPriorityWorkerPool(
    gbuffer.WithWorkers(8),
    gbuffer.WithQueueSize(10000),
)

This separates responsibilities:

Batcher / Aggregator: shape pending data and decide when to flush
WorkerPool: schedule submitted flushes by priority
Sinker: persist one completed payload

The request path does not need to know database batch syntax or worker scheduling details.

Middleware: Input and Output Hooks

Batch and aggregate operations benefit from optional behavior at two layers: before values enter the buffer (add middleware) and before flushed payloads are written (sink middleware).

Add middleware wraps the Add call:

type AddMiddleware[T any] func(AddFunc[T]) AddFunc[T]

Use it to skip or transform values before they enter memory:

logs := gbuffer.NewBatcher(
    writeLogs,
    gbuffer.UseAdd(
        SkipBefore[Log](startAt),
        Dedupe[Log](logKey),
        Validate[Log](validateLog),
    ),
)

Sink middleware wraps the Sinker:

type Middleware[T any] func(Sinker[T]) Sinker[T]

Use it for cross-cutting payload-level behavior:

gbuffer.UseSink(
    Retry[[]Log](retryPolicy),
    Metrics[[]Log]("logs"),
)

For aggregators, the same pattern applies with map payloads:

gbuffer.UseSink(
    Retry[map[string]int64](retryPolicy),
    Metrics[map[string]int64]("video_views"),
)

Middleware order is predictable: UseSink(A, B, C) runs A -> B -> C -> sink. Errors remain inspectable: ErrFull, ErrClosed, ErrDropped, or context.Canceled.

Keep add middleware for per-value decisions and sink middleware for payload durability and observability. This keeps the fast path clean while allowing production hardening without changing core buffer configuration.

Performance Baseline

Local benchmarks on Apple M1, macOS arm64, measure core in-memory paths:

BenchmarkBatcherAdd-8                  81818955    14.41 ns/op    8 B/op    0 allocs/op
BenchmarkAggregatorAdd-8               23008322    50.87 ns/op   72 B/op    0 allocs/op
BenchmarkPriorityWorkerPoolSubmit-8     3748939   322.7 ns/op    48 B/op    2 allocs/op

These are baselines, not production guarantees. Hardware, payload shape, contention, flush thresholds, and sink latency all change real-world results.

The comparison still explains design tradeoffs:

• Batcher.Add is cheapest: mutex-protected append until a threshold flushes.
• Aggregator.Add costs more: map access plus value combination.
• WorkerPool.Submit costs more again: priority heap work and worker coordination.

The hot path stays small by avoiding copies until flush, rotating pending storage rather than copying each input, and keeping Redis, Kafka, retries, and observability outside core execution.

Run benchmarks locally before changing behavior:

go test -bench=. -benchmem ./...

Bounded Memory and Shutdown Matter

Buffering trades immediate I/O for memory. That trade needs limits.

gbuffer keeps pending data bounded and returns ErrFull when an aggregator cannot accept another distinct key. It also exposes explicit lifecycle methods:

Flush(context.Context) error
Close(context.Context) error

Close stops new values, flushes pending data, submits remaining jobs, and waits until completion or context expiry. Future Add calls return ErrClosed.

For workloads that need stronger durability, combine memory with a spill store. When the in-memory aggregator reaches its key threshold, raw increments spill to Redis instead of returning ErrFull:

type RedisSpill[K comparable, V any] struct {
    mem    *gbuffer.Aggregator[K, V]
    client *redis.Client
}

func (s *RedisSpill[K, V]) Add(ctx context.Context, key K, value V) error {
    err := s.mem.Add(ctx, key, value)
    if err == gbuffer.ErrFull {
        return s.client.HIncrBy(ctx, "spill", fmt.Sprint(key), int64(value)).Err()
    }
    return err
}

The consumer can replay spilled keys and merge them into the next aggregation cycle. Same pattern applies for batcher overflow — spill individual values when memory fills, then drain on recovery.

When to Use Each Durability Level

Code: Memory Only

Lowest latency, lossy on restart. Good for logs, analytics, metrics.

buf, _ := gbuffer.NewBatcher(writeLogs,
    gbuffer.WithBatchSize[Log](500),
    gbuffer.WithBatchFlushInterval[Log](time.Second),
)
defer buf.Close(ctx)
buf.Add(ctx, Log{Line: "request"})

agg, _ := gbuffer.NewAggregator(writeViews, gbuffer.SumInt64,
    gbuffer.WithEventThreshold[string, int64](1000),
    gbuffer.WithKeyThreshold[string, int64](500),
)
defer agg.Close(ctx)
agg.Add(ctx, "video-42", 1)

Code: Memory + Redis Spill

Fast memory path until full, then spills to Redis. Good for views, impressions, inventory.

type SpillAgg struct {
    agg    *gbuffer.Aggregator[string, int64]
    client *redis.Client
}

func (s *SpillAgg) Add(ctx context.Context, key string, value int64) error {
    if err := s.agg.Add(ctx, key, value); err == gbuffer.ErrFull {
        return s.client.HIncrBy(ctx, "spill", key, value).Err()
    }
    return nil
}

views := &SpillAgg{agg, rdb}
views.Add(ctx, "video-42", 1)

Code: Durable Before Ack (Redis Only)

Every write persists before returning. No data loss. Good for orders, payments.

client.LPush(ctx, "orders", orderJSON)

Cost Comparison Per 1M Events

Interpretation:

• Memory-only paths are sub-microsecond per event. Redis or disk I/O dominates when spills happen.
• Durable before ack adds 2–3 orders of magnitude latency but guarantees zero data loss.
• Spill-only-on-full gives the best of both: most events stay in the fast memory path; only burst traffic pays the I/O cost.
• For aggregators, spilling raw increments is cheaper than spilling merged values because the downstream consumer can re-aggregate without coordination.

Durable overflow to Redis, Kafka, or disk is intentionally an extension pattern, not a hidden core dependency. Add it when in-memory bounds are not enough for delivery requirements.

When to Use It

Use gbuffer when writes can tolerate controlled delay and benefit from batching or coalescing. Do not use it for operations that require each request to synchronously confirm durable storage before responding.

Start with measured bottlenecks. Add batching for independent records. Add aggregation only where multiple updates safely merge. Keep thresholds and flush intervals tied to latency and memory budgets.

Links

• gbuffer source code
• Package documentation
• Benchmark notes
• Chhung's published profile