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 be XA ROLLBACK-ed, so the rows accumulated so far in that transaction must be discarded, not delivered (ADR-0003). The halt is keyed on the Decoder's type-byte signal, never re-derived here.
• Any mid-transaction failure — an unmapped table_id, a Rows.decode/2 error, or a Decoder {: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; the COMMIT is the terminator, not a change.
• a self-committing DDL QUERY — a QUERY that is not BEGIN/COMMIT and arrives with no open BEGIN. Yields a %SchemaChange{} and advances the position. BEGIN opens a DML block; COMMIT closes 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.