Replicant.Connection (Replicant v0.3.1)

Copy Markdown View Source

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.

Summary

Types

step()
t()

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

classify_slot_status(list)

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()

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

encode_status_update(lsn)

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(config)

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

via(slot_name)

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

Types

step()

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

t()

@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
}

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

classify_slot_status(list)

@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()

@spec default_max_inflight_lag() :: pos_integer()

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

encode_status_update(lsn)

@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(config)

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

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

via(slot_name)

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

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