The composition point: the GenServer that turns a Capstan.Connection's frame stream
into durable, fail-closed, at-least-once delivery (lib-owned checkpoint mode — the
checkpoint advances only after the sink's {:ok, _}, so a crash in between re-delivers;
see Capstan and ADR-0004. Effect-once is the deferred sink-owned atomic path).
Connection owns ONLY the socket and forwards {:binlog_event, raw} (the raw 19-byte
header + body + CRC — exactly Capstan.Binlog.Event.parse/1's input) and
{:capstan_halt, reason}. This server owns everything downstream of the wire:
the assembly fold, the sink delivery, the checkpoint, and every fail-closed halt.
It never touches the socket. That split makes the reconnect path auditable — the
recurring OTP-async-lifetime bug family lives exactly at a process owning both.
Flow per binlog event
Event.parse/1 # header + CRC verify; a failure HALTS fail-closed
-> Assembler.step/2 # the pure three-terminator fold (ADR-0003)
{:cont, [output], next} # for each output, the delivery + checkpoint below
{:halt, reason} # XA_PREPARE -> HALT, do not advance
{:error, reason} # any refused/desynced event -> HALT, do not advanceA %Capstan.Transaction{} or %Capstan.SchemaChange{} output runs, in order:
- Dedup — if its GTID is already in the processed set (
Capstan.Gtid.member?/2, never hand-rolled), the output is SKIPPED: the sink is not called and the checkpoint is not touched, emitting:already_processed(effect-once). - Filtered / empty transaction (
changes: [], ADR-0003) — advances the checkpoint with NO sink call. A long run of filtered transactions therefore never stalls the watermark. - Delivered transaction —
Sink.handle_transaction/1FIRST; the checkpoint is written ONLY after it returns{:ok, _}. A{:error, _}HALTS fail-closed WITHOUT advancing, so the failed transaction is re-delivered on restart (at-least-once). This is the checkpoint-after-{:ok, _}ordering. - Schema change (self-committing DDL, Q13) —
Sink.handle_schema_change/2, then the advanced watermark is checkpointed.{:error, _}HALTS without advancing.
Every halt — a {:capstan_halt, _} from the Connection, an assembler {:halt, _} /
{:error, _}, an event-parse failure, a sink error, or an exhausted checkpoint-write
budget — stops the pipeline with {:shutdown, {:halt, reason}} and never checkpoints
past the failure. Fail closed is the whole point: silently continuing past any of these
is the data-loss class this server exists to prevent.
Checkpoint mode
C1 implements lib-owned checkpoint mode: a Capstan.CheckpointStore persists the
processed gtid_set. The server resumes from read_position/2 at startup (a read fault
fails closed rather than resume from a guessed position; nil is a fresh empty start)
and advances via write_position/3. A write fault is retried per the shared budget
(retry_decision/2) and then halts fail-closed. Sink-owned checkpoint mode (the sink
persisting the position atomically with its own write and returning it via
Capstan.Sink.checkpoint/0) is not wired in this server.
Why a separate processed set
The pure Capstan.Assembler unions each transaction's GTID into its running fold and
returns the transaction's position INCLUDING it, so the assembler's own set can never
answer "was this already processed?". This server keeps the durable processed watermark
separately: it is the resumed checkpoint, advanced only when a delivery is durably
checkpointed, and it is what dedup consults BEFORE delivering.
Summary
Functions
Returns a specification to start this module under a supervisor.
Starts the assembler server.
Types
@type checkpoint_store() :: {module(), Capstan.CheckpointStore.store()}
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec start_link(keyword()) :: GenServer.on_start()
Starts the assembler server.
Options:
:sink(required) — theCapstan.Sinkcallback module.:checkpoint_store(required) —{impl_module, store_handle}for lib-owned mode.:tables— the table filter passed toCapstan.Assembler(:all, the default, or an allowlist of{schema, table}pairs).:max_retries— the checkpoint-write retry budget (defaultCapstan.CheckpointStore.default_max_retries/0).:name— an optional registered name.