The socket-owning GenServer: handshake, auth, the fail-closed gap gate, the dump, and frame forwarding.

Connection owns ONLY the TCP/TLS socket. It never delivers to a sink and never writes the checkpoint — Capstan.AssemblerServer is the sole checkpoint writer, so there is no split-brain over the durable position. It does read the durable resume position at each (re)establish (see "Resume position" below); reading is not writing, and it is what makes a reconnect resume from the current watermark. That split keeps the reconnect seam auditable: every socket-lifetime primitive (the reconnect timer, the streaming reader) lives in this one module and is torn down with the connection it serves.

Resume position

The dump resumes from the durable checkpoint, RE-READ from the checkpoint store on every establish — the initial connect AND every reconnect. Freezing the resume position at start-up would be a correctness bug: after a transient drop the AssemblerServer has advanced the durable watermark, so a reconnect that replayed from the frozen start-up position would (a) needlessly re-stream everything since start-up (the AssemblerServer dedups it, but at O(run-length) cost) and, worse, (b) once source retention purged past that stale position, feed gap_check/3 a checkpoint whose unapplied remainder intersects gtid_purged — a false :data_gap halt on a perfectly healthy pipeline, re-opening the very silent-loss vector the gate exists to close. Re-reading makes the resume correct by construction. The Connection is given the store as a {impl, handle} and calls Capstan.CheckpointStore.read_position/2 (read-only — the single-writer invariant holds). A store read fault on refresh keeps the last-known position and proceeds (never worse than the frozen behaviour; the next reconnect refreshes); a store is optional (its absence — in a unit test wired only with :start_position — keeps the injected position).

Lifecycle

connect + auth  (via the injected connect_fun; default: gen_tcp + Handshake)
  -> Config.check_preconditions/1        (fail-closed server gate, ADR-0002)
  -> read @@gtid_executed / @@gtid_purged
  -> gap_check/3                          (proactive retention gap)
  -> SET @master_binlog_checksum          (required BEFORE the dump)
  -> SET @master_heartbeat_period         (liveness — required BEFORE the dump)
  -> COM_BINLOG_DUMP_GTID                 (resume from the start position)
  -> stream frames -> receiver

Streaming liveness (half-open partition detection)

A silent half-open partition mid-stream would otherwise hang the pipeline forever — the reader blocks on recv and MySQL only heartbeats an IDLE stream. Two mechanisms bound the detection window: (1) SET @master_heartbeat_period asks the master to emit a HEARTBEAT_LOG_EVENT after :heartbeat_period_ms of idleness (self-contained — not dependent on the server's slave_net_timeout), so even a quiet-but-healthy stream keeps delivering frames; (2) a parent-side liveness timer fires if NO frame (event OR heartbeat) arrives within :stream_timeout_ms (default 4× the heartbeat period, tolerating 3 missed heartbeats) — on fire it emits [:capstan, :connection, :stream_timeout] (the stall is no longer silent), kills the recv-blocked reader, and reconnects; a persistent partition halts :stream_stalled via the established-then-dropped budget. keepalive: true on the socket is the OS backstop for a fully-dead peer (its ~2h default idle is too slow to be primary). The reader itself keeps reading with no recv timeout; liveness is the parent's job.

Receiver contract

Each streamed binlog event is forwarded as {:binlog_event, raw_event_bytes} where raw_event_bytes is the 19-byte header + body + CRC — exactly what Capstan.Binlog.Event.parse/1 consumes (this module never decodes; that is the assembler's job). A fail-closed halt is delivered as {:capstan_halt, reason} and then the process stops with {:shutdown, {:halt, reason}}.

The forwarding is a bare send/2, so it never raises when the receiver is dead — but a send into a dead receiver is a silent no-op, and the receiver's OWN fail-closed halt does not message back here. So this module monitors the receiver: on its :DOWN the Connection stops fail-closed (:receiver_down) rather than keep streaming events into a dead pid. Monitoring is not linking — a :shutdown stop of a :temporary child never restarts, so this closes the silent stall without the restart livelock the plain-send wiring exists to avoid.

Two independent counters

• Command budget (max_command_retries, default 5): counts failures that occur before the connection establishes (connect/auth/query/dump-send errors). It is RESET when a frame arrives — replicant's reset-on-frame policy, correct for transient pre-establish faults. Halts :command_retries_exhausted on the max + 1-th failure.
• Cycle counter: counts established-then-dropped cycles. It is NOT reset by frame arrival, and is reset only by clean shutdown. A duplicate server_id makes MySQL evict this replica after each establish — authenticating and receiving frames before dying — so a frame-reset counter would livelock forever while emitting healthy :established telemetry. Halts :server_id_conflict on the max + 1-th cycle.

Error 1236 is OVERLOADED (ADR-0003)

A dump refusal (error 1236) is discriminated on its message text: checksum-negotiation text -> :checksum_negotiation_failed; purged-logs text (with OR without a named range) -> :data_gap; a duplicate-replica message (server_uuid/server_id) -> :server_id_conflict (MySQL 8.0.x evicts a same-server_id replica via 1236, not a socket drop); anything else -> :unrecognized_dump_error, never :data_gap — mapping 1236 unconditionally to a gap would tell an operator to reprovision past what is actually a config bug, manufacturing the silent loss the gate exists to prevent.