# `Replicant.Connection`
[🔗](https://github.com/baselabs/replicant/blob/v0.1.0/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 every reply-requested keepalive with the **last durably-checkpointed
    LSN** as the flush position — never the received `wal_end` (walex's
    fire-and-forget `wal_end+1` is the at-most-once bug this fixes), so the slot
    never advances past un-persisted data;
  * 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
  | :invalidation_check
  | :create_slot
  | :create_export_slot
  | :snapshotting
  | :streaming
```

# `t`

```elixir
@type t() :: %Replicant.Connection{
  batch_delivery: keyword() | nil,
  checkpoint_lsn: Replicant.lsn(),
  checkpoint_state: :present | :empty | :fault | :fault_permanent,
  checkpoint_store: keyword() | nil,
  connection: keyword(),
  go_forward_only: boolean(),
  in_stream: boolean(),
  max_inflight_lag: pos_integer(),
  max_spill_bytes: non_neg_integer() | nil,
  publication: String.t(),
  received_lsn: Replicant.lsn(),
  sink: module(),
  slot_name: String.t(),
  snapshot: boolean(),
  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}
```

Classify a `pg_replication_slots` invalidation-status result (spec §8, PG16
columns): `[]` → `:absent`; `wal_status = "lost"` → `{:invalidated, :wal_lost}`;
`conflicting = true` → `{:invalidated, :conflict}`; otherwise `: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*