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 advance

A %Capstan.Transaction{} or %Capstan.SchemaChange{} output runs, in order:

1. 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).
2. Filtered / empty transaction (changes: [], ADR-0003) — advances the checkpoint with NO sink call. A long run of filtered transactions therefore never stalls the watermark.
3. Delivered transaction — Sink.handle_transaction/1 FIRST; 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.
4. 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.