From 2cba82efabbcf84f7af2f47518c5ec94bcdec48b Mon Sep 17 00:00:00 2001 From: "david.yapfest.club" Date: Sat, 17 Oct 2026 02:02:46 -0400 Subject: [PATCH] Stream bounded CI progress and enforce owned cleanup deadlines --- docs/CI.md | 168 +++- tests/python/test_ci_workflow.py | 1040 +++++++++++++++++++++- tests/python/test_release_local.py | 81 ++ tools/run_ci.py | 1310 +++++++++++++++++++++++++--- tools/verify_release_local.py | 102 ++- 5 files changed, 2546 insertions(+), 155 deletions(-) diff --git a/docs/CI.md b/docs/CI.md index df80579..7f31371 100644 --- a/docs/CI.md +++ b/docs/CI.md @@ -10,8 +10,11 @@ files under `.tangled/workflows` and the helper they call, `tools/run_ci.py`. > 4 GiB RAM, two vCPUs and a 24 GiB writable volume; admission includes a 96 MiB > runner allowance and the server permits 30 minutes. These are allocations, > not measured peak usage. `check.yml` now uses the same observed image and -> guest dependencies, but has not passed hosted execution. Emulator and physical -> acceptance remain separate. Local success is not hosted-CI evidence. +> guest dependencies, but has not passed hosted execution. The helper spends at +> most 1500 of those seconds (section 6), so a cold container build plus the +> whole gate may end in a bounded budget failure instead of a verdict. Emulator +> and physical acceptance remain separate. Local success is not hosted-CI +> evidence. ## 1. What exists @@ -106,6 +109,10 @@ python3 tools/run_ci.py emulator --require-vice-version 3.8 python3 tools/run_ci.py package ``` +Every action also accepts `--budget-seconds N` for a directly invoked run. It +can only shorten the 1500s budget, never extend it; the budget is derived from +the workflow step ceiling below. + ### `smoke` Builds the prepared image, then runs inside it: @@ -196,13 +203,156 @@ runtime harness on the guest. It never runs the `release` preflight mode and never invokes the publisher (`tools/publish_tangled_artifacts.py`). Publication remains a separate, separately-approved step. -### Bounded timeouts and logs - -Every command carries a bounded timeout (quick checks get seconds, cold builds -get tens of minutes) and is killed if it exceeds it, which is reported as a -failure rather than a hang. Command output is **spooled to a regular temp log**, -and only a bounded head and tail are read back, so a noisy command cannot -exhaust memory and the log stays bounded. +### Deadlines, logs, and what a bounded failure means + +One monotonic budget covers a whole action. It is created once, spent as the +action runs, never reset by a stage, and every child timeout is the smaller of +that stage's own ceiling and the time still left. + +| Limit | Seconds | Meaning | +| --- | --- | --- | +| workflow step ceiling | 1800 | the server's admission limit for one step; unchanged by this scaffold | +| reserve inside that ceiling | 300 | guest setup, teardown and image removal | +| `CI_BUDGET_SECONDS` | 1500 | the helper's budget for one action | +| container handoff margin | 30 | the inner helper's budget is the `docker run` timeout minus this | +| minimum usable remainder | 5 | below this no further stage is started | +| cleanup reserve | 60 | the most any one command may spend stopping and removing what it owns | + +Per-command ceilings, all at or below the 1500s budget: validate 120, guest +docker probe 15, VICE version query 120, build-environment descriptor 120, +first-run demo 300, `make demo` 900, `npm ci` 900, debug build 900, release +check 600, package 1200, runtime harness 1200, and 1500 for `make check-all`, +the release build, the VICE acceptance run, `docker build` and `docker run`. + +So the deadlines nest: the innermost stage ceiling first, then what is left of +the helper budget, then the outer `docker run`, then the workflow step. The +container is given a budget 30s shorter than the outer `docker run` timeout, so +the helper that knows which stage it was prints the expiry before the outer +process group is terminated. + +When a command exceeds its bound, the helper prints that it is about to clean up +**before** it starts the risky part, so a run killed during cleanup still leaves +the reason in the log. It then creates **one** monotonic cleanup deadline of +60s and spends it on everything that follows: the cooperative `SIGTERM`, the +forced `SIGKILL`, the reap, the output drain, and the container listing, +removal and confirmation. Every wait and every Docker call inside that window is +capped by the time still left on it, and every poll is truncated to it, so no +step can run past the deadline. A step that no longer fits is skipped and +reported instead. A timed-out stage is never retried. + +The stop is only *claimed* when the exit was confirmed. If the group could not +be reaped inside the deadline — a process stuck in an uninterruptible sleep, for +instance — the helper reports `cleanup unconfirmed` and says that anything the +stage started in the Docker daemon may still exist. A reaped process is also +**not** treated as proof that its containers are gone: resource cleanup is +reported on its own evidence. + +The Docker calls the cleanup makes run through the same runner, and they are +handed the *same* deadline rather than a fresh allowance, so a listing or +removal that itself times out spends the time that is left — including its own +stop and drain — and cannot extend the cleanup past its end. + +What this cannot promise: if the helper *itself* is killed with `SIGKILL`, no +cleanup code runs at all. Nothing here claims otherwise; a container left that +way is identifiable by its label (see below) and is not swept by a later run. + +When the budget itself runs out the helper prints the stage it was about to +run and how little time was left, and that the stage and every later one +did not run. That is a **bounded failure, not a stage verdict**: a cold +container build plus the whole gate may not fit 1500s, and the honest result is +that the remaining stages were not attempted. Raising the workflow duration to +fit a cold run is a separate change that needs its own approval; this helper +never raises its own limit, and nothing here changes the workflow files. + +Command output is **streamed live** as the child writes it, so a step that is +still running is visible and a workflow that kills the helper still leaves the +log behind. Volume is bounded twice: + +- at most 256 KiB is echoed live per command; +- at most a 64 KiB head and a 64 KiB tail are retained in memory for the + failure diagnostic, and the number of bytes withheld is always reported. + +Nothing is spooled to an unbounded temp file, and nothing is printed twice: the +diagnostic prints only the retained tail that the live echo did not already +show, which is also what keeps a nested log (the inner `_container` helper +inside the outer `docker run`) from being replayed. Output is decoded +incrementally, so a multi-byte character split across two reads is echoed whole; +the returned head and tail are cut at byte offsets and can still split a +character at the truncation point, which is why the withheld byte count is +reported. The helper never prints the environment. + +### Container ownership and teardown + +Killing the local `docker` client does not stop the container the daemon +started, and `--rm` only removes a container that exits, so neither one is +enough on its own. Every container this helper starts is created with both +ownership marks: + +- a unique name, `ultimate-tool-ci---<12 hex>`; +- the labels `ultimate-tool-ci.run=-<12 hex>` and + `ultimate-tool-ci.stage=`. + +The token is generated per invocation, so no two runs can address each other's +containers. It is a namespace, not a credential: it selects containers, and it +authorises nothing. + +Removal happens in a `finally` block, independently of whether the client +exited cleanly, failed, or was killed, and inside the single cleanup deadline: + +1. list the containers carrying *this* invocation's label + (`docker ps --all --quiet --no-trunc --filter label=ultimate-tool-ci.run=`); +2. remove exactly the identifiers that listing printed + (`docker rm --force ...`), accepting only values that match an + identifier pattern and never a guess; +3. re-probe until nothing carrying the token remains, then report `removed`. + +The outcome is `removed`, `absent`, or `unconfirmed`, and only `removed` and +`absent` are treated as confirmed. `unconfirmed` is an error (exit 3), not a +warning, and it is what you get when: + +- the daemon cannot be asked, or its listing is truncated or contains something + that is not an identifier (fail closed: nothing is removed on a guess); +- a `docker rm` returns an ambiguous status — it is **never retried**, because a + second attempt could remove a container that appeared after the first; +- the process that owned the containers could not be confirmed stopped; +- the command timed out at all. A confirmed, unforced stop is a clean *process* + fact, not a clean *resource* fact: the run may have been killed with a + container half-created; +- an observation window closes without ever seeing a container. A container + killed mid-create can appear later, so "nothing listed yet" is a bounded + observation, not proof of absence. + +There is **no `docker system prune`, no filter wildcard, and no cross-invocation +cleanup**: the only identifiers this can touch are ones the daemon printed for +this invocation's own random label. A label prefix a teardown may filter on must +be on the helper's allowlist, so a typo cannot turn into a broad sweep. + +### The runtime harness's own containers + +`tools/verify_release_local.py` labels the containers it starts with the +key-only label `ultimate-tool-verify-local.`. Its id is normally minted +inside its own process, which the caller cannot know — so the CI action supplies +one and makes the division of ownership explicit: + +- `--invocation-id `: the caller-chosen id. It is validated (1-64 + characters of `a-z`, `0-9` and `-`) **before any journal, scratch directory, + image or container exists**, so an unacceptable id cannot leave a resource + behind under a name it chose. Omitted, the harness generates its own id + exactly as before. +- `--external-container-cleanup`: requires `--mode runtime` and an explicit + `--invocation-id`. In this mode the supervisor is the **sole remover**: the + harness skips its three container-removal paths (pre-clean, the registered + cleanup action, and the post-run sweep) and keeps its read-only checks and its + scratch-directory cleanup. Two removers would make every `docker rm` result + ambiguous, which is exactly what this switch prevents. Standalone runs are + unchanged. + +The CI helper therefore passes both flags, keeps the harness's scratch `HOME` +alive until the teardown has finished, and removes exactly +`ultimate-tool-verify-local.` inside the same cleanup deadline that +stopped the harness. A harness that had to be forced, or whose stop could not be +confirmed, leaves its resource cleanup explicitly unconfirmed even when the +listing is empty. ## 7. Source manifest gate diff --git a/tests/python/test_ci_workflow.py b/tests/python/test_ci_workflow.py index 07b8147..f7baa77 100644 --- a/tests/python/test_ci_workflow.py +++ b/tests/python/test_ci_workflow.py @@ -13,10 +13,16 @@ scalars, and booleans. There is no dependency on PyYAML. from __future__ import annotations +import contextlib +import io +import itertools import json import os +import signal +import subprocess import sys import tempfile +import time import unittest from unittest import mock from pathlib import Path @@ -231,7 +237,7 @@ def all_commands(): for action in run_ci.CONTAINER_ACTIONS: commands.extend(run_ci.container_steps(action)) commands.extend(run_ci.emulator_steps(dummy)) - commands.extend(run_ci.runtime_steps()) + commands.extend(run_ci.runtime_steps("1234-abcd")) commands.extend(run_ci.release_steps()) return commands @@ -415,7 +421,7 @@ class StepGuardTests(unittest.TestCase): run_ci.assert_safe_steps(run_ci.container_steps("check")) run_ci.assert_safe_steps(run_ci.container_steps("package")) run_ci.assert_safe_steps(run_ci.emulator_steps("/usr/bin/x64sc")) - run_ci.assert_safe_steps(run_ci.runtime_steps()) + run_ci.assert_safe_steps(run_ci.runtime_steps("1234-abcd")) def test_every_command_has_a_bounded_timeout(self): for command in all_commands(): @@ -484,26 +490,993 @@ class StepExecutionTests(unittest.TestCase): self.assertEqual(caught.exception.code, 3) -class BoundedOutputTests(unittest.TestCase): - def _spool(self, text: str) -> str: - handle, path = tempfile.mkstemp(prefix="run_ci_test_", suffix=".log") - with os.fdopen(handle, "wb") as fp: - fp.write(text.encode("utf-8")) - self.addCleanup(lambda: os.path.exists(path) and os.unlink(path)) - return path +class OutputRetainerTests(unittest.TestCase): + """The bounded live echo and the bounded retained head and tail.""" + + def test_small_output_is_whole_and_echoed_exactly_once(self): + sink = io.StringIO() + retainer = run_ci.OutputRetainer(stream=sink, live_bytes=1000, head_bytes=1000, tail_bytes=1000) + retainer.feed(b"hello world\n") + retainer.finish() + self.assertEqual(retainer.text(), "hello world\n") + self.assertFalse(retainer.suppressed) + retainer.report("stage") + # the live echo and the diagnostic never both print the same line + self.assertEqual(sink.getvalue().count("hello world"), 1) + self.assertIn("all shown above", sink.getvalue()) + + def test_large_output_retains_a_bounded_head_and_tail(self): + sink = io.StringIO() + retainer = run_ci.OutputRetainer( + stream=sink, live_bytes=200, head_bytes=400, tail_bytes=400 + ) + for _ in range(5000): + retainer.feed(b"x" * 10) + retainer.finish() + text = retainer.text() + self.assertLessEqual(len(text), 400 + 400 + 200) + self.assertIn("truncated", text) + self.assertTrue(text.startswith("x")) + self.assertTrue(text.endswith("x")) + self.assertEqual(retainer.total, 50000) + self.assertEqual(retainer.echoed, 200) + self.assertTrue(retainer.suppressed) + + def test_diagnostic_only_prints_the_part_the_live_echo_did_not_show(self): + sink = io.StringIO() + retainer = run_ci.OutputRetainer( + stream=sink, live_bytes=200, head_bytes=400, tail_bytes=400 + ) + for _ in range(5000): + retainer.feed(b"x" * 10) + retainer.finish() + echoed = sink.getvalue() + retainer.report("stage") + diagnostic = sink.getvalue()[len(echoed):] + # the retained tail is 400 bytes, the echo covered the first 200 + self.assertIn("not shown live", diagnostic) + self.assertLess(len(diagnostic), 400 + 400) + # the dropped middle is reported, never silently lost + self.assertIn("not shown", diagnostic) + + def test_overlapping_head_and_tail_are_not_duplicated(self): + retainer = run_ci.OutputRetainer(head_bytes=100, tail_bytes=100, live_bytes=10) + retainer.feed(b"a" * 150) + retainer.finish() + self.assertEqual(retainer.dropped, 0) + self.assertEqual(retainer.text(), "a" * 150) + + def test_a_child_that_never_emits_a_newline_still_shows_progress(self): + sink = io.StringIO() + retainer = run_ci.OutputRetainer( + stream=sink, live_bytes=100_000, head_bytes=1000, tail_bytes=1000 + ) + retainer.feed(b"y" * (run_ci.LIVE_LINE_FLUSH_BYTES * 2)) + self.assertTrue(sink.getvalue()) + self.assertTrue(sink.getvalue().startswith("y")) + + def test_no_environment_is_ever_read_or_dumped(self): + sink = io.StringIO() + retainer = run_ci.OutputRetainer(stream=sink) + retainer.feed(b"PATH=/bin\nC64U_PASSWORD=hunter2\n") + retainer.finish() + retainer.report("stage") + # output is echoed as the child wrote it; the helper adds no environment + self.assertNotIn("environ", sink.getvalue()) + + +def _child_env() -> dict[str, str]: + return {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "CI": "true"} + + +class StreamingRunnerTests(unittest.TestCase): + """The real runner streams a child's output while it is still running.""" + + def test_short_child_output_is_streamed_before_the_child_exits(self): + with tempfile.TemporaryDirectory() as tmp: + marker = Path(tmp, "released") + script = Path(tmp, "child.py") + script.write_text( + "import pathlib, sys, time\n" + "marker = pathlib.Path(sys.argv[1])\n" + "print('first', flush=True)\n" + "deadline = time.monotonic() + 20\n" + "while not marker.exists():\n" + " if time.monotonic() > deadline:\n" + " print('blocked', flush=True)\n" + " raise SystemExit(9)\n" + " time.sleep(0.01)\n" + "print('second', flush=True)\n", + encoding="utf-8", + ) + sink = _TriggerSink(marker, "first") + with contextlib.redirect_stdout(sink): + code, output = run_ci.default_runner( + [sys.executable, "-u", str(script), str(marker)], + env=_child_env(), cwd=ROOT, timeout=30, + ) + self.assertEqual(code, 0) + # the child only prints "second" once the sink has seen "first"; if the + # runner held output until exit the child would exit 9 instead + self.assertNotIn("blocked", output) + self.assertIn("first", output) + self.assertIn("second", output) + # each line reaches the log once: no replay at exit + self.assertEqual(sink.getvalue().count("second"), 1) + + def test_large_child_output_is_bounded_on_the_stream_and_in_memory(self): + child = "for i in range(60000): print('line-%d-' % i + 'z' * 40)" + sink = io.StringIO() + with contextlib.redirect_stdout(sink): + code, output = run_ci.default_runner( + [sys.executable, "-u", "-c", child], + env=_child_env(), cwd=ROOT, timeout=120, + ) + self.assertEqual(code, 0) + live = len(sink.getvalue()) + self.assertLessEqual( + live, + run_ci.LIVE_OUTPUT_BYTES + run_ci.RETAINED_TAIL_BYTES + run_ci.LIVE_LINE_FLUSH_BYTES + 4096, + ) + self.assertLessEqual( + len(output), + run_ci.RETAINED_HEAD_BYTES + run_ci.RETAINED_TAIL_BYTES + 256, + ) + self.assertIn("truncated", output) + self.assertIn("not shown live", sink.getvalue()) + + def test_failure_status_and_output_are_returned(self): + with contextlib.redirect_stdout(io.StringIO()): + code, output = run_ci.default_runner( + [sys.executable, "-u", "-c", "import sys; print('boom'); raise SystemExit(3)"], + env=_child_env(), cwd=ROOT, timeout=30, + ) + self.assertEqual(code, 3) + self.assertIn("boom", output) + + +class _TriggerSink(io.StringIO): + """A stdout sink that releases the child once its first line appears.""" + + def __init__(self, marker: Path, needle: str) -> None: + super().__init__() + self.marker = marker + self.needle = needle + self.released = False + + def write(self, text: str) -> int: + written = super().write(text) + if not self.released and self.needle in self.getvalue(): + self.released = True + self.marker.write_text("released", encoding="utf-8") + return written + + +class UnreapableChild: + """A stand-in child that never exits and never gives up its exit status. + + It models a process stuck in an uninterruptible sleep: it cannot be reaped, + so nothing may claim it is gone. ``wait`` fails the test if it is ever + called without a bound. + """ + + def __init__(self, sink: io.StringIO) -> None: + read_fd, write_fd = os.pipe() + os.close(write_fd) # the pipe is already at end of file + self.sink = sink + self.stdout = os.fdopen(read_fd, "rb", buffering=0) + self.pid = os.getpid() + self.returncode = None + self.signals: list[int] = [] + self.bounds: list[float] = [] + self.sink_at_first_signal: str | None = None + + def poll(self): + return None + + def wait(self, timeout=None): + assert timeout is not None, "wait() must always carry a bound" + self.bounds.append(timeout) + raise subprocess.TimeoutExpired("child", timeout) + + def send_signal(self, signum): + self.signals.append(signum) + if self.sink_at_first_signal is None: + self.sink_at_first_signal = self.sink.getvalue() + + def kill(self): # pragma: no cover - only used if send_signal is unavailable + self.signals.append(signal.SIGKILL) - def test_small_log_is_returned_whole(self): - path = self._spool("hello world\n") - self.assertEqual(run_ci.read_bounded_text(path), "hello world\n") - def test_large_log_is_bounded_and_marked(self): - path = self._spool("x" * 5000) - bounded = run_ci.read_bounded_text(path, limit=1000) - self.assertLess(len(bounded), 5000) - self.assertIn("truncated", bounded) - # the head and the tail are both retained - self.assertTrue(bounded.startswith("x")) - self.assertTrue(bounded.endswith("x")) +class TimeoutCleanupTests(unittest.TestCase): + """A bounded deadline terminates the whole child process group, once.""" + + def _alive(self, pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: # pragma: no cover - not expected here + return True + return True + + def test_timeout_kills_children_and_returns_a_bounded_failure(self): + with tempfile.TemporaryDirectory() as tmp: + pid_file = Path(tmp, "pids") + child = ( + "import os, pathlib, subprocess, sys, time\n" + f"pid_file = pathlib.Path({str(pid_file)!r})\n" + "grand = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" + "pid_file.write_text('%d %d' % (os.getpid(), grand.pid))\n" + "print('started', flush=True)\n" + "time.sleep(60)\n" + ) + started = time.monotonic() + with contextlib.redirect_stdout(io.StringIO()) as sink: + code, output = run_ci.default_runner( + [sys.executable, "-u", "-c", child], + env=_child_env(), cwd=ROOT, timeout=2, + ) + elapsed = time.monotonic() - started + self.assertEqual(code, run_ci.EXIT_TIMEOUT) + self.assertLess(elapsed, 30, "the deadline must return promptly") + self.assertIn("started", output) + self.assertIn("its process group was stopped", output) + self.assertIn("its process group was stopped", sink.getvalue()) + child_pid, grandchild_pid = (int(part) for part in pid_file.read_text().split()) + for pid in (child_pid, grandchild_pid): + for _ in range(100): # bounded wait, at most ~5s + if not self._alive(pid): + break + time.sleep(0.05) + self.assertFalse(self._alive(pid), f"pid {pid} outlived its deadline") + + def test_a_child_that_closes_stdout_still_respects_the_deadline(self): + child = ( + "import os, time\n" + "os.dup2(os.open(os.devnull, os.O_WRONLY), 1)\n" + "time.sleep(60)\n" + ) + started = time.monotonic() + with contextlib.redirect_stdout(io.StringIO()): + code, _ = run_ci.default_runner( + [sys.executable, "-u", "-c", child], + env=_child_env(), cwd=ROOT, timeout=2, + ) + elapsed = time.monotonic() - started + self.assertEqual(code, run_ci.EXIT_TIMEOUT) + self.assertLess(elapsed, 30, "a closed pipe must not defeat the deadline") + + def test_a_cooperative_child_is_stopped_on_sigterm_first(self): + with tempfile.TemporaryDirectory() as tmp: + marker = Path(tmp, "released") + script = Path(tmp, "child.py") + script.write_text( + "import pathlib, signal, sys, time\n" + "marker = pathlib.Path(sys.argv[1])\n" + "def on_term(signum, frame):\n" + " marker.write_text('released', encoding='utf-8')\n" + " raise SystemExit(0)\n" + "signal.signal(signal.SIGTERM, on_term)\n" + "print('ready', flush=True)\n" + "time.sleep(60)\n", + encoding="utf-8", + ) + with contextlib.redirect_stdout(io.StringIO()): + outcome = run_ci.default_runner( + [sys.executable, "-u", str(script), str(marker)], + env=_child_env(), cwd=ROOT, timeout=2, + ) + self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) + self.assertFalse(outcome.cleanup_unconfirmed) + self.assertIn("SIGTERM", outcome.detail) + # the child ran its own release path: that is what the grace is for + self.assertTrue(marker.is_file()) + + def test_a_child_that_ignores_sigterm_is_forced_and_confirmed(self): + child = ( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "print('ready', flush=True)\n" + "time.sleep(60)\n" + ) + with contextlib.redirect_stdout(io.StringIO()): + proc = subprocess.Popen( + [sys.executable, "-u", "-c", child], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + self.addCleanup(lambda: proc.poll() is None and proc.kill()) + time.sleep(0.3) # let the child install its handler + budget = run_ci.Deadline(30, label="test cleanup") + confirmed, forced, detail = run_ci.stop_process_group(proc, budget=budget, grace=1, reap=5) + self.assertTrue(confirmed) + self.assertTrue(forced, "the escalation must be reported, not assumed") + self.assertIn("SIGKILL", detail) + self.assertIsNotNone(proc.poll()) + + def test_a_child_that_cannot_be_reaped_reports_cleanup_unconfirmed(self): + sink = io.StringIO() + created: list[UnreapableChild] = [] + + def factory(*args, **kwargs): + created.append(UnreapableChild(sink)) + return created[-1] + + with mock.patch.object(run_ci.subprocess, "Popen", factory), \ + mock.patch.object(run_ci.os, "getpgid", side_effect=ProcessLookupError), \ + contextlib.redirect_stdout(sink): + outcome = run_ci.default_runner(["/nonexistent-child"], env={}, cwd=ROOT, timeout=1) + child = created[0] + self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) + self.assertTrue(outcome.cleanup_unconfirmed) + self.assertIn("could NOT be confirmed", outcome.output) + self.assertIn("may still exist", outcome.output) + self.assertEqual(child.signals[:2], [signal.SIGTERM, signal.SIGKILL]) + # the diagnostic was printed before the process group was signalled + self.assertIn("exceeded its", child.sink_at_first_signal or "") + # and nothing waited without a bound + self.assertTrue(child.bounds) + self.assertTrue(all(bound is not None for bound in child.bounds)) + + def test_a_failing_stage_reports_cleanup_unconfirmed_through_run_checked(self): + def runner(argv, *, env, cwd, timeout=None): + return run_ci.RunOutcome(run_ci.EXIT_TIMEOUT, "", True, "still running 5s after SIGKILL") + with self.assertRaises(run_ci.CommandFailed) as caught: + run_ci.run_checked(run_ci.cmd(["slow"], 5), runner, {"CI": "true"}, stage="slow stage") + self.assertTrue(caught.exception.cleanup_unconfirmed) + self.assertIn("may still exist", str(caught.exception)) + + def test_no_wait_is_issued_without_a_bound(self): + source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") + self.assertNotIn(".wait()", source) + self.assertNotIn("wait(timeout=None)", source) + self.assertIn("def wait_bounded", source) + + def test_one_cleanup_deadline_bounds_every_step(self): + # The per-step ceilings must fit inside the single deadline that covers + # the stop, the drain and the container work. + self.assertLessEqual( + run_ci.CHILD_DRAIN_SECONDS + run_ci.CHILD_GRACE_SECONDS + run_ci.CHILD_REAP_SECONDS, + run_ci.CLEANUP_RESERVE_SECONDS, + ) + self.assertLessEqual(run_ci.T_CONTAINER_TEARDOWN, run_ci.CLEANUP_RESERVE_SECONDS) + self.assertGreater(run_ci.MIN_CLEANUP_SECONDS, 0) + + def test_a_timed_out_command_spends_no_more_than_its_cleanup_deadline(self): + child = ( + "import signal, time\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN)\n" + "print('ready', flush=True)\n" + "time.sleep(60)\n" + ) + started = time.monotonic() + with contextlib.redirect_stdout(io.StringIO()): + outcome = run_ci.default_runner( + [sys.executable, "-u", "-c", child], + env=_child_env(), cwd=ROOT, timeout=1, cleanup_seconds=3, + ) + elapsed = time.monotonic() - started + self.assertEqual(outcome.code, run_ci.EXIT_TIMEOUT) + self.assertTrue(outcome.forced) + self.assertFalse(outcome.cleanup_unconfirmed) + self.assertIsNotNone(outcome.cleanup) + self.assertLessEqual(outcome.cleanup.elapsed(), 3 + 0.5, "cleanup stayed inside its deadline") + self.assertLess(elapsed, 1 + 3 + 5, "the whole command is bound plus one cleanup deadline") + + def test_the_runner_does_not_retry_a_timed_out_stage(self): + calls = [] + def runner(argv, *, env, cwd, timeout=None): + calls.append(timeout) + return run_ci.EXIT_TIMEOUT, "" + with self.assertRaises(run_ci.CommandFailed) as caught: + run_ci.execute_steps( + [run_ci.cmd(["slow"], 5), run_ci.cmd(["later"], 5)], + runner, {"CI": "true"}, + ) + self.assertTrue(caught.exception.timed_out) + self.assertEqual(len(calls), 1) + + +class FakeClock: + def __init__(self, start: float = 0.0) -> None: + self.value = start + + def __call__(self) -> float: + return self.value + + def advance(self, seconds: float) -> None: + self.value += max(0.0, float(seconds)) + + +def cleanup_budget(clock=None, seconds: int = run_ci.CLEANUP_RESERVE_SECONDS): + """The single cleanup deadline a test hands to a teardown.""" + return run_ci.Deadline(seconds, clock=clock or time.monotonic, label="test cleanup") + + +class ClockedChild: + """A fake child that consumes the injected clock and can ignore SIGTERM. + + ``wait`` always takes a bound and spends part of it, exactly like a real + command against a slow daemon. ``returncode=None`` means the child ignores + SIGTERM and only stops when SIGKILL arrives. + """ + + def __init__(self, clock, *, output: bytes = b"", exit_code=0, wait_cost: float = 0.0) -> None: + read_fd, write_fd = os.pipe() + if output: + os.write(write_fd, output) + os.close(write_fd) + self.clock = clock + self.stdout = os.fdopen(read_fd, "rb", buffering=0) + self.pid = os.getpid() + self.returncode = exit_code + self.wait_cost = wait_cost + self.signals: list[int] = [] + self.waits: list[object] = [] + + def poll(self): + return self.returncode + + def wait(self, timeout=None): + assert timeout is not None, "wait() must always carry a bound" + offered = float(timeout) + self.waits.append(offered) + if self.returncode is None: + self.clock.advance(offered) # still running: it uses what it was given + raise subprocess.TimeoutExpired("child", timeout) + self.clock.advance(min(offered, self.wait_cost)) + return self.returncode + + def send_signal(self, signum): + self.signals.append(signum) + if signum == signal.SIGKILL: + self.returncode = -signal.SIGKILL + self.wait_cost = 2.0 # a killed child still takes a moment to reap + + +class FakeDaemon: + """A fake Docker daemon whose containers outlive the client that made them. + + ``visible_after_seconds`` models a client killed while the daemon was still + creating the container: nothing is listable until that much simulated time + has passed, so a single probe right after the kill sees nothing. + ``call_seconds`` charges every successful list/remove call against the + injected clock, the way a slow daemon consumes a real deadline. + """ + + def __init__(self, *, clock=time.monotonic, advance=None, visible_after_seconds: float = 0.0, + run_status: int = 0, call_seconds: float = 0.0) -> None: + self.clock = clock + self.advance = advance or (lambda seconds: None) + self.visible_after_seconds = visible_after_seconds + self.run_status = run_status + self.call_seconds = call_seconds + self.containers: dict[str, dict] = {} + self.calls: list[tuple[list[str], object]] = [] + self.removed: list[str] = [] + self.selectors: list[str] = [] + + def start(self, selector: str, name: str = "container") -> str: + """Register a container that exists independently of any client.""" + identifier = os.urandom(8).hex() + self.containers[identifier] = { + "name": name, + "label": selector, + "visible_at": self.clock() + self.visible_after_seconds, + } + return identifier + + def _charge(self) -> None: + if self.call_seconds: + self.advance(self.call_seconds) + + # -- the fake docker CLI --------------------------------------------- + def runner(self, argv, *, env, cwd, timeout=None): + argv = list(argv) + self.calls.append((argv, timeout)) + if "ps" in argv: + selector = argv[argv.index("--filter") + 1][len("label="):] + self.selectors.append(selector) + listed = "".join( + f"{identifier}\n" + for identifier, info in self.containers.items() + if info["label"] == selector and self.clock() >= info["visible_at"] + ) + self._charge() + return 0, listed + if "rm" in argv: + identifiers = argv[argv.index("--force") + 1:] + status = 0 + for identifier in identifiers: + if identifier in self.containers: + self.removed.append(self.containers.pop(identifier)["name"]) + else: + status = 1 + self._charge() + return status, "" + if "run" in argv: + name = argv[argv.index("--name") + 1] + labels = [argv[index + 1] for index, value in enumerate(argv) if value == "--label"] + self.start(labels[0], name) + return self.run_status, "" + return 0, "" + + def stages(self): + """The stage commands, without the teardown probes.""" + return [(argv, timeout) for argv, timeout in self.calls if "ps" not in argv] + + +class ContainerOwnershipTests(unittest.TestCase): + """M1: killing the client must not orphan the daemon's container.""" + + def test_the_container_is_created_with_a_unique_name_and_label(self): + first = run_ci.new_ownership_token() + second = run_ci.new_ownership_token() + self.assertNotEqual(first, second) + name = f"{run_ci.CONTAINER_NAME_PREFIX}-check-{first}" + argv = run_ci.docker_run_argv( + "check", 600, name=name, + labels=[f"{run_ci.CONTAINER_LABEL_KEY}={first}", + f"{run_ci.CONTAINER_STAGE_LABEL_KEY}=check"], + ) + self.assertEqual(argv[argv.index("--name") + 1], name) + labels = [argv[index + 1] for index, value in enumerate(argv) if value == "--label"] + self.assertIn(f"{run_ci.CONTAINER_LABEL_KEY}={first}", labels) + self.assertIn(f"{run_ci.CONTAINER_STAGE_LABEL_KEY}=check", labels) + # the ownership marks are this invocation's, not a fixed name + other = run_ci.docker_run_argv( + "check", 600, name=f"{run_ci.CONTAINER_NAME_PREFIX}-check-{second}", + labels=[f"{run_ci.CONTAINER_LABEL_KEY}={second}"], + ) + self.assertNotEqual(argv[argv.index("--name") + 1], other[other.index("--name") + 1]) + + def test_the_two_label_shapes_are_allowlisted_and_the_runtime_one_is_key_only(self): + token = "1234-abcd" + self.assertEqual(run_ci.ci_owned_label(token).selector(), f"{run_ci.CONTAINER_LABEL_KEY}={token}") + # the harness labels its containers with a key only, not key=value + self.assertEqual( + run_ci.runtime_owned_label(token).selector(), + f"{run_ci.RUNTIME_LABEL_KEY_PREFIX}.{token}", + ) + with self.assertRaises(run_ci.ConfigError): + run_ci.OwnedLabel("some-other-project", token) + with self.assertRaises(run_ci.ConfigError): + # an unsafe token is refused before any docker command is built + run_ci.ci_owned_label("../../etc") + + def test_the_runtime_label_prefix_is_the_harness_constant_and_the_token_is_valid(self): + # The source builder changes its imported harness's label namespace. + # Load this contract independently, as a runtime subprocess would. + import importlib.util + + name = "runtime_ownership_contract_under_test" + spec = importlib.util.spec_from_file_location(name, TOOLS / "verify_release_local.py") + harness = importlib.util.module_from_spec(spec) + with mock.patch.dict(sys.modules, {name: harness}): + spec.loader.exec_module(harness) + self.assertEqual(run_ci.RUNTIME_LABEL_KEY_PREFIX, harness.CONTAINER_LABEL) + token = run_ci.new_ownership_token() + self.assertEqual(harness.validate_invocation_id(token), token) + + def test_teardown_removes_exactly_the_container_this_invocation_labelled(self): + daemon = FakeDaemon() + token = "1234-abcd" + daemon.start(run_ci.ci_owned_label(token).selector(), "owned") + # another invocation's container must survive untouched + daemon.start(run_ci.ci_owned_label("9999-ffff").selector(), "other") + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label(token), runner=daemon.runner, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "removed") + self.assertEqual(daemon.removed, ["owned"]) + self.assertEqual(len(daemon.containers), 1, "only the owned container may be removed") + self.assertEqual([info["name"] for info in daemon.containers.values()], ["other"]) + self.assertEqual(daemon.selectors[0], run_ci.ci_owned_label(token).selector()) + + def test_teardown_never_prunes_and_only_addresses_identifiers_it_was_given(self): + daemon = FakeDaemon() + token = "5555-eeee" + daemon.start(run_ci.ci_owned_label(token).selector(), "owned") + run_ci.teardown_owned_containers( + run_ci.ci_owned_label(token), runner=daemon.runner, env={}, budget=cleanup_budget() + ) + for argv, _ in daemon.calls: + self.assertNotIn("prune", argv) + self.assertNotIn("system", argv) + self.assertNotIn("container", argv) + for part in argv: + self.assertNotIn("$(", part) + self.assertNotIn("*", part) + listing = [argv for argv, _ in daemon.calls if "ps" in argv][0] + self.assertIn("--no-trunc", listing, "full identifiers, so removal names the same container") + + def test_teardown_follows_a_delayed_start_inside_the_same_deadline(self): + clock = FakeClock() + # the daemon registers the container 35s after the client was killed + daemon = FakeDaemon(clock=clock, visible_after_seconds=35) + token = "7777-1111" + daemon.start(run_ci.ci_owned_label(token).selector(), "late") + budget = cleanup_budget(clock) + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label(token), runner=daemon.runner, env={}, + budget=budget, settle=True, sleep=clock.advance, + ) + self.assertEqual(result.outcome, "removed") + self.assertGreaterEqual(clock.value, 35, "the observation covered a delayed start") + self.assertLessEqual(budget.elapsed(), run_ci.CLEANUP_RESERVE_SECONDS) + self.assertEqual(daemon.containers, {}) + + def test_teardown_reports_absent_without_touching_anything(self): + daemon = FakeDaemon() + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("nothing-here"), runner=daemon.runner, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "absent") + self.assertTrue(result.confirmed) + self.assertEqual([argv for argv, _ in daemon.calls if "rm" in argv], []) + + def test_a_listing_that_cannot_be_trusted_is_reported_not_used(self): + def unreachable(argv, *, env, cwd, timeout=None): + return 1, "Cannot connect to the Docker daemon" + + def garbage(argv, *, env, cwd, timeout=None): + return 0, "all\n" + + def truncated(argv, *, env, cwd, timeout=None): + return run_ci.RunOutcome(0, "aabbccddee", truncated=True) + + for runner, expected in ( + (unreachable, "could not list"), + (garbage, "not a container identifier"), + (truncated, "was truncated"), + ): + seen = [] + + def recording(argv, *, env, cwd, timeout=None, _inner=runner): + seen.append(list(argv)) + return _inner(argv, env=env, cwd=cwd, timeout=timeout) + + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("x1"), runner=recording, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "unconfirmed", expected) + self.assertIn(expected, result.detail) + self.assertEqual([argv for argv in seen if "rm" in argv], [], "nothing removed on a guess") + + def test_a_container_that_survives_its_removal_is_reported_not_chased(self): + calls = [] + + def runner(argv, *, env, cwd, timeout=None): + calls.append(list(argv)) + if "ps" in argv: + return 0, "aabbccddeeff\n" # the daemon keeps listing it + return 0, "" + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "unconfirmed") + self.assertIn("still listed after removal", result.detail) + self.assertEqual(len([argv for argv in calls if "rm" in argv]), 1, "no chase loop") + + def test_an_ambiguous_removal_result_is_not_retried(self): + def runner(argv, *, env, cwd, timeout=None): + if "ps" in argv: + return 0, "aabbccddeeff\n" + return 1, "Error response from daemon: No such container" # who removed it? + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "unconfirmed") + self.assertIn("ambiguous removal result", result.detail) + self.assertIn("not retried", result.detail) + + def test_a_daemon_that_keeps_minting_containers_stops_at_the_round_cap(self): + counter = itertools.count() + + def runner(argv, *, env, cwd, timeout=None): + if "ps" in argv: + return 0, f"{next(counter):015x}\n" + return 0, "" + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("x"), runner=runner, env={}, budget=cleanup_budget() + ) + self.assertEqual(result.outcome, "unconfirmed") + self.assertTrue( + "teardown rounds" in result.detail or "deadline" in result.detail, result.detail + ) + + def test_a_nested_runner_timeout_cannot_extend_the_shared_cleanup_deadline(self): + # The teardown's docker calls run through the real default_runner. A + # nested timeout there must spend what is left of the shared deadline, + # not open a fresh 60s allowance of its own. + clock = FakeClock() + budget = run_ci.Deadline(45, clock=clock, label="partly spent cleanup") + created: list[tuple[list[str], ClockedChild]] = [] + + def factory(argv, **kwargs): + argv = list(argv) + if "ps" in argv: + child = ClockedChild(clock, output=b"abcdef012345\n", exit_code=0, wait_cost=20) + elif "rm" in argv: + child = ClockedChild(clock, output=b"", exit_code=None) # ignores SIGTERM + else: # pragma: no cover - no other docker call is expected + child = ClockedChild(clock, output=b"", exit_code=0) + created.append((argv, child)) + return child + + with mock.patch.object(run_ci.subprocess, "Popen", factory), \ + mock.patch.object(run_ci.os, "getpgid", side_effect=ProcessLookupError), \ + contextlib.redirect_stdout(io.StringIO()): + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("tok123"), + runner=run_ci.default_runner, + env={}, + budget=budget, + settle=True, + ) + self.assertEqual(result.outcome, "unconfirmed") + self.assertIn("not retried", result.detail) + removal = [child for argv, child in created if "rm" in argv] + self.assertEqual(len(removal), 1) + child = removal[0] + self.assertIn(signal.SIGTERM, child.signals) + self.assertIn(signal.SIGKILL, child.signals, "a TERM-ignoring client needs the kill") + # every nested wait was bounded, and the whole cleanup ended at the + # deadline it was handed instead of overshooting it + self.assertTrue(all(wait is not None for wait in child.waits)) + self.assertLessEqual( + clock.value, budget.total_seconds, + "the nested stop must end at the shared cleanup deadline", + ) + + def test_slow_docker_calls_are_bounded_by_the_one_cleanup_deadline(self): + clock = FakeClock() + # every successful ps/rm costs 25s of the 60s deadline + daemon = FakeDaemon(clock=clock, advance=clock.advance, call_seconds=25) + counter = itertools.count() + budget = cleanup_budget(clock) + + def runner(argv, *, env, cwd, timeout=None): + # each listing returns a fresh container, so nothing is ever settled + if "ps" in argv: + return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout)[0], f"{next(counter):016x}\n" + return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout) + + result = run_ci.teardown_owned_containers( + run_ci.ci_owned_label("x"), runner=runner, env={}, budget=budget, settle=True + ) + self.assertEqual(result.outcome, "unconfirmed") + self.assertLessEqual(budget.elapsed(), run_ci.CLEANUP_RESERVE_SECONDS, "cleanup never outlives its deadline") + # no docker call was ever given more than the per-call ceiling + for _argv, timeout in daemon.calls: + self.assertLessEqual(timeout, run_ci.T_CONTAINER_TEARDOWN) + + def test_a_killed_client_still_loses_its_container(self): + daemon = FakeDaemon(run_status=run_ci.EXIT_TIMEOUT, visible_after_seconds=0) + with mock.patch.object(run_ci, "require_docker", return_value=None), \ + mock.patch.object(run_ci, "TEARDOWN_POLL_SECONDS", 0), \ + contextlib.redirect_stdout(io.StringIO()) as sink: + with self.assertRaises(run_ci.CommandFailed) as caught: + run_ci.run_container("check", daemon.runner, {"PATH": "/bin"}) + self.assertTrue(caught.exception.timed_out) + self.assertEqual(daemon.containers, {}, "the daemon-owned container must be removed") + self.assertEqual(len(daemon.removed), 1) + self.assertIn("container teardown", sink.getvalue()) + + def test_unconfirmed_container_cleanup_is_reported_not_hidden(self): + def runner(argv, *, env, cwd, timeout=None): + if "build" in argv: + return 0, "" + if "run" in argv: + return 0, "" + return 1, "Cannot connect to the Docker daemon" + with mock.patch.object(run_ci, "require_docker", return_value=None), \ + contextlib.redirect_stdout(io.StringIO()) as sink: + with self.assertRaises(run_ci.CleanupUnconfirmed) as caught: + run_ci.run_container("smoke", runner, {"PATH": "/bin"}) + self.assertIn("cleanup unconfirmed", str(caught.exception)) + self.assertIn("not removed by a broad prune", str(caught.exception)) + self.assertIn("unconfirmed", sink.getvalue()) + + def test_a_cooperative_timeout_with_no_container_listed_is_still_unconfirmed(self): + # EXIT_TIMEOUT with a confirmed, unforced stop: the process fact is clean + # but the resource fact is not, so an empty listing must not be called + # absent. The settle stays bounded by the cleanup deadline. + daemon = FakeDaemon() + + def runner(argv, *, env, cwd, timeout=None): + if any(part.endswith("verify_release_local.py") for part in argv): + return run_ci.RunOutcome( + run_ci.EXIT_TIMEOUT, "", False, + "stopped on SIGTERM within 10s of the cleanup deadline", + run_ci.Deadline(2, label="harness cleanup"), # already nearly spent + ) + return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout) + + with contextlib.redirect_stdout(io.StringIO()) as sink: + with self.assertRaises(run_ci.CommandFailed) as caught: + run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) + self.assertTrue(caught.exception.timed_out) + self.assertFalse(caught.exception.cleanup_unconfirmed) + self.assertFalse(caught.exception.forced) + self.assertIn("unconfirmed", sink.getvalue()) + self.assertIn("a container may still appear", sink.getvalue()) + + def test_the_runtime_harness_container_is_removed_by_its_exact_label_after_a_kill(self): + daemon = FakeDaemon() + selectors = [] + + def runner(argv, *, env, cwd, timeout=None): + if any(part.endswith("verify_release_local.py") for part in argv): + # the harness started a container and was then killed at its bound + self.assertIn("--external-container-cleanup", argv) + token = argv[argv.index("--invocation-id") + 1] + selector = f"{run_ci.RUNTIME_LABEL_KEY_PREFIX}.{token}" + selectors.append(selector) + daemon.start(selector, "verify-run") + return run_ci.RunOutcome( + run_ci.EXIT_TIMEOUT, "", False, + "stopped on SIGTERM within 10s of the cleanup deadline", + run_ci.Deadline(run_ci.CLEANUP_RESERVE_SECONDS, label="harness cleanup"), + ) + return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout) + + with contextlib.redirect_stdout(io.StringIO()) as sink: + with self.assertRaises(run_ci.CommandFailed): + run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) + self.assertEqual(daemon.removed, ["verify-run"], "the harness container must be removed exactly") + self.assertEqual(daemon.containers, {}) + # the harness's containers are addressed by its own key-only label + self.assertEqual( + set(s for s in daemon.selectors if s.startswith(run_ci.RUNTIME_LABEL_KEY_PREFIX)), + set(selectors), + ) + self.assertIn("runtime container teardown", sink.getvalue()) + + def test_a_reaped_harness_with_no_visible_container_is_not_called_clean(self): + daemon = FakeDaemon() + + def runner(argv, *, env, cwd, timeout=None): + if any(part.endswith("verify_release_local.py") for part in argv): + # the process was reaped, but it died before the daemon created + # anything: that is not proof that nothing will appear + return run_ci.RunOutcome( + run_ci.EXIT_TIMEOUT, "", False, "stopped on SIGKILL within 5s", + run_ci.Deadline(2, label="harness cleanup"), # already nearly spent + False, True, # forced + ) + return daemon.runner(argv, env=env, cwd=cwd, timeout=timeout) + + with contextlib.redirect_stdout(io.StringIO()) as sink: + with self.assertRaises(run_ci.CommandFailed): + run_ci.run_runtime_steps(runner, {"PATH": "/bin"}) + self.assertIn("unconfirmed", sink.getvalue()) + self.assertIn("a container may still appear", sink.getvalue()) + + +class BudgetTests(unittest.TestCase): + """The action's budget is one monotonic allowance, never reset by a stage.""" + + def test_budget_stays_inside_the_workflow_step_ceiling(self): + self.assertEqual( + run_ci.CI_BUDGET_SECONDS + run_ci.WORKFLOW_RESERVE_SECONDS, + run_ci.WORKFLOW_STEP_SECONDS, + ) + self.assertEqual(run_ci.WORKFLOW_STEP_SECONDS, 1800) + self.assertLessEqual(run_ci.CI_BUDGET_SECONDS, 1500) + + def test_every_declared_ceiling_fits_the_shared_budget(self): + # the previously inconsistent ceilings all nested inside the step ceiling + for command in all_commands(): + self.assertLessEqual( + command.timeout_seconds, run_ci.CI_BUDGET_SECONDS, + f"{command.note or command.argv[0]} declares more than the budget", + ) + + def test_remaining_time_caps_every_command_and_is_never_reset(self): + ticks = [0.0] + + def clock(): + return ticks[0] + + deadline = run_ci.Deadline(100, clock=clock) + runner, calls = scripted_runner([0, 0, 0]) + run_ci.run_checked(run_ci.cmd(["a"], 1000), runner, {}, deadline=deadline) + ticks[0] = 60.0 + run_ci.run_checked(run_ci.cmd(["b"], 1000), runner, {}, deadline=deadline) + ticks[0] = 94.0 + run_ci.run_checked(run_ci.cmd(["c"], 7), runner, {}, deadline=deadline) + self.assertEqual([call["timeout"] for call in calls], [100, 40, 6]) + + def test_expired_budget_stops_before_the_next_stage_and_says_so(self): + ticks = [0.0] + + def clock(): + return ticks[0] + + deadline = run_ci.Deadline(100, clock=clock) + runner, calls = scripted_runner([0]) + run_ci.run_checked( + run_ci.cmd(["a"], 100), runner, {}, deadline=deadline, stage="first" + ) + ticks[0] = 150.0 + with self.assertRaises(run_ci.BudgetExpired) as caught: + run_ci.run_checked( + run_ci.cmd(["b"], 100), runner, {}, deadline=deadline, stage="second" + ) + message = str(caught.exception) + self.assertIn("second", message) + self.assertIn("bounded failure", message) + self.assertIn("separately approved", message) + self.assertEqual(len(calls), 1, "no stage starts after the budget expires") + + def test_later_stages_draw_on_the_same_budget_as_earlier_ones(self): + ticks = [0.0] + + def clock(): + return ticks[0] + + deadline = run_ci.Deadline(900, clock=clock) + timeouts = [] + + def runner(argv, *, env, cwd, timeout=None): + # stage commands only; the container teardown also calls docker + if "ps" in argv or "rm" in argv: + return 0, "" + timeouts.append(timeout) + ticks[0] += 200 # every stage consumes budget + return 0, "" + + with mock.patch.object(run_ci, "require_docker", return_value=None), \ + contextlib.redirect_stdout(io.StringIO()): + run_ci.run_check(runner, {"PATH": "/bin"}, deadline=deadline) + # docker build, docker run, then the runtime harness + self.assertEqual(len(timeouts), 3, timeouts) + self.assertGreater(timeouts[0], timeouts[1]) + self.assertGreater(timeouts[1], timeouts[2]) + + def test_container_handoff_is_shorter_than_the_outer_timeout(self): + seen = [] + + def runner(argv, *, env, cwd, timeout=None): + seen.append((list(argv), timeout)) + return 0, "" + + with mock.patch.object(run_ci, "require_docker", return_value=None), \ + contextlib.redirect_stdout(io.StringIO()): + run_ci.run_container("smoke", runner, {"PATH": "/bin"}) + run_argv, run_timeout = seen[1] + inner = int(run_argv[run_argv.index("--budget-seconds") + 1]) + self.assertEqual(inner, run_timeout - run_ci.CONTAINER_HANDOFF_SECONDS) + self.assertLess(inner, run_timeout) + + def test_a_direct_run_cannot_raise_the_budget(self): + parser = run_ci.build_parser() + args = parser.parse_args(["check", "--budget-seconds", "7200"]) + self.assertEqual(run_ci.action_budget(args.budget_seconds), run_ci.CI_BUDGET_SECONDS) + self.assertEqual(run_ci.action_budget(None), run_ci.CI_BUDGET_SECONDS) + self.assertEqual(run_ci.action_budget(600), 600) + + def test_the_gate_keeps_every_stage(self): + check_flat = flatten(run_ci.container_steps("check")) + for token in ("npm", "ci", "check-all", "demo", "tools/package_release.py", "tools/check_release.py"): + self.assertIn(token, check_flat) + seen = [] + + def runner(argv, *, env, cwd, timeout=None): + seen.append(list(argv)) + return 0, "" + + with mock.patch.object(run_ci, "require_docker", return_value=None), \ + contextlib.redirect_stdout(io.StringIO()): + run_ci.run_check(runner, {"PATH": "/bin"}) + joined = " ".join(" ".join(argv) for argv in seen) + # the gate runs inside the container, and the runtime harness still runs + self.assertIn("_container", joined) + self.assertIn("check", seen[1]) + self.assertIn("tools/verify_release_local.py", joined) class CommandDefinitionTests(unittest.TestCase): @@ -575,9 +1548,17 @@ class CommandDefinitionTests(unittest.TestCase): self.assertEqual(argv[argv.index("--vice-executable") + 1], "/usr/bin/x64sc") def test_runtime_harness_is_wired(self): - flat = flatten(run_ci.runtime_steps()) + token = "run-123-abc" + flat = flatten(run_ci.runtime_steps(token)) self.assertIn("tools/verify_release_local.py", flat) self.assertIn("runtime", flat) + # the helper is the sole remover and names itself, so no docker rm race + self.assertIn("--external-container-cleanup", flat) + self.assertEqual(flat[flat.index("--invocation-id") + 1], token) + self.assertIn( + run_ci.RUNTIME_LABEL_KEY_PREFIX, + run_ci.runtime_owned_label(token).selector(), + ) def test_no_command_uses_release_preflight_mode(self): for command in all_commands(): @@ -731,6 +1712,23 @@ class SourceHygieneTests(unittest.TestCase): self.assertNotIn("capture_output", source) self.assertNotIn("subprocess.run(", source) + def test_output_is_not_spooled_to_a_temp_file_until_exit(self): + # A spool that is only read at exit loses every log when the workflow + # kills the helper, so the runner streams instead. + source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") + self.assertNotIn("mkstemp", source) + self.assertIn("select.select", source) + + def test_child_processes_are_put_in_their_own_group_and_killed_by_group(self): + source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") + self.assertIn("start_new_session=True", source) + self.assertIn("os.killpg", source) + + def test_helper_never_prints_the_environment(self): + source = (TOOLS / "run_ci.py").read_text(encoding="utf-8") + self.assertNotIn("os.environ.items", source) + self.assertNotIn("os.environ.copy", source) + if __name__ == "__main__": unittest.main() diff --git a/tests/python/test_release_local.py b/tests/python/test_release_local.py index 1ec9016..3445337 100644 --- a/tests/python/test_release_local.py +++ b/tests/python/test_release_local.py @@ -2065,5 +2065,86 @@ class IsolatedRuntimeGateTests(TemporaryDirectoryCase): self.assertEqual(document["runtime_environment"]["installed_by_recipe"], ["python3"]) +class InvocationOwnershipTests(TemporaryDirectoryCase): + """The caller may own container removal only by naming the run explicitly.""" + + def setUp(self) -> None: + super().setUp() + original = verify.INVOCATION + self.addCleanup(setattr, verify, "INVOCATION", original) + + def _main_without_docker(self, argv) -> int: + with mock.patch.object(verify, "docker_available", return_value=(False, "no docker in this test")): + return verify.main(argv) + + def test_unsafe_invocation_ids_are_refused_before_anything_exists(self) -> None: + base = ["--mode", "runtime", "--release-dir", str(self.root / "release")] + for unsafe in ("../../etc", "UPPER", "with space", "a/b", "", "x" * 65, "-leading", "a.b"): + with self.assertRaises(SystemExit) as caught: + verify.main(base + ["--invocation-id", unsafe]) + self.assertEqual(caught.exception.code, 2, unsafe) + self.assertEqual(list(self.root.iterdir()), [], "nothing may be created for a refused id") + + def test_a_safe_id_is_accepted_and_labels_the_run(self) -> None: + code = self._main_without_docker([ + "--mode", "runtime", "--release-dir", str(self.root / "release"), + "--invocation-id", "ci-42-abcdef", + ]) + self.assertEqual(code, verify.EXIT_ENVIRONMENT) + self.assertEqual(verify.INVOCATION, "ci-42-abcdef") + self.assertEqual(verify.container_label(), f"{verify.CONTAINER_LABEL}.ci-42-abcdef") + + def test_the_default_id_is_still_generated_when_none_is_given(self) -> None: + code = self._main_without_docker([ + "--mode", "runtime", "--release-dir", str(self.root / "release"), + ]) + self.assertEqual(code, verify.EXIT_ENVIRONMENT) + self.assertTrue(verify.INVOCATION_ID.fullmatch(verify.INVOCATION), verify.INVOCATION) + + def test_external_cleanup_requires_runtime_and_an_explicit_id(self) -> None: + with self.assertRaises(SystemExit) as caught: + verify.main([ + "--mode", "source-rebuild", "--release-dir", str(self.root), + "--invocation-id", "abc", "--external-container-cleanup", + ]) + self.assertEqual(caught.exception.code, 2) + with self.assertRaises(SystemExit) as caught: + verify.main([ + "--mode", "runtime", "--release-dir", str(self.root), + "--external-container-cleanup", + ]) + self.assertEqual(caught.exception.code, 2) + + def test_external_cleanup_skips_container_removal_and_keeps_scratch_cleanup(self) -> None: + for external, expected_removals in ((True, 0), (False, 1)): + token = f"ci-{int(external)}-abcdef" + argv = [ + "--mode", "runtime", "--release-dir", str(self.root), + "--work-dir", str(self.root), "--invocation-id", token, + "--checkout-root", str(self.root / "unmounted" / "checkout"), + ] + if external: + argv.append("--external-container-cleanup") + args = verify.build_parser().parse_args(argv) + verify.INVOCATION = verify.validate_invocation_id(token) + cleanup, journal = verify.Cleanup(), verify.Journal() + with mock.patch.object(verify, "docker_available", return_value=(True, "test")), \ + mock.patch.object(verify, "verify_release_directory", + return_value={"binary_archive": {"name": "a.tar"}, "problems": []}), \ + mock.patch.object(verify, "remove_containers_with_label", return_value=[]) as remover, \ + mock.patch.object(verify, "extract_verified_archive", + side_effect=verify.HarnessError("stop here")): + with self.assertRaises(verify.HarnessError): + verify.run_runtime(args, journal, cleanup) + self.assertEqual(remover.call_count, expected_removals, external) + registered = " ".join(cleanup.as_json()) + self.assertIn("scratch directory", registered, "scratch cleanup is always kept") + if external: + self.assertNotIn("containers labelled", registered) + self.assertIn("--external-container-cleanup", " ".join(journal.notes)) + else: + self.assertIn("containers labelled", registered) + + if __name__ == "__main__": unittest.main() diff --git a/tools/run_ci.py b/tools/run_ci.py index df62b41..2e0f1c0 100644 --- a/tools/run_ci.py +++ b/tools/run_ci.py @@ -42,24 +42,56 @@ The helper does not read secrets, does not contact a device, and does not push a tag or artifact. It passes child processes a small allowlisted environment; credential-shaped variables are dropped rather than forwarded. +One monotonic budget covers a whole action: ``CI_BUDGET_SECONDS`` (1500s) inside +the workflow step's 1800s ceiling, with the remainder reserved for the guest's +own setup and teardown. Every child timeout is capped by the time still left in +that budget, and the container is given a budget slightly shorter than the +outer ``docker run`` timeout so the inner helper always reports first. The +nesting order is therefore always stage, then helper budget, then workflow step. + +Child output is streamed live as it arrives, up to a bounded number of bytes, +and only a bounded head and tail are retained in memory for the failure +diagnostic. Nothing is spooled to an unbounded temp file, so a workflow that +kills the helper still leaves the live log behind, and a noisy command cannot +exhaust memory or the log volume. + +A bounded timeout failure is not evidence that a stage is broken. A cold build +that does not fit the budget fails closed, says so plainly, and stops. Making +the workflow duration longer is a separate, separately approved change. + +Killing the local Docker client does not stop the container the daemon started, +and ``--rm`` only removes a container that exits. Every container this helper +starts therefore carries a unique name and a unique per-invocation label, and it +is removed by exactly that identity in a ``finally`` block, after the client has +exited, failed, or been killed. The runtime harness can be addressed the same +way: it is given ``--invocation-id`` and ``--external-container-cleanup``, so it +labels its containers with this invocation's token and leaves their removal to +this helper, which is then the only party removing them. Both removals spend the +same single cleanup deadline, address only what this invocation labelled, and +never prune: a broad prune could delete another invocation's containers. + What this helper does not do: it does not build a microVM image, does not configure a spindle, and does not prove that any hosted CI ran. A successful local action is not hosted-CI evidence. Exit codes: 0 success, 1 a precondition or check failed, 2 usage error, -3 internal error. +3 internal error or an owned resource that could not be confirmed removed. """ from __future__ import annotations import argparse +import codecs import os import platform import re +import select import shutil +import signal import subprocess import sys import tempfile +import time from dataclasses import dataclass, field from pathlib import Path @@ -108,27 +140,100 @@ DEVELOPMENT_LABEL = "ci-local" RUNTIME_RECORD = "/tmp/ultimate-tool-runtime-verification.json" BUILD_ENV_WRITER = "/usr/local/bin/ultimate-tool-build-env" -# Bounded timeouts, in seconds, chosen per command: cold builds get minutes, -# quick checks get seconds. Every child command is given one; none runs forever. +# Time budget, in seconds. The workflow step ceiling is the server-reported +# 1800s admission limit; the helper keeps a reserve for the guest's own setup, +# teardown and image removal, so it never relies on the ceiling being reached. +# Nothing here raises a limit: CI_BUDGET_SECONDS is strictly inside the ceiling. +WORKFLOW_STEP_SECONDS = 1800 +WORKFLOW_RESERVE_SECONDS = 300 +CI_BUDGET_SECONDS = WORKFLOW_STEP_SECONDS - WORKFLOW_RESERVE_SECONDS + +# The inner container helper gets a slightly shorter budget than the outer +# `docker run` timeout, so the stage diagnostic is printed by the helper that +# knows which stage it was, before the outer process group is terminated. +CONTAINER_HANDOFF_SECONDS = 30 + +# A command is not started with less time than this left in the budget; the +# action reports the budget expiry instead of starting something that cannot +# finish and cannot report why. +MIN_COMMAND_SECONDS = 5 + +# Per-command ceilings, in seconds: quick checks get seconds, cold builds get +# minutes. Every one is at or below CI_BUDGET_SECONDS, and the timeout actually +# used is the smaller of its ceiling and the budget still remaining, so no +# declared ceiling can outlive the action that runs it. T_VALIDATE = 120 +T_GUEST_PROBE = 15 T_BUILD_ENV = 120 T_DEMO = 300 T_MAKE_DEMO = 900 T_NPM_CI = 900 T_BUILD_DEBUG = 900 T_CHECK_RELEASE = 600 -T_CHECK_ALL = 1800 -T_BUILD_RELEASE = 1800 -T_PACKAGE = 1800 -T_VICE_ACCEPTANCE = 1800 -T_RUNTIME = 1800 -T_DOCKER_BUILD = 1800 -T_DOCKER_RUN = 2400 - -# Echo at most this many bytes of a command's output. Beyond it the middle is -# dropped and marked. Output is spooled to a temp file first, so this bounds -# what is read back into memory, not what the command is allowed to write. -MAX_OUTPUT_BYTES = 1_000_000 +T_PACKAGE = 1200 +T_RUNTIME = 1200 +T_CHECK_ALL = CI_BUDGET_SECONDS +T_BUILD_RELEASE = CI_BUDGET_SECONDS +T_VICE_ACCEPTANCE = CI_BUDGET_SECONDS +T_DOCKER_BUILD = CI_BUDGET_SECONDS +T_DOCKER_RUN = CI_BUDGET_SECONDS + +# Log bounds, in bytes. Output is echoed live up to LIVE_OUTPUT_BYTES, and only +# a bounded head and tail are retained for the diagnostic, so a noisy command +# cannot exhaust memory or the workflow log. Byte counts are reported for +# whatever is not shown, and nothing here reads or prints the environment. +READ_CHUNK_BYTES = 65536 +LIVE_OUTPUT_BYTES = 256 * 1024 +RETAINED_HEAD_BYTES = 64 * 1024 +RETAINED_TAIL_BYTES = 64 * 1024 +# Flush a partial line once this much has accumulated, so a child that never +# emits a newline still makes progress visible while it runs. +LIVE_LINE_FLUSH_BYTES = 8 * 1024 + +# Cleanup bounds, in seconds. One absolute monotonic cleanup deadline of +# CLEANUP_RESERVE_SECONDS covers everything that happens after a command exceeds +# its bound: stopping its process group, draining what it wrote, and listing, +# removing and confirming the containers it owned. Each step also has its own +# ceiling, and every wait and every docker call is capped by the time still left +# on that one deadline, so the cleanup as a whole cannot outlive it. A step that +# cannot be started with enough time left is reported unconfirmed rather than +# run past the deadline. +CLEANUP_RESERVE_SECONDS = 60 +CHILD_DRAIN_SECONDS = 5 +CHILD_GRACE_SECONDS = 10 +CHILD_REAP_SECONDS = 5 +# Poll interval while a killed client's container may still be appearing. +TEARDOWN_POLL_SECONDS = 2 +# Per-call ceiling for one docker list/remove call inside the deadline. +T_CONTAINER_TEARDOWN = 20 +# Hard stop for the teardown sweep, so a daemon that keeps re-listing a +# container is reported rather than chased. +MAX_TEARDOWN_ROUNDS = 64 +# The smallest useful slice of the cleanup deadline. +MIN_CLEANUP_SECONDS = 1 + +# Container ownership. Killing the local Docker client does not stop the +# container the daemon started, and `--rm` only removes a container that +# exits. Every container this helper starts therefore carries a unique name and +# a unique per-invocation label, and teardown addresses exactly those and +# nothing else: no `docker system prune`, no wildcards, no other invocation's +# containers. +CONTAINER_NAME_PREFIX = "ultimate-tool-ci" +CONTAINER_LABEL_KEY = "ultimate-tool-ci.run" +CONTAINER_STAGE_LABEL_KEY = "ultimate-tool-ci.stage" + +# The runtime harness labels its containers `ultimate-tool-verify-local.` +# (verify_release_local.CONTAINER_LABEL). This helper passes that id with the +# harness's `--invocation-id` option, so it can address exactly the containers +# that harness run started. These are the only label keys a teardown may filter +# on: an allowlist, so a typo cannot turn into a broad sweep. +RUNTIME_LABEL_KEY_PREFIX = "ultimate-tool-verify-local" +ALLOWED_LABEL_KEY_PREFIXES = (CONTAINER_LABEL_KEY, RUNTIME_LABEL_KEY_PREFIX) + +# The shape an ownership token may take. It is embedded in a Docker label, in a +# container name and in the runtime image tag, and the harness validates the id +# it is given with the same rule before it creates anything. +TOKEN_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$") # Child processes receive only these variables. The allowlist is the primary # defence: anything not named here (a password, a token, a personal override) is @@ -177,10 +282,161 @@ class ConfigError(Exception): class CommandFailed(Exception): """A child command returned a non-zero status.""" - def __init__(self, argv, code: int) -> None: + def __init__( + self, + argv, + code: int, + *, + stage: str = "", + timed_out: bool = False, + cleanup_unconfirmed: bool = False, + cleanup=None, + forced: bool = False, + ) -> None: self.argv = list(argv) self.code = code - super().__init__(f"command failed with status {code}: {' '.join(self.argv)}") + self.stage = stage + self.timed_out = timed_out + self.cleanup_unconfirmed = cleanup_unconfirmed + self.cleanup = cleanup + self.forced = forced + if timed_out: + message = ( + f"stage {stage!r} exceeded its bounded timeout: " + f"{' '.join(self.argv)}" + ) + else: + message = f"command failed with status {code}: {' '.join(self.argv)}" + if stage: + message = f"stage {stage!r} failed with status {code}: {' '.join(self.argv)}" + if cleanup_unconfirmed: + message += ( + "; its process group could not be confirmed stopped, so " + "resources it started in the Docker daemon may still exist" + ) + super().__init__(message) + + +class CleanupUnconfirmed(Exception): + """An owned resource could not be confirmed released. + + Reported instead of claiming a clean release. The exact resource is named; + nothing broad is ever pruned to compensate. + """ + + def __init__(self, resource: str, detail: str) -> None: + self.resource = resource + self.detail = detail + super().__init__( + f"cleanup unconfirmed for {resource}: {detail}. The resource may " + "still exist; it was not removed by a broad prune because that " + "could touch another invocation's containers" + ) + + +@dataclass(frozen=True) +class RunOutcome: + """What one child command reported. + + Unpacks as ``(code, output)`` so an injected runner may still return a plain + two-tuple. ``cleanup_unconfirmed`` is true when the child's process group was + signalled but its exit could not be confirmed, which means anything it + started in the Docker daemon may survive the deadline. + """ + + code: int + output: str = "" + cleanup_unconfirmed: bool = False + detail: str = "" + # The one cleanup deadline this command's stop/drain spent, so a caller can + # continue the same cleanup (container or harness teardown) inside it instead + # of starting a fresh allowance. + cleanup: object = None + # True when the retained output lost bytes, so a caller must not parse it as + # a complete listing. + truncated: bool = False + # True when the group had to be forced with SIGKILL, which means anything the + # child started in the Docker daemon may have survived it. + forced: bool = False + + def __iter__(self): + return iter((self.code, self.output)) + + +def as_outcome(value) -> RunOutcome: + """Accept either a ``RunOutcome`` or a plain ``(code, output)`` pair.""" + if isinstance(value, RunOutcome): + return value + code, output = value + return RunOutcome(int(code), output) + + +class BudgetExpired(Exception): + """The action's overall time budget has no usable time left.""" + + def __init__(self, deadline: "Deadline", stage: str, detail: str = "") -> None: + self.deadline = deadline + self.stage = stage + remaining = deadline.remaining() + reason = detail or ( + f"only {remaining:.1f}s of the {deadline.total_seconds}s budget for this " + f"action is left, below the {MIN_COMMAND_SECONDS}s minimum for a stage" + ) + super().__init__( + f"{reason}, so stage {stage!r} did not run: bounded failure, not a stage " + "verdict. That stage and every later stage did not run. A cold build may " + "not fit this budget; a longer workflow duration is a separately approved " + "change, not something this helper can decide" + ) + + +class Deadline: + """One monotonic time budget shared by every command in an action. + + The budget is created once per helper invocation and is never reset, so a + later stage draws whatever the earlier stages left instead of receiving a + fresh allowance. ``clock`` is injectable so a test can drive elapsed time + without sleeping. + """ + + def __init__(self, total_seconds: int, *, clock=time.monotonic, label: str = "ci helper") -> None: + if not isinstance(total_seconds, int) or isinstance(total_seconds, bool): + raise ConfigError("a time budget needs a whole number of seconds") + if total_seconds <= 0: + raise ConfigError("a time budget must be a positive number of seconds") + self.total_seconds = total_seconds + self.label = label + self._clock = clock + self._started = clock() + + def elapsed(self) -> float: + return max(0.0, self._clock() - self._started) + + def remaining(self) -> float: + return max(0.0, self.total_seconds - self.elapsed()) + + def now(self) -> float: + """The clock this budget measures, so callers do not mix clocks.""" + return self._clock() + + def cap(self, declared_seconds) -> int: + """Return the timeout to use: the declared ceiling, capped by what remains.""" + remaining = self.remaining() + if declared_seconds is None: + limit = remaining + else: + limit = min(float(declared_seconds), remaining) + return max(0, int(limit)) + + def require(self, stage: str) -> None: + """Refuse to start a stage that no longer has usable time.""" + if self.remaining() < MIN_COMMAND_SECONDS: + raise BudgetExpired(self, stage) + + +def default_deadline() -> Deadline: + """The budget a direct library call gets when it is not part of an action.""" + return Deadline(CI_BUDGET_SECONDS) @dataclass(frozen=True) @@ -352,8 +608,14 @@ def emulator_steps(vice_executable: str) -> list[Command]: ] -def runtime_steps() -> list[Command]: - """The runtime harness, run on the guest where the Docker daemon lives.""" +def runtime_steps(token: str) -> list[Command]: + """The runtime harness, run on the guest where the Docker daemon lives. + + The harness is given this invocation's id and told that the caller removes + the containers it starts, so exactly one party removes them: the harness + keeps its read-only checks and its scratch cleanup, and this helper owns + container removal. + """ return [ cmd( [ @@ -363,6 +625,8 @@ def runtime_steps() -> list[Command]: "--out", RUNTIME_RECORD, "--replace", "--json", + "--invocation-id", token, + "--external-container-cleanup", ], T_RUNTIME, note="runtime verification of the packaged binaries", @@ -466,7 +730,8 @@ def require_docker(env: dict[str, str]) -> None: ) -def verify_guest(env: dict[str, str], runner) -> None: +def verify_guest(env: dict[str, str], runner, deadline=None) -> None: + deadline = default_deadline() if deadline is None else deadline facts = read_guest_facts() problems = guest_problems(facts, docker_present(env)) if problems: @@ -474,9 +739,10 @@ def verify_guest(env: dict[str, str], runner) -> None: "guest preconditions unmet: " + "; ".join(problems) ) # Test the guest-local daemon, not a remote or inherited Docker endpoint. + deadline.require("guest docker probe") code, _ = runner( ["docker", "--host", "unix:///var/run/docker.sock", "info", "--format", "{{.ServerVersion}}"], - env=env, cwd=ROOT, timeout=15, + env=env, cwd=ROOT, timeout=max(1, deadline.cap(T_GUEST_PROBE)), ) if code != 0: raise PreconditionError("guest-local Docker daemon is unavailable to the workflow user") @@ -487,82 +753,439 @@ def verify_guest(env: dict[str, str], runner) -> None: # -------------------------------------------------------------------------- -def read_bounded_text(path, limit: int = MAX_OUTPUT_BYTES) -> str: - """Read at most ``limit`` bytes from a spooled log, marking what was dropped. +class OutputRetainer: + """Echo bounded live output and retain a bounded head and tail in memory. + + ``feed`` is called with whatever the child wrote. Every byte is counted, + the first ``head_bytes`` and last ``tail_bytes`` are kept, and at most + ``live_bytes`` are written through to ``stream`` while the child runs. - The whole file is never held in memory: for a large log only the head and - the tail are read, and the dropped byte count is reported. + The same bytes reach the operator at most once: ``report`` only writes the + retained tail that the live echo did not already show. That is what keeps a + nested container log (whose child output is echoed live by the inner helper + and again by the outer ``docker run``) from being replayed twice. """ - path = str(path) - size = os.path.getsize(path) - with open(path, "rb") as handle: - if size <= limit: - return handle.read().decode("utf-8", "replace") - half = limit // 2 - head = handle.read(half) - handle.seek(size - half) - tail = handle.read() - dropped = size - len(head) - len(tail) - return ( - head.decode("utf-8", "replace") - + f"\n...[truncated {dropped} bytes]...\n" - + tail.decode("utf-8", "replace") - ) + + def __init__( + self, + *, + live_bytes: int = LIVE_OUTPUT_BYTES, + head_bytes: int = RETAINED_HEAD_BYTES, + tail_bytes: int = RETAINED_TAIL_BYTES, + stream=None, + ) -> None: + self.live_bytes = max(0, int(live_bytes)) + self.head_bytes = max(0, int(head_bytes)) + self.tail_bytes = max(0, int(tail_bytes)) + self.stream = stream + self.total = 0 + self.echoed = 0 + self.head = bytearray() + self.tail = bytearray() + self._pending = b"" + # Incremental, so a multi-byte character split across two reads is + # echoed as one character instead of two replacement marks. + self._decoder = codecs.getincrementaldecoder("utf-8")("replace") + + # -- capture ---------------------------------------------------------- + + def feed(self, chunk: bytes) -> None: + if not chunk: + return + self.total += len(chunk) + room = self.head_bytes - len(self.head) + if room > 0: + self.head += chunk[:room] + self.tail += chunk + if len(self.tail) > self.tail_bytes: + del self.tail[: len(self.tail) - self.tail_bytes] + self._echo(chunk) + + def finish(self) -> None: + """Flush a trailing partial line, then any incomplete character.""" + if self._pending: + self._write(self._pending) + self._pending = b"" + self._write(b"", final=True) + + # -- live echo -------------------------------------------------------- + + def _sink(self): + return sys.stdout if self.stream is None else self.stream + + def _write(self, block: bytes, *, final: bool = False) -> None: + sink = self._sink() + text = self._decoder.decode(block, final) + if text: + sink.write(text) + try: + sink.flush() + except (AttributeError, ValueError): # pragma: no cover - closed sink + pass + + def _echo(self, chunk: bytes) -> None: + if self.echoed >= self.live_bytes: + return + piece = chunk[: self.live_bytes - self.echoed] + self.echoed += len(piece) + self._pending += piece + cut = self._pending.rfind(b"\n") + if cut == -1: + # A child that never emits a newline still has to show progress. + if len(self._pending) >= LIVE_LINE_FLUSH_BYTES: + self._write(self._pending) + self._pending = b"" + return + block, self._pending = self._pending[: cut + 1], self._pending[cut + 1 :] + self._write(block) + + # -- retained diagnostic ---------------------------------------------- + + @property + def _retained_start(self) -> int: + """Absolute offset of the first retained tail byte.""" + return max(len(self.head), self.total - len(self.tail)) + + def _tail_from(self, start: int) -> bytes: + """The retained tail from absolute offset ``start`` onwards.""" + offset = start - (self.total - len(self.tail)) + if offset < 0: # pragma: no cover - defensive, never left of the buffer + offset = 0 + return bytes(self.tail[offset:]) + + @property + def dropped(self) -> int: + """Bytes that neither the head nor the tail retains.""" + return max(0, self._retained_start - len(self.head)) + + @property + def suppressed(self) -> bool: + """True when more output arrived than was echoed live.""" + return self.total > self.echoed + + def text(self) -> str: + """The retained diagnostic: head, a dropped-byte marker, and tail.""" + head = self.head.decode("utf-8", "replace") + tail = self._tail_from(self._retained_start).decode("utf-8", "replace") + dropped = self.dropped + if dropped == 0: + return head + tail + return head + f"\n...[truncated {dropped} bytes]...\n" + tail + + def unsuppressed_tail(self) -> str: + """The retained tail from the first byte that was not echoed live.""" + start = max(self.echoed, self._retained_start) + return self._tail_from(start).decode("utf-8", "replace") + + def report(self, stage: str) -> None: + """Print the bounded diagnostic exactly once, without replaying the echo.""" + sink = self._sink() + if self.suppressed: + hidden = self.total - self.echoed + block = ( + f"[run_ci] {stage}: {self.total} bytes of output; {self.echoed} " + f"shown live, {hidden} not shown live. Retained tail:\n" + + self.unsuppressed_tail() + + f"\n...[end of retained tail, {hidden} bytes not shown]...\n" + ) + else: + block = f"[run_ci] {stage}: {self.total} bytes of output, all shown above\n" + sink.write(block) + try: + sink.flush() + except (AttributeError, ValueError): # pragma: no cover - closed sink + pass + + +def _signal_group(proc, signum: int) -> bool: + """Signal the child's whole process group; True when a signal was delivered.""" + try: + pgid = os.getpgid(proc.pid) + except OSError: + pgid = None + if pgid is not None and pgid != os.getpgid(0): + try: + os.killpg(pgid, signum) + return True + except OSError: + return False + try: + proc.send_signal(signum) + except OSError: # pragma: no cover - the child is already gone + return False + return True -def default_runner(argv, *, env, cwd, timeout=None): - """Run ``argv`` without a shell and return ``(exit_code, bounded_output)``. +def wait_bounded(proc, seconds): + """Wait at most ``seconds`` and return the exit code, or None if still running. - Output is spooled to a regular temp file, so a noisy command cannot exhaust - memory; only a bounded head and tail are read back. The command is killed - and reported as failed if it exceeds ``timeout``. + ``wait()`` is never called without a bound: a child stuck in an + uninterruptible sleep would otherwise hang the helper past every deadline it + has, including the workflow step's own limit. """ - print("+ " + " ".join(str(part) for part in argv), flush=True) - handle, log_path = tempfile.mkstemp(prefix="run_ci_log_", suffix=".log") - os.close(handle) - code = EXIT_TIMEOUT + limit = max(0.0, float(seconds)) try: - with open(log_path, "wb") as log: - proc = subprocess.Popen( - argv, - env=env, - cwd=cwd, - stdout=log, - stderr=subprocess.STDOUT, - ) + return proc.wait(timeout=limit) + except subprocess.TimeoutExpired: + return None + + +def stop_process_group(proc, *, budget: Deadline, grace: int = CHILD_GRACE_SECONDS, reap: int = CHILD_REAP_SECONDS): + """Stop the child's process group inside the cleanup deadline. + + Every wait is capped by the time still left on ``budget`` — the one deadline + that also covers the drain and any container or harness teardown — so the + cleanup as a whole cannot outlive it. When the deadline cannot cover the next + step, that step is skipped and the stop is reported unconfirmed: the process + may still be running, and nothing here pretends otherwise. + + The cooperative window is what lets a child that owns daemon resources + release them itself: the runtime harness removes exactly its own labelled + containers from a SIGTERM handler. SIGKILL is only the fallback. ``grace=0`` + is the last-chance form used when a group already had its window. + Returns ``(confirmed, forced, detail)``. + """ + if proc.poll() is not None: + return True, False, "already exited" + if budget.remaining() < MIN_CLEANUP_SECONDS: + return False, False, "the cleanup deadline expired before the process group could be signalled" + if grace >= MIN_CLEANUP_SECONDS: + if not _signal_group(proc, signal.SIGTERM): + return (wait_bounded(proc, budget.cap(reap)) is not None), False, "no signal could be delivered" + # Never let the cooperative window eat the whole deadline: keep room for + # the forced stop and its confirmation, or a short deadline would always + # end unconfirmed without ever escalating. + forced_reserve = min(float(reap), budget.remaining() / 2.0) + window = budget.cap( + min(float(grace), max(float(MIN_CLEANUP_SECONDS), budget.remaining() - forced_reserve)) + ) + if window >= MIN_CLEANUP_SECONDS and wait_bounded(proc, window) is not None: + return True, False, f"stopped on SIGTERM within {window:g}s of the cleanup deadline" + if budget.remaining() < MIN_CLEANUP_SECONDS: + return False, True, "the cleanup deadline expired before a forced stop could be sent" + if not _signal_group(proc, signal.SIGKILL): + return False, True, "SIGKILL could not be delivered" + window = budget.cap(reap) + if window < MIN_CLEANUP_SECONDS: + return False, True, "the cleanup deadline expired before the forced stop could be confirmed" + if wait_bounded(proc, window) is not None: + return True, True, f"stopped on SIGKILL within {window:g}s of the cleanup deadline" + return False, True, f"still running {window:g}s after SIGKILL" + + +def _drain(fd: int, retainer: OutputRetainer, *, budget: Deadline) -> None: + """Read whatever a stopped child left in the pipe, inside the cleanup deadline.""" + window = budget.cap(CHILD_DRAIN_SECONDS) + if window < MIN_CLEANUP_SECONDS: + return + end = budget.now() + window + while True: + left = end - budget.now() + if left <= 0: + return + try: + ready, _, _ = select.select([fd], [], [], left) + except OSError: # pragma: no cover - the descriptor is already closed + return + if not ready: + return + try: + chunk = os.read(fd, READ_CHUNK_BYTES) + except OSError: # pragma: no cover - the descriptor is already closed + return + if not chunk: + return + retainer.feed(chunk) + + +def default_runner( + argv, + *, + env, + cwd, + timeout=None, + cleanup_seconds: int = CLEANUP_RESERVE_SECONDS, + cleanup_budget: Deadline = None, +) -> RunOutcome: + """Run ``argv`` without a shell and return a bounded ``RunOutcome``. + + Output is streamed live up to a bounded number of bytes and only a bounded + head and tail are retained, so a noisy command cannot exhaust memory and a + workflow that kills the helper still leaves the live log behind. + + The child runs in its own process group. When it exceeds its bound the helper + spends a single monotonic cleanup deadline on everything that follows — the + cooperative SIGTERM, the forced SIGKILL, the reap and the output drain — so + no step can run past that deadline. A step that no longer fits is skipped and + reported: ``cleanup_unconfirmed`` is set and nothing claims the group is + gone. + + ``cleanup_budget`` is an *existing* deadline to spend instead of a fresh + ``cleanup_seconds`` allowance. A caller that is already inside a cleanup + (this helper's own container teardown, whose docker calls run through this + runner) passes it so a nested timeout cannot overshoot the shared end. + """ + stage = " ".join(str(part) for part in argv) + print("+ " + stage, flush=True) + retainer = OutputRetainer() + timed_out = False + stop_detail = "" + unconfirmed = False + forced = False + cleanup: Deadline | None = cleanup_budget + proc = subprocess.Popen( + [str(part) for part in argv], + env=env, + cwd=cwd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + start_new_session=True, + bufsize=0, + ) + fd = proc.stdout.fileno() + expires_at = None if timeout is None else time.monotonic() + float(timeout) + + def expired() -> None: + """Say what is about to happen *before* the risky cleanup, so a helper + killed during cleanup still leaves the reason in the log.""" + bound = cleanup.remaining() if cleanup is not None else cleanup_seconds + print( + f"[run_ci] stage {stage!r} exceeded its {timeout}s bounded timeout; " + f"cleaning up inside the {bound:g}s left of its cleanup deadline " + "(SIGTERM, then SIGKILL, then any owned container)", + flush=True, + ) + + def begin_cleanup() -> Deadline: + """The one deadline this command's stop and drain are spent from. + + A caller that is already inside a cleanup hands over the remaining + deadline; only a command with none of its own opens a fresh allowance. + """ + nonlocal cleanup + if cleanup is None: + cleanup = Deadline(cleanup_seconds, label=f"cleanup of {stage!r}") + return cleanup + + try: + while True: + if expires_at is None: + wait = None + else: + wait = expires_at - time.monotonic() + if wait <= 0: + timed_out = True + break + try: + ready, _, _ = select.select([fd], [], [], wait) + except OSError: # pragma: no cover - the descriptor is already closed + break + if not ready: + continue try: - code = proc.wait(timeout=timeout) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() - code = EXIT_TIMEOUT - output = read_bounded_text(log_path) - if code == EXIT_TIMEOUT: - output += f"\ncommand exceeded its {timeout}s timeout: {' '.join(argv)}" + chunk = os.read(fd, READ_CHUNK_BYTES) + except OSError: # pragma: no cover - the descriptor is already closed + break + if not chunk: + break + retainer.feed(chunk) + if timed_out: + begin_cleanup() + expired() + else: + # A child can close stdout and keep running; the deadline applies to + # the exit, not only to the reads, and the wait stays bounded even + # when no timeout was requested. + left = CHILD_REAP_SECONDS + if expires_at is not None: + left = max(0.0, expires_at - time.monotonic()) + if wait_bounded(proc, left) is None: + timed_out = True + begin_cleanup() + expired() + if timed_out: + confirmed, forced, stop_detail = stop_process_group(proc, budget=cleanup) + unconfirmed = not confirmed + _drain(fd, retainer, budget=cleanup) finally: try: - os.unlink(log_path) - except OSError: # pragma: no cover - best effort cleanup + proc.stdout.close() + except OSError: # pragma: no cover - already closed pass - if output: - sys.stdout.write(output) - sys.stdout.flush() - return code, output + if proc.poll() is None: # pragma: no cover - defensive cleanup + # The group already had its cooperative window; do not grant another. + confirmed, forced_again, final_detail = stop_process_group( + proc, budget=begin_cleanup(), grace=0 + ) + unconfirmed = unconfirmed or not confirmed + forced = forced or forced_again + stop_detail = stop_detail or final_detail + retainer.finish() + status = proc.returncode + if timed_out: + code = EXIT_TIMEOUT + else: + code = EXIT_INTERNAL if status is None else status + retainer.report(stage) + output = retainer.text() + if timed_out: + if unconfirmed: + message = ( + f"stage {stage!r} exceeded its {timeout}s bounded timeout; its " + f"process group could NOT be confirmed stopped ({stop_detail}), " + "so resources it started in the Docker daemon may still exist" + ) + else: + message = ( + f"stage {stage!r} exceeded its {timeout}s bounded timeout; its " + f"process group was stopped ({stop_detail})" + ) + print(f"[run_ci] {message}", flush=True) + output = f"{output}\n{message}" if output else message + return RunOutcome( + code=code, + output=output, + cleanup_unconfirmed=unconfirmed, + detail=stop_detail, + cleanup=cleanup, + truncated=retainer.dropped > 0, + forced=forced, + ) -def run_checked(command, runner, env, cwd=ROOT): +def run_command_under(command, runner, env, cwd=ROOT, *, deadline=None, stage=""): + """Run one command under the shared action budget and return its outcome.""" + deadline = default_deadline() if deadline is None else deadline argv = _argv_of(command) - timeout = command.timeout_seconds if isinstance(command, Command) else None - code, _ = runner(argv, env=env, cwd=cwd, timeout=timeout) - if code != 0: - raise CommandFailed(argv, code) - return code + label = stage or (command.note if isinstance(command, Command) else "") or argv[0] + declared = command.timeout_seconds if isinstance(command, Command) else None + deadline.require(label) + timeout = deadline.cap(declared) + outcome = as_outcome(runner(argv, env=env, cwd=cwd, timeout=max(1, timeout))) + if outcome.code != 0: + raise CommandFailed( + argv, + outcome.code, + stage=label, + timed_out=(outcome.code == EXIT_TIMEOUT), + cleanup_unconfirmed=outcome.cleanup_unconfirmed, + cleanup=outcome.cleanup, + forced=outcome.forced, + ) + return outcome -def execute_steps(steps, runner, env, cwd=ROOT): +def run_checked(command, runner, env, cwd=ROOT, *, deadline=None, stage=""): + """Run one command under the shared budget, failing closed on any status.""" + return run_command_under(command, runner, env, cwd=cwd, deadline=deadline, stage=stage).code + + +def execute_steps(steps, runner, env, cwd=ROOT, *, deadline=None): """Run each command in order, stopping at the first non-zero status.""" + deadline = default_deadline() if deadline is None else deadline for command in steps: - run_checked(command, runner, env, cwd=cwd) + run_checked(command, runner, env, cwd=cwd, deadline=deadline) # -------------------------------------------------------------------------- @@ -580,23 +1203,293 @@ def docker_build_argv() -> list[str]: ] -def docker_run_argv(container_action: str) -> list[str]: +# -------------------------------------------------------------------------- +# Container ownership: what this invocation started, and only that +# -------------------------------------------------------------------------- + + +def new_ownership_token() -> str: + """A per-invocation token that no other invocation can share.""" + return f"{os.getpid()}-{os.urandom(6).hex()}" + + +@dataclass(frozen=True) +class OwnedLabel: + """The exact label that names one invocation's containers. + + Two shapes exist and both are keyed on a per-invocation token: this helper's + own containers carry ``ultimate-tool-ci.run=`` (key *and* value), + while the runtime harness labels its containers with the key + ``ultimate-tool-verify-local.`` (key only). Only the allowlisted + prefixes can be expressed, so a typo cannot turn a teardown into a broad + sweep, and the token is a namespace, not a credential. + """ + + prefix: str + token: str + valued: bool = True + + def __post_init__(self) -> None: + if self.prefix not in ALLOWED_LABEL_KEY_PREFIXES: + raise ConfigError( + f"label prefix {self.prefix!r} is not allowlisted; refusing to filter on it" + ) + if not isinstance(self.token, str) or not TOKEN_PATTERN.fullmatch(self.token): + raise ConfigError( + f"ownership token {self.token!r} is not safe to use in a Docker label" + ) + + def selector(self) -> str: + """What goes after ``--filter label=``.""" + return f"{self.prefix}={self.token}" if self.valued else f"{self.prefix}.{self.token}" + + +def ci_owned_label(token: str) -> OwnedLabel: + """This helper's own container label for one invocation.""" + return OwnedLabel(CONTAINER_LABEL_KEY, token, valued=True) + + +def runtime_owned_label(token: str) -> OwnedLabel: + """The runtime harness's container label for one invocation.""" + return OwnedLabel(RUNTIME_LABEL_KEY_PREFIX, token, valued=False) + + +def docker_owned_list_argv(label: OwnedLabel) -> list[str]: + """List the containers carrying exactly this invocation's label. + + A label filter, never a prune. ``--no-trunc`` keeps the identifiers whole so + each removal names the same container the listing showed. + """ + return [ + "docker", "--host", "unix:///var/run/docker.sock", "ps", + "--all", "--quiet", "--no-trunc", "--filter", f"label={label.selector()}", + ] + + +def docker_owned_remove_argv(identifiers: list[str]) -> list[str]: + """Remove exactly the named identifiers, with no wildcard and no prune.""" + return [ + "docker", "--host", "unix:///var/run/docker.sock", "rm", + "--force", *identifiers, + ] + + +def _container_identifier(text: str) -> bool: + """Accept only what Docker prints for an identifier; never a guess.""" + return bool(re.fullmatch(r"[0-9a-f]{12,64}", text)) + + +@dataclass(frozen=True) +class TeardownResult: + """What became of the containers this invocation owned.""" + + outcome: str + detail: str = "" + identifiers: tuple[str, ...] = () + + @property + def confirmed(self) -> bool: + """True only when the absence of owned containers was established.""" + return self.outcome in ("removed", "absent") + + +def call_runner(runner, argv, *, env, cwd, timeout, cleanup_budget=None): + """Call a command runner, handing our own runner the shared cleanup deadline. + + ``cleanup_budget`` is an *existing* absolute deadline: the teardown's docker + calls must finish — including their own stop and drain if they time out — + before the same end that the rest of the cleanup obeys. A nested timeout that + opened a second allowance would overshoot that end, so the object is threaded + through. + + Only ``default_runner`` understands that parameter. An injected runner keeps + exactly the calling convention it already has, so custom runners are neither + surprised by a new keyword nor required to accept one. + """ + if cleanup_budget is not None and runner is default_runner: + return runner( + argv, env=env, cwd=cwd, timeout=timeout, cleanup_budget=cleanup_budget + ) + return runner(argv, env=env, cwd=cwd, timeout=timeout) + + +def _owned_identifiers(label: OwnedLabel, runner, env, cwd, budget: Deadline): + """The owned container identifiers, or a reason they cannot be trusted.""" + if budget.remaining() < MIN_CLEANUP_SECONDS: + return None, "the cleanup deadline expired before the containers could be listed" + outcome = as_outcome( + call_runner( + runner, + docker_owned_list_argv(label), + env=env, + cwd=cwd, + timeout=budget.cap(T_CONTAINER_TEARDOWN), + cleanup_budget=budget, + ) + ) + if outcome.code != 0: + return None, f"the daemon could not list containers carrying {label.selector()}" + if outcome.truncated: + return None, f"the container listing for {label.selector()} was truncated, so it cannot be trusted" + identifiers = [line.strip() for line in outcome.output.splitlines() if line.strip()] + if any(not _container_identifier(identifier) for identifier in identifiers): + return None, f"the daemon printed something that is not a container identifier for {label.selector()}" + return identifiers, "" + + +def teardown_owned_containers( + label: OwnedLabel, + *, + runner, + env, + budget: Deadline, + settle: bool = False, + producer_stop_unconfirmed: bool = False, + cwd=ROOT, + sleep=time.sleep, +) -> TeardownResult: + """Remove exactly the containers carrying ``label``, inside one clean-up deadline. + + This is independent of the child that started them: it runs after the Docker + client has exited, failed, or been killed, because a killed client does not + stop the container the daemon started and ``--rm`` only removes a container + that exits. + + ``budget`` is the single absolute deadline that also covered the process + stop and the drain; every docker call here is capped by the time left on it + and every poll is truncated to it, so the whole cleanup cannot outlive it. A + step that no longer fits is skipped and reported, never run late. + + ``settle`` keeps polling while the deadline allows, because a client killed + while the daemon was still creating the container leaves nothing to remove + *yet*. When the observation window closes without proof, the answer is + ``unconfirmed``: a bounded observation is not proof of eventual absence, and + neither is a stopped process. + + An ambiguous ``docker rm`` result is reported and never retried: a second + attempt could remove a container that appeared after the first. + """ + removed: list[str] = [] + attempted: set[str] = set() + rounds = 0 + while True: + rounds += 1 + if rounds > MAX_TEARDOWN_ROUNDS: + return TeardownResult( + "unconfirmed", + f"gave up after {MAX_TEARDOWN_ROUNDS} teardown rounds; the daemon " + f"kept answering about {label.selector()}", + tuple(removed), + ) + if budget.remaining() < MIN_CLEANUP_SECONDS: + return TeardownResult( + "unconfirmed", + f"the cleanup deadline expired with {len(removed)} container(s) removed for " + f"{label.selector()}", + tuple(removed), + ) + identifiers, problem = _owned_identifiers(label, runner, env, cwd, budget) + if identifiers is None: + return TeardownResult("unconfirmed", problem, tuple(removed)) + if identifiers: + if any(identifier in attempted for identifier in identifiers): + return TeardownResult( + "unconfirmed", + f"still listed after removal: {', '.join(identifiers)}", + tuple(removed), + ) + window = budget.cap(T_CONTAINER_TEARDOWN) + if window < MIN_CLEANUP_SECONDS: + return TeardownResult( + "unconfirmed", + f"the cleanup deadline expired before {', '.join(identifiers)} could be removed", + tuple(removed), + ) + outcome = as_outcome( + call_runner( + runner, + docker_owned_remove_argv(identifiers), + env=env, + cwd=cwd, + timeout=window, + cleanup_budget=budget, + ) + ) + if outcome.code != 0: + return TeardownResult( + "unconfirmed", + f"ambiguous removal result (status {outcome.code}) for " + f"{', '.join(identifiers)}; not retried", + tuple(removed), + ) + attempted.update(identifiers) + removed.extend(identifiers) + continue # re-probe: removal is only claimed once it is confirmed + if removed: + return TeardownResult( + "removed", + f"removed {len(removed)} owned container(s) and confirmed none remain", + tuple(removed), + ) + if not settle: + return TeardownResult( + "absent", + f"no container carrying {label.selector()} was present", + (), + ) + if producer_stop_unconfirmed: + return TeardownResult( + "unconfirmed", + f"the process that owned {label.selector()} could not be confirmed stopped, " + "so no container carrying that label was seen yet is not proof that none exists", + (), + ) + if budget.remaining() <= TEARDOWN_POLL_SECONDS: + return TeardownResult( + "unconfirmed", + f"bounded observation of {label.selector()} ended with the cleanup deadline " + "and saw nothing; a container may still appear", + (), + ) + sleep(min(TEARDOWN_POLL_SECONDS, budget.remaining())) + + +def docker_run_argv(container_action: str, budget_seconds: int = None, name: str = None, labels=()) -> list[str]: """Run a container action against the mounted source, with no host socket. The only bind mount is the source tree. The container uses the invoking guest user's numeric identity so a root-owned CI checkout remains writable without broad chmod or host access. There is no Docker socket or privileged mode; network access uses the default bridge for explicit dependency setup. + + ``budget_seconds`` is the inner helper's own budget. It is always shorter + than the outer ``docker run`` timeout, so the helper that knows which stage + it was reports the expiry before the outer process group is terminated. + + ``name`` and ``labels`` are this invocation's ownership marks: a unique name + and a unique per-invocation label, so teardown can address exactly this + container after the client is gone. """ - return [ - "docker", "--host", "unix:///var/run/docker.sock", "run", "--rm", + if budget_seconds is None: + budget_seconds = CI_BUDGET_SECONDS - CONTAINER_HANDOFF_SECONDS + argv = ["docker", "--host", "unix:///var/run/docker.sock", "run"] + if name: + argv += ["--name", name] + argv += [ + "--rm", "--user", f"{os.getuid()}:{os.getgid()}", "--network", "default", "-v", f"{ROOT}:/work", "-w", "/work", + ] + for label in labels or (): + argv += ["--label", label] + argv += [ RELEASE_ENV_IMAGE, "python3", "tools/run_ci.py", "_container", container_action, + "--budget-seconds", str(int(budget_seconds)), ] + return argv # -------------------------------------------------------------------------- @@ -604,17 +1497,101 @@ def docker_run_argv(container_action: str) -> list[str]: # -------------------------------------------------------------------------- -def run_container(action: str, runner, env): - """Build the prepared image, then run an action inside it.""" +def run_container(action: str, runner, env, deadline=None): + """Build the prepared image, then run an action inside it. + + The build draws on the same budget as everything else, and the container + gets the remaining time minus the handoff margin, so the inner helper's + deadline always fires before the outer ``docker run`` is killed. + + The container is created with a unique name and a unique per-invocation + label, and it is removed by that identity in a ``finally`` block. A killed + ``docker run`` client does not stop the container the daemon started, so + the removal does not depend on the client exiting cleanly. + """ + deadline = default_deadline() if deadline is None else deadline steps = container_steps(action) assert_safe_steps(steps) require_docker(env) - run_checked(cmd(docker_build_argv(), T_DOCKER_BUILD), runner, env) - run_checked(cmd(docker_run_argv(action), T_DOCKER_RUN), runner, env) + run_checked( + cmd(docker_build_argv(), T_DOCKER_BUILD), + runner, env, deadline=deadline, stage=f"docker build ({action})", + ) + outer_timeout = deadline.cap(T_DOCKER_RUN) + inner_budget = outer_timeout - CONTAINER_HANDOFF_SECONDS + if inner_budget < MIN_COMMAND_SECONDS: + raise BudgetExpired( + deadline, + f"container action {action!r}", + detail=( + f"only {outer_timeout}s of the {deadline.total_seconds}s budget is left " + f"for the outer docker run, which leaves {inner_budget}s for the " + f"{CONTAINER_HANDOFF_SECONDS}s-shorter inner helper budget" + ), + ) + token = new_ownership_token() + label = ci_owned_label(token) + name = f"{CONTAINER_NAME_PREFIX}-{action}-{token}" + print( + f"[run_ci] container action {action!r} owns {name} " + f"(label {label.selector()})", + flush=True, + ) + cleanup = None + resource_uncertain = False + producer_unconfirmed = False + failed = False + try: + run_command_under( + cmd( + docker_run_argv( + action, + inner_budget, + name=name, + labels=[ + f"{CONTAINER_LABEL_KEY}={token}", + f"{CONTAINER_STAGE_LABEL_KEY}={action}", + ], + ), + outer_timeout, + ), + runner, env, deadline=deadline, stage=f"container action {action!r}", + ) + except CommandFailed as error: + failed = True + # The client was killed, forced, or could not be confirmed stopped: the + # daemon may still be creating or running the container, so absence + # cannot be inferred from the process outcome. + cleanup = error.cleanup + resource_uncertain = error.timed_out or error.cleanup_unconfirmed or error.forced + producer_unconfirmed = error.cleanup_unconfirmed + raise + except BaseException: + failed = True + resource_uncertain = True + producer_unconfirmed = True + raise + finally: + budget = cleanup or Deadline(CLEANUP_RESERVE_SECONDS, label="container cleanup") + result = teardown_owned_containers( + label, + runner=runner, + env=env, + budget=budget, + settle=resource_uncertain, + producer_stop_unconfirmed=producer_unconfirmed, + ) + print( + f"[run_ci] container teardown for {name}: {result.outcome}: {result.detail}", + flush=True, + ) + if not failed and not result.confirmed: + raise CleanupUnconfirmed(name, result.detail) -def run_container_steps(action: str, runner, env): +def run_container_steps(action: str, runner, env, deadline=None): """Use provisioned toolchains and an owned home for arbitrary guest UIDs.""" + deadline = default_deadline() if deadline is None else deadline steps = container_steps(action) assert_safe_steps(steps) # Fixed image paths are not inherited overrides from the workflow environment. @@ -627,40 +1604,105 @@ def run_container_steps(action: str, runner, env): "RUSTUP_HOME": "/usr/local/rustup", "PLAYWRIGHT_BROWSERS_PATH": "/opt/ms-playwright", }) - execute_steps(steps, runner, child_env, cwd=ROOT) + execute_steps(steps, runner, child_env, cwd=ROOT, deadline=deadline) + + +def run_smoke(runner, env, deadline=None): + run_container("smoke", runner, env, deadline=deadline) -def run_smoke(runner, env): - run_container("smoke", runner, env) +def run_runtime_steps(runner, env, deadline=None): + """Run the runtime harness, owning the containers it starts. + The harness is given this invocation's id and told that the caller removes + its containers, so no two parties race for the same ``docker rm``. Its + scratch HOME stays alive until the teardown below has finished, and that + teardown spends the same cleanup deadline that stopped the process. -def run_runtime_steps(runner, env): - """Keep runtime-harness Docker configuration local and credential-free.""" + A reaped harness is not by itself proof that its containers are gone: an + empty listing after a forced or unconfirmed stop is reported as unconfirmed. + """ + deadline = default_deadline() if deadline is None else deadline + token = new_ownership_token() + label = runtime_owned_label(token) + print( + f"[run_ci] runtime harness owns containers labelled {label.selector()}", + flush=True, + ) + cleanup = None + resource_uncertain = False + producer_unconfirmed = False + failed = False with tempfile.TemporaryDirectory(prefix="ultimate-tool-ci-runtime-home-") as home: child_env = dict(env) child_env["HOME"] = home - execute_steps(runtime_steps(), runner, child_env, cwd=ROOT) + try: + for command in runtime_steps(token): + run_command_under(command, runner, child_env, cwd=ROOT, deadline=deadline) + except CommandFailed as error: + failed = True + cleanup = error.cleanup + # A timeout is enough to make the container fate uncertain: the + # harness may have been killed with a container half-created. The + # producer flag stays separate, because a cooperative stop can be + # perfectly confirmed while its containers still are not. + resource_uncertain = ( + error.timed_out or error.cleanup_unconfirmed or error.forced + ) + producer_unconfirmed = error.cleanup_unconfirmed + raise + except BaseException: + failed = True + resource_uncertain = True + producer_unconfirmed = True + raise + finally: + budget = cleanup or Deadline(CLEANUP_RESERVE_SECONDS, label="runtime container cleanup") + result = teardown_owned_containers( + label, + runner=runner, + env=child_env, + budget=budget, + settle=resource_uncertain, + producer_stop_unconfirmed=producer_unconfirmed, + ) + print( + f"[run_ci] runtime container teardown for {label.selector()}: " + f"{result.outcome}: {result.detail}", + flush=True, + ) + if not failed and not result.confirmed: + raise CleanupUnconfirmed(label.selector(), result.detail) -def run_check(runner, env): - run_container("check", runner, env) - run_runtime_steps(runner, env) +def run_check(runner, env, deadline=None): + """The full gate: container stages, then the runtime harness, one budget.""" + deadline = default_deadline() if deadline is None else deadline + run_container("check", runner, env, deadline=deadline) + run_runtime_steps(runner, env, deadline=deadline) -def run_package(runner, env): - run_container("package", runner, env) - run_runtime_steps(runner, env) +def run_package(runner, env, deadline=None): + deadline = default_deadline() if deadline is None else deadline + run_container("package", runner, env, deadline=deadline) + run_runtime_steps(runner, env, deadline=deadline) -def detect_vice_version(exe: str, runner, env): - code, output = runner([exe, "--version"], env=env, cwd=ROOT, timeout=T_VALIDATE) +def detect_vice_version(exe: str, runner, env, deadline=None): + deadline = default_deadline() if deadline is None else deadline + deadline.require("VICE version query") + code, output = runner( + [exe, "--version"], env=env, cwd=ROOT, + timeout=max(1, deadline.cap(T_VALIDATE)), + ) if code != 0: raise PreconditionError(f"could not query the VICE version from {exe!r}") match = re.search(r"(\d+\.\d+)", output) return match.group(1) if match else None -def run_emulator(args, runner, env): +def run_emulator(args, runner, env, deadline=None): + deadline = default_deadline() if deadline is None else deadline exe = args.vice_executable or env.get("VICE_EXECUTABLE") if not exe: raise PreconditionError( @@ -670,7 +1712,7 @@ def run_emulator(args, runner, env): if not os.path.exists(exe) and shutil.which(exe, path=env.get("PATH")) is None: raise PreconditionError(f"VICE executable not found: {exe!r}") if args.require_vice_version: - found = detect_vice_version(exe, runner, env) + found = detect_vice_version(exe, runner, env, deadline=deadline) if found != args.require_vice_version: raise PreconditionError( f"required VICE {args.require_vice_version}, found {found!r}; " @@ -682,11 +1724,11 @@ def run_emulator(args, runner, env): ) steps = emulator_steps(exe) assert_safe_steps(steps) - execute_steps(steps, runner, env, cwd=ROOT) + execute_steps(steps, runner, env, cwd=ROOT, deadline=deadline) -def run_guest_check(runner, env): - verify_guest(env, runner) +def run_guest_check(runner, env, deadline=None): + verify_guest(env, runner, deadline=deadline) # -------------------------------------------------------------------------- @@ -701,39 +1743,67 @@ def build_parser() -> argparse.ArgumentParser: ) sub = parser.add_subparsers(dest="action", required=True) - sub.add_parser("guest-check", help="assert the CI guest preconditions") - sub.add_parser("smoke", help="build, validate, and run the first-run fake") - sub.add_parser("check", help="the whole gate in the prepared image") - sub.add_parser("package", help="the release path and runtime harness") + def action(name: str, help_text: str): + child = sub.add_parser(name, help=help_text) + child.add_argument( + "--budget-seconds", + type=int, + default=None, + help=( + "override this action's time budget for a directly invoked run; " + f"never more than the {CI_BUDGET_SECONDS}s default" + ), + ) + return child + + action("guest-check", "assert the CI guest preconditions") + action("smoke", "build, validate, and run the first-run fake") + action("check", "the whole gate in the prepared image") + action("package", "the release path and runtime harness") - emulator = sub.add_parser("emulator", help="the VICE acceptance run") + emulator = action("emulator", "the VICE acceptance run") emulator.add_argument("--vice-executable", default=None) emulator.add_argument("--require-vice-version", default="3.8") internal = sub.add_parser("_container", help=argparse.SUPPRESS) internal.add_argument("container_action", choices=CONTAINER_ACTIONS) + internal.add_argument("--budget-seconds", type=int, default=None) return parser +def action_budget(requested) -> int: + """Return the action's budget, never above the helper's own ceiling. + + The outer action hands the container a shorter budget so the inner helper + reports first; a larger request cannot raise the ceiling above + ``CI_BUDGET_SECONDS``, because that ceiling is derived from the workflow + step and from nothing this helper is allowed to change. + """ + if requested is None: + return CI_BUDGET_SECONDS + return max(MIN_COMMAND_SECONDS, min(int(requested), CI_BUDGET_SECONDS)) + + def main(argv=None) -> int: parser = build_parser() args = parser.parse_args(argv) env = safe_env(os.environ) runner = default_runner try: + deadline = Deadline(action_budget(args.budget_seconds)) if args.action == "guest-check": - run_guest_check(runner, env) + run_guest_check(runner, env, deadline=deadline) elif args.action == "smoke": - run_smoke(runner, env) + run_smoke(runner, env, deadline=deadline) elif args.action == "check": - run_check(runner, env) + run_check(runner, env, deadline=deadline) elif args.action == "package": - run_package(runner, env) + run_package(runner, env, deadline=deadline) elif args.action == "emulator": - run_emulator(args, runner, env) + run_emulator(args, runner, env, deadline=deadline) elif args.action == "_container": - run_container_steps(args.container_action, runner, env) + run_container_steps(args.container_action, runner, env, deadline=deadline) else: # pragma: no cover - argparse enforces the choice set raise UsageError(f"unknown action {args.action!r}") except UsageError as error: @@ -745,6 +1815,12 @@ def main(argv=None) -> int: except CommandFailed as error: print(f"run_ci: {error}", file=sys.stderr) return EXIT_FAILED + except CleanupUnconfirmed as error: + print(f"run_ci: {error}", file=sys.stderr) + return EXIT_INTERNAL + except BudgetExpired as error: + print(f"run_ci: {error}", file=sys.stderr) + return EXIT_TIMEOUT except (OSError, subprocess.SubprocessError) as error: print(f"run_ci: internal error: {error}", file=sys.stderr) return EXIT_INTERNAL diff --git a/tools/verify_release_local.py b/tools/verify_release_local.py index 6b8670e..61cc103 100644 --- a/tools/verify_release_local.py +++ b/tools/verify_release_local.py @@ -29,6 +29,14 @@ packager-facing record: directory by calling the packager's own validator, so the record is proven consumable instead of assumed consumable. +``--invocation-id `` + Own this run under a caller-supplied id instead of the fresh id derived from + the process id and random bytes. The id labels every container this run + starts (``ultimate-tool-verify-local.``) and is validated before any + journal, scratch directory, image or container exists, so a caller that + knows the id can address exactly this run's containers from outside — which + is what makes an independent cleanup possible after this process was killed. + What these gates do not claim ----------------------------- @@ -105,11 +113,33 @@ BINARY_ARCHIVE_KINDS = ("host-binary",) # scoped to the containers this invocation actually started. INVOCATION = "unset" +# An invocation id becomes a Docker label suffix, part of a runtime image tag and +# part of a scratch directory name. It is validated before any of those exist, +# because an id that is not lower-case alphanumeric plus ``-`` could address a +# different invocation's containers or step outside the scratch parent. +INVOCATION_ID = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$") + def new_invocation_id() -> str: return f"{os.getpid()}-{os.urandom(6).hex()}" +def validate_invocation_id(value: object) -> str: + """Return ``value`` when it is safe as an invocation id, else refuse it. + + A caller that supplies the id (see ``--invocation-id``) can then address + exactly this run's containers from outside, which is what makes an + independent cleanup possible after this process was killed. The default is + still a fresh id per invocation. + """ + if not isinstance(value, str) or not INVOCATION_ID.fullmatch(value): + raise ValueError( + f"invocation id {value!r} is not permitted: use 1-64 characters of " + "'a-z', '0-9' and '-', starting with a letter or digit" + ) + return value + + def container_label() -> str: return f"{CONTAINER_LABEL}.{INVOCATION}" @@ -2515,16 +2545,26 @@ def run_runtime(args: argparse.Namespace, journal: Journal, cleanup: Cleanup) -> lambda root=removable_root: shutil.rmtree(root, ignore_errors=True), condition=lambda: not scratch_state["preserve"], ) - cleanup.add( - f"remove containers labelled {container_label()}", - lambda: remove_containers_with_label(container_label()), - ) + if args.external_container_cleanup: + # The caller is the sole remover for this run's containers, so nothing + # here may race it: two removers would make every ``docker rm`` result + # ambiguous. + journal.note( + f"container removal for label {container_label()} is owned by the calling " + "supervisor (--external-container-cleanup)" + ) + else: + cleanup.add( + f"remove containers labelled {container_label()}", + lambda: remove_containers_with_label(container_label()), + ) # A container left behind by an earlier invocation carries a different label # and is deliberately left alone; only this invocation's label is cleared. - removed = remove_containers_with_label(container_label()) - if removed: - journal.note(f"removed {len(removed)} leftover container(s) carrying this invocation's label") + if not args.external_container_cleanup: + removed = remove_containers_with_label(container_label()) + if removed: + journal.note(f"removed {len(removed)} leftover container(s) carrying this invocation's label") runtime_root = work_root / "runtime" extraction = extract_verified_archive( @@ -2682,9 +2722,17 @@ def run_runtime(args: argparse.Namespace, journal: Journal, cleanup: Cleanup) -> observed = load_json_file(report_path, "runtime report") leftovers = docker_containers_with_label(container_label()) - if leftovers: + if leftovers and not args.external_container_cleanup: removed_after = remove_containers_with_label(container_label()) journal.note(f"removed leftover containers after the runtime run: {removed_after}") + elif leftovers: + # Observed, not removed, and recorded as such: this run's containers are + # the calling supervisor's to remove, and the record must not imply they + # were cleaned here. + journal.note( + f"{len(leftovers)} container(s) carrying {container_label()} are left for the " + "calling supervisor to remove (--external-container-cleanup)" + ) failures: list[str] = list(observed.get("failures") or []) if driver_run.exit_code not in (0, 1): @@ -2925,6 +2973,26 @@ def build_parser() -> argparse.ArgumentParser: help="a path that must not be reachable from a verification container (repeatable)", ) parser.add_argument("--json", action="store_true", help="print the record to stdout") + parser.add_argument( + "--invocation-id", + default=None, + help=( + "own this run under a caller-supplied invocation id instead of a " + "fresh one; it labels every container this run starts, so a caller " + "that knows it can address exactly this run's containers" + ), + ) + parser.add_argument( + "--external-container-cleanup", + action="store_true", + help=( + "leave removal of this run's labelled containers to the caller. " + "Requires --mode runtime and an explicit --invocation-id. The " + "harness then skips its own container removal (pre-clean, registered " + "cleanup and post-run sweep) and keeps its read-only checks and its " + "scratch-directory cleanup, so exactly one party removes containers" + ), + ) return parser @@ -2943,6 +3011,24 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) global INVOCATION INVOCATION = new_invocation_id() + if args.invocation_id is not None: + # Before any journal, scratch directory, image or container exists, so an + # unacceptable id cannot leave a resource behind under a name it chose. + try: + INVOCATION = validate_invocation_id(args.invocation_id) + except ValueError as error: + parser.error(str(error)) + if args.external_container_cleanup: + # The supervisor that supplies the id is the only party that removes this + # run's containers; it cannot do that for a mode that starts none, and it + # cannot address them at all without an explicit id. + if args.mode != "runtime": + parser.error("--external-container-cleanup requires --mode runtime") + if args.invocation_id is None: + parser.error( + "--external-container-cleanup requires an explicit --invocation-id " + "so the caller can address exactly this run's containers" + ) args.release_dir = args.release_dir.resolve() args.checkout_root = args.checkout_root.resolve() args.forbidden = [path.resolve() for path in (args.forbid or default_forbidden(args.checkout_root))] -- 2.54.0