Capstan.Assembler (Capstan v0.1.0)

Copy Markdown View Source

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.

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.

t()

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

filter()

@type filter() :: :all | MapSet.t({String.t(), String.t()})

The table filter: everything, or an allowlist of {schema, table} pairs.

output()

An emitted committed unit.

run_result()

@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.

step_result()

@type step_result() ::
  {:cont, [output()], t()}
  | {:halt, :unsupported_transaction_shape}
  | {:error, term()}

The result of step/2.

t()

@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

new(start_position, opts \\ [])

@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.

position(assembler)

@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).

run(events, start_position, opts \\ [])

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.

step(state, event)

@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).