feat(memory): remove the watermark self-recycle

README.md

@@ -193,7 +193,7 @@ There is **no data loss path** — machine-checked, not just argued: every consi
 | Crash after destination commit, before cursor persist | Destination has data, cursor stale | At-least-once: range re-read and re-applied idempotently (full CDC); tombstones remove any since-deleted rows the crashed commit wrote. | Per-destination |
 | Cursor persist failure (PG down) | Destination commit already landed | 3 in-process retries ([`delivery.py:_advance_cursor_with_retry`](viaduck/delivery.py)); on exhaustion, same path as flush failure — range re-read, idempotent re-apply. | Per-destination |
 | Destination apply failure (full CDC) | Delete + upsert transaction rolled back | No partial state on destination. Buffer dropped, range re-read. | Per-destination |
-| SIGTERM with data buffered | Shutdown drain | `drain()` flushes everything buffered (trigger=shutdown), bounded by a 60s deadline; anything abandoned is re-read on restart. Note: the 60s deadline exceeds K8s's default 30s `terminationGracePeriodSeconds` — raise the grace period or expect SIGKILL to cut the drain short (safe, just re-read). The watermark self-recycle exit uses a 300s drain budget instead (no grace clock is ticking — see Deployment). | — |
+| SIGTERM with data buffered | Shutdown drain | `drain()` flushes everything buffered (trigger=shutdown), bounded by a 60s deadline; anything abandoned is re-read on restart. Note: the 60s deadline exceeds K8s's default 30s `terminationGracePeriodSeconds` — raise the grace period or expect SIGKILL to cut the drain short (safe, just re-read). | — |
 | Destination at its buffer cap | Backpressure (by design) | The destination's queue (buffer + in-flight) hit its per-destination cap: its buffer force-flushes (trigger=`memory`) and its CDC reads pause until the flush drains — `viaduck_delivery_reads_paused` gauges it. Healthy peers keep reading and flushing. | **Per-destination** |
 | Destination failing flushes repeatedly | Circuit breaker opens after `flush_circuit_failures` consecutive failures | Flush submissions pause behind an exponential backoff; reads continue under the buffer cap. A probe after each backoff closes the circuit on success. `viaduck_delivery_circuit_open` / `viaduck_delivery_circuit_opens_total` gauge/count it; logs WARN on open. | **Per-destination** |
 | Routing field missing from source | `RoutingError` halts group processing | Error metricked, logged. Requires config or schema fix. | All destinations in group |
@@ -289,10 +289,7 @@ delivery:
   # flush_adaptive_reprobe_after: 50        # consecutive in-band full flushes before one upward re-probe
 
 memory:
-  self_recycle_enabled: true                # watermark self-recycle (see Deployment)
-  # self_recycle_rss_gib: 0                 # absolute RSS watermark in GiB (0 = derive from fraction)
-  # self_recycle_rss_fraction: 0.75         # watermark as fraction of the cgroup memory limit
-  # self_recycle_min_uptime_seconds: 3600   # never recycle a young (catching-up) process
+  # dest_conn_max_age_seconds: 600        # close/reopen pooled destination connections past this age (0 disables)
 
 server:
   port: 8000                                # metrics, health checks, status UI
@@ -492,7 +489,8 @@ The web UI (`/ui`) and status API (`/status`) report a per-destination operation
 | `viaduck_delivery_buffers_dropped_total` | Counter | destination | Buffers dropped on flush failure |
 | `viaduck_delivery_covered_replays_dropped_total` | Counter | destination | Buffered replay entries dropped at flush commit (already covered by it) |
 | `viaduck_retention_clamp_total` | Counter | destination, outcome | Retention-edge cursor clamps (`lost` = unrecoverable, alert; `at_risk` = pending flush) |
-| `viaduck_self_recycles_total` | Counter | — | Clean watermark-triggered restarts (drain + exit 0 on RSS watermark) |
+| `viaduck_rss_bytes` | Gauge | — | Process RSS, exported once per poll cycle — the memory-safety signal |
+| `viaduck_dest_conn_sweeps_total` | Counter | — | Destination-pool age sweeps performed (the memory bound; see Deployment) |
 | `viaduck_cdc_routing_mutations_total` | Counter | — | Cross-tenant routing value changes |
 | `viaduck_cdc_conflicts_resolved_total` | Counter | — | Rowid-level conflicts resolved in Phase 2 |
 | `viaduck_cdc_tombstones_emitted_total` | Counter | — | Deletes surviving from insert+delete pairs (write cost of phantom healing; churn signal) |
@@ -601,11 +599,9 @@ kubectl apply -f k8s/deployment.yaml
 
 Viaduck runs as a K8s Deployment (not StatefulSet — no ordinal-based identity needed). For horizontal scaling, deploy multiple instances with different `instance.partition` configs. See [`k8s/deployment.yaml`](k8s/deployment.yaml) for manifests.
 
-### Watermark self-recycle
+### Destination connection age sweep
 
-Long-lived processes accrue untracked native memory in the ducklake extension (roughly proportional to catalog metadata volume, freed only on connection close). An OOM-kill mid-flight rewinds every destination to its durable cursor and scatters the cursor groups; a clean exit after a drain leaves cursors tight at the read position. So viaduck preempts the OOM: after `memory.self_recycle_min_uptime_seconds` (default 3600 — a post-restart catch-up legitimately runs hot), each poll cycle checks RSS against a watermark; on crossing it, the process finishes the cycle, drains with a 300s budget (longer than the SIGTERM drain — no grace clock is ticking), and exits 0 for an in-place kubelet restart (`restartPolicy: Always`). Signals: the `[SELF-RECYCLE]` WARN, `viaduck_self_recycles_total`, and the container's last-state `Completed`/exit 0 (vs `OOMKilled`/137).
-
-Knobs ([`config.py`](viaduck/config.py) `memory.*`): `self_recycle_enabled` (default true), `self_recycle_rss_gib` (absolute watermark, default 0 = derive), `self_recycle_rss_fraction` (default 0.75 × the cgroup memory limit; disabled with a startup log if no limit is readable). Sizing: the watermark must clear the deployment's legitimate peak (≈ `delivery.buffer_total_max_bytes` + pool native footprint + baseline) or leak-free load recycles the pod every min-uptime — sized deployments should set the absolute knob.
+Long-lived destination connections accrue per-connection native memory that is only reclaimed on close. The poll loop sweeps the destination pool: connections older than `memory.dest_conn_max_age_seconds` (default 600s; 0 disables; values below 60 are rejected) are force-evicted and recreated lazily on their next flush — capped at two evictions per cycle so a mass-expiry event can't storm the destination catalog. A pinned (in-flight) connection is skipped rather than closed mid-apply. Signal: `viaduck_dest_conn_sweeps_total`.
 
 ## Error Handling and Retries
 

docs/runbook-offset-reset.md

@@ -80,9 +80,9 @@ kubectl -n argocd patch application <viaduck-app> --type merge \
   -p '{"spec":{"syncPolicy":{"automated":{"prune":true,"selfHeal":true}}}}'
 ```
 
-Verify on startup: `Self-recycle watermark` line present, first poll
-cycles show small lags, `viaduck_dest_lag_snapshots` near zero for reset
-destinations, no `retention clamp` warnings.
+Verify on startup: first poll cycles show small lags,
+`viaduck_dest_lag_snapshots` near zero for reset destinations, no
+`retention clamp` warnings.
 
 ## Hazards
 

log-consumer-proposal.md

@@ -68,6 +68,11 @@ production log, 2026-08-15, viaduck 0.0.70.)
 | Destination append cliff (team-2) | 13.2s @30–60k rows, 56.5s @60–90k, 164.4s @90–120k (240s deadline) |
 | Memory | ~2.9 GiB/h untracked native residual; 82 GiB self-recycle watermark |
 
+*(Erratum, 2026-09-02: the watermark self-recycle was removed in PR #85
+after the dest-connection age sweep (#84) held prod flat for ~19h; the
+sweep is the bound now. The dated rows above/below record the
+2026-08-15 state verbatim.)*
+
 Structural findings:
 
 1. **Group-scan amplification**: N cursor groups each pay full scan cost
@@ -237,8 +242,9 @@ whether anyone does.
 
 - **Reliable**: at-least-once (unchanged contract). Recovery state is the
   durable per-destination cursor. In-flight state may be volatile:
-  crashes are rare (clean self-recycle drain for the known RSS residual),
-  and §6.5 prices the rewind honestly.
+  crashes are rare (clean self-recycle drain for the known RSS residual;
+  that drain path was removed 2026-09-02 in PR #85 — the dest-connection
+  age sweep is the bound now), and §6.5 prices the rewind honestly.
 - **Performant**: a consumer's pace is bounded only by its destination's
   append capacity, never by peers.
 - **Simple**: delete more than we add; any mechanism that arbitrates
@@ -428,7 +434,7 @@ position-grid chunking, group fairness machinery.
 
 | Event | Cost |
 |---|---|
-| Clean restart / self-recycle | Drain flushes buffers; cursors tight; resume near head. Seconds. |
+| Clean restart | Drain flushes buffers; cursors tight; resume near head. Seconds. |
 | Hard crash, healthy fleet | Rewind = buffered window ≤ flush cadence (~120s × 2 snap/s ≈ ~240 snapshots). Re-read at feed speed: seconds. Clustering re-glues in one cycle. |
 | Hard crash with an at-cap destination | Worst case rewind = the full per-destination cap: 4GiB ≈ ~4.3M team-2 rows ≈ **~6.7k snapshots ≈ ~1h of head**. Re-read is minutes (feed is cheap); **re-delivery is append-bound, ~16–20 min** at 3.6–4.5k rows/s. Note the crash postures correlate with fat buffers — plan for the at-cap case, not the healthy one. Still not absorbing: the 08-14 catastrophe required the divergence regime. |
 | One destination down hours | Cap freezes its position; on recovery it sweeps at its own append rate. Fleet unaffected. |
@@ -532,8 +538,8 @@ bridge split (§10.3), not schedule compression.
    churn this implies).
 5. **Shadow** (validation scaffolding, owned as a time-boxed third image,
    not a compat shim): old image + feed in dual-read mode, divergence
-   counter, 24–48h. Note the shadow pod still leaks ~2.9GiB/h toward its
-   self-recycle — comparison continuity across recycles is scripted, not
+   counter, 24–48h. Note the shadow pod still carries the ~2.9GiB/h
+   residual — comparison continuity across any restarts is scripted, not
    assumed. Shadow doubles source read load briefly; acceptable.
    **Operational gate for the shadow window: no source-table DDL.** The
    feed has no `mapping_id`/rename story yet (§11.4) — a mid-shadow rename

pyproject.toml

@@ -19,8 +19,7 @@ dependencies = [
     # that worked around 1.0.16's ==1.5.2 pin is gone). 1.0.18 adds the
     # DETACH-before-close fix for hypothesis-2's Leak A (~5.5MB orphaned
     # per conflicted close) — wheel-verified 2026-08-14: Catalog.close()
-    # now DETACHes best-effort before conn.close(). The watermark
-    # self-recycle remains for the distinct in-lifetime Leak B residual.
+    # now DETACHes best-effort before conn.close().
     "pyducklake>=1.0.18",
     # 1.5.2 -> 1.5.5 (2026-07-31): the duckdb-1.5.2 extension channel is
     # frozen at ducklake build 415a9ebd (2026-04-09), whose per-connection

tests/unit/test_config.py

@@ -1092,66 +1092,44 @@ def test_destination_buffer_max_bytes_default_zero(config_file: Path):
     assert cfg.destinations[0].buffer_max_bytes == 0
 
 
-# --- memory / self-recycle ---
-
-
-def test_memory_defaults(config_file: Path):
-    cfg = load(config_file)
-    assert cfg.memory.self_recycle_enabled is True
-    assert cfg.memory.self_recycle_rss_fraction == 0.75
-    assert cfg.memory.self_recycle_rss_gib == 0.0
-    assert cfg.memory.self_recycle_min_uptime_seconds == 3600.0
+def test_memory_dest_conn_max_age_default_and_override(tmp_path: Path):
+    p = tmp_path / "viaduck.yaml"
+    p.write_text(MINIMAL_YAML)
+    assert load(p).memory.dest_conn_max_age_seconds == 600.0
 
+    p.write_text(MINIMAL_YAML + "\nmemory:\n  dest_conn_max_age_seconds: 120\n")
+    assert load(p).memory.dest_conn_max_age_seconds == 120.0
 
-def test_memory_explicit_values(tmp_path: Path):
-    p = tmp_path / "viaduck.yaml"
-    p.write_text(
-        MINIMAL_YAML
-        + """
-memory:
-  self_recycle_enabled: false
-  self_recycle_rss_fraction: 0.5
-  self_recycle_rss_gib: 70
-  self_recycle_min_uptime_seconds: 0
-"""
-    )
-    cfg = load(p)
-    assert cfg.memory.self_recycle_enabled is False
-    assert cfg.memory.self_recycle_rss_fraction == 0.5
-    assert cfg.memory.self_recycle_rss_gib == 70.0
-    assert cfg.memory.self_recycle_min_uptime_seconds == 0.0
+    # off is allowed (0 disables the sweep)
+    p.write_text(MINIMAL_YAML + "\nmemory:\n  dest_conn_max_age_seconds: 0\n")
+    assert load(p).memory.dest_conn_max_age_seconds == 0.0
 
 
 @pytest.mark.parametrize(
     "snippet",
     [
-        "  self_recycle_rss_fraction: 0",
-        "  self_recycle_rss_fraction: 1",
-        "  self_recycle_rss_fraction: 1.5",
-        "  self_recycle_rss_gib: -1",
-        "  self_recycle_min_uptime_seconds: -5",
         "  dest_conn_max_age_seconds: -1",
-        "  dest_conn_max_age_seconds: 30",
+        "  dest_conn_max_age_seconds: 30",  # below the 60s floor (connect-storm guard)
     ],
 )
-def test_memory_validation_rejects(tmp_path: Path, snippet: str):
+def test_memory_dest_conn_max_age_rejects(tmp_path: Path, snippet: str):
     p = tmp_path / "viaduck.yaml"
     p.write_text(MINIMAL_YAML + "\nmemory:\n" + snippet + "\n")
     with pytest.raises(ConfigError):
         load(p)
 
 
-def test_memory_dest_conn_max_age_default_and_override(tmp_path: Path):
+def test_memory_retired_self_recycle_keys_warn_not_refuse(tmp_path: Path, caplog):
+    """A stale chart carrying the removed self_recycle_* keys must load
+    (refusing would CrashLoop the rollout) but must WARN (silence would
+    let an operator believe the deleted backstop exists)."""
     p = tmp_path / "viaduck.yaml"
-    p.write_text(MINIMAL_YAML)
-    assert load(p).memory.dest_conn_max_age_seconds == 600.0
-
-    p.write_text(MINIMAL_YAML + "\nmemory:\n  dest_conn_max_age_seconds: 120\n")
-    assert load(p).memory.dest_conn_max_age_seconds == 120.0
-
-    # off is allowed (0 disables the sweep)
-    p.write_text(MINIMAL_YAML + "\nmemory:\n  dest_conn_max_age_seconds: 0\n")
-    assert load(p).memory.dest_conn_max_age_seconds == 0.0
+    p.write_text(MINIMAL_YAML + "\nmemory:\n  self_recycle_enabled: true\n  self_recycle_rss_gib: 82\n")
+    with caplog.at_level("WARNING"):
+        cfg = load(p)
+    assert cfg.memory.dest_conn_max_age_seconds == 600.0  # defaults intact
+    warnings = [r for r in caplog.records if "self_recycle" in r.getMessage()]
+    assert len(warnings) == 2  # one per retired key present
 
 
 def test_to_libpq_conninfo_translation():

tests/unit/test_main.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import time
 from datetime import UTC
 from unittest.mock import MagicMock, call, patch
 
@@ -3397,91 +3396,10 @@ def test_poll_cycle_survives_membership_smaller_than_assigned():
     delivery.maybe_flush.assert_called()
 
 
-# --- watermark self-recycle ---
-
-
-def _mem(enabled=True, fraction=0.75, gib=0.0, min_uptime=3600.0):
-    """Real MemoryConfig (not a stand-in): field renames must break these
-    tests, not just run()'s startup."""
-    from viaduck.config import MemoryConfig
-
-    return MemoryConfig(
-        self_recycle_enabled=enabled,
-        self_recycle_rss_fraction=fraction,
-        self_recycle_rss_gib=gib,
-        self_recycle_min_uptime_seconds=min_uptime,
-    )
-
-
-def test_recycle_watermark_absolute_wins():
-    from viaduck.main import resolve_recycle_watermark
-
-    with patch("viaduck.main._cgroup_memory_limit_gib", return_value=96.0):
-        assert resolve_recycle_watermark(_mem(gib=70.0, fraction=0.5)) == 70.0
-
-
-def test_recycle_watermark_fraction_of_cgroup_limit():
-    from viaduck.main import resolve_recycle_watermark
-
-    with patch("viaduck.main._cgroup_memory_limit_gib", return_value=96.0):
-        assert resolve_recycle_watermark(_mem(fraction=0.75)) == pytest.approx(72.0)
-
-
-def test_recycle_watermark_disabled_by_config():
-    from viaduck.main import resolve_recycle_watermark
-
-    assert resolve_recycle_watermark(_mem(enabled=False)) == 0.0
-
-
-def test_recycle_watermark_disabled_without_cgroup_limit():
-    """Bare-metal/dev: no readable limit and no absolute knob -> disabled."""
-    from viaduck.main import resolve_recycle_watermark
-
-    with patch("viaduck.main._cgroup_memory_limit_gib", return_value=0.0):
-        assert resolve_recycle_watermark(_mem()) == 0.0
-
-
-def test_should_self_recycle_trips_above_watermark_after_uptime():
-    from viaduck.main import _should_self_recycle
-
-    with patch("viaduck.main._read_rss_gib", return_value=73.0):
-        started = time.monotonic() - 7200
-        assert _should_self_recycle(72.0, started, 3600.0) is True
-
-
-def test_should_self_recycle_respects_min_uptime():
-    """A young process never recycles, however hot — post-restart catch-up
-    runs high legitimately, and an eager watermark would flap-restart."""
-    from viaduck.main import _should_self_recycle
-
-    with patch("viaduck.main._read_rss_gib", return_value=95.0):
-        assert _should_self_recycle(72.0, time.monotonic(), 3600.0) is False
-
-
-def test_should_self_recycle_below_watermark_and_disabled():
-    from viaduck.main import _should_self_recycle
-
-    started = time.monotonic() - 7200
-    with patch("viaduck.main._read_rss_gib", return_value=40.0):
-        assert _should_self_recycle(72.0, started, 3600.0) is False
-    # watermark 0 = disabled: RSS is never even read
-    with patch("viaduck.main._read_rss_gib", side_effect=AssertionError("must not be called")):
-        assert _should_self_recycle(0.0, started, 3600.0) is False
-
-
-def test_should_self_recycle_never_trips_on_read_failure():
-    """/proc absent (macOS dev) or transient read error must not restart."""
-    from viaduck.main import _should_self_recycle
-
-    started = time.monotonic() - 7200
-    with patch("viaduck.main._read_rss_gib", side_effect=OSError("no procfs")):
-        assert _should_self_recycle(72.0, started, 3600.0) is False
-
-
 def test_read_rss_gib_raises_when_vmrss_absent(tmp_path):
-    """A 'successful' 0.0 read would silently disarm the watermark on Linux;
-    absence of VmRSS must surface as a failure (caught + warn-limited by
-    _should_self_recycle)."""
+    """A 'successful' 0.0 read would silently zero the per-cycle RSS gauge
+    on Linux; absence of VmRSS must surface as a failure (the gauge export
+    catches + keeps the last value)."""
     from unittest.mock import mock_open
 
     from viaduck.main import _read_rss_gib