gbuffer

README

gbuffer

gbuffer is a generic buffering design for batching, aggregating, and scheduling typed work through a shared worker pool.

go get github.com/foxie-io/gbuffer

The core is intentionally small:

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

T is the flushed payload. For normal batching, T can be a slice. For aggregation, T can be a map.

Sinker[[]Log]
Sinker[map[string]int64]

Core Concepts

Concept	Responsibility
Sinker[T]	Writes one flushed payload
Batcher[T]	Converts many T values into []T payloads
Aggregator[K,V]	Converts many (K,V) inputs into map[K]V payloads
Job	Wraps a flushed payload so a pool can run it
WorkerPool	Schedules and runs jobs by priority
Memory store	Holds pending values and owns memory overflow behavior

Sinker

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

Sinker is a function type, not an interface. It writes the payload that is ready to flush.

Batch sink:

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

Aggregate sink:

writeViews := Sinker[map[string]int64](func(ctx context.Context, views map[string]int64) error {
    return repo.IncrementVideoViews(ctx, views)
})

Batcher

Batcher[T] accepts one value at a time and flushes slices.

logs := NewBatcher[Log](
    writeLogs,
    WithBatchPool[Log](pool),
    WithBatchPriority[Log](10),
    WithBatchSize[Log](500),
    WithBatchFlushInterval[Log](time.Second),
)

err := logs.Add(ctx, log)

Flow:

Log -> []Log -> SinkJob[[]Log] -> WorkerPool -> Sinker[[]Log]

Use Batcher[T] for logs, orders, audit events, analytics events, and other records that should be written in batches.

Aggregator

Aggregator[K,V] accepts key/value updates and flushes a merged map.

views := NewAggregator[string, int64](
    writeViews,
    SumInt64,
    WithAggregatorPool[string, int64](pool),
    WithAggregatorPriority[string, int64](5),
    WithEventThreshold(1000),
    WithKeyThreshold(500),
    WithAggregatorFlushInterval[string, int64](time.Second),
)

err := views.Add(ctx, videoID, int64(1))

Flow:

videoID + 1 -> map[videoID]count -> SinkJob[map[string]int64] -> WorkerPool -> Sinker[map[string]int64]

Use Aggregator[K,V] for counters and coalescing workloads, such as video views, likes, impressions, metric increments, or inventory deltas.

Worker Pool

The worker pool is not generic. This allows different typed buffers to share one scheduler.

type Job interface {
    Priority() int
    Run(context.Context) error
}

type WorkerPool interface {
    Submit(context.Context, Job) error
    Close(context.Context) error
}

Typed work is wrapped into a non-generic job.

type SinkJob[T any] struct {
    priority int
    payload  T
    sink     Sinker[T]
}

func (j SinkJob[T]) Priority() int {
    return j.priority
}

func (j SinkJob[T]) Run(ctx context.Context) error {
    return j.sink(ctx, j.payload)
}

This supports one shared pool for many types:

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

logs := NewBatcher[Log](writeLogs, WithBatchPool[Log](pool), WithBatchPriority[Log](10))
orders := NewBatcher[Order](writeOrders, WithBatchPool[Order](pool), WithBatchPriority[Order](1))
views := NewAggregator[string, int64](writeViews, SumInt64, WithAggregatorPool[string, int64](pool), WithAggregatorPriority[string, int64](5))

Memory And Overflow

Memory is a bounded pending store. Overflow belongs to memory because overflow only happens when a bounded store cannot accept more data.

Memory should own policies such as:

• Return ErrFull
• Drop overflow
• Block until space is available
• Spill overflow to Redis, Kafka, or disk

Flush policy belongs to Batcher or Aggregator, because only they know how to convert pending data into a flushed payload.

Memory overflow policy: what happens when memory is full
Batcher flush policy: when []T becomes a job
Aggregator flush policy: when map[K]V becomes a job

Shutdown

Buffers should expose explicit shutdown methods.

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

Close should:

• Stop accepting new values
• Flush pending memory
• Submit remaining jobs
• Wait for submitted jobs or return when the context expires
• Return ErrClosed for future Add calls

Design Rule

Keep responsibilities separate:

Batcher / Aggregator = shape and flush data
Memory / Redis / Kafka = hold pending data
WorkerPool = schedule jobs
Overflow = bounded store behavior
Middleware = optional behavior around add or sink

More detailed extension design lives in docs/:

• docs/benchmarks.md
• docs/middleware.md
• docs/overflow.md
• docs/errors.md
• docs/safety.md
• docs/external-scaling.md

Runnable examples live in examples/:

• examples/batcher
• examples/aggregator
• examples/shared_pool
• examples/middleware
• examples/close_flush
• examples/redis
• examples/kafka
• examples/split

Status

The core Sinker, Batcher, Aggregator, and WorkerPool design is implemented. Redis, Kafka, spill stores, and durable external scaling are documented as extension patterns.

Contributing

Contributions are welcome. See CONTRIBUTING.md, CODE_OF_CONDUCT.md, and SECURITY.md.

This project is licensed under the MIT License. See LICENSE.

Documentation

Overview

Package gbuffer provides generic batching, aggregation, and priority worker scheduling.

Index

• Variables
• func SumInt64(old, next int64) int64
• type AddFunc
• type AddMiddleware
• type AggregateAddFunc
• type AggregateAddMiddleware
• type Aggregator
  • func NewAggregator[K comparable, V any](sink Sinker[map[K]V], combine Combiner[V], opts ...AggregatorOption[K, V]) (*Aggregator[K, V], error)
  • func (a *Aggregator[K, V]) Add(ctx context.Context, key K, value V) error
  • func (a *Aggregator[K, V]) Close(ctx context.Context) error
  • func (a *Aggregator[K, V]) Flush(ctx context.Context) error
• type AggregatorOption
  • func WithAggregatorFlushInterval[K comparable, V any](d time.Duration) AggregatorOption[K, V]
  • func WithAggregatorOnError[K comparable, V any](onError func(error)) AggregatorOption[K, V]
  • func WithAggregatorPool[K comparable, V any](pool WorkerPool) AggregatorOption[K, V]
  • func WithAggregatorPriority[K comparable, V any](priority int) AggregatorOption[K, V]
  • func WithEventThreshold[K comparable, V any](n int) AggregatorOption[K, V]
  • func WithKeyThreshold[K comparable, V any](n int) AggregatorOption[K, V]
• type Batcher
  • func NewBatcher[T any](sink Sinker[[]T], opts ...BatcherOption[T]) (*Batcher[T], error)
  • func (b *Batcher[T]) Add(ctx context.Context, value T) error
  • func (b *Batcher[T]) Close(ctx context.Context) error
  • func (b *Batcher[T]) Flush(ctx context.Context) error
• type BatcherOption
  • func WithBatchFlushInterval[T any](d time.Duration) BatcherOption[T]
  • func WithBatchMaxPending[T any](n int) BatcherOption[T]
  • func WithBatchOnError[T any](onError func(error)) BatcherOption[T]
  • func WithBatchPool[T any](pool WorkerPool) BatcherOption[T]
  • func WithBatchPriority[T any](priority int) BatcherOption[T]
  • func WithBatchSinkMiddleware[T any](middleware ...Middleware[[]T]) BatcherOption[T]
  • func WithBatchSize[T any](size int) BatcherOption[T]
• type Combiner
• type Job
• type Middleware
• type PoolOption
  • func WithQueueSize(n int) PoolOption
  • func WithWorkers(n int) PoolOption
• type PriorityWorkerPool
  • func NewPriorityWorkerPool(opts ...PoolOption) *PriorityWorkerPool
  • func (p *PriorityWorkerPool) Close(ctx context.Context) error
  • func (p *PriorityWorkerPool) Submit(ctx context.Context, job Job) error
• type SinkJob
  • func NewSinkJob[T any](priority int, payload T, sink Sinker[T]) SinkJob[T]
  • func (j SinkJob[T]) Priority() int
  • func (j SinkJob[T]) Run(ctx context.Context) error
• type Sinker
  • func Use[T any](sink Sinker[T], middleware ...Middleware[T]) Sinker[T]
• type WorkerPool

Variables

var (
	ErrClosed        = errors.New("gbuffer: closed")
	ErrFull          = errors.New("gbuffer: full")
	ErrDropped       = errors.New("gbuffer: dropped")
	ErrRejected      = errors.New("gbuffer: rejected")
	ErrInvalidConfig = errors.New("gbuffer: invalid config")
)

Functions

func SumInt64

func SumInt64(old, next int64) int64

Types

type AddFunc

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

type AddMiddleware

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

type AggregateAddFunc

type AggregateAddFunc[K comparable, V any] func(context.Context, K, V) error

type AggregateAddMiddleware

type AggregateAddMiddleware[K comparable, V any] func(AggregateAddFunc[K, V]) AggregateAddFunc[K, V]

type Aggregator

type Aggregator[K comparable, V any] struct {
	// contains filtered or unexported fields
}

func NewAggregator

func NewAggregator[K comparable, V any](sink Sinker[map[K]V], combine Combiner[V], opts ...AggregatorOption[K, V]) (*Aggregator[K, V], error)

NewAggregator returns an Aggregator. combine must not call back into the aggregator: it runs while the internal mutex is held and re-entrant Add or Flush calls will deadlock.

func (*Aggregator[K, V]) Add

func (a *Aggregator[K, V]) Add(ctx context.Context, key K, value V) error

func (*Aggregator[K, V]) Close

func (a *Aggregator[K, V]) Close(ctx context.Context) error

Close stops accepting items, waits for the periodic flush loop to exit, performs a final flush, and then waits for all pooled flushes submitted before Close started to finish running. If ctx is cancelled during the final flush or the wait, ctx.Err() is returned.

func (*Aggregator[K, V]) Flush

func (a *Aggregator[K, V]) Flush(ctx context.Context) error

Flush drains aggregated values and delivers them to the sink (inline or via the configured pool). Calls are serialized against Close so Close observes every in-flight pooled job before returning.

type AggregatorOption

type AggregatorOption[K comparable, V any] func(*Aggregator[K, V])

func WithAggregatorFlushInterval

func WithAggregatorFlushInterval[K comparable, V any](d time.Duration) AggregatorOption[K, V]

func WithAggregatorOnError

func WithAggregatorOnError[K comparable, V any](onError func(error)) AggregatorOption[K, V]

WithAggregatorOnError installs a callback invoked for errors that occur outside of explicit Flush calls (the periodic flush loop). A nil callback drops errors. The callback runs on the flush loop goroutine: it must not call Close or Flush on the same Aggregator, or it will deadlock.

func WithAggregatorPool

func WithAggregatorPool[K comparable, V any](pool WorkerPool) AggregatorOption[K, V]

func WithAggregatorPriority

func WithAggregatorPriority[K comparable, V any](priority int) AggregatorOption[K, V]

func WithEventThreshold

func WithEventThreshold[K comparable, V any](n int) AggregatorOption[K, V]

func WithKeyThreshold

func WithKeyThreshold[K comparable, V any](n int) AggregatorOption[K, V]

type Batcher

type Batcher[T any] struct {
	// contains filtered or unexported fields
}

func NewBatcher

func NewBatcher[T any](sink Sinker[[]T], opts ...BatcherOption[T]) (*Batcher[T], error)

func (*Batcher[T]) Add

func (b *Batcher[T]) Add(ctx context.Context, value T) error

func (*Batcher[T]) Close

func (b *Batcher[T]) Close(ctx context.Context) error

Close stops accepting items, waits for the periodic flush loop to exit, performs a final flush, and then waits for all pooled flushes submitted before Close started to finish running. If ctx is cancelled during the final flush or the wait, ctx.Err() is returned.

func (*Batcher[T]) Flush

func (b *Batcher[T]) Flush(ctx context.Context) error

type BatcherOption

type BatcherOption[T any] func(*Batcher[T])

func WithBatchFlushInterval

func WithBatchFlushInterval[T any](d time.Duration) BatcherOption[T]

func WithBatchMaxPending

func WithBatchMaxPending[T any](n int) BatcherOption[T]

WithBatchMaxPending caps how many undrained items may sit in the internal batch; Add returns ErrFull beyond it. Note: if maxPending < batchSize and no flush interval is configured, Add will report ErrFull before a size-based flush can trigger — pair such configs with WithBatchFlushInterval.

func WithBatchOnError

func WithBatchOnError[T any](onError func(error)) BatcherOption[T]

WithBatchOnError installs a callback invoked for errors that occur outside of explicit Flush calls (the periodic flush loop). A nil callback drops errors. The callback runs on the flush loop goroutine: it must not call Close or Flush on the same Batcher, or it will deadlock.

func WithBatchPool

func WithBatchPool[T any](pool WorkerPool) BatcherOption[T]

func WithBatchPriority

func WithBatchPriority[T any](priority int) BatcherOption[T]

func WithBatchSinkMiddleware

func WithBatchSinkMiddleware[T any](middleware ...Middleware[[]T]) BatcherOption[T]

func WithBatchSize

func WithBatchSize[T any](size int) BatcherOption[T]

type Combiner

type Combiner[V any] func(old V, next V) V

type Job

type Job interface {
	Priority() int
	Run(context.Context) error
}

type Middleware

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

type PoolOption

type PoolOption func(*PriorityWorkerPool)

func WithQueueSize

func WithQueueSize(n int) PoolOption

func WithWorkers

func WithWorkers(n int) PoolOption

type PriorityWorkerPool

type PriorityWorkerPool struct {
	// contains filtered or unexported fields
}

func NewPriorityWorkerPool

func NewPriorityWorkerPool(opts ...PoolOption) *PriorityWorkerPool

func (*PriorityWorkerPool) Close

func (p *PriorityWorkerPool) Close(ctx context.Context) error

Close stops accepting new jobs and waits for queued and running jobs to finish, returning every collected job error joined into one error (errors.Join). If ctx is cancelled first, it returns ctx.Err() while workers keep draining.

func (*PriorityWorkerPool) Submit

func (p *PriorityWorkerPool) Submit(ctx context.Context, job Job) error

Submit enqueues a job as long as ctx is still live at enqueue time and the queue is not full. The ctx is not propagated into job execution: jobs always run with context.Background.

type SinkJob

type SinkJob[T any] struct {
	// contains filtered or unexported fields
}

func NewSinkJob

func NewSinkJob[T any](priority int, payload T, sink Sinker[T]) SinkJob[T]

func (SinkJob[T]) Priority

func (j SinkJob[T]) Priority() int

func (SinkJob[T]) Run

func (j SinkJob[T]) Run(ctx context.Context) error

type Sinker

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

func Use

func Use[T any](sink Sinker[T], middleware ...Middleware[T]) Sinker[T]

type WorkerPool

type WorkerPool interface {
	Submit(context.Context, Job) error
	Close(context.Context) error
}