diff --git a/docs/adr/0247-worker-cgroup-memory-evidence.md b/docs/adr/0247-worker-cgroup-memory-evidence.md new file mode 100644 index 000000000..4982052f9 --- /dev/null +++ b/docs/adr/0247-worker-cgroup-memory-evidence.md @@ -0,0 +1,72 @@ +# ADR 0247: Worker cgroup memory evidence before capacity limits + +- Status: Accepted +- Date: 2026-08-27 + +## Context + +The canonical worker was once observed with exit code 137 and was later +recreated healthy. Exit 137 establishes a `SIGKILL`, not its cause. Recreation +also discards the prior container's Docker state and cgroup counters, so the +new container's `OOMKilled=false` cannot disprove a historical OOM. + +The base Compose service has no worker-specific memory limit or reservation. +Docker therefore exposes the Docker Desktop VM capacity, not an accepted +worker capacity envelope. Setting `mem_limit` from the current idle footprint, +an arbitrary percentage, or an undocumented headroom multiplier would turn an +unrepresentative observation into a production failure boundary. + +## Decision + +`scripts/capture_worker_memory_evidence.py` is the canonical worker-memory +measurement procedure. It captures two snapshots around an explicitly chosen +representative workload window from the unchanged `lineageweave` worker: + +- Docker status, exit code, `OOMKilled`, restart count, and configured memory + limit/reservation; +- cgroup v2 `memory.current`, `memory.peak`, `memory.max`, and the keyed local + event counters in `memory.events.local`. + +The procedure rejects a container replacement, unavailable cgroup v2 +evidence, decreasing counters, and non-positive windows. It classifies OOM as +confirmed only when Docker records `OOMKilled` or the kernel's local +`oom_kill` counter increases. Exit 137 without either signal remains +`sigkill_unattributed`. `high`, `max`, or `oom` deltas establish memory +pressure without inventing an OOM kill. + +The core `low`, `high`, `max`, `oom`, and `oom_kill` counters are required. +`oom_group_kill` is recorded when the host exposes it, but its absence remains +an explicit `null` delta because neither OOM confirmation nor pressure +classification depends on that optional group counter. + +If the unchanged worker exits during the window, Compose discovery includes +stopped containers and Docker inspection preserves its terminal state. The +terminated cgroup is no longer readable, so ending current usage and event +deltas remain `null`; the output retains only the peak captured before exit +and labels that limited scope. Docker `OOMKilled` may still confirm OOM and an +otherwise unattributed exit 137 remains distinguishable. Every other terminal +state without ending cgroup evidence is rejected rather than classified. + +No observation emits a memory-limit proposal. `memory.peak` is a measured +maximum for that cgroup lifetime, but neither Docker nor the kernel defines a +universal safety margin that turns it into a safe hard limit. A future limit +requires an accepted representative workload/capacity envelope and a separate +decision that names the workload, concurrency, host capacity, observation +window, zero-OOM acceptance, and rollback procedure. Disabling the OOM killer +is prohibited. + +## Consequences + +- Operators must capture evidence before recreating a failed worker. +- A healthy idle sample proves only that the sampled window had no new local + pressure events; it is not capacity acceptance. +- Canonical Compose remains unchanged until representative workload evidence + supports a bounded configuration. + +## References + +Docker, Inc. (2026a). *Define services in Docker Compose*. https://docs.docker.com/reference/compose-file/services/ + +Docker, Inc. (2026b). *Resource constraints*. https://docs.docker.com/engine/containers/resource_constraints/ + +The Linux Kernel Organization. (2026). *Control group v2*. https://docs.kernel.org/admin-guide/cgroup-v2.html diff --git a/docs/adr/README.md b/docs/adr/README.md index 67e621ec2..56dbb3dcc 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -30,6 +30,8 @@ decision from them. | Product semantic catalog and typed evidence relations | [0228](0228-evidence-bound-product-semantic-catalog.md) | | Source-preserving voice semantic taxonomy | [0244](0244-source-preserving-voice-semantic-taxonomy.md) | | Expanded Voice-of-X post lookup and ontology | [0246](0246-expanded-voice-of-x-post-taxonomy.md) | +| Worker cgroup memory evidence | [0247](0247-worker-cgroup-memory-evidence.md) | +| [`WORKER_CGROUP_MEMORY_REFERENCES.md`](../doctoring/WORKER_CGROUP_MEMORY_REFERENCES.md) | [0247](0247-worker-cgroup-memory-evidence.md) | [0011](0011-prov-o-standard-relations.md) and [0065](0065-prov-o-provenance-boundary.md) cite the dated W3C PROV-O and PROV-DM Recommendations (https://www.w3.org/TR/2013/REC-prov-o-20130430/ and https://www.w3.org/TR/2013/REC-prov-dm-20130430/). diff --git a/docs/doctoring/WORKER_CGROUP_MEMORY_REFERENCES.md b/docs/doctoring/WORKER_CGROUP_MEMORY_REFERENCES.md new file mode 100644 index 000000000..e9fe9c9d7 --- /dev/null +++ b/docs/doctoring/WORKER_CGROUP_MEMORY_REFERENCES.md @@ -0,0 +1,22 @@ +# Worker cgroup memory references + +This supporting register documents the evidence boundary adopted by ADR 0247. +Docker Compose defines `mem_limit` as a hard allocation limit and +`mem_reservation` as a reservation. Docker Engine documents that the kernel +kills container processes on OOM by default and warns against disabling that +behavior without a hard memory limit. Linux cgroup v2 defines `memory.peak` as +the maximum observed usage and `memory.events.local` as the non-hierarchical +counter source; `oom_kill` counts processes killed by an OOM killer. + +These contracts do not specify a universal multiplier or percentage for +turning one observed peak into a safe service limit. LineageWeave therefore +records measured evidence and leaves the limit unset until a representative +capacity acceptance is approved. + +## References — APA 7th + +Docker, Inc. (2026a). *Define services in Docker Compose*. https://docs.docker.com/reference/compose-file/services/ + +Docker, Inc. (2026b). *Resource constraints*. https://docs.docker.com/engine/containers/resource_constraints/ + +The Linux Kernel Organization. (2026). *Control group v2*. https://docs.kernel.org/admin-guide/cgroup-v2.html diff --git a/docs/operability/postgresql-observed-tuning.md b/docs/operability/postgresql-observed-tuning.md index 5433548d4..84d24b6f2 100644 --- a/docs/operability/postgresql-observed-tuning.md +++ b/docs/operability/postgresql-observed-tuning.md @@ -46,3 +46,20 @@ uv run python scripts/plan_postgres_tuning.py rollback \ The base `docker-compose.yml` contains no tuned command. Removing the tuning overlay and recreating PostgreSQL is the secondary rollback path. + +## Non-identifying canonical observation — 2026-08-27 + +Since the 2026-08-24 statistics reset, the canonical PostgreSQL 16 instance +reported 25,308 requested checkpoints versus 382 timed checkpoints, 336.7 GB +of WAL, 7,598,680 `wal_buffers_full` events, 81,194,401 backend buffer writes, +and no lock waiter at capture. The running configuration retained +`wal_level=replica`, `max_wal_size=1GB`, and `shared_buffers=128MB` under read +committed isolation. + +This snapshot confirms severe cumulative pressure, not an apply value. +PostgreSQL documents that `max_wal_size` pressure can start a checkpoint before +`checkpoint_timeout`, that high WAL output can require more WAL buffers, and +that its own WAL recycling estimate adapts to prior checkpoint cycles. Run the +aligned planner across the representative write workload before applying its +segment-aligned proposal. The snapshot supplies no evidence for changing +`shared_buffers`, durability, isolation, or storage concurrency. diff --git a/docs/operability/worker-memory-evidence.md b/docs/operability/worker-memory-evidence.md new file mode 100644 index 000000000..1a65cb1b4 --- /dev/null +++ b/docs/operability/worker-memory-evidence.md @@ -0,0 +1,23 @@ +# Worker memory evidence procedure + +Run this before restarting or recreating a worker, over a declared window that +contains the workload and concurrency being accepted: + +```bash +uv run python scripts/capture_worker_memory_evidence.py \ + --sample-seconds "$OBSERVATION_SECONDS" \ + --output /tmp/lineageweave-worker-memory-evidence.json +``` + +The output contains aggregates and no container identifier or record content. +Preserve it outside git with the workload definition and host capacity. An +`oom_confirmed` result requires Docker `OOMKilled` or a local kernel +`oom_kill` delta. `sigkill_unattributed` requires further host/runtime logs; +do not relabel it OOM. A container change invalidates the window. If the same +container exits, ending cgroup values remain unavailable and the retained +pre-exit peak is labeled as such; it is not a whole-window maximum. + +Acceptance requires the declared representative workload to finish on one +unchanged container with zero `high`, `max`, `oom`, and `oom_kill` deltas. +The observed peak is evidence, not a proposed Compose limit. Any future +`mem_limit`/`mem_reservation` change needs a separate ADR and rollback test. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index c64fd0a9b..62007ddee 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,17 +1,23 @@ # Product & Technical Gap Baseline -> Dashboard delivery snapshot: 2026-08-27 01:44 KST. Protected `main` was +> Dashboard delivery snapshot: 2026-08-27. Protected `main` was > `ff7431bd1851c03e737808d22c6a2d43968582f9`. Dashboard PR #640 exact -> observed head was `5594029c801263a7f629c287ce41580ecf4e0739`; this branch is not -> protected-main release evidence. The queue contained 29 open PRs (22 -> `BLOCKED`, five `UNSTABLE`, two `CLEAN`) and no exact-head approval. PR #715 +> observed head is `b3befa8bec8dd2807994444299df7eafdd1c7781`; this branch is not +> protected-main release evidence. The queue contained 35 open PRs (23 +> `BLOCKED`, six `UNSTABLE`, four `CLEAN`, and two `UNKNOWN`) and no exact-head +> approval. PR #715 > merged normally into #640 and repaired the four stale HTTP transport test > doubles plus one Python-before-3.7 Semgrep false positive that contradicted -> the repository's Python >=3.12 contract. Stacked PR #722 pre-documentation -> head `eed7cabd` restores semantic-query and opt-in public-verification -> factories in the dedicated Ask worker and the production-equivalent -> concurrent-migration fixture path; its focused evidence is 46 unit tests and -> one live Keycloak/PostgreSQL public-verification integration test. +> the repository's Python >=3.12 contract. PR #722 also merged normally into +> #640, restoring semantic-query and opt-in public-verification factories in +> the dedicated Ask worker and the production-equivalent concurrent-migration +> fixture path. PR #727 merged normally as `353dfd01`, replacing general-reader +> implementation wording with evidence actions. Worker evidence PR #725's +> implementation exact is `40283c9b`; subsequent evidence-only refreshes do not +> change that code and its fresh hosted gates are pending. Storybook follow-up +> #730 merged normally at `b3befa8b` from exact head `c21cac5d`. +> Current #640 required checks are queued and independent approval remains +> absent, so the candidate stays blocked. ## Operations Dashboard PRD/TRD traceability @@ -25,8 +31,8 @@ the current #640 head. The canonical containers currently return HTTP 200 from backend `/healthz` and the frontend root, but their Compose labels do not prove the source commit; therefore neither the running stack nor the historical k6 run is exact-head authenticated acceptance. Exact-head desktop/mobile -screenshots and k6 remain required after #722 is incorporated and #640 is -rebuilt. Historical test projects are retired only by their exact Compose +screenshots and k6 remain required after #640 is rebuilt from its current +exact head. Historical test projects are retired only by their exact Compose project label and without named-volume deletion. PR #678 implementation head `da98de07` fixes the default project name; its follow-up exact-label audit also removed the remaining identifiable isolated test containers while preserving @@ -45,9 +51,10 @@ named volumes. | Similar VOC, customer cohort, prior action | Persisted repeat-issue candidate semantics plus orchestrator pair adjudication and extractive evidence | Candidate live post endpoint and post-detail UI implemented; authenticated runtime acceptance pending | | TEPP independent Event Lineage anchor | Accepted, persisted TEPP criterion bound to exact snapshot/cutoff before fast-mlsirm activation | Consumer PR #606 is on protected main; TEPP producer PR #237 remains open, so no end-to-end accepted artifact is release evidence yet | | Temporal Lineage topics and multilevel important posts | ADR 0210; TEPP posterior topic/plausible-value contract followed by fast-mlsirm observed-information case-deletion influence | This stacked candidate adds normalized persistence, exact run/snapshot/cutoff binding, pre-aggregation scope authorization, API diagnostics, and populated/unavailable Storybook surfaces. TEPP PR #247 remains open at `063f10f3`; stacked #251–#254 provide fail-closed input validation, full joint precision, deterministic joint plausible-value draws, and the canonical research register, while complete provenance assembly remains gated. fast-mlsirm PR #1418 validates the Rust consumer envelope but intentionally returns `EstimatorUnavailable` until the scientific estimator lands. Runtime therefore remains honestly unavailable with no local Python or fallback score. | -| PostgreSQL WAL/checkpoint pressure | ADR 0227; aligned two-snapshot `pg_stat_wal`/checkpoint deltas, PostgreSQL WAL-segment and checkpoint constraints, cgroup memory, and data-volume space | Candidate procedure emits a content-authenticated plan and Compose environment, retains unmeasured memory/I/O/compression settings, preserves durability, validates the overlay without mutation, rejects stale preconditions, and requires an approved service recreation for apply or rollback. The observed CPU-bound GIN scan remains distinct from historical checkpoint pressure; canonical runtime application waits for the active migration to complete. | +| PostgreSQL WAL/checkpoint and worker memory pressure | ADR 0227 and ADR 0247; aligned PostgreSQL counter deltas plus unchanged-container Docker/cgroup v2 evidence | Candidate PostgreSQL procedure emits a content-authenticated restart/rollback plan while retaining unmeasured settings and durability. The current cumulative counters establish sustained historical WAL/checkpoint pressure but do not replace a representative aligned apply window. A prior worker exit 137 is not attributable after container recreation: the current healthy worker has no configured service memory limit/reservation, and a one-second non-identifying observation showed an approximately 109 MiB cgroup lifetime peak with no new local pressure/OOM event. That idle window is not capacity acceptance. Capture the declared representative workload before recreation; do not add a limit or headroom multiplier until that evidence supports a separately accepted capacity boundary. | -Customer-copy audit at #640 exact `c142c4ea` retained the ADR-required +Customer-copy audit began at #640 exact `c142c4ea` and was delivered by #727 +at merge `353dfd01`; it retained the ADR-required measurement-administrator terms and explicit ontology/provenance inspection labels. Two general-reader gaps were isolated: Customer Master explained the ontology/semantic implementation boundary instead of the evidence action, and @@ -55,7 +62,9 @@ Global Ask called authorized workspace evidence "internal" posts. The stacked copy repair tells the reader to compare the source identifier with related posts and organization evidence, and reuses the authorized-citation action. Five-locale consistency, rendered component tests, and desktop/narrow -Storybook scenes cover the repair. `프로젝트별 관측 Event` and ADR 0210's exact +Storybook scenes cover the Customer Master repair. Follow-up #730 renders the +Global Ask no-public-claim result and its next action in desktop and narrow +scenes delivered at merge `b3befa8b`. `프로젝트별 관측 Event` and ADR 0210's exact `model influence` estimand name remain unchanged. ### Technical contract and flow diff --git a/scripts/capture_worker_memory_evidence.py b/scripts/capture_worker_memory_evidence.py new file mode 100755 index 000000000..e7d1dc2ad --- /dev/null +++ b/scripts/capture_worker_memory_evidence.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +"""Capture and compare non-identifying worker cgroup memory evidence.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import time +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +SERVICE = "backend-worker" +PROJECT = "lineageweave" +REQUIRED_EVENT_KEYS = ("low", "high", "max", "oom", "oom_kill") +OPTIONAL_EVENT_KEYS = ("oom_group_kill",) + + +class MemoryEvidenceError(ValueError): + """Reject incomplete or incomparable worker memory evidence.""" + + +def _integer(value: Any, field: str) -> int: + """Return a non-negative integer evidence field.""" + try: + result = int(value) + except (TypeError, ValueError) as exc: + raise MemoryEvidenceError(f"{field} must be an integer") from exc + if result < 0: + raise MemoryEvidenceError(f"{field} must not be negative") + return result + + +def parse_flat_keys(value: str) -> dict[str, int]: + """Parse a cgroup flat-keyed file without relying on line positions.""" + result: dict[str, int] = {} + for line in value.splitlines(): + parts = line.split() + if len(parts) != 2: + raise MemoryEvidenceError("invalid cgroup flat-key evidence") + try: + result[parts[0]] = int(parts[1]) + except ValueError as exc: + raise MemoryEvidenceError("invalid cgroup flat-key evidence") from exc + return result + + +def _events(snapshot: Mapping[str, Any]) -> Mapping[str, Any]: + """Return the required local cgroup memory-event mapping.""" + events = snapshot.get("memory_events_local") + if not isinstance(events, Mapping): + raise MemoryEvidenceError("memory.events.local is unavailable") + missing = [key for key in REQUIRED_EVENT_KEYS if key not in events] + if missing: + raise MemoryEvidenceError( + "memory.events.local is missing required keys: " + ", ".join(missing) + ) + return events + + +def compare_snapshots( + before: Mapping[str, Any], after: Mapping[str, Any], *, elapsed_seconds: float +) -> dict[str, Any]: + """Classify one unchanged-container observation without proposing a limit.""" + if elapsed_seconds <= 0: + raise MemoryEvidenceError("elapsed_seconds must be positive") + if not before.get("container_started_at") or ( + before.get("container_started_at") != after.get("container_started_at") + ): + raise MemoryEvidenceError("container changed during the observation") + peak = after.get("memory_peak_bytes") + if peak is None and after.get("memory_events_local") is not None: + raise MemoryEvidenceError("memory.peak is unavailable") + if peak is None: + observed_peak = _integer(before.get("memory_peak_bytes"), "memory_peak_bytes") + peak_scope = "before_terminal_exit" + else: + observed_peak = _integer(peak, "memory_peak_bytes") + peak_scope = "cgroup_lifetime_at_window_end" + current = after.get("memory_current_bytes") + ending_current = ( + None if current is None else _integer(current, "memory_current_bytes") + ) + docker_oom_killed = bool(after.get("container_oom_killed")) + exit_code = _integer(after.get("container_exit_code", 0), "container_exit_code") + before_events = _events(before) + deltas: dict[str, int | None] | None = None + if after.get("memory_events_local") is not None: + after_events = _events(after) + deltas = {} + for key in REQUIRED_EVENT_KEYS: + earlier = _integer(before_events[key], f"memory.events.local.{key}") + later = _integer(after_events[key], f"memory.events.local.{key}") + if later < earlier: + raise MemoryEvidenceError(f"memory.events.local.{key} decreased") + deltas[key] = later - earlier + for key in OPTIONAL_EVENT_KEYS: + if key not in before_events or key not in after_events: + deltas[key] = None + continue + earlier = _integer(before_events[key], f"memory.events.local.{key}") + later = _integer(after_events[key], f"memory.events.local.{key}") + if later < earlier: + raise MemoryEvidenceError(f"memory.events.local.{key} decreased") + deltas[key] = later - earlier + elif not docker_oom_killed and exit_code != 137: + raise MemoryEvidenceError("ending cgroup evidence is unavailable") + + if docker_oom_killed or (deltas is not None and deltas["oom_kill"] > 0): + classification = "oom_confirmed" + elif exit_code == 137: + classification = "sigkill_unattributed" + elif deltas is not None and any(deltas[key] for key in ("high", "max", "oom")): + classification = "memory_pressure_observed" + else: + classification = "observed_without_memory_pressure" + + return { + "contract_version": 1, + "elapsed_seconds": elapsed_seconds, + "classification": classification, + "observed_peak_bytes": observed_peak, + "observed_peak_scope": peak_scope, + "ending_current_bytes": ending_current, + "configured_memory_limit_bytes": after.get("memory_limit_bytes"), + "configured_memory_reservation_bytes": after.get("memory_reservation_bytes"), + "event_deltas": deltas, + # A representative peak does not establish safe headroom. Operators must + # not turn it into a Compose limit by adding an undocumented multiplier. + "memory_limit_proposal": None, + } + + +def _run(command: Sequence[str], *, timeout: float = 15) -> str: + """Run one bounded local command and return standard output.""" + try: + completed = subprocess.run( + list(command), text=True, capture_output=True, check=False, timeout=timeout + ) + except subprocess.TimeoutExpired as exc: + raise MemoryEvidenceError("container evidence command timed out") from exc + if completed.returncode: + raise MemoryEvidenceError(completed.stderr.strip() or "container evidence command failed") + return completed.stdout.strip() + + +def capture_snapshot() -> dict[str, Any]: + """Capture Docker state and cgroup v2 counters for the canonical worker.""" + container_ids = [ + value + for value in _run( + ["docker", "compose", "-p", PROJECT, "ps", "--all", "-q", SERVICE] + ).splitlines() + if value + ] + if not container_ids: + raise MemoryEvidenceError("canonical backend-worker container is unavailable") + if len(container_ids) != 1: + raise MemoryEvidenceError( + "canonical backend-worker evidence requires exactly one container" + ) + container_id = container_ids[0] + inspected = json.loads(_run(["docker", "inspect", container_id])) + if not isinstance(inspected, list) or len(inspected) != 1: + raise MemoryEvidenceError("Docker inspection is incomplete") + state = inspected[0].get("State", {}) + host = inspected[0].get("HostConfig", {}) + if not isinstance(state, Mapping) or not isinstance(host, Mapping): + raise MemoryEvidenceError("Docker state is incomplete") + if not state.get("StartedAt") or not isinstance(state.get("Status"), str): + raise MemoryEvidenceError("Docker state is incomplete") + if state.get("Status") != "running": + return { + "captured_at": datetime.now(UTC).isoformat(), + "container_started_at": state.get("StartedAt"), + "container_status": state.get("Status"), + "container_oom_killed": bool(state.get("OOMKilled")), + "container_exit_code": _integer( + state.get("ExitCode", 0), "container_exit_code" + ), + "container_restart_count": _integer( + inspected[0].get("RestartCount", 0), "container_restart_count" + ), + "memory_limit_bytes": _integer( + host.get("Memory", 0), "memory_limit_bytes" + ) + or None, + "memory_reservation_bytes": ( + _integer( + host.get("MemoryReservation", 0), "memory_reservation_bytes" + ) + or None + ), + "memory_current_bytes": None, + "memory_peak_bytes": None, + "memory_max_bytes": None, + "memory_events_local": None, + } + cgroup = _run( + [ + "docker", "exec", container_id, "sh", "-eu", "-c", + ( + "test -r /sys/fs/cgroup/memory.current; " + "test -r /sys/fs/cgroup/memory.peak; " + "test -r /sys/fs/cgroup/memory.max; " + "test -r /sys/fs/cgroup/memory.events.local; " + "cat /sys/fs/cgroup/memory.current; " + "cat /sys/fs/cgroup/memory.peak; " + "cat /sys/fs/cgroup/memory.max; " + "cat /sys/fs/cgroup/memory.events.local" + ), + ] + ).splitlines() + if len(cgroup) < 4: + raise MemoryEvidenceError("cgroup v2 memory evidence is incomplete") + memory_max = None if cgroup[2] == "max" else _integer(cgroup[2], "memory.max") + events = parse_flat_keys("\n".join(cgroup[3:])) + return { + "captured_at": datetime.now(UTC).isoformat(), + "container_started_at": state.get("StartedAt"), + "container_status": state.get("Status"), + "container_oom_killed": bool(state.get("OOMKilled")), + "container_exit_code": _integer(state.get("ExitCode", 0), "container_exit_code"), + "container_restart_count": _integer( + inspected[0].get("RestartCount", 0), "container_restart_count" + ), + "memory_limit_bytes": _integer(host.get("Memory", 0), "memory_limit_bytes") or None, + "memory_reservation_bytes": ( + _integer(host.get("MemoryReservation", 0), "memory_reservation_bytes") or None + ), + "memory_current_bytes": _integer(cgroup[0], "memory.current"), + "memory_peak_bytes": _integer(cgroup[1], "memory.peak"), + "memory_max_bytes": memory_max, + "memory_events_local": events, + } + + +def observe(sample_seconds: float) -> dict[str, Any]: + """Capture an explicitly sized same-container observation window.""" + if sample_seconds <= 0: + raise MemoryEvidenceError("sample_seconds must be positive") + before = capture_snapshot() + started = time.monotonic() + time.sleep(sample_seconds) + after = capture_snapshot() + result = compare_snapshots( + before, after, elapsed_seconds=time.monotonic() - started + ) + result["before_captured_at"] = before["captured_at"] + result["after_captured_at"] = after["captured_at"] + return result + + +def main(argv: Sequence[str] | None = None) -> int: + """Capture one worker-memory observation as non-identifying JSON.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--sample-seconds", type=float, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args(argv) + args.output.write_text( + json.dumps(observe(args.sample_seconds), indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/test_worker_memory_evidence.py b/tests/test_worker_memory_evidence.py new file mode 100644 index 000000000..d87dd1f35 --- /dev/null +++ b/tests/test_worker_memory_evidence.py @@ -0,0 +1,365 @@ +"""Worker cgroup memory evidence contracts.""" + +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +from pathlib import Path +from types import ModuleType, SimpleNamespace + +import pytest + +_ROOT = Path(__file__).resolve().parents[1] +_SCRIPT = _ROOT / "scripts" / "capture_worker_memory_evidence.py" + + +def _load_module() -> ModuleType: + spec = importlib.util.spec_from_file_location("capture_worker_memory_evidence", _SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +worker_memory = _load_module() + + +def _snapshot(**changes: object) -> dict[str, object]: + value: dict[str, object] = { + "container_started_at": "2026-08-27T00:00:00Z", + "container_status": "running", + "container_oom_killed": False, + "container_exit_code": 0, + "container_restart_count": 0, + "memory_limit_bytes": None, + "memory_reservation_bytes": None, + "memory_current_bytes": 80 * 1024 * 1024, + "memory_peak_bytes": 120 * 1024 * 1024, + "memory_max_bytes": None, + "memory_events_local": { + "low": 0, + "high": 0, + "max": 0, + "oom": 0, + "oom_kill": 0, + "oom_group_kill": 0, + }, + } + value.update(changes) + return value + + +def test_compare_confirms_only_kernel_or_docker_oom_evidence() -> None: + before = _snapshot() + after = _snapshot( + container_status="exited", + container_oom_killed=True, + container_exit_code=137, + memory_events_local={ + "low": 0, + "high": 0, + "max": 1, + "oom": 1, + "oom_kill": 1, + "oom_group_kill": 0, + }, + ) + + evidence = worker_memory.compare_snapshots(before, after, elapsed_seconds=60) + + assert evidence["classification"] == "oom_confirmed" + assert evidence["event_deltas"]["oom_kill"] == 1 + assert evidence["observed_peak_bytes"] == 120 * 1024 * 1024 + assert evidence["memory_limit_proposal"] is None + + +def test_compare_does_not_call_exit_137_an_oom() -> None: + evidence = worker_memory.compare_snapshots( + _snapshot(), + _snapshot(container_status="exited", container_exit_code=137), + elapsed_seconds=60, + ) + + assert evidence["classification"] == "sigkill_unattributed" + + +def test_compare_accepts_representative_window_without_pressure() -> None: + evidence = worker_memory.compare_snapshots( + _snapshot(), + _snapshot(memory_peak_bytes=160 * 1024 * 1024), + elapsed_seconds=60, + ) + + assert evidence["classification"] == "observed_without_memory_pressure" + assert evidence["memory_limit_proposal"] is None + + +def test_compare_rejects_container_replacement_and_counter_reset() -> None: + with pytest.raises(worker_memory.MemoryEvidenceError, match="container changed"): + worker_memory.compare_snapshots( + _snapshot(), + _snapshot(container_started_at="2026-08-27T00:01:00Z"), + elapsed_seconds=60, + ) + with pytest.raises(worker_memory.MemoryEvidenceError, match="decreased"): + worker_memory.compare_snapshots( + _snapshot( + memory_events_local={ + "low": 0, + "high": 0, + "max": 0, + "oom": 0, + "oom_kill": 1, + "oom_group_kill": 0, + } + ), + _snapshot(), + elapsed_seconds=60, + ) + + +def test_compare_rejects_invalid_window_or_missing_evidence() -> None: + with pytest.raises(worker_memory.MemoryEvidenceError, match="elapsed_seconds"): + worker_memory.compare_snapshots(_snapshot(), _snapshot(), elapsed_seconds=0) + with pytest.raises(worker_memory.MemoryEvidenceError, match="memory.peak"): + worker_memory.compare_snapshots( + _snapshot(), _snapshot(memory_peak_bytes=None), elapsed_seconds=1 + ) + + +def test_parse_flat_keys_uses_names_not_line_positions() -> None: + assert worker_memory.parse_flat_keys("oom_kill 2\nlow 1\n") == { + "oom_kill": 2, + "low": 1, + } + with pytest.raises(worker_memory.MemoryEvidenceError, match="invalid cgroup"): + worker_memory.parse_flat_keys("oom_kill nope\n") + with pytest.raises(worker_memory.MemoryEvidenceError, match="invalid cgroup"): + worker_memory.parse_flat_keys("oom_kill\n") + + +def test_integer_and_event_validation_fail_closed() -> None: + for value, message in (("bad", "integer"), (-1, "negative")): + with pytest.raises(worker_memory.MemoryEvidenceError, match=message): + worker_memory._integer(value, "field") + with pytest.raises(worker_memory.MemoryEvidenceError, match="events.local"): + worker_memory.compare_snapshots( + _snapshot(memory_events_local=None), _snapshot(), elapsed_seconds=1 + ) + missing_key_events = dict(_snapshot()["memory_events_local"]) + del missing_key_events["oom_kill"] + with pytest.raises(worker_memory.MemoryEvidenceError, match="required keys"): + worker_memory.compare_snapshots( + _snapshot(memory_events_local=missing_key_events), + _snapshot(), + elapsed_seconds=1, + ) + + +def test_compare_preserves_unavailable_optional_group_oom_counter() -> None: + before_events = dict(_snapshot()["memory_events_local"]) + after_events = dict(_snapshot()["memory_events_local"]) + del before_events["oom_group_kill"] + del after_events["oom_group_kill"] + + evidence = worker_memory.compare_snapshots( + _snapshot(memory_events_local=before_events), + _snapshot(memory_events_local=after_events), + elapsed_seconds=1, + ) + + assert evidence["classification"] == "observed_without_memory_pressure" + assert evidence["event_deltas"]["oom_group_kill"] is None + + decreasing_before = dict(_snapshot()["memory_events_local"]) + decreasing_before["oom_group_kill"] = 1 + after_events["oom_group_kill"] = 0 + with pytest.raises(worker_memory.MemoryEvidenceError, match="decreased"): + worker_memory.compare_snapshots( + _snapshot(memory_events_local=decreasing_before), + _snapshot(memory_events_local=after_events), + elapsed_seconds=1, + ) + + +def test_compare_reports_pressure_without_claiming_oom() -> None: + after = _snapshot( + memory_events_local={ + "low": 0, + "high": 1, + "max": 0, + "oom": 0, + "oom_kill": 0, + "oom_group_kill": 0, + } + ) + evidence = worker_memory.compare_snapshots(_snapshot(), after, elapsed_seconds=1) + assert evidence["classification"] == "memory_pressure_observed" + + +def test_run_is_bounded_and_reports_failures(monkeypatch) -> None: + monkeypatch.setattr( + worker_memory.subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(returncode=0, stdout=" ok \n", stderr=""), + ) + assert worker_memory._run(["command"]) == "ok" + monkeypatch.setattr( + worker_memory.subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(returncode=1, stdout="", stderr="bad"), + ) + with pytest.raises(worker_memory.MemoryEvidenceError, match="bad"): + worker_memory._run(["command"]) + monkeypatch.setattr( + worker_memory.subprocess, + "run", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + subprocess.TimeoutExpired("command", 1) + ), + ) + with pytest.raises(worker_memory.MemoryEvidenceError, match="timed out"): + worker_memory._run(["command"]) + + +def test_capture_snapshot_reads_docker_and_keyed_cgroup_evidence(monkeypatch) -> None: + inspection = json.dumps( + [ + { + "State": { + "StartedAt": "2026-08-27T00:00:00Z", + "Status": "running", + "OOMKilled": False, + "ExitCode": 0, + }, + "HostConfig": {"Memory": 1024, "MemoryReservation": 512}, + "RestartCount": 1, + } + ] + ) + outputs = iter( + [ + "container-id", + inspection, + "100\n200\n300\nlow 0\nhigh 0\nmax 0\noom 0\noom_kill 0\noom_group_kill 0\n", + ] + ) + monkeypatch.setattr(worker_memory, "_run", lambda *_args, **_kwargs: next(outputs)) + + snapshot = worker_memory.capture_snapshot() + + assert snapshot["memory_current_bytes"] == 100 + assert snapshot["memory_peak_bytes"] == 200 + assert snapshot["memory_max_bytes"] == 300 + assert snapshot["memory_limit_bytes"] == 1024 + assert snapshot["container_restart_count"] == 1 + + +@pytest.mark.parametrize( + ("oom_killed", "exit_code", "classification"), + [(True, 137, "oom_confirmed"), (False, 137, "sigkill_unattributed")], +) +def test_capture_and_compare_classify_worker_that_exits_mid_window( + monkeypatch, oom_killed: bool, exit_code: int, classification: str +) -> None: + inspection = json.dumps( + [ + { + "State": { + "StartedAt": "2026-08-27T00:00:00Z", + "Status": "exited", + "OOMKilled": oom_killed, + "ExitCode": exit_code, + }, + "HostConfig": {"Memory": 0, "MemoryReservation": 0}, + "RestartCount": 0, + } + ] + ) + outputs = iter(["container-id", inspection]) + monkeypatch.setattr(worker_memory, "_run", lambda *_args, **_kwargs: next(outputs)) + + after = worker_memory.capture_snapshot() + evidence = worker_memory.compare_snapshots(_snapshot(), after, elapsed_seconds=1) + + assert evidence["classification"] == classification + assert evidence["observed_peak_bytes"] == 120 * 1024 * 1024 + assert evidence["observed_peak_scope"] == "before_terminal_exit" + assert evidence["ending_current_bytes"] is None + assert evidence["event_deltas"] is None + + +def test_compare_rejects_other_exit_without_ending_cgroup_evidence() -> None: + after = _snapshot( + container_status="exited", + container_exit_code=1, + memory_current_bytes=None, + memory_peak_bytes=None, + memory_max_bytes=None, + memory_events_local=None, + ) + with pytest.raises(worker_memory.MemoryEvidenceError, match="ending cgroup"): + worker_memory.compare_snapshots(_snapshot(), after, elapsed_seconds=1) + + +@pytest.mark.parametrize( + ("outputs", "message"), + [ + ([""], "unavailable"), + (["id-one\nid-two"], "exactly one"), + (["id", "[]"], "inspection"), + (["id", '[{"State": [], "HostConfig": {}}]'], "state"), + ( + ["id", '[{"State": {"Status": "running"}, "HostConfig": {}}]'], + "state", + ), + ( + [ + "id", + ( + '[{"State": {"StartedAt": "start", "Status": "running"}, ' + '"HostConfig": {}, "RestartCount": 0}]' + ), + "1\n2\nmax", + ], + "cgroup v2", + ), + ], +) +def test_capture_snapshot_rejects_incomplete_boundaries( + monkeypatch, outputs: list[str], message: str +) -> None: + values = iter(outputs) + monkeypatch.setattr(worker_memory, "_run", lambda *_args, **_kwargs: next(values)) + with pytest.raises(worker_memory.MemoryEvidenceError, match=message): + worker_memory.capture_snapshot() + + +def test_observe_and_main_write_non_identifying_result(monkeypatch, tmp_path: Path) -> None: + snapshots = iter( + [ + {**_snapshot(), "captured_at": "before"}, + {**_snapshot(memory_peak_bytes=130 * 1024 * 1024), "captured_at": "after"}, + ] + ) + clocks = iter([10.0, 12.0]) + sleeps: list[float] = [] + monkeypatch.setattr(worker_memory, "capture_snapshot", lambda: next(snapshots)) + monkeypatch.setattr(worker_memory.time, "monotonic", lambda: next(clocks)) + monkeypatch.setattr(worker_memory.time, "sleep", sleeps.append) + + result = worker_memory.observe(2) + + assert sleeps == [2] + assert result["before_captured_at"] == "before" + assert result["after_captured_at"] == "after" + with pytest.raises(worker_memory.MemoryEvidenceError, match="sample_seconds"): + worker_memory.observe(0) + + output = tmp_path / "evidence.json" + monkeypatch.setattr(worker_memory, "observe", lambda _seconds: result) + assert worker_memory.main(["--sample-seconds", "2", "--output", str(output)]) == 0 + assert json.loads(output.read_text())["classification"] == result["classification"]