Documentation
¶
Overview ¶
Package writebatcher provides a generic, transaction-batching write serializer for database operations. It collects items of any type T from multiple concurrent goroutines and flushes them in batched transactions through a single background worker, eliminating write contention on single-writer databases like SQLite.
This package simplifies batched database writes by:
- Serializing all writes through one goroutine (no lock contention)
- Batching items into single transactions (fewer round-trips)
- Triggering flushes by count or timeout (configurable latency vs throughput)
- Providing a generic API over any item type T
Usage ¶
Create a batcher with a BeginTx function and a FlushFunc:
wb, err := writebatcher.New[MyItem](ctx, writebatcher.Config[MyItem]{
BeginTx: func(ctx context.Context) (*sql.Tx, error) {
return db.BeginTx(ctx, nil)
},
Flush: func(ctx context.Context, tx *sql.Tx, batch []MyItem) error {
for _, item := range batch {
if _, err := tx.ExecContext(ctx, "INSERT ...", item.Val); err != nil {
return err
}
}
return nil
},
OnError: func(err error, batch []MyItem) { log.Println(err) },
MaxBatchSize: 50,
})
Submit items from any goroutine:
if err := wb.Submit(item); err != nil {
// ErrFull (channel at capacity) or ErrClosed (batcher shut down)
}
Close stops the worker and releases resources. In-memory channel items are flushed best-effort; the dque overflow queue is left on disk for recovery on the next process start (Close does not drain dque):
wb.Close()
Flush Triggers ¶
A flush occurs when any of these conditions is met:
- The batch reaches MaxBatchSize items (default 50)
- FlushInterval elapses since the first item entered the current batch (default 200ms)
- The batch's cumulative size (via SizeFunc) reaches MaxBatchBytes (when SizeFunc and MaxBatchBytes > 0)
- Close() is called
Transaction Lifecycle ¶
The batcher calls BeginTx, passes the *sql.Tx to FlushFunc, then calls Commit on success or Rollback on failure. FlushFunc should only execute SQL statements -- it must not call Commit or Rollback itself.
Thread Safety ¶
Submit is safe for concurrent use by multiple goroutines. Close is safe to call multiple times. All other methods are internal to the worker goroutine.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( ErrClosed = errors.New("writebatcher: closed") ErrFull = errors.New("writebatcher: channel full") ErrQuotaExceeded = errors.New("writebatcher: dque disk quota exceeded") )
Sentinel errors returned by Submit.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config[T any] struct { BeginTx func(ctx context.Context) (*sql.Tx, error) // how to start a tx Flush FlushFunc[T] // business logic OnError OnErrorFunc[T] // called on flush failure (nil = log only) OnSuccess OnSuccessFunc[T] // called after successful commit OnAfterCommit OnAfterCommitFunc[T] // called after commit, no tx active MaxBatchSize int // flush at this count (default 50) FlushInterval time.Duration // flush after this duration (default 200ms) MaintenanceInterval time.Duration // run OnAfterCommit periodically (0 = disabled) ChannelSize int // buffered channel capacity (default 1024) SizeFunc func(T) int64 // returns byte cost of an item (nil = size tracking disabled) MaxBatchBytes int64 // flush when cumulative batch bytes >= this (0 = no byte limit) // DQueDirPath specifies a file system path for a durable queue (dque) // used as disk-backed overflow storage for crash recovery. When // non-empty, the batcher creates or opens a dque at this path. DQueDirPath string // DQueItemsPerSegment is the number of items per dque segment file. // When zero or negative, defaults to 250. DQueItemsPerSegment int // MaxDiskBytes sets a maximum disk usage for the dque overflow queue. // When the dque directory exceeds this threshold, Submit returns // ErrQuotaExceeded instead of overflowing to disk. 0 means unlimited. MaxDiskBytes int64 // DeferDQueDrain when true keeps persisted dque items on disk until // StartDQueDrain is called. Channel submits and flushes work normally. DeferDQueDrain bool // contains filtered or unexported fields }
Config holds all parameters for a WriteBatcher. BeginTx and Flush are required; other fields have defaults (MaxBatchSize 50, FlushInterval 200ms, ChannelSize 1024).
type FlushFunc ¶
FlushFunc executes the batch within the provided transaction. The batcher calls BeginTx before Flush and Commit or Rollback after; FlushFunc must only run SQL statements and must not call Commit or Rollback. The batch slice is valid only for the duration of the call; do not retain it.
type OnAfterCommitFunc ¶ added in v0.3.0
type OnAfterCommitFunc[T any] func(ctx context.Context, lastWalCheckpointTime time.Time, lastOptimizeTime time.Time, totalCommitted int64)
OnAfterCommitFunc is called after a successful commit, before the next batch. No transaction is active, making it safe for operations like WAL checkpointing or PRAGMA optimize that require no active transactions. Receives the context, time of last WAL checkpoint, time of last PRAGMA optimize, and total committed count.
type OnErrorFunc ¶
OnErrorFunc is called when a flush fails (after Rollback). The batch is a copy and is safe to retain or use for retry logic. If OnError is nil, the batcher logs the error with slog.Error instead.
type OnSuccessFunc ¶
type OnSuccessFunc[T any] func(batch []T)
OnSuccessFunc is called after a successful flush and commit. The batch is passed as a slice; the caller must not retain it as it may be reused.
type Stats ¶
type Stats struct {
ChannelSize int `json:"channel_size"`
MaxBatchSize int `json:"max_batch_size"`
FlushInterval time.Duration `json:"flush_interval"`
IsClosed bool `json:"is_closed"`
TotalFlushed int64 `json:"total_flushed"`
TotalErrors int64 `json:"total_errors"`
OverflowCount int64 `json:"overflow_count"`
DQueEnabled bool `json:"dque_enabled"`
DQueSize int `json:"dque_size"`
DiskBytes int64 `json:"disk_bytes"`
MaxDiskBytes int64 `json:"max_disk_bytes"`
}
Stats holds statistics about the WriteBatcher.
type WriteBatcher ¶
type WriteBatcher[T any] struct { // contains filtered or unexported fields }
WriteBatcher collects items of type T and flushes them in batched transactions through a single background worker. A WriteBatcher must be created using New and should not be copied after first use. The zero value is not usable.
func New ¶
New creates a WriteBatcher for type T and starts its background worker.
BeginTx must start a new transaction; it is called by the worker for each flush. Flush is called with that transaction and the current batch; it must execute the SQL (e.g. INSERT/UPSERT) and return. OnError is optional; if nil, errors are logged with slog. MaxBatchSize, FlushInterval, and ChannelSize use defaults when zero or negative (50, 200ms, 1024).
SizeFunc and MaxBatchBytes are optional. If both are set (MaxBatchBytes > 0 and SizeFunc non-nil), the batcher tracks cumulative batch size and flushes when the total reaches MaxBatchBytes. If MaxBatchBytes is 0, size tracking runs but never triggers a flush.
The worker runs until the context is cancelled or the input channel is closed. The caller must call Close to shut down the batcher and release resources; closing the context without calling Close leaves the channel open.
New returns an error if BeginTx or Flush is nil.
func (*WriteBatcher[T]) Close ¶
func (wb *WriteBatcher[T]) Close() error
Close signals shutdown: it closes the input channel, waits for the worker to flush in-memory channel items and exit, cancels the context, then closes the dque handle (if configured) without draining persisted overflow items. After Close returns, all subsequent Submit calls return ErrClosed.
Close is safe to call multiple times; after the first call it returns nil immediately without blocking.
func (*WriteBatcher[T]) GetStats ¶
func (wb *WriteBatcher[T]) GetStats() Stats
GetStats returns the current statistics of the WriteBatcher.
func (*WriteBatcher[T]) PendingCount ¶
func (wb *WriteBatcher[T]) PendingCount() int64
PendingCount returns the number of items currently enqueued or in the current batch and not yet flushed. It is intended for completion checks (e.g. consider processing done only when PendingCount is zero in addition to worker in-flight).
func (*WriteBatcher[T]) SetMaxDiskBytes ¶ added in v0.13.0
func (wb *WriteBatcher[T]) SetMaxDiskBytes(n int64)
SetMaxDiskBytes updates the dque disk quota in bytes at runtime (0 = unlimited). The new value takes effect on subsequent Submit calls and is reflected in GetStats. It is safe to call concurrently with Submit.
func (*WriteBatcher[T]) StartDQueDrain ¶ added in v0.10.0
func (wb *WriteBatcher[T]) StartDQueDrain()
StartDQueDrain begins draining persisted dque overflow items into batches. It is a no-op when dque is disabled or draining was not deferred.
func (*WriteBatcher[T]) Submit ¶
func (wb *WriteBatcher[T]) Submit(item T) error
Submit enqueues an item for inclusion in a future flush. It does not block: the item may be flushed later when the batch reaches MaxBatchSize, when FlushInterval elapses, or when Close is called.
Submit returns nil on success. It returns ErrFull if the internal channel is at capacity (caller may retry or drop). It returns ErrClosed if the batcher has been closed or the context passed to New was cancelled.
When a dque is configured and the channel is full, Submit overflows the item to the dque instead of returning ErrFull.
Submit is safe to call concurrently from multiple goroutines.
Source Files