Bound queuing thread with hard limit instead of soft

plans/config.md

@@ -51,8 +51,12 @@ matches steady state.
 
 Bootstrap fixed points stay on `EmitterConfig`, boot-only, never republished:
 connection params (`[ch] host/port/user/password/database/secure`), toast store,
-`soft_delete`. These describe how to reach CH or wire into pipeline stages at
-spawn; a live swap would mean reconnecting or rebuilding the pipeline.
+`soft_delete`, and the source physical replication slot (`[source] slot` →
+`source_slot`, with a `--slot` CLI override applied at parse time à la
+`--ch-flush-timeout-ms`; created idempotently before pre-flight, streamed
+from, and resumed against on reconnect — see [source.md](source.md)). These describe how
+to reach CH/source or wire into pipeline stages at spawn; a live swap would mean
+reconnecting or rebuilding the pipeline.
 `target_database` and `soft_delete` thread into the DDL applicator at
 construction and carry across refreshes unchanged.
 

plans/future/ch_bounce_recovery.md

@@ -78,7 +78,12 @@ emitter's retry path
   matters only if the workload demands lower-latency recovery than
   "operator restart + WAL re-read"
 * WAL re-read from cursor is fast — bounded by source-side retention
-  and walshadow's decode throughput, not by spill replay
+  and walshadow's decode throughput, not by spill replay. A physical
+  slot + the `flush_lsn = min(durable, apply_ceiling)` cap now hold
+  source WAL through the CH outage (slot `restart_lsn` sticks at the
+  CH-durable point), so re-read from cursor stays available unless the
+  slot goes `lost` / source disk fills — that edge is fatal and recovers
+  via config `initial_load` re-seed (see [../source.md](../source.md))
 * Spill format extension would re-open the spill-format-version-bump
   debt; better paid once than twice
 

plans/future/pipeline_backpressure_and_scaling.md

@@ -12,28 +12,28 @@ feed inserters. Out-of-order INSERTs are fine — `_lsn` plus
 ([[project_walshadow_eventual_consistency]]) — but slot feedback still
 needs the strict contiguous durable watermark.
 
-## Pump→worker bound (wire/record split)
-
-The pump→worker channel (`queueing_record_sink.rs`) is soft-capped, not
-hard-bounded. `on_record` and shadow-wire delivery run lockstep in
-`wal_stream.rs::drain_records`, wire chunk before record, so every
-`ShadowCatalog::wait_for_replay` target sits behind the wire head: the
-gate waits on shadow *apply* of already-sent bytes, never on bytes the
-pump has yet to produce. Delivery of sent bytes is pump-independent
-(listener task drains `send_queues`, idle keepalives force walreceiver
-flush, `wire_buf` backfill covers in-segment reconnects,
-`restore_command` older ranges), so a pump parked in a hard-bounded
-`on_record` cannot starve a pending gate. It still freezes for the full
-send → walreceiver flush → replay → poll round-trip each time the queue
-fills while a gate is pending, and it couples wire progress to decode
-progress, turning any delivery path that does need fresh pump bytes
-into a deadlock. The soft cap (yield past `soft_cap`) keeps wire
-delivery independent of decode; the cost is an unbounded buffer: under
-sustained CH-slower-than-WAL it grows, holding WAL in walshadow RAM
-rather than letting the PG slot hold it on disk.
-
-Fix: split wire delivery (runs ahead, paced by shadow apply) from record
-dispatch (blocks on a bounded queue). Architectural, not a channel swap.
+## Pump→worker bound (implemented — hard cap)
+
+The pump→worker channel (`queueing_record_sink.rs`) is hard-bounded: a
+`mpsc::channel(max_records / batch_size)` whose `send().await` blocks the
+pump when full. Real backpressure — a sustained CH-slower-than-WAL run
+parks the pump on the source socket (TCP backpressure → with a physical
+slot + the `flush_lsn` cap, the source holds WAL on its disk, see
+[source.md](../source.md)) instead of growing walshadow RAM without limit.
+
+Deadlock-safe, even though `on_record` and shadow-wire delivery run
+lockstep in `wal_stream.rs::drain_records` and
+`ReorderSink::maybe_sweep_dropped` → `ShadowCatalog::wait_for_replay`
+couples decode to shadow apply. Two facts break the feared deadlock:
+`on_wire_chunk` fires *before* `on_record` per record, and the walsender
+feeds shadow on an **independent** per-connection task (+keepalive-on-idle,
+`shadow_stream.rs`), not from the pump. So a parked pump only withholds
+*future* wire (`> resume`); every LSN a worker `wait_for_replay` can target
+is a dispatched record's own `source_lsn`, always ≤ the highest wire
+already enqueued (dispatch is FIFO ascending, wire precedes dispatch), and
+that wire reaches shadow independently → replay advances → the wait clears.
+The bytes always lead the wait target; the wait is never for bytes shadow
+hasn't received.
 
 The other two ingress channels need nothing: bootstrap page-walk→drain
 is bounded (`BOOTSTRAP_TUPLE_CHANNEL_CAP`, a full channel parks the page

plans/source.md

@@ -51,9 +51,35 @@ have not reached and let source recycle un-filtered WAL. Cadence
 `wal_sender_timeout` default 60s gives 6× headroom
 
 `slot: Option<&str>` on `start_physical_replication`: Some = bind
-permanent physical slot, pins source's `pg_wal/` until our `apply_lsn`
-advances; None = slotless, source recycles on `wal_keep_size` schedule.
-Slot caps catch-up, slotless caps source disk burn
+permanent physical slot, pins source's `pg_wal/`; None = slotless, source
+recycles on `wal_keep_size` schedule. Slot caps catch-up, slotless caps
+source disk burn. Slot name comes from config (`[source] slot` →
+`EmitterConfig.source_slot`), with a `--slot` CLI override applied at parse
+time (CLI > TOML, boot-only — same idiom as `--ch-flush-timeout-ms`).
+`SourceFeed::ensure_physical_slot(name)` creates it idempotently with
+immediate WAL reservation (`pg_create_physical_replication_slot(name,
+true)`), run before pre-flight (which requires the slot to exist). A
+physical slot's `restart_lsn` tracks the reported **flush** LSN, so the
+pump caps `flush_lsn` at `apply_ceiling = min(shadow_replay, emitter_ack)`
+(see [ops.md](ops.md)): the source retains WAL until CH has durably
+ingested it, not merely until walshadow fsynced its filter output.
+
+`SourceFeed::reconnect(cfg, slot, resume_lsn, timeline, interval)`
+re-establishes a dropped replication connection (e.g. source
+`wal_sender_timeout` fired during a CH-stall backpressure) and resumes at
+`resume_lsn = stream.next_lsn()` — the byte-contiguous resume point, never
+`dispatched_lsn` (which lags by any buffered in-progress record →
+misaligned push). `reconnect` is one attempt; the caller
+(`reconnect_or_fatal`) drives `backon` exponential-backoff retry to ride
+out a transient drop until the source returns. A recycled segment (SQLSTATE
+`58P01`, classified by `is_wal_segment_removed` on the code, not the
+locale-dependent message → typed `WalSegmentRemoved`) stops the retry and
+is **fatal**: the
+resume point is gone, so the daemon exits and recovery is a re-seed via
+config `initial_load` (`base_backup`/`object_store`) on restart, not an
+in-place reconnect. The slot + `flush_lsn` cap make this recycle path an
+edge (slot `lost` / dropped / source disk full), not the normal CH-stall
+path
 
 TLS / SCRAM via wal-rus's
 `walrus::pg::replication::tls`: modes `disable / allow
@@ -189,18 +215,21 @@ decode, turning any delivery path that needs fresh pump bytes into a
 deadlock
 
 `QueueingRecordSink::spawn` takes inner `RecordSink + Send + 'static`,
-`batch_size`, `soft_cap`. Pump-side `on_record` clones record to
+`batch_size`, `max_records`. Pump-side `on_record` clones record to
 `'static` via `record.parsed.clone().into_owned()`, pushes onto local
-`Vec`, ships when `len >= batch_size` onto unbounded
-`mpsc::UnboundedSender<Vec<Record<'static>>>`. Worker drains at its own
-pace through inner sink
-
-Backpressure soft only. `in_flight: AtomicU64` tracks records in channel
-+ pump buffer; crossing `soft_cap` triggers `tokio::task::yield_now()`.
-Hard cap open-ended — permanently stalled worker surfaces catalog
-`wait_for_replay` timeout on shared err slot, pump's next `on_record`
-returns parked error, daemon exits cleanly with real root cause rather
-than hanging
+`Vec`, ships when `len >= batch_size` onto a **bounded**
+`mpsc::Sender<Vec<Record<'static>>>` sized `max_records / batch_size`.
+Worker drains at its own pace through inner sink
+
+Backpressure hard. `flush_buf`'s `tx.send().await` blocks the pump when
+the channel is full, so a CH-slower-than-WAL run parks the pump on the
+source socket (→ the physical slot holds WAL on source disk) instead of
+growing walshadow RAM. Deadlock-safe despite the wire/record lockstep
+above: `on_wire_chunk` fires before `on_record`, and the walsender feeds
+shadow on an independent per-connection task (+keepalive-on-idle), so a
+parked pump only withholds *future* wire — never the ≤`resume` bytes any
+pending `wait_for_replay` needs (bytes lead the wait target). See
+[future/pipeline_backpressure_and_scaling.md](future/pipeline_backpressure_and_scaling.md)
 
 Worker uses `tokio::time::timeout(idle_interval, rx.recv())`. On timeout
 calls `inner.on_idle()`; channel close fires `inner.on_close()` for one
@@ -371,10 +400,13 @@ walsender listener before shadow's walreceiver attaches; construct
 `QueueingRecordSink::spawn(DecoderXactPair { decoder, xact_drain },
 batch_size, capacity)`); open `WalStream` + `DirSegmentSink`, attach
 `ShadowStreamSink` via `set_bytes_sink`; spawn metrics endpoint, SIGHUP
-handler, retention sweeper, cursor write loop; wait for walsender
+handler, retention sweeper, cursor write loop; ensure the configured physical slot before pre-flight; wait for walsender
 connect barrier; pump loop: `feed.next_chunk()` → `stream.push(lsn,
 bytes, &mut record_sink, &mut segment_sink)` → cursor advance → status
-update → repeat
+update → repeat. A `next_chunk` error routes through `reconnect_or_fatal`:
+transient drop → `SourceFeed::reconnect` at `stream.next_lsn()`; recycled
+segment (`58P01`) → fatal exit (re-seed via config `initial_load` on
+restart)
 
 `DecoderXactPair` order is fixed: decoder absorbs heap record into xact
 buffer *before* xact_drain flushes matching commit/abort.

src/bin/stream.rs

@@ -200,7 +200,8 @@ struct Args {
     /// `restore_command` reads from here
     #[arg(long)]
     out_dir: PathBuf,
-    /// Optional permanent physical slot name on source PG
+    /// CLI override for the TOML's `[source] slot` (physical replication
+    /// slot). Unset defers to config; unset in both = slotless.
     #[arg(long)]
     slot: Option<String>,
     /// Start LSN in `X/Y` hex form. Defaults to source's current
@@ -514,10 +515,17 @@ async fn run(args: Args) -> Result<()> {
         if let Some(ms) = args.ch_flush_timeout_ms {
             cfg.flush_timeout = std::time::Duration::from_millis(ms);
         }
+        // CLI override wins over TOML `[source] slot` (CLI > config).
+        if args.slot.is_some() {
+            cfg.source_slot = args.slot.clone();
+        }
         Some(cfg)
     } else {
         None
     };
+    // Effective physical replication slot (`[source] slot` + --slot override);
+    // None = slotless.
+    let source_slot: Option<String> = ch_config.as_ref().and_then(|c| c.source_slot.clone());
     let bootstrap_end_lsn: Option<u64> = if matches!(args.bootstrap_mode, BootstrapMode::Off) {
         None
     } else {
@@ -689,6 +697,14 @@ async fn run(args: Args) -> Result<()> {
         .await
         .set_invalidation_epoch(invalidation_epoch.clone());
 
+    // Create the configured slot before preflight, which requires it to exist.
+    if let Some(slot) = source_slot.as_deref() {
+        feed.ensure_physical_slot(slot)
+            .await
+            .with_context(|| format!("ensure physical replication slot {slot}"))?;
+        tracing::info!(target: "walshadow", slot, "physical replication slot ready");
+    }
+
     // Pre-flight validators run after both source + shadow SQL clients
     // are up so every check has its connection.
     if !args.skip_preflight {
@@ -708,7 +724,7 @@ async fn run(args: Args) -> Result<()> {
             source_version_num,
             source_sql,
             shadow_sql: &shadow_sql,
-            slot: args.slot.as_deref(),
+            slot: source_slot.as_deref(),
             ch_config: ch_config.as_ref(),
         })
         .await
@@ -766,7 +782,7 @@ async fn run(args: Args) -> Result<()> {
         }
     };
 
-    feed.start_physical_replication(args.slot.as_deref(), aligned, ident.timeline)
+    feed.start_physical_replication(source_slot.as_deref(), aligned, ident.timeline)
         .await
         .context("START_REPLICATION")?;
     // Spill dir wiped every startup: cursor file commits drains
@@ -1194,9 +1210,12 @@ async fn run(args: Args) -> Result<()> {
                 .context("write resume cursor")?;
             last_cursor_write = Some(Instant::now());
         }
+        // flush drives a physical slot's restart_lsn (what the source retains),
+        // so cap it at apply_ceiling: the source keeps WAL until CH has durably
+        // applied it (and shadow replayed), never recycling un-consumed WAL.
         let status = StandbyStatus {
             write_lsn: received,
-            flush_lsn: durable,
+            flush_lsn: durable.min(apply_ceiling),
             apply_lsn: apply_ceiling,
         };
         let dispatched_before = stream.dispatched_lsn();
@@ -1210,10 +1229,34 @@ async fn run(args: Args) -> Result<()> {
             // Idle tick so metrics/cursor keep tracking the draining pipeline
             // when no new WAL arrives.
             _ = tokio::time::sleep(metrics_tick) => None,
-            res = feed.next_chunk(status, &mut chunk_buf) => Some(match res? {
-                Some(c) => c,
-                None => break "CopyDone",
-            }),
+            res = feed.next_chunk(status, &mut chunk_buf) => match res {
+                Ok(Some(c)) => Some(c),
+                Ok(None) => break "CopyDone",
+                Err(e) => {
+                    let resume = stream.next_lsn();
+                    tracing::warn!(
+                        target: "walshadow",
+                        error = %e,
+                        resume_lsn = format_pg_lsn(resume).to_string(),
+                        "source stream error — recovering",
+                    );
+                    feed = reconnect_or_fatal(
+                        e,
+                        &cfg,
+                        source_slot.as_deref(),
+                        resume,
+                        ident.timeline,
+                        Duration::from_secs(args.status_interval),
+                    )
+                    .await?;
+                    tracing::info!(
+                        target: "walshadow",
+                        resume_lsn = format_pg_lsn(resume).to_string(),
+                        "source reconnected — resuming replication",
+                    );
+                    None
+                }
+            },
         };
         let server_end = chunk.as_ref().map(|c| c.server_wal_end).unwrap_or(received);
         if let Some(chunk) = chunk {
@@ -1963,6 +2006,47 @@ async fn populate_metrics(
     registry.set(snap).await;
 }
 
+/// Recover from a source stream error. A transient drop retries the reconnect
+/// with exponential backoff until the source is back; a recycled segment
+/// (58P01) is fatal — the resume point is gone, so the daemon exits and
+/// recovery is a re-seed via config `initial_load` on restart, not a reconnect.
+async fn reconnect_or_fatal(
+    e: anyhow::Error,
+    cfg: &PgConfig,
+    slot: Option<&str>,
+    resume_lsn: u64,
+    timeline: u32,
+    status_interval: Duration,
+) -> Result<SourceFeed> {
+    use backon::{ExponentialBuilder, Retryable};
+
+    let recycled = |e: anyhow::Error| {
+        e.context(
+            "source WAL segment recycled past the resume point; \
+             re-seed the affected tables via config initial_load \
+             (base_backup/object_store), then restart",
+        )
+    };
+    if walshadow::source_feed::is_wal_segment_removed(&e) {
+        return Err(recycled(e));
+    }
+    // Ride out a transient drop (source restart, wal_sender_timeout, brief
+    // network blip); only a recycled segment stops the retry and is fatal.
+    (|| SourceFeed::reconnect(cfg, slot, resume_lsn, timeline, status_interval))
+        .retry(
+            ExponentialBuilder::default()
+                .with_min_delay(Duration::from_millis(200))
+                .with_max_delay(Duration::from_secs(10))
+                .without_max_times(),
+        )
+        .when(|e: &anyhow::Error| !walshadow::source_feed::is_wal_segment_removed(e))
+        .notify(|e: &anyhow::Error, d: Duration| {
+            tracing::warn!(target: "walshadow", error = %e, retry_in_ms = d.as_millis() as u64, "source reconnect failed — retrying");
+        })
+        .await
+        .map_err(recycled)
+}
+
 /// Orchestrate BASE_BACKUP into a fresh shadow data dir; returns the
 /// backup's `end_lsn` so the caller rebinds the WAL pump past it.
 ///

src/ch_emitter.rs

@@ -161,6 +161,10 @@ pub struct EmitterConfig {
     /// overlay tables. `None` (field empty or omitted) disables the whole
     /// overlay subsystem — no boot seed, no config_decoder, pure TOML+CLI.
     pub runtime_config_schema: Option<String>,
+    /// `[source] slot`: physical replication slot to create + stream from on
+    /// the source. `Some` reserves WAL so a stalled/disconnected consumer
+    /// resumes without recycling; `None` runs slotless. Boot-only.
+    pub source_slot: Option<String>,
 }
 
 pub const DEFAULT_INSERT_TIMEOUT_SECS: u64 = 30;
@@ -211,6 +215,7 @@ impl Default for EmitterConfig {
             toast: crate::toast::ToastConfig::default(),
             decode_chunk_rows: crate::pipeline::decode::DECODE_CHUNK_ROWS,
             runtime_config_schema: None,
+            source_slot: None,
         }
     }
 }
@@ -448,6 +453,13 @@ impl EmitterConfig {
             // Empty string == omitted == overlay disabled.
             out.runtime_config_schema = Some(schema.into());
         }
+        if let Some(src) = root.get("source").and_then(Value::as_table)
+            && let Some(slot) = src.get("slot").and_then(Value::as_str)
+            && !slot.is_empty()
+        {
+            // Empty string == omitted == slotless.
+            out.source_slot = Some(slot.into());
+        }
         if let Some(nss) = root.get("namespace").and_then(Value::as_table) {
             for (k, v) in nss {
                 let t = v.as_table().ok_or_else(|| {