processor

func FlushBatch ¶

func FlushBatch[V any](
	ctx context.Context,
	cfg *Config,
	pipelineName string,
	items []FlushItem[V],
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
	concurrency int,
) error

FlushBatch applies every FlushItem in `items` to (store, cache) using the processor's retry/backoff/metrics surface, spreading the work across at most `concurrency` goroutines.

Semantics:

• concurrency <= 1: items are applied sequentially in the order given, same as a plain `for _, it := range items { mergeKeyWithRetry(...) }`. This is the default path and preserves the pre-concurrency behavior of every caller.

• concurrency > 1: a worker pool of size `min(concurrency, len(items))` drains a channel of FlushItems and runs MergeUpdate in parallel. **Each FlushItem must address a distinct (entity, bucket) Key** — callers are expected to coalesce deltas per-key BEFORE calling FlushBatch (additive monoids have already-collapsed all contributions to one delta per key, so parallelizing the per-key flushes is safe regardless of arrival order). FlushBatch does not enforce distinctness — passing duplicate keys with concurrency>1 races the store's MergeUpdate semantics and is undefined.

• At-least-once preservation: if any worker returns a non-nil error from mergeKeyWithRetry, FlushBatch records the first such error and returns it after all in-flight items finish. Subsequent items that haven't been dispatched yet are skipped (cancellation propagates through the worker context). The caller must treat a non-nil return as "this batch failed" — earlier-completed items have already written to the store, but the batch as a whole has not been Ack'd so the source will redeliver everything. Idempotent-merge + dedup keeps the redelivered batch correct.

• Ordering: because the worker pool is unordered, callers must NOT rely on items being applied in slice order. This is the explicit tradeoff that unlocks parallelism for additive monoids; for non- additive monoids (First/Last/Min/Max), per-key ordering still matters at the routing layer (see streaming.WithConcurrency), but within a single coalesced flush the deltas are already collapsed so there is at most one delta per key in `items`.

FlushBatch always invokes mergeKeyWithRetry for each item — the same retry budget, backoff schedule, dedup-skip handling, and metrics that MergeMany uses. Pass the same *Config you'd pass to MergeMany.

func MergeBatch ¶

func MergeBatch[T any, V any](
	ctx context.Context,
	cfg *Config,
	coalesceCfg CoalesceConfig,
	pipelineName string,
	mon monoid.Monoid[V],
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
	events []T,
	eventIDFn func(T) string,
	eventTimeFn func(T) time.Time,
	keysFn func(T) []string,
	valueFn func(T) V,
) error

MergeBatch is a convenience wrapper around NewCoalescer + a loop over events + Flush. It is the single-line entrypoint drivers can call when they hold an entire source batch in memory (Kinesis Lambda Records, bootstrap snapshot chunk, etc).

The events slice is iterated in order; each event's keys are derived via keysFn, each delta via valueFn. After the loop, MergeBatch calls Flush. Returns nil on full success, or a wrapped *FlushError listing failed keys.

MergeBatch is a zero-allocation passthrough for non-additive monoids (it just loops over MergeMany), so drivers can call it unconditionally regardless of the pipeline's monoid choice.

func MergeMany ¶

func MergeMany[V any](
	ctx context.Context,
	cfg *Config,
	pipelineName string,
	eventID string,
	eventTime time.Time,
	keys []string,
	delta V,
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
) error

MergeMany is the multi-key entry point. It claims the EventID via Dedup (once, regardless of how many keys), then for each key runs the store and cache MergeUpdate with the supplied delta. Used by hierarchical- rollup pipelines where one event contributes to many aggregation keys (e.g. "likes for this post", "likes for this post per country", "global likes" — three keys, one delta).

Failure semantics:

• All-or-nothing per record (with respect to dead-lettering): if ANY key's merge fails after retries, MergeMany returns an error. Earlier keys that succeeded keep their writes — there is no rollback. The idempotent-merge contract (with Dedup) ensures a redelivery folds correctly: the dedup row prevents the keys-that-succeeded from double-counting on retry, while the keys-that-failed get another attempt.

Pass a one-element keys slice for the single-key case; that's exactly what MergeOne does internally. The retry/backoff loop is per-key — each key gets its own MaxAttempts budget.

func MergeOne ¶

func MergeOne[T any, V any](
	ctx context.Context,
	cfg *Config,
	pipelineName string,
	eventID string,
	eventTime time.Time,
	value T,
	keyFn func(T) string,
	valueFn func(T) V,
	store state.Store[V],
	cache state.Cache[V],
	window *windowed.Config,
) error

MergeOne is the single-key per-record processing entry point. It applies the keyFn / valueFn extractors, claims the EventID via Dedup (if configured), runs the store and cache MergeUpdates, and retries on transient failure with exponential backoff.

Return semantics:

• nil: the record was processed successfully OR was a duplicate skip. The caller should Ack the record (for sources that have an Ack) or do nothing (for Lambda handlers that report only failures).
• non-nil: every retry was exhausted. The error wraps the last underlying failure. The caller decides what to do — typically: add to BatchItemFailures (Lambda) or record + Ack-and-skip (streaming).

MergeOne also short-circuits to a non-nil error on context cancellation during a retry backoff so the caller can return promptly rather than burning the remaining budget.

MergeOne does NOT call Ack. Sources with an Ack callback should invoke it themselves on a nil return.

Implementation note: MergeOne is now a thin wrapper around MergeMany — it extracts the single key + value and delegates. Use MergeMany directly for hierarchical-rollup pipelines where one record contributes to many keys.