batchgo

README

📦 batchgo

A generic, bounded, in-memory batching engine for Go.

batchgo collects items from concurrent producers and processes them in size- or time-bounded batches. It is designed for bulk database operations, search indexing, message publishing, API batching, push delivery, and analytics pipelines.

---

📥 Installation

batchgo supports Go 1.22 and newer.

go get github.com/codenaline/batchgo

---

🚀 Quick Start

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/codenaline/batchgo"
)

func main() {
	handler := func(_ context.Context, messages []string) error {
		fmt.Println(messages)
		return nil
	}

	batcher, err := batchgo.New(batchgo.Config{
		MaxSize:        100,
		MaxWait:        50 * time.Millisecond,
		Workers:        4,
		QueueSize:      1_000,
		ErrorQueueSize: 64,
	}, handler)
	if err != nil {
		panic(err)
	}

	if err := batcher.Add(context.Background(), "hello"); err != nil {
		panic(err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := batcher.Close(ctx); err != nil {
		panic(err)
	}
}

---

✨ Features

• Generic Batcher[T] for any item type
• Size- and time-triggered batch dispatch
• Safe concurrent submission
• Bounded queues and worker backpressure
• Fixed worker pool with concurrent handler execution
• Blocking Add and non-blocking TryAdd
• Ordered flush barriers
• Graceful, idempotent shutdown
• Best-effort asynchronous handler errors
• Panic recovery with stack information
• No required third-party dependencies

---

⚙️ Configuration

Every field is required and must be greater than zero:

Field	Purpose
MaxSize	Flush a batch as soon as it contains this many items.
MaxWait	Flush a partial batch after this duration from its first item.
Workers	Maximum number of handler calls that may run concurrently.
QueueSize	Bound items and control commands waiting for the coordinator.
ErrorQueueSize	Bound errors waiting in the best-effort error stream.

New returns an error for invalid configuration or a nil handler.

---

📤 Submitting Items

Add(ctx, item) waits for queue capacity. It returns the context error if the context expires before acceptance, or ErrClosed after shutdown begins.

if err := batcher.Add(ctx, item); err != nil {
	return err
}

TryAdd(item) makes one non-blocking attempt. It returns ErrQueueFull when no capacity is immediately available:

if err := batcher.TryAdd(item); errors.Is(err, batchgo.ErrQueueFull) {
	// Apply application-specific overload policy.
}

Once either method succeeds, the item is dispatched to the handler exactly once during the lifetime of the batcher. A handler error or panic still counts as that single dispatch; the library does not retry.

---

🧰 Handler Contract

Handlers may run concurrently when Workers is greater than one. Item order is preserved within a batch, but separate batches may finish out of order.

The item slice is valid only for the duration of the handler call. Copy it before retaining it:

handler := func(ctx context.Context, items []Message) error {
	retained := append([]Message(nil), items...)
	return store(ctx, retained)
}

---

🧯 Error Handling

Handler errors and recovered panics are published through Errors(). Recovered panics are reported as *batchgo.PanicError with the recovered value and stack.

go func() {
	for err := range batcher.Errors() {
		log.Printf("batch handler failed: %v", err)
	}
}()

Reporting never blocks workers. When the configured buffer is full, the newest error is dropped and DroppedErrors() is incremented. Retry, logging, metrics, alerting, and recovery policy belong to the application.

---

🚿 Flushing

Flush(ctx) inserts an ordered barrier. It flushes the current partial batch and waits until every batch ordered before that barrier finishes handler execution. Items submitted after the barrier are not included.

if err := batcher.Flush(ctx); err != nil {
	return err
}

If the context expires, Flush returns its error while processing continues internally.

---

🛑 Graceful Shutdown

Close(ctx) stops admission, drains every accepted item, waits for active handlers, closes Errors(), and finally closes Done(). It is safe to call concurrently and repeatedly.

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

if err := batcher.Close(ctx); err != nil {
	log.Printf("batchgo is still draining: %v", err)
}

If the context expires, graceful draining continues in the background. Wait on Done() only when the caller is prepared to wait for draining to finish.

---

🔒 Guarantees and Scope

• Queues and batches are bounded and held in memory.
• Accepted items are not durable across process crashes or restarts.
• Item order is preserved within each batch.
• Separate batches may execute and complete concurrently.
• Handler panics are recovered with stack information.
• There are no automatic retries, priorities, keyed batching, dynamic worker scaling, forced cancellation, or per-item results.

---

📊 Benchmarks

Run benchmarks locally:

go test -run=^$ -bench=. -benchmem ./...

The benchmark verifies that every submitted item is handled before reporting parallel Add throughput for several batch sizes.

---

🧪 Testing

go vet ./...
go test ./...
go test -race ./...

The suite covers validation, admission, backpressure, batch timing, flush barriers, out-of-order completion, panic recovery, dropped errors, concurrent shutdown, bounded completion tracking, and exactly-once dispatch.

---

📌 Status

batchgo is preparing for its first v0.1.0 release. The API, lifecycle semantics, documentation, benchmarks, race tests, and stress tests are in place.

---

🤝 Contributing

Pull requests are welcome. See CONTRIBUTING.md for details.

👨🏻‍💻 Credits

• Mahdi Rezaei

📄 License

The MIT License. See LICENSE for details.

Documentation

Overview

Package batchgo collects concurrently submitted items and processes them in bounded, in-memory batches.

A Batcher flushes when a batch reaches Config.MaxSize, when Config.MaxWait elapses after the first item, or when Flush or Close requests a barrier.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/codenaline/batchgo"
)

func main() {
	handler := func(_ context.Context, items []int) error {
		fmt.Println(items)
		return nil
	}

	batcher, err := batchgo.New(batchgo.Config{
		MaxSize:        2,
		MaxWait:        time.Second,
		Workers:        1,
		QueueSize:      16,
		ErrorQueueSize: 4,
	}, handler)
	if err != nil {
		panic(err)
	}

	if err := batcher.Add(context.Background(), 1); err != nil {
		panic(err)
	}
	if err := batcher.Add(context.Background(), 2); err != nil {
		panic(err)
	}
	if err := batcher.Close(context.Background()); err != nil {
		panic(err)
	}

}

Output:
[1 2]

Index

• Variables
• type Batcher
  • func New[T any](cfg Config, handler Handler[T]) (*Batcher[T], error)
  • func (b *Batcher[T]) Add(ctx context.Context, item T) error
  • func (b *Batcher[T]) Close(ctx context.Context) error
  • func (b *Batcher[T]) Done() <-chan struct{}
  • func (b *Batcher[T]) DroppedErrors() uint64
  • func (b *Batcher[T]) Errors() <-chan error
  • func (b *Batcher[T]) Flush(ctx context.Context) error
  • func (b *Batcher[T]) TryAdd(item T) error
• type Config
• type Handler
• type PanicError
  • func (e *PanicError) Error() string

Examples

• Package

Variables

var (
	// ErrClosed indicates that shutdown has begun and no new command is accepted.
	ErrClosed = errors.New("batchgo: closed")
	// ErrQueueFull indicates that TryAdd could not enqueue an item immediately.
	ErrQueueFull = errors.New("batchgo: queue full")
)

Types

type Batcher

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

Batcher collects values of type T and delivers completed batches to a Handler. Its methods are safe for concurrent use.

func New

func New[T any](cfg Config, handler Handler[T]) (*Batcher[T], error)

New validates cfg and starts a batcher, its coordinator, and worker pool.

func (*Batcher[T]) Add

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

Add waits until item is accepted, ctx is canceled, or shutdown begins. A nil return means the item will be dispatched exactly once during this batcher lifetime.

func (*Batcher[T]) Close

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

Close begins graceful shutdown and waits for all accepted items and active handlers. If ctx expires, graceful shutdown continues and Done eventually closes. Close is safe to call concurrently and repeatedly.

func (*Batcher[T]) Done

func (b *Batcher[T]) Done() <-chan struct{}

Done returns a channel closed after workers stop and Errors is closed.

func (*Batcher[T]) DroppedErrors

func (b *Batcher[T]) DroppedErrors() uint64

DroppedErrors returns the number of errors omitted because Errors was full.

func (*Batcher[T]) Errors

func (b *Batcher[T]) Errors() <-chan error

Errors returns the best-effort handler error stream. Reporting never blocks workers; errors are dropped when the configured buffer is full.

func (*Batcher[T]) Flush

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

Flush waits for all batches ordered before its coordinator barrier to finish. If ctx expires, processing continues internally.

func (*Batcher[T]) TryAdd

func (b *Batcher[T]) TryAdd(item T) error

TryAdd attempts to accept item without waiting for queue capacity.

type Config

type Config struct {
	// MaxSize is the number of items that causes an immediate batch flush.
	MaxSize int
	// MaxWait is the maximum age of a partial batch, measured from its first item.
	MaxWait time.Duration
	// Workers is the number of handlers that may run concurrently.
	Workers int
	// QueueSize bounds item and control commands waiting for the coordinator.
	QueueSize int
	// ErrorQueueSize bounds errors waiting to be read from Batcher.Errors.
	ErrorQueueSize int
}

Config controls batching, worker concurrency, and bounded queue capacities. Every field must be greater than zero.

type Handler

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

Handler processes one batch. Calls may run concurrently when Config.Workers is greater than one. The item slice is valid only for the duration of the call; a handler that retains data must copy it.

type PanicError

type PanicError struct {
	// Value is the value recovered from the handler panic.
	Value any
	// Stack is the stack captured when the panic was recovered.
	Stack []byte
}

PanicError describes a panic recovered from a Handler.

func (*PanicError) Error

func (e *PanicError) Error() string

Error returns a concise description of the recovered panic.