The three-terminator fold — a pure assembler over a %Capstan.Binlog.Event{}
stream (ADR-0003; the owning GenServer is Capstan.AssemblerServer).
The fold turns raw binlog events into the committed units a Sink consumes:
%Capstan.Transaction{} for DML, %Capstan.SchemaChange{} for self-committing
DDL. It threads a running Capstan.Position (the processed GTID set) from a
caller-supplied start, and a Capstan.Binlog.TableRegistry built as TABLE_MAP
events arrive.
Why it folds %Event{}, not decoded terms
Capstan.Binlog.Decoder.decode/1 returns only the per-type body term and drops
the event header. But a Transaction's commit_ts is the terminator event's
header timestamp and its diagnostic pos is the header log_pos. So the fold
decodes each %Event{} via Decoder.decode/1 for the body term and reads the
header off the same struct for timestamp/log_pos.
The single biggest failure mode this fold is built AROUND (risk-first)
A phantom / rolled-back transaction delivered effect-once — the silent one. Two shapes of it, both defended structurally rather than only by convention:
- An
XA_PREPARE(Decoder{:halt, :unsupported_transaction_shape}): its row images may later beXA ROLLBACK-ed, so the rows accumulated so far in that transaction must be discarded, not delivered (ADR-0003). The halt is keyed on theDecoder's type-byte signal, never re-derived here. - Any mid-transaction failure — an unmapped
table_id, aRows.decode/2error, or aDecoder{:error, _}— aborts fail-closed. A partial transaction carrying only the rows decoded before the failure is never emitted.
Both are guaranteed by the fold's shape: rows accumulate in the in-flight
transaction buffer and are released only when a terminator is reached. A
{:halt, _}/{:error, _} short-circuits with that buffer still un-released, so it
is structurally impossible to leak. The happy path (row → %Change{}) is secondary.
Stream desyncs fail closed the same way rather than being silently absorbed: a
GTID arriving while a transaction is still open (:gtid_within_open_transaction),
or a QUERY / row / terminator with no open GTID scope (:begin_without_gtid,
:query_without_gtid, :rows_without_transaction, :terminator_without_transaction)
— each aborts rather than dropping or overwriting an open buffer, since a silently
absorbed desync can mask exactly the loss this fold exists to prevent.
The three terminators (ADR-0003) — anchored on the proven fixture_capture
classify/apply_event precedent so a terminator is never misclassified (itself a
silent-loss risk: a COMMIT read as DDL would drop the transaction's rows):
XID— a transactional (InnoDB) commit.QUERY("COMMIT")— a non-transactional (MyISAM) commit; theCOMMITis the terminator, not a change.- a self-committing DDL
QUERY— aQUERYthat is notBEGIN/COMMITand arrives with no openBEGIN. Yields a%SchemaChange{}and advances the position.BEGINopens a DML block;COMMITcloses it.
Fully-filtered transactions keep the watermark moving (ADR-0003)
The checkpoint is a processed-GTID watermark, so a committed transaction whose
changes are all filtered (or which has none) still advances the position and is
delivered as a %Transaction{} with empty changes. Capstan.AssemblerServer
checkpoints it
without calling the sink. A long run of filtered transactions therefore never
stalls the position.
Capstan.Config carries no table filter in C1, so the filter is a parameter:
tables: :all (default — nothing filtered) or an allowlist of {schema, table}
pairs, applied before row decode.
Rule 1
No %Change{} value, %SchemaChange{} DDL text, or error tuple ever carries a row
value or statement literal. DDL is reduced to schema/table/kind here; the raw
SQL is inspected only to classify and is then dropped.
Interface
new/2+step/2+position/1— the incremental primitive the owning GenServer (Capstan.AssemblerServer) drives to checkpoint per transaction.run/3— folds a whole event sequence, returning every output plus the final watermark; the batch entry the tests drive.
Summary
Types
The table filter: everything, or an allowlist of {schema, table} pairs.
An emitted committed unit.
The result of run/3. On a {:halt, _}/{:error, _} short-circuit the
already-COMMITTED outputs (and the watermark up to the failure) are still returned —
a transaction that reached its terminator before the halt is genuinely committed and
must not be lost, only the in-flight one is discarded.
The result of step/2.
Functions
Builds a fresh assembler state resuming from start_position.
The running processed-GTID watermark as a Capstan.Position.
Folds an entire event sequence from start_position.
Folds one %Capstan.Binlog.Event{} into the state.
Types
The table filter: everything, or an allowlist of {schema, table} pairs.
@type output() :: Capstan.Transaction.t() | Capstan.SchemaChange.t()
An emitted committed unit.
@type run_result() :: {:ok, [output()], Capstan.Position.t()} | {:halt, :unsupported_transaction_shape, [output()], Capstan.Position.t()} | {:error, term(), [output()], Capstan.Position.t()}
The result of run/3. On a {:halt, _}/{:error, _} short-circuit the
already-COMMITTED outputs (and the watermark up to the failure) are still returned —
a transaction that reached its terminator before the halt is genuinely committed and
must not be lost, only the in-flight one is discarded.
@type step_result() :: {:cont, [output()], t()} | {:halt, :unsupported_transaction_shape} | {:error, term()}
The result of step/2.
@type t() :: %Capstan.Assembler{ file: String.t() | nil, filter: filter(), gtid_set: Capstan.Gtid.t(), pos: non_neg_integer() | nil, registry: Capstan.Binlog.TableRegistry.t(), txn: inflight() | nil }
Functions
@spec new( Capstan.Position.t(), keyword() ) :: t()
Builds a fresh assembler state resuming from start_position.
opts accepts tables: :all (default) or tables: [{schema, table}, ...] — the
allowlist of tables whose rows are delivered; everything else is filtered before
row decode.
@spec position(t()) :: Capstan.Position.t()
The running processed-GTID watermark as a Capstan.Position.
Capstan.AssemblerServer reads this after a step/2 to checkpoint a transaction — including a
self-committing DDL or a fully-filtered transaction, whose position advances even
though the emitted output carries none (or an empty changes).
@spec run([Capstan.Binlog.Event.t()], Capstan.Position.t(), keyword()) :: run_result()
Folds an entire event sequence from start_position.
Returns {:ok, outputs, final_position} when every event is consumed. On the first
{:halt, _}/{:error, _} a step/2 produces, returns {:halt, reason, committed, watermark} / {:error, reason, committed, watermark} — the transactions that
already terminated before the failure are genuinely committed and returned; only the
in-flight (e.g. XA-prepared) transaction is discarded. opts is as for new/2.
@spec step(t(), Capstan.Binlog.Event.t()) :: step_result()
Folds one %Capstan.Binlog.Event{} into the state.
Returns {:cont, outputs, state} (outputs is [] for every event except a
terminator), or short-circuits fail-closed with {:halt, :unsupported_transaction_shape} or {:error, reason} — in either case the
in-flight transaction is discarded and nothing is emitted (see the module doc).