# `Capstan.Connection`
[🔗](https://github.com/baselabs/capstan/blob/v1.3.1/lib/capstan/connection.ex#L1)

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 + 0013)
      -> read @@gtid_executed / @@gtid_purged
      -> gap_check/3                          (proactive retention gap)
      -> read @@server_uuid                   (source identity — ADR-0013 cycle reset)
      -> 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 — or, since ADR-0013, by a
    **server change**: a reconnect that passes `gap_check/3` while observing a
    `@@server_uuid` DIFFERENT from the previous cycle's resets the counter to 0 (a
    failover promoted a new writer — the drop was the failover's DNS change, not an
    eviction), and the observed UUID becomes the comparison baseline. A cycle against
    the SAME `@@server_uuid` still counts: 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, and a uuid-reset would never fire (the evicting server's
    UUID never changes). 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.

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `classify_dump_error`

```elixir
@spec classify_dump_error(non_neg_integer(), binary()) ::
  :checksum_negotiation_failed
  | :data_gap
  | :server_id_conflict
  | :unrecognized_dump_error
  | {:dump_failed, non_neg_integer()}
```

Discriminates a dump-refusal error into a halt reason (ADR-0003).

Error 1236 is overloaded: a checksum-negotiation message -> `:checksum_negotiation_failed`;
a purged-logs message (with or without a named range) -> `:data_gap`; a duplicate-replica
message (`server_uuid`/`server_id`) -> `:server_id_conflict`; any other 1236 ->
`:unrecognized_dump_error` (never `:data_gap`). A non-1236 code is surfaced as
`{:dump_failed, code}`.

# `gap_check`

```elixir
@spec gap_check(String.t(), String.t(), String.t()) ::
  :ok | {:halt, :data_gap | :source_identity_mismatch}
```

The proactive retention-gap predicate.

Given the server's `gtid_executed` and `gtid_purged` and the resume `checkpoint`
(all canonical GTID-set strings), returns `:ok` when the pipeline is healthy, or a
fail-closed halt. The predicate is on the **unapplied remainder**, not the checkpoint
itself:

  * a checkpoint carrying GTIDs the server never executed -> `:source_identity_mismatch`
  * `(gtid_executed − checkpoint) ∩ gtid_purged ≠ ∅` -> `:data_gap`

An **empty** checkpoint is a fresh start with no durable position to lose, so it never
halts — otherwise every fresh start against a server that has ever purged (i.e. every
real server) would falsely halt `:data_gap`, an over-rejection this gate must not commit.

# `parse_recover_rows`

```elixir
@spec parse_recover_rows([[binary()]]) :: {:ok, [binary()]} | :error
```

Parses `XA RECOVER` resultset rows into XID digests — `{:ok, [digest]}` or `:error`.

The pure, RED-capable seam `maybe_xa_recover/1` refuses on: a row that is not the
documented 4-column binary shape, a NON-INTEGER-NUMERIC cell (span finding O6:
`Integer.parse/1` accepts a PARTIAL match like `"12abc"`, which `String.to_integer/1`
then crashes on — a malformed row must be the named `:xa_recover_unexpected_shape`
establish fault, never an ArgumentError crash), or a `gtrid_length` beyond the data
column (`binary_part/3` would raise). Simple-query results are ALL text
(Critical Rule 2): every numeric cell is coerced as a FULL integer string.

# `start_link`

```elixir
@spec start_link(keyword()) :: GenServer.on_start()
```

Starts the connection.

Options: `:server_id` (required), `:connection` (keyword passed to the connect
function), `:receiver` (pid/name that frames and halts are sent to), `:start_position`
(`%Capstan.Position{}` or `nil` for a fresh start — the resume position used until the
first store refresh, and the only source when no store is given), `:checkpoint_store`
(an optional `{impl_module, handle}` re-read for the current resume position on every
establish; see "Resume position"), `:max_command_retries` (default 5), `:connect_fun` (a
1-arg override for the connect+auth step, default the real `gen_tcp` + `Handshake` path),
`:reconnect_backoff` ms, `:heartbeat_period_ms` (default 15_000 — the master heartbeat
interval; see "Streaming liveness"), `:stream_timeout_ms` (default 60_000 — the liveness
window; MUST be `> :heartbeat_period_ms` or start-up fails closed), and `:name`. Every
liveness/backoff value must be a positive integer within the schedulable
`Process.send_after` ceiling (2^32-1 ms), mirroring `Capstan.Config.validate/1`.

---

*Consult [api-reference.md](api-reference.md) for complete listing*