Chhung's Portfolio

Generic Go buffering for batching typed records, aggregating keyed updates, and scheduling flushes through a shared priority worker pool.

Overview

gbuffer keeps high-frequency writes out of request paths. It accepts typed values, turns them into a flush payload, then runs sinks directly or through one shared priority worker pool.

The public sink is deliberately small:

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

T is the payload ready to write. A batcher flushes []T; an aggregator flushes map[K]V.

Why It Exists

Applications often need both of these workloads:

Write records together: logs, orders, audit events, analytics events
Merge repeated keyed updates: views, likes, impressions, inventory deltas

These workloads should share scheduling without forcing every caller into untyped queues or separate worker pools.

Core Pieces

Piece	Job
`Sinker[T]`	Writes one flushed payload
`Batcher[T]`	Turns many values into `[]T`
`Aggregator[K, V]`	Turns keyed updates into `map[K]V`
`WorkerPool`	Runs submitted jobs by priority
`SinkJob[T]`	Bridges typed payloads to shared pool jobs

Batch Records

Use Batcher[T] when individual records should reach storage 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)

Aggregate Updates

Use Aggregator[K, V] when repeated updates to same key should be coalesced before writing.

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))

Shared Priority Pool

Typed buffers can share one non-generic WorkerPool. Each flushed payload becomes a Job, preserving typed sinks while centralizing concurrency and priority.

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

logs, _ := gbuffer.NewBatcher(writeLogs, gbuffer.WithBatchPool[Log](pool))
views, _ := gbuffer.NewAggregator(writeViews, gbuffer.SumInt64, gbuffer.WithAggregatorPool[string, int64](pool))

Shutdown And Limits

Flush(context.Context) submits pending work. Close(context.Context) stops new writes, flushes remaining data, and waits for scheduled work until context expiry. Future Add calls return ErrClosed.

Pending in-memory values are bounded. ErrFull signals that an aggregator cannot accept another distinct key. Durable overflow stores such as Redis, Kafka, or disk are documented extension patterns, not core runtime dependencies.

gbuffer

Tech Stack