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

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

func (*Aggregator[K, V]) Flush

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

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

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]

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

func (*PriorityWorkerPool) Submit

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

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
}