Documentation
¶
Overview ¶
Package batching provides a write-coalescing decorator for the EntityStateRepository port. It collects Ingest calls into an in-memory buffer and flushes them to the underlying store as a single Save transaction whenever the buffer fills up, the deadline fires, or Close is invoked.
Use case: streaming adapters that submit one observation per gRPC/HTTP call against a SQLite backend. Per-call commits cost one fsync each; batching coalesces N writes into one fsync, trading some per-write latency for substantially higher write throughput.
The decorator is opt-in via the operator config; default is "no batching", preserving the original per-call commit semantics for deployments that prefer low latency over throughput.
Index ¶
- type Config
- type Repo
- func (r *Repo) Close(ctx context.Context) error
- func (r *Repo) Count(ctx context.Context, adapterName string) (int64, error)
- func (r *Repo) DeleteOlderThan(ctx context.Context, cutoff time.Time, adapterName string) error
- func (r *Repo) Flush(ctx context.Context) error
- func (r *Repo) Ingest(ctx context.Context, adapterName string, state chronos.EntityState) error
- func (r *Repo) ListByEntity(ctx context.Context, entityID uuid.UUID) ([]chronos.EntityState, error)
- func (r *Repo) ListByScope(ctx context.Context, scopeID uuid.UUID) ([]chronos.EntityState, error)
- func (r *Repo) ListScopes(ctx context.Context) ([]uuid.UUID, error)
- func (r *Repo) Save(ctx context.Context, adapterName string, states []chronos.EntityState) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
Config tunes the batcher. Both must be positive; otherwise New returns an error rather than silently disabling.
type Repo ¶
type Repo struct {
// contains filtered or unexported fields
}
Repo wraps an upstream EntityStateRepository with a buffered Ingest path. Reads and the explicit batch Save bypass the buffer and go straight to the upstream — this matches Chronos's design that reads should always see a consistent snapshot, while writes can tolerate small batching latency.
func New ¶
func New(upstream ports.EntityStateRepository, cfg Config) (*Repo, error)
New wraps upstream with batched-Ingest semantics. Callers must call Close when shutting down to drain the buffer; otherwise the most recent buffered writes are lost.
func (*Repo) Close ¶
Close drains the buffer and prevents further Ingest calls. Reads remain available because they go straight to the upstream.
func (*Repo) DeleteOlderThan ¶
DeleteOlderThan passes through to upstream.
func (*Repo) Flush ¶
Flush drains the buffer immediately. Safe to call concurrently with Ingest; the timer is also driven through this path.
func (*Repo) Ingest ¶
Ingest buffers the state. The flush is triggered when the buffer hits maxBatch; otherwise a timer drains it at maxWait. Returns an error only when validation fails — successful buffering is reported as nil even though no row has been persisted yet.
func (*Repo) ListByEntity ¶
ListByEntity passes through to upstream.
func (*Repo) ListByScope ¶
ListByScope passes through to upstream.
func (*Repo) ListScopes ¶
ListScopes passes through to upstream.
Source Files