diff --git a/README.md b/README.md index 96bde88..dfd300b 100644 --- a/README.md +++ b/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 diff --git a/docs/runbook-offset-reset.md b/docs/runbook-offset-reset.md index 570e363..c7245e9 100644 --- a/docs/runbook-offset-reset.md +++ b/docs/runbook-offset-reset.md @@ -80,9 +80,9 @@ kubectl -n argocd patch application --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 diff --git a/log-consumer-proposal.md b/log-consumer-proposal.md index 73b458f..674300d 100644 --- a/log-consumer-proposal.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index bbfc224..f08fe9a 100644 --- a/pyproject.toml +++ b/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 diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 5db4b24..e2dbe76 100644 --- a/tests/unit/test_config.py +++ b/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(): diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 03a7542..b63d79c 100644 --- a/tests/unit/test_main.py +++ b/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 diff --git a/viaduck/config.py b/viaduck/config.py index 0769e54..f848e38 100644 --- a/viaduck/config.py +++ b/viaduck/config.py @@ -397,33 +397,15 @@ def __post_init__(self): @dataclass(frozen=True) class MemoryConfig: - # Watermark self-recycle: the shipped ducklake extension still accrues - # untracked native memory (~2.5-4 GiB/h on prod; see hypothesis-1.md - # residual) with a horizon of roughly a day per pod. A mid-flight OOM - # rewinds every destination to its durable cursor and tips the - # cursor-group system into the degraded scattered regime (2026-07-31 and - # 2026-08-14 incidents); a CLEAN exit after drain() leaves cursors tight - # at the read position, so the kubelet restart resumes with no rewind. - # When RSS crosses the watermark, finish the current poll cycle, drain, - # exit 0. - self_recycle_enabled: bool = True - # Watermark as a fraction of the cgroup memory limit. Used when - # self_recycle_rss_gib is 0 and the limit is readable; if the limit is - # unreadable/unlimited the recycle is disabled (logged at startup). - # SIZING: the watermark must clear the deployment's LEGITIMATE peak — - # roughly delivery.buffer_total_max_bytes + the pool's native footprint - # + process baseline — or leak-free load recycles the pod every - # min-uptime. On sized deployments prefer the absolute knob, set from - # that envelope plus headroom (prod-us: ~74GiB envelope on a 96Gi pod - # → 0.75 x limit is BELOW it; the chart sets self_recycle_rss_gib - # explicitly instead). - self_recycle_rss_fraction: float = 0.75 - # Absolute watermark override in GiB; 0 derives from the fraction. - self_recycle_rss_gib: float = 0.0 - # Never recycle a young process: a post-restart catch-up legitimately - # runs hot, and a too-eager watermark would flap-restart into the exact - # churn this feature exists to avoid. - self_recycle_min_uptime_seconds: float = 3600.0 + # History signpost: the watermark self-recycle (drain + exit 0 at an RSS + # watermark) was REMOVED 2026-09-02 (PR #85) after the dest-connection + # age sweep held prod flat ~19h against a prior ~70-90min seppuku + # cadence. Do not re-add casually — the revert pointer is + # `git revert 1440dc3 f34627e`, and it exists only for a leak class that + # dodges the dest pool. The Duckling component keeps its own + # RSS-limit exit (single_destination.py _maybe_recycle): one connection + # per pod, nothing to sweep — the pod is the unit of recycling there. + # # EXPERIMENTAL source-connection recycle (the Leak-2 discriminating # test, persistent_oom.md §The discriminating experiment): 0 disables; # >0 closes and reopens the long-lived SOURCE catalog connection at a @@ -433,6 +415,8 @@ class MemoryConfig: # instantaneous before/after (allocator slack confounds the latter — # see delivery-endgame C3 [R1-2]); viaduck_rss_bytes is exported every # cycle so the slope is measurable from metrics alone. + # FALSIFIED 2026-09-02 (RSS slope unaffected, twice — persistent_oom.md); + # kept as a default-off bisection probe for future leak classes. source_conn_recycle_interval_seconds: float = 0.0 # Dest-connection age sweep (persistent_oom.md Leak-2): force-evict # pooled destination connections older than this many seconds — the @@ -444,21 +428,10 @@ class MemoryConfig: # dest-scope for the prod leak rate. The falsifier (does a 60s sweep # zero the slope?) is the next measurement. 600s default bounds the # storm layer hypothesis-2 measured (~460MB/90s/contended connection) - # at ~10 min of accumulation per connection. 0 disables; the watermark - # self-recycle above stays the backstop for anything the sweep misses. + # at ~10 min of accumulation per connection. 0 disables. dest_conn_max_age_seconds: float = 600.0 def __post_init__(self): - if not 0.0 < self.self_recycle_rss_fraction < 1.0: - raise ConfigError( - f"memory.self_recycle_rss_fraction must be in (0, 1), got {self.self_recycle_rss_fraction}" - ) - if self.self_recycle_rss_gib < 0: - raise ConfigError(f"memory.self_recycle_rss_gib must be >= 0, got {self.self_recycle_rss_gib}") - if self.self_recycle_min_uptime_seconds < 0: - raise ConfigError( - f"memory.self_recycle_min_uptime_seconds must be >= 0, got {self.self_recycle_min_uptime_seconds}" - ) if self.source_conn_recycle_interval_seconds < 0: raise ConfigError( f"memory.source_conn_recycle_interval_seconds must be >= 0, " @@ -1082,12 +1055,25 @@ def load(path: str | Path) -> ViaduckConfig: server_raw = raw.get("server", {}) server = ServerConfig(port=server_raw.get("port", 8000)) - memory_raw = raw.get("memory", {}) + memory_raw = raw.get("memory", {}) or {} + # Retired with the watermark self-recycle (PR #85, 2026-09-02): WARN- + # ignored, never refused — prod's chart still carries these keys, so a + # refusal would CrashLoop the rollout, but silence would let an operator + # believe the 82GiB preemption bound exists when it doesn't. The dest- + # connection age sweep (dest_conn_max_age_seconds) is the bound now. + for retired in ( + "self_recycle_enabled", + "self_recycle_rss_fraction", + "self_recycle_rss_gib", + "self_recycle_min_uptime_seconds", + ): + if retired in memory_raw: + log.warning( + "config: memory.%s was removed with the watermark self-recycle (PR #85); " + "the dest-connection age sweep is the memory bound now; ignoring it", + retired, + ) memory = MemoryConfig( - self_recycle_enabled=bool(memory_raw.get("self_recycle_enabled", True)), - self_recycle_rss_fraction=float(memory_raw.get("self_recycle_rss_fraction", 0.75)), - self_recycle_rss_gib=float(memory_raw.get("self_recycle_rss_gib", 0.0)), - self_recycle_min_uptime_seconds=float(memory_raw.get("self_recycle_min_uptime_seconds", 3600.0)), source_conn_recycle_interval_seconds=float( os.environ.get( "SOURCE_CONN_RECYCLE_INTERVAL_SECONDS", diff --git a/viaduck/main.py b/viaduck/main.py index 95b66f3..478a66c 100644 --- a/viaduck/main.py +++ b/viaduck/main.py @@ -1031,10 +1031,6 @@ def run(cfg: config.ViaduckConfig) -> None: shutdown = False - recycle_watermark_gib = resolve_recycle_watermark(cfg.memory) - loop_started_at = time.monotonic() - recycling = False - def _signal_handler(signum, frame): nonlocal shutdown log.info("Received signal %s, shutting down", signal.Signals(signum).name) @@ -1178,22 +1174,6 @@ def _site(stat) -> str: log.exception("Fatal error in poll cycle") break - # Watermark self-recycle: preempt the residual-leak OOM with a clean - # exit through the SAME graceful path SIGTERM takes — drain() flushes - # every buffer so cursors land tight at the read position, and the - # kubelet restarts a fresh process with no rewind and no - # cursor-group scatter. - if not shutdown and _should_self_recycle( - recycle_watermark_gib, loop_started_at, cfg.memory.self_recycle_min_uptime_seconds - ): - # Best-effort metric: the process exits shortly after and the - # counter resets, so a scrape can miss it. The durable signals - # are the [SELF-RECYCLE] WARN and the container's last-state - # `Completed`/exit 0 (vs `OOMKilled`/137). - metrics.self_recycles_total.inc() - recycling = True - shutdown = True - if not shutdown: # Chunked sleep so SIGTERM is honored within ~1s rather than # waiting up to `interval_seconds`. With long poll intervals (e.g. @@ -1204,20 +1184,13 @@ def _site(stat) -> str: # Graceful shutdown: flush everything buffered (the spec's # shutdown-trigger FlushStart), wait for workers, then close. # - # The recycle path gets a longer drain budget than SIGTERM: SIGTERM is - # bounded by terminationGracePeriodSeconds (kubelet SIGKILLs at the - # deadline, so a drain longer than the grace just dies mid-close), but a - # self-recycle has NO grace clock — nothing external is killing the - # process — so it can afford to flush a fat catch-up buffer instead of - # abandoning it to a cursor rewind. The bound that DOES apply here is - # the liveness probe: with the poll loop stopped, /healthz goes stale - # after ~300s poll-age plus the probe's 10x30s failure budget (~600s - # total), which is also the backstop that reaps a flush worker wedged - # in a native call (the interpreter's exit-join would otherwise wait - # forever — kubelet SIGKILL via failed liveness is the way out of that - # already-pathological state). + # The drain budget is bounded by terminationGracePeriodSeconds (kubelet + # SIGKILLs at the deadline, so a drain longer than the grace just dies + # mid-close). A flush worker wedged in a native call is reaped the same + # way via the liveness probe (the interpreter's exit-join would + # otherwise wait forever). log.info("Shutting down...") - delivery.drain(timeout_s=_RECYCLE_DRAIN_TIMEOUT_S if recycling else 60.0) + delivery.drain(timeout_s=60.0) dest_pool.close_all() state_mgr.close() if feed_reader is not None: @@ -2151,9 +2124,9 @@ def log_startup_memory() -> None: def _read_rss_gib() -> float: """Process RSS in GiB from /proc; raises on platforms without procfs. - A missing VmRSS line raises too: on Linux with the watermark armed that - is abnormal, and returning 0.0 would silently disable the recycle while - looking like a successful read. + A missing VmRSS line raises too: on Linux that is abnormal, and + returning 0.0 would silently zero the per-cycle viaduck_rss_bytes gauge + while looking like a successful read. """ for line in open("/proc/self/status"): if line.startswith("VmRSS:"): @@ -2161,92 +2134,6 @@ def _read_rss_gib() -> float: raise RuntimeError("VmRSS not present in /proc/self/status") -def _cgroup_memory_limit_gib() -> float: - """Container memory limit in GiB from the cgroup (v2 then v1); 0 when - unreadable or unlimited ("max" / the v1 no-limit sentinel).""" - for path in ("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"): - try: - raw = open(path).read().strip() - except OSError: - continue - if raw == "max": - return 0.0 - limit = int(raw) / 1024**3 - # cgroup v1 reports ~8 EiB when unlimited; anything implausibly - # large is "no limit". - return limit if limit < 4096 else 0.0 - return 0.0 - - -def resolve_recycle_watermark(cfg_memory) -> float: - """Resolve the self-recycle RSS watermark in GiB; 0 = disabled. - - Logged once at startup so the effective threshold (and why it is what it - is) is always on the record: absolute knob wins, else fraction x cgroup - limit, else disabled when no limit is readable (bare-metal/dev runs). - """ - if not cfg_memory.self_recycle_enabled: - log.info("Self-recycle disabled by config") - return 0.0 - if cfg_memory.self_recycle_rss_gib > 0: - log.info("Self-recycle watermark: %.1fGiB (absolute)", cfg_memory.self_recycle_rss_gib) - return cfg_memory.self_recycle_rss_gib - limit = _cgroup_memory_limit_gib() - if limit <= 0: - log.info("Self-recycle disabled: no cgroup memory limit readable and no absolute watermark set") - return 0.0 - watermark = limit * cfg_memory.self_recycle_rss_fraction - log.info( - "Self-recycle watermark: %.1fGiB (%.0f%% of %.1fGiB cgroup limit)", - watermark, - cfg_memory.self_recycle_rss_fraction * 100, - limit, - ) - return watermark - - -# RSS read failures never trip a restart, but a PERSISTENT failure with the -# watermark armed means the recycle is silently dark (the leak then runs to -# OOM unpreempted) — worth a rate-limited WARN. Same count-and-log-every-Nth -# shape as _MEM_STATS_FAILURES; at the ~5s cycle interval, every 360th ≈ -# one line per half hour. -_RSS_READ_FAILURES = 0 -_RSS_READ_LOG_EVERY_NTH_FAILURE = 360 - -# Drain budget for the self-recycle path. Must stay comfortably under the -# liveness reap window (~600s: 300s poll-age staleness + 10x30s probe -# failures) so a healthy long drain is never killed mid-flush. -_RECYCLE_DRAIN_TIMEOUT_S = 300.0 - - -def _should_self_recycle(watermark_gib: float, started_at: float, min_uptime_s: float) -> bool: - """One cheap check per poll cycle. Failures never trip a restart.""" - global _RSS_READ_FAILURES - if watermark_gib <= 0 or time.monotonic() - started_at < min_uptime_s: - return False - try: - rss = _read_rss_gib() - except Exception: - _RSS_READ_FAILURES += 1 - if _RSS_READ_FAILURES == 1 or _RSS_READ_FAILURES % _RSS_READ_LOG_EVERY_NTH_FAILURE == 0: - log.warning( - "[SELF-RECYCLE] RSS read failed (occurrence #%d) — watermark armed but not checking", - _RSS_READ_FAILURES, - exc_info=True, - ) - return False - if rss < watermark_gib: - return False - log.warning( - "[SELF-RECYCLE] rss=%.1fGiB >= watermark %.1fGiB; draining and exiting 0 for a clean restart. " - "Whatever the drain flushes lands tight; anything past the drain deadline re-reads from " - "persisted cursors — still strictly less rewind than the mid-flight OOM this preempts", - rss, - watermark_gib, - ) - return True - - def _log_watermark_paused(kind: str) -> None: """Rate-limited buffer-watermark WARN. Fires once when the stall starts, then a heartbeat every `_WATERMARK_HEARTBEAT_INTERVAL_S` for as long as diff --git a/viaduck/metrics.py b/viaduck/metrics.py index be4aab7..dc9fa0b 100644 --- a/viaduck/metrics.py +++ b/viaduck/metrics.py @@ -25,11 +25,6 @@ def remove(self, *label_values): "Poll cycles executed", ["pipeline"], ) -_self_recycles_total = Counter( - "viaduck_self_recycles_total", - "Clean watermark-triggered restarts (drain + exit 0 on RSS watermark)", - ["pipeline"], -) _source_conn_recycles_total = Counter( "viaduck_source_conn_recycles_total", "Source catalog connection close/reopen cycles (Leak-2 experiment; memory.source_conn_recycle_interval_seconds)", @@ -467,7 +462,6 @@ def remove(self, *label_values): # --- Public names (replaced by init() with pipeline-bound instances) --- polls_total = _polls_total -self_recycles_total = _self_recycles_total source_conn_recycles_total = _source_conn_recycles_total dest_conn_sweeps_total = _dest_conn_sweeps_total rss_bytes = _rss_bytes @@ -556,7 +550,7 @@ def init(pipeline: str): global polls_total, cdc_read_seconds, cdc_rows_read_total, source_snapshot_id global cdc_feed_query_seconds, cdc_feed_files_total, cdc_feed_inlined_rows_total, cdc_feed_replans_total global read_clusters, read_pool_inflight - global self_recycles_total, source_conn_recycles_total, rss_bytes + global source_conn_recycles_total, rss_bytes global dest_conn_sweeps_total, arrow_pool_allocated_bytes, arrow_pool_max_bytes global source_columns_excluded_total global dest_write_seconds, dest_rows_written_total, dest_last_snapshot_id, dest_lag_snapshots @@ -630,7 +624,6 @@ def init(pipeline: str): # Metrics with no other labels — pre-label to get direct .inc()/.set()/.observe() polls_total = _polls_total.labels(pipeline=pipeline) - self_recycles_total = _self_recycles_total.labels(pipeline=pipeline) source_conn_recycles_total = _source_conn_recycles_total.labels(pipeline=pipeline) dest_conn_sweeps_total = _dest_conn_sweeps_total.labels(pipeline=pipeline) rss_bytes = _rss_bytes.labels(pipeline=pipeline)