Pace replication intake without releasing slot ownership

CHANGELOG.md

@@ -7,6 +7,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- `max_inflight_lag` now bounds queued payload bytes and separately retained
+  transaction/batch payload, rather than WAL-position distance. The historical
+  option name and 64 MiB default remain. Queue intake also pauses at 512 messages
+  and resumes below half of both thresholds. This is not a process-memory cap.
+- Slow sinks pause instead of halting with `:sink_too_slow`. Migrate alerts to
+  `[:replicant, :connection, :stalled]` plus slot-retention alerts. Queued frames
+  with no completed assembler frame for five seconds emit telemetry and a warning,
+  paused or not, repeating every five seconds without progress. `duration` is time
+  since the last completion (or initial enqueue), in milliseconds. This replaces
+  the draft `:pause_ongoing` event. Slow but progressing sinks still require monitoring.
+  Monitor
+  PostgreSQL slot WAL retention and disk space separately: feedback keeps the
+  slot alive but never acknowledges undelivered data. There is no automatic
+  pause deadline; configure a server-side WAL retention limit for disk protection.
+- Retained payload exhaustion halts with `:buffer_limit_exceeded`. With spill
+  enabled, non-streamed transactions retain the compatibility ceiling of
+  `max_inflight_lag + max_spill_bytes`; they use memory, not spill files. Configure
+  this allowance deliberately, since PostgreSQL can send transactions smaller
+  than `logical_decoding_work_mem` without streaming even with protocol v2 enabled.
+
+### Fixed
+
+- Drain old assembler work before reconnect replay, preventing queue multiplication
+  and spill-file cleanup races when a sink is blocked.
+- Track retained batch bytes incrementally, release aborted subtransaction row-payload
+  accounting, and spill reclaimable row buffers until within budget.
+- Streamed messages carry pgoutput's top-level xid even inside a savepoint; do
+  not claim to discard sub-aborted messages using that field. Their resident bytes
+  remain charged. Savepoint-message delivery semantics require separate work.
+- Do not re-track an already received transaction on a buffer-triggered batch flush.
+- Idle open transactions no longer indefinitely block snapshot windows.
+- Pace replication intake using queued payload bytes and message counts, rather
+  than WAL-position distance. Pause without releasing slot ownership, continue
+  durable-only feedback, and reject stale drain credits after reconnects.
+- Pace incremental snapshot windows using queued stream work. WAL archive
+  segment gaps no longer halt streaming or indefinitely defer snapshot reads.
+- Fail closed on actual buffer/spill exhaustion without confusing a slow sink
+  with a large WAL-position gap. Streamed message bytes remain charged after
+  changes are spilled.
+
 ## [1.5.1] - 2026-10-10
 
 ### Fixed

README.md

@@ -260,6 +260,16 @@ detected at the next change or within `schema_check_interval` (default 30 s).
 
 ## Configuration reference
 
+**Backpressure.** `max_inflight_lag` retains its historical name but now limits
+queued and retained payload bytes rather than WAL-position distance. Replication
+input pauses while the sink drains and resumes without disconnecting. This is a
+payload budget, not an exact process-memory limit. With spill enabled, non-streamed
+transactions retain a memory allowance of `max_inflight_lag + max_spill_bytes`;
+they cannot use the stream spill reader. Tune this alongside PostgreSQL's
+`logical_decoding_work_mem`, since enabling streaming does not force every
+transaction to stream. Incremental snapshots yield to queued WAL, not idle open
+transactions. Reconnects wait for the previous assembler queue before replay.
+
 All options go to `Replicant.start_link/1`; `Replicant.Config` validates them.
 
 | Option | Default | Purpose |
@@ -274,7 +284,7 @@ All options go to `Replicant.start_link/1`; `Replicant.Config` validates them.
 | `checkpoint_store:` | off | `[connection:, table:, progress_table:, max_retries: 5, retry_backoff_ms: 1000, batch: [...]]`: library-owned checkpoint for non-transactional sinks |
 | `batch_delivery:` | off | `[max_transactions: 100, max_delay_ms: 1000]`: atomic batches through `handle_batch/1` |
 | `streaming:` | off | `[max_concurrent_txns: 64, spill: [dir:, max_spill_bytes:]]`: in-progress streaming and disk spill |
-| `max_inflight_lag:` | 64 MiB | WAL the connection may run ahead of the durable checkpoint before halting `:sink_too_slow` |
+| `max_inflight_lag:` | 64 MiB | Queued payload budget for pause/resume and a separate retained transaction/batch budget; oversized unspillable transactions halt `:buffer_limit_exceeded` |
 | `max_command_retries:` | `5` | connect cycles a pre-stream replication command may fail before the pipeline halts instead of reconnecting forever |
 | `messages:` | `false` | deliver `pg_logical_emit_message` payloads; the sink must implement `handle_message/2` |
 | `failover:` | `false` | create a `FAILOVER` slot on PostgreSQL 17+ |
@@ -296,8 +306,26 @@ rejections are in [`usage-rules.md`](usage-rules.md). Two to alert on:
 
 - `[:replicant, :pipeline, :halted]` with the structural `reason` atom, on every
   fail-closed halt;
-- `[:replicant, :connection, :disconnected]` with `reason: :sink_too_slow` and a signed
-  `lag` measurement, when the sink falls behind the in-flight window.
+- `[:replicant, :buffer, :exhausted]` with `reason: :buffer_limit_exceeded`, when
+  retained transaction data cannot fit within the configured budget.
+
+`[:replicant, :connection, :paused]` and `:resumed` report normal backpressure, with
+queued payload bytes and message counts. Paused connections send durable feedback
+every five seconds; configure `wal_sender_timeout` comfortably above that interval.
+`[:replicant, :connection, :stalled]` and a warning log fire after five seconds
+with queued frames but no completed assembler frame, then repeat every five seconds
+without progress. This includes quiet publications below the pause thresholds and
+queues waiting across reconnects. Measurements are queued `byte_size`, `change_count`
+(frames), and `duration` (milliseconds since the last completion, or initial enqueue).
+The log remains visible without a telemetry handler; neither signal contains row or
+message payloads. Completion resets the stall clock; draining cancels the timer.
+
+Replace `:sink_too_slow` alerts with stall alerts **and** PostgreSQL slot-retention /
+primary-disk monitoring. A slow sink that keeps completing frames, including short
+pause/resume cycles, does not trigger a stall alert but can still accumulate WAL.
+Configure `max_slot_wal_keep_size` for a server-side retention bound. A
+retention-limit invalidation requires rebuilding the consumer; disconnecting alone
+does not release retained WAL.
 
 ## Documentation
 

docs/adr/0005-spill-is-ephemeral-scratch.md

@@ -23,11 +23,16 @@ of the transaction. Two design questions follow:
   reconnect. There is **no fsync** — spill is not a WAL; a crash re-streams the transaction from
   the durable `confirmed_flush` LSN (the spill is discarded, never replayed).
   (`lib/replicant/spill.ex`, `lib/replicant/spill/reader.ex`.)
-- **Two ceilings**, not one: resident RAM is bounded by `max_inflight_lag` (the spill trigger),
-  and disk is bounded by `max_spill_bytes` (a transaction exceeding it halts `:spill_exhausted`).
-  The §4 in-flight numerator is `received − floor − spilled` compared to `max_inflight_lag +
-  max_spill_bytes`, so a legitimately-spilling transaction is not double-counted toward the RAM
-  halt.
+- **Independent queue, resident, and disk budgets.** Socket intake pauses on actual
+  queued payload bytes at `max_inflight_lag` (or 512 frames); completed assembler
+  frames release credits. WAL-position distance is not used as a memory estimate.
+  Streamed resident payload spills reclaimable row buffers above `max_inflight_lag`;
+  messages remain in memory and charged. Spill repeats until the budget is met or
+  no row payload can be reclaimed. `max_spill_bytes` separately limits spill files
+  (exceeding it halts `:spill_exhausted`). With spill configured, non-streamed
+  transactions retain a `max_inflight_lag + max_spill_bytes` **memory** allowance
+  for compatibility, since they cannot use the streamed spill path. None of these
+  budgets is an exact RSS limit: decoded-term overhead and socket buffers are extra.
 - **No in-lib encryption.** Key management is a separable subsystem with its own operational
   surface (KMS, rotation, envelope keys) that this library does not own. The operator places the
   spill `dir` on a secure/encrypted volume and cleans it on decommission; the default `dir` is a

examples/replication_pipeline/mix.exs

@@ -26,8 +26,9 @@ defmodule ReplicationPipeline.MixProject do
     [
       # The CDC engine, from THIS checkout (the example is part of its repo).
       {:replicant, path: "../.."},
-      # Destination client (mirrors the library's own floor).
-      {:postgrex, "~> 0.22.4"},
+      # Match the CDC engine's pause/resume-capable transport pin.
+      {:postgrex,
+       github: "elixir-ecto/postgrex", ref: "a209c8b96c5d92cfff9a35df8ed31ed123943bc2"},
       # Static gates (same tooling as the library itself; dev/test only).
       {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
       {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false}

examples/replication_pipeline/mix.lock

@@ -7,6 +7,6 @@
   "erlex": {:hex, :erlex, "0.2.9", "7debbbaa9f4f368b8cd648983e0f1d7963028508e9c59e9d4ed504e94ef52a55", [:mix], [], "hexpm", "8cfffc0ec7159e6d73de2ab28a588064de80f88b2798d5cbe4482cbbc200178b"},
   "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"},
   "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"},
-  "postgrex": {:hex, :postgrex, "0.22.4", "d271f595dfd25230b6398354e19d17bb5e2d20130fd2d9bdca7e15f125d43552", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "4aae45a2d60e35b04eea2602440be152fae332901f1fc7a60fc7cb7f0f9a9c5a"},
+  "postgrex": {:git, "https://github.com/elixir-ecto/postgrex.git", "a209c8b96c5d92cfff9a35df8ed31ed123943bc2", [ref: "a209c8b96c5d92cfff9a35df8ed31ed123943bc2"]},
   "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
 }

lib/replicant/assembler.ex

@@ -86,6 +86,7 @@ defmodule Replicant.Assembler do
           batch: keyword() | nil,
           batch_count: non_neg_integer(),
           batch_txns: [Transaction.t()],
+          batch_bytes: non_neg_integer(),
           batch_spill_paths: [String.t()],
           last_buffered_changes: [Change.t()] | :spilled,
           pending_lsn: lsn() | nil,
@@ -112,6 +113,8 @@ defmodule Replicant.Assembler do
           messages: [Message.t()],
           byte_size: non_neg_integer(),
           resident_bytes: non_neg_integer(),
+          change_bytes: %{non_neg_integer() => non_neg_integer()},
+          message_bytes: %{non_neg_integer() => non_neg_integer()},
           spilled_bytes: non_neg_integer(),
           spilled_by_subxid: %{non_neg_integer() => non_neg_integer()},
           aborted: MapSet.t(non_neg_integer()),
@@ -136,6 +139,7 @@ defmodule Replicant.Assembler do
     batch: nil,
     batch_count: 0,
     batch_txns: [],
+    batch_bytes: 0,
     batch_spill_paths: [],
     last_buffered_changes: [],
     pending_lsn: nil,
@@ -176,10 +180,10 @@ defmodule Replicant.Assembler do
   end
 
   @doc """
-  Accumulate the raw WAL payload byte-size of the message about to be handled into
+  Accumulate the raw WAL payload byte-size of the message just handled into
   the open transaction buffer, for the `byte_size` metadata on
   `[:replicant, :transaction, :assembled]` (spec §10). `Replicant.AssemblerServer`
-  calls this with `byte_size(payload)` before each `handle_message/2`. It is a
+  calls this with `byte_size(payload)` after each non-terminal `handle_message/2`. It is a
   no-op before any `Begin` (bytes arriving outside a transaction — a lone
   Relation/Type/Origin — have no buffer to attribute to), so a stray pre-Begin
   payload never crashes here.
@@ -193,7 +197,7 @@ defmodule Replicant.Assembler do
   def observe_bytes(%__MODULE__{current_stream_xid: xid, stream_txns: streams} = asm, bytes)
       when is_integer(xid) and is_integer(bytes) and bytes >= 0 and is_map_key(streams, xid) do
     buf = Map.fetch!(streams, xid)
-    buf = %{buf | byte_size: buf.byte_size + bytes, resident_bytes: buf.resident_bytes + bytes}
+    buf = __MODULE__.Streaming.account_bytes(buf, xid, bytes)
     asm = %{asm | stream_txns: Map.put(streams, xid, buf)}
 
     case __MODULE__.Streaming.maybe_spill(asm) do
@@ -209,6 +213,16 @@ defmodule Replicant.Assembler do
     %{asm | txn: %{buffer | byte_size: buffer.byte_size + bytes}}
   end
 
+  @doc false
+  def buffered_bytes(asm) do
+    transaction_bytes = if asm.txn, do: asm.txn.byte_size, else: 0
+
+    streamed_bytes =
+      Enum.reduce(asm.stream_txns, 0, fn {_xid, buf}, n -> n + buf.resident_bytes end)
+
+    transaction_bytes + streamed_bytes + asm.batch_bytes
+  end
+
   @doc """
   Handle one decoded message. Returns:
     * `{:ok, t()}` — accumulated, no boundary crossed.
@@ -1045,6 +1059,7 @@ defmodule Replicant.Assembler do
       | batch_count: 0,
         pending_lsn: nil,
         batch_txns: [],
+        batch_bytes: 0,
         batch_spill_paths: [],
         last_buffered_changes: []
     }

lib/replicant/assembler/batch.ex

@@ -82,6 +82,7 @@ defmodule Replicant.Assembler.Batch do
       | batch_count: count,
         pending_lsn: lsn,
         batch_txns: [txn | asm.batch_txns],
+        batch_bytes: asm.batch_bytes + :erlang.external_size(txn),
         batch_spill_paths: prepend_path(asm.batch_spill_paths, spill_path)
     }
 
@@ -157,6 +158,7 @@ defmodule Replicant.Assembler.Batch do
            | batch_count: 0,
              pending_lsn: nil,
              batch_txns: [],
+             batch_bytes: 0,
              batch_spill_paths: [],
              lib_checkpoint: max(asm.lib_checkpoint || 0, lsn)
          }}