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.