Group-commit coordinator for step outcomes.
Instead of every worker Task committing its own outcome (one complete_*
round-trip per instance per step), a Task hands its outcome here and blocks
until it is durably written — commit_before_proceed is preserved; the Task
sits and waits for the write, it does not run ahead of it. The flusher
coalesces the outcomes of all the Tasks waiting on it into one transaction
(GenDurable.Queries.flush/2: a single guarded UPDATE … FROM unnest(...)
plus the consume/notify riders) and releases every waiter when it commits.
Two triggers, whichever fires first:
max_batch(default 100) — flush as soon as this many outcomes are buffered;max_delay_ms(default 100) — flush this long after the first buffered outcome, bounding the durability/latency window under light load.
Under load the batch auto-grows: while a flush transaction runs, the flusher's mailbox fills, so the next flush is bigger — the single serialization point does not become a linear bottleneck. Step execution stays fully parallel across Tasks; only the commit is coalesced.
A running engine wires one flusher per flushers: spec (see
GenDurable.Supervisor), each owning a disjoint set of queues; the
Testing.drain/1 path flushes synchronously without one.
Summary
Types
One buffered outcome. The set_* flags + precomputed values drive the batched
UPDATE (Queries.flush/2); slot/notify drive the post-flush side effects.
Functions
Returns a specification to start this module under a supervisor.
Submit entry and block until the batch containing it commits. Returns
:committed (its state landed), :stale (the ownership guard failed — the row
was reclaimed, a new claimant redoes the step), or {:error, reason} (the flush
transaction failed; the row stays executing for the reaper).
Flush entries synchronously in the calling process — no coordinator, no
batching — and run the side effects. The Testing.drain/1 path uses this
(batch of one per step); returns the MapSet of committed ids.
Types
@type entry() :: %{ id: integer(), worker: String.t(), slot: term() | nil, notify: boolean(), status: String.t(), attempt: integer(), delay_ms: integer(), set_step: boolean(), step: String.t() | nil, set_state: boolean(), state: String.t() | nil, set_result: boolean(), result: String.t() | nil, set_error: boolean(), error: String.t() | nil, clear_awaits: boolean(), set_rate: boolean(), rate_limit: String.t() | nil, weight: float(), set_ck: boolean(), ck_value: String.t() | nil, consumed_ids: [integer()] }
One buffered outcome. The set_* flags + precomputed values drive the batched
UPDATE (Queries.flush/2); slot/notify drive the post-flush side effects.
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec commit(GenServer.server(), entry()) :: :committed | :stale | {:error, term()}
Submit entry and block until the batch containing it commits. Returns
:committed (its state landed), :stale (the ownership guard failed — the row
was reclaimed, a new claimant redoes the step), or {:error, reason} (the flush
transaction failed; the row stays executing for the reaper).
Flush entries synchronously in the calling process — no coordinator, no
batching — and run the side effects. The Testing.drain/1 path uses this
(batch of one per step); returns the MapSet of committed ids.