# `Replicant.Connection`
[🔗](https://github.com/baselabs/replicant/blob/v0.3.1/lib/replicant/connection.ex#L1)

The `Postgrex.ReplicationConnection` that owns the replication slot and closes
the exactly-once seam (spec §2/§4/§8). It:

  * replies to a reply-requested keepalive with the **last durably-checkpointed LSN**
    while any published transaction is in flight — never advancing the slot past
    un-persisted publication data (walex's fire-and-forget `wal_end+1` is the
    at-most-once bug this fixes). When IDLE (no open transaction and the checkpoint
    caught up), it advances the slot to the keepalive `wal_end` so a quiet-but-filtered
    publication stops pinning WAL (spec A1) — the intervening WAL carries nothing for
    the publication;
  * decodes each XLogData payload behind Plan 1's value-free boundary and forwards
    the decoded message to `Replicant.AssemblerServer` — it never applies the sink,
    so it is always free to answer keepalives;
  * tracks a **bounded in-flight window** (spec §4): the high-water received LSN
    (`received_lsn`, the latest XLogData `wal_end`) minus the confirmed-durable
    floor is the in-flight WAL lag — un-drained WAL, a proxy for the transaction
    BACKLOG accumulating ahead of the sink. A single non-blocking integer
    comparison in `handle_data/2` compares it to `max_inflight_lag` (a
    backlog-sized ceiling, default 64 MiB);
  * when the in-flight lag exceeds the bound the sink is genuinely lagging: it
    **halts fail-closed** with `{:sink_too_slow, lag}` (surfaced telemetry +
    `Replicant.Supervisor.halt/2` + `{:disconnect, :sink_too_slow}`) — never a
    silent reconnect livelock or an unbounded-mailbox OOM. On restart it resumes
    from the checkpoint (loss=0 by the §6 idempotent dedup);
  * advances the ack asynchronously on `{:sink_committed, L}` from the AssemblerServer;
  * on connect/reconnect detects an invalidated slot (`wal_status = 'lost'` or
    `conflicting` on PG16) and halts the pipeline fail-closed — never silently
    dropping and recreating the slot (spec §8 R-ISO).

The connect chain is a `handle_result/2` state machine:
`:recovery_check` → `:invalidation_check` → (`:create_slot` if absent) → `:streaming`.

## Why a fail-closed halt and not soft pacing (Postgrex flow-control)

`Postgrex.ReplicationConnection` re-arms the replication socket (`active: :once`)
**automatically** after each `handle_data/2` batch — the handler's return value has
no way to defer socket re-activation, so there is no lever to pause TCP reading
from the handler. `:max_messages` bounds only Postgrex's per-batch socket buffer,
not the downstream `AssemblerServer` mailbox. So true socket-level pacing is not
available; the bounded in-flight window is enforced by the fail-closed halt at the
ceiling — which bounds memory (the pipeline tears down before the mailbox grows
unbounded) and surfaces the overload rather than silently livelocking.

# `step`

```elixir
@type step() ::
  :disconnected
  | :recovery_check
  | :publication_check
  | :invalidation_check
  | :create_slot
  | :create_export_slot
  | :create_incremental_slot
  | :snapshotting
  | :streaming
```

# `t`

```elixir
@type t() :: %Replicant.Connection{
  backfill_floor: Replicant.lsn() | nil,
  batch_delivery: keyword() | nil,
  checkpoint_lsn: Replicant.lsn(),
  checkpoint_state: :present | :empty | :fault | :fault_permanent,
  checkpoint_store: keyword() | nil,
  command_error: %{
    count: non_neg_integer(),
    max_retries: non_neg_integer(),
    store_paced: boolean()
  },
  connection: keyword(),
  failover: boolean(),
  frontier_epoch: non_neg_integer(),
  go_forward_only: boolean(),
  in_recovery: boolean(),
  in_stream: boolean(),
  in_txn: boolean(),
  last_commit_lsn: Replicant.lsn(),
  last_frontier_cast: non_neg_integer(),
  max_inflight_lag: pos_integer(),
  max_spill_bytes: non_neg_integer() | nil,
  messages: boolean(),
  open_streams: MapSet.t(),
  publication: [String.t()],
  reader_pid: pid() | nil,
  received_lsn: Replicant.lsn(),
  server_version_num: non_neg_integer(),
  sink: module(),
  slot_name: String.t(),
  snapshot: boolean() | keyword(),
  spilled_bytes: non_neg_integer(),
  step: step(),
  store_retry_count: non_neg_integer(),
  stream_floor_lsn: Replicant.lsn() | nil,
  streaming: keyword() | nil
}
```

# `child_spec`

Returns a specification to start this module under a supervisor.

See `Supervisor`.

# `classify_slot_status`

```elixir
@spec classify_slot_status([[term()]]) ::
  :absent
  | :ok
  | {:invalidated,
     :wal_lost
     | :conflict
     | :rows_removed
     | :wal_level_insufficient
     | :invalidated}
```

Classify a `pg_replication_slots` invalidation-status result (spec §5/§8). `[]` →
`:absent`. On the **PG16 2-col** row `[wal_status, conflicting]`: `wal_status = "lost"` →
`{:invalidated, :wal_lost}`; `conflicting = true` → `{:invalidated, :conflict}`; otherwise
`:ok`. On the **PG17 4-col** row `[wal_status, conflicting, invalidation_reason, synced]`:
the legacy signals classify first (same as above), then any non-empty `invalidation_reason`
→ `{:invalidated, <fixed atom>}` via `invalidation_reason_atom/1` (never `String.to_atom`);
a `nil`/`""` reason is `:ok`.

# `default_max_inflight_lag`

```elixir
@spec default_max_inflight_lag() :: pos_integer()
```

The default in-flight-lag ceiling (WAL bytes) when the config omits it.

# `encode_status_update`

```elixir
@spec encode_status_update(Replicant.lsn()) :: binary()
```

Encode a Standby Status Update reporting `lsn` as the write/flush/apply position
(spec §2 — the slot never advances past the durable checkpoint). `reply_requested`
is 0 (we volunteer status). `clock` is microseconds since the PG 2000 epoch.

# `start_link`

```elixir
@spec start_link(Replicant.Config.t()) :: {:ok, pid()} | {:error, term()}
```

Start the replication connection for a pipeline (called by `Replicant.Pipeline`).

# `via`

```elixir
@spec via(String.t()) :: {:via, module(), term()}
```

The Registry via-name a pipeline's Connection registers under.

---

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