fix(gateway): contain output pump shutdown races

bot_bottle/backend/docker/cleanup.py

@@ -28,7 +28,6 @@ import subprocess
 
 from ...paths import bot_bottle_root
 from ...log import info, warn
-from .. import EnumerationError
 from . import util as docker_mod
 from .bottle_cleanup_plan import DockerBottleCleanupPlan
 from ...bottle_state import bottle_state_dir, is_preserved
@@ -37,17 +36,15 @@ from .compose import COMPOSE_PROJECT_PREFIX, list_compose_projects
 
 def _list_prefixed_containers() -> list[str]:
     """All bot-bottle-prefixed containers, running or stopped."""
-    try:
-        result = subprocess.run(
-            ["docker", "ps", "-a",
-             "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}",
-             "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
-            capture_output=True, text=True, check=False,
-        )
-    except OSError as exc:
-        raise EnumerationError(f"docker ps failed: {exc}") from exc
+    result = subprocess.run(
+        ["docker", "ps", "-a",
+         "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}",
+         "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
+        capture_output=True, text=True, check=False,
+    )
     if result.returncode != 0:
-        raise EnumerationError(f"docker ps failed: {result.stderr.strip()}")
+        warn(f"docker ps failed: {result.stderr.strip()}")
+        return []
     out: list[str] = []
     for line in (result.stdout or "").splitlines():
         if not line:
@@ -66,19 +63,15 @@ def _list_prefixed_networks() -> list[str]:
     to a compose project. Compose-managed networks have a
     `com.docker.compose.project` label; bare ones (from pre-compose
     code paths) don't."""
-    try:
-        result = subprocess.run(
-            ["docker", "network", "ls",
-             "--filter", f"name={COMPOSE_PROJECT_PREFIX}",
-             "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
-            capture_output=True, text=True, check=False,
-        )
-    except OSError as exc:
-        raise EnumerationError(f"docker network ls failed: {exc}") from exc
+    result = subprocess.run(
+        ["docker", "network", "ls",
+         "--filter", f"name={COMPOSE_PROJECT_PREFIX}",
+         "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
+        capture_output=True, text=True, check=False,
+    )
     if result.returncode != 0:
-        raise EnumerationError(
-            f"docker network ls failed: {result.stderr.strip()}"
-        )
+        warn(f"docker network ls failed: {result.stderr.strip()}")
+        return []
     out: list[str] = []
     for line in (result.stdout or "").splitlines():
         if not line:
@@ -127,10 +120,7 @@ def prepare_cleanup() -> DockerBottleCleanupPlan:
     `enumerate_active_agents()` so the orphan-state-dir bucket
     doesn't include slugs whose non-docker bottle is still up."""
     docker_mod.require_docker()
-    projects = list_compose_projects(
-        warn_on_error=False,
-        raise_on_error=True,
-    )
+    projects = list_compose_projects()
     project_set = set(projects)
     # Late import to avoid a circular at module-load time —
     # the backend package's __init__ imports this module.

bot_bottle/backend/firecracker/cleanup.py

@@ -11,9 +11,10 @@ Reaps *orphans* only — resources with no live VM behind them:
     — a VMM left lingering after its dir was removed.
 
 A run dir with a *live* firecracker process is a running bottle and is
-left strictly alone: it is neither killed nor removed. Active-agent
-enumeration uses this same process snapshot, so cleanup and generic
-backend consumers agree about which bottles are running.
+left strictly alone: it is neither killed nor removed. (The backend's
+`enumerate_active` registry is still a stub — #354 — so a live process
+is the only reliable "this bottle is in use" signal we have. Once the
+registry lands, registry-orphaned-but-running VMs can be reaped too.)
 
 TAP slots free themselves (the flock drops when the launcher exits), so
 there is nothing to reclaim there.
@@ -21,7 +22,6 @@ there is nothing to reclaim there.
 
 from __future__ import annotations
 
-from collections.abc import Sequence
 import os
 import shutil
 import signal
@@ -38,7 +38,7 @@ def _run_root() -> Path:
     return util.cache_dir() / "run"
 
 
-def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None:
+def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
     """The bottle run dir a firecracker cmdline belongs to, or None.
 
     A bottle VM is launched with `--config-file <run_root>/<slug>/config.json`,
@@ -46,35 +46,15 @@ def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None:
     the run root. Anything else (a builder VM, the infra VM elsewhere) is
     not ours to reap here.
     """
-    for i, arg in enumerate(args):
-        if arg == "--config-file" and i + 1 < len(args):
-            parent = Path(args[i + 1]).parent
+    toks = cmd.split()
+    for i, tok in enumerate(toks):
+        if tok == "--config-file" and i + 1 < len(toks):
+            parent = Path(toks[i + 1]).parent
             if parent.parent == run_root:
                 return parent
     return None
 
 
-def _decode_cmdline(raw: bytes) -> tuple[str, ...]:
-    """Decode Linux's NUL-delimited argv without losing embedded spaces."""
-    return tuple(
-        value.decode(errors="surrogateescape")
-        for value in raw.split(b"\0") if value
-    )
-
-
-def _process_args(pid: int) -> tuple[str, ...] | None:
-    """Read one process's lossless argv, or None when it exited meanwhile."""
-    try:
-        raw = Path(f"/proc/{pid}/cmdline").read_bytes()
-    except FileNotFoundError:
-        return None
-    except OSError as exc:
-        raise EnumerationError(
-            f"could not inspect Firecracker pid {pid}: {exc}"
-        ) from exc
-    return _decode_cmdline(raw)
-
-
 def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     """Inspect running firecracker VMs under ``run_root``.
 
@@ -85,7 +65,7 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     """
     try:
         result = subprocess.run(
-            ["pgrep", "firecracker"],
+            ["pgrep", "-a", "firecracker"],
             capture_output=True, text=True, check=False,
         )
     except OSError as exc:
@@ -103,14 +83,14 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     live: set[str] = set()
     orphan_pids: list[int] = []
     for line in result.stdout.splitlines():
+        parts = line.split(None, 1)
+        if len(parts) != 2:
+            continue
         try:
-            pid = int(line.strip())
+            pid = int(parts[0])
         except ValueError:
             continue
-        args = _process_args(pid)
-        if args is None:
-            continue
-        run_dir = _run_dir_of(args, run_root)
+        run_dir = _run_dir_of(parts[1], run_root)
         if run_dir is None:
             continue
         if run_dir.is_dir():
@@ -177,8 +157,8 @@ def _terminate_orphan(pid: int, run_root: Path) -> None:
             raise EnumerationError(
                 f"could not revalidate Firecracker pid {pid}: {exc}"
             ) from exc
-        args = _decode_cmdline(raw)
-        run_dir = _run_dir_of(args, run_root)
+        command = raw.replace(b"\0", b" ").decode(errors="replace")
+        run_dir = _run_dir_of(command, run_root)
         if run_dir is None or run_dir.is_dir():
             return
         info(f"kill firecracker VM pid {pid}")

bot_bottle/backend/firecracker/infra_artifact.py

@@ -41,8 +41,6 @@ from ... import resources
 from ...log import die, info
 from . import util
 
-ARTIFACT_HTTP_TIMEOUT_SECONDS = 30.0
-
 # Bump if the on-disk artifact *format* changes (compression, layout) so a new
 # scheme can't collide with a cached/published artifact of the old one.
 _ARTIFACT_FORMAT = "1"
@@ -166,9 +164,7 @@ def _download(url: str, dest: Path) -> None:
     """Stream `url` to `dest` (atomic via a `.part` sibling)."""
     tmp = dest.with_suffix(dest.suffix + ".part")
     try:
-        with urllib.request.urlopen(
-            _open(url), timeout=ARTIFACT_HTTP_TIMEOUT_SECONDS,
-        ) as resp, open(tmp, "wb") as out:
+        with urllib.request.urlopen(_open(url)) as resp, open(tmp, "wb") as out:
             shutil.copyfileobj(resp, out, _CHUNK)
     except urllib.error.HTTPError as e:
         tmp.unlink(missing_ok=True)

bot_bottle/backend/firecracker/publish_infra.py

@@ -34,7 +34,6 @@ from pathlib import Path
 from . import infra_artifact, infra_vm, util
 
 _CHUNK = 1 << 20
-_REGISTRY_HTTP_TIMEOUT_SECONDS = 30.0
 
 _GZ_NAME = "rootfs.ext4.gz"
 _SHA_NAME = "rootfs.ext4.gz.sha256"
@@ -92,9 +91,7 @@ def _put(url: str, body: "bytes | Path", token: str) -> None:
         req.add_header("Authorization", f"token {token}")
     req.add_header("Content-Type", "application/octet-stream")
     try:
-        with urllib.request.urlopen(
-            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
-        ) as resp:
+        with urllib.request.urlopen(req) as resp:
             print(f"  uploaded {url} (HTTP {resp.status})")
     except urllib.error.HTTPError as e:
         if e.code == 409:
@@ -115,9 +112,7 @@ def _delete(url: str, token: str) -> None:
     if token:
         req.add_header("Authorization", f"token {token}")
     try:
-        with urllib.request.urlopen(
-            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
-        ):
+        with urllib.request.urlopen(req):
             pass
     except urllib.error.HTTPError as e:
         if e.code != 404:
@@ -156,10 +151,7 @@ def _try_download_published(role: str, role_dir: Path) -> str | None:
     version = _role_version(role)
     sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role)
     try:
-        with urllib.request.urlopen(
-            infra_artifact._open(sha_url),
-            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
-        ):
+        with urllib.request.urlopen(infra_artifact._open(sha_url)):
             pass
     except urllib.error.HTTPError as e:
         if e.code == 404:
@@ -203,10 +195,7 @@ def _publish_bundle(role: str, role_dir: Path, token: str) -> str:
     # present, a re-publish is a no-op. Otherwise clear any partial upload left
     # by an interrupted prior attempt and upload the complete set.
     try:
-        with urllib.request.urlopen(
-            infra_artifact._open(sha_url),
-            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
-        ) as resp:
+        with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp:
             remote_sha = resp.read().decode("utf-8").split()[0].strip().lower()
     except urllib.error.HTTPError as e:
         if e.code != 404:

docs/prds/0082-authoritative-failure-boundaries.md

@@ -13,9 +13,7 @@ resource-consuming boundary revalidate the assumptions it acts on. This
 finishes the focused quality work begun under #444 without broad rewrites:
 cleanup cannot act on stale identities, policy introspection cannot publish a
 fabricated empty policy, gateway servers bound untrusted work, and daemon
-shutdown does not emit uncaught background-thread failures. Shared
-control-plane storage and gateway credential provisioning also enforce their
-filesystem security contract before sensitive data is written.
+shutdown does not emit uncaught background-thread failures.
 
 ## Problem
 
@@ -47,18 +45,6 @@ misleading behavior:
 8. Firecracker artifact downloads and registry publication have no network
    deadline, so an unresponsive registry can hold setup or release work
    indefinitely.
-9. Authenticated secret blobs select the unauthenticated legacy decoder when
-   their in-band version prefix is changed, allowing storage tampering to
-   bypass tag verification.
-10. Cleanup executes the entire post-confirmation snapshot rather than the
-    intersection with what the operator saw, and mutation failures are not
-    reflected in the command result.
-11. Git smart-HTTP can retain sixteen 100 MiB request bodies concurrently,
-    cleanup mutations have no subprocess deadline, and Firecracker signalling
-    failures bypass shared mutation accounting.
-12. SQLite creates the shared control-plane database before its mode is
-    restricted, then suppresses permission-repair failures. Gateway transports
-    also differ in whether copied deploy-key modes are preserved.
 
 These are one design problem: state used to authorize deletion, replacement,
 or resource allocation must be authoritative at the point of use.
@@ -87,20 +73,6 @@ or resource allocation must be authoritative at the point of use.
   as completion while preserving diagnostics for unexpected failures.
 - Artifact pull, existence-check, and publication requests use explicit
   network deadlines.
-- Persisted secrets accept only the authenticated format. The schema migration
-  intentionally clears legacy rows; local agents are reprovisioned rather
-  than retaining a ciphertext-controlled downgrade path.
-- Cleanup executes only resources present in both the displayed and current
-  authoritative plans, attempts every approved mutation, and returns failure
-  when any mutation does not complete.
-- Git request bodies spool to disk behind a separate heavy-work semaphore;
-  cleanup commands have configurable deadlines; Firecracker signalling
-  failures aggregate while identity-verification uncertainty still aborts.
-- The shared database directory and file are private before SQLite writes any
-  control-plane state; an inability to enforce those modes aborts startup.
-- Gateway credential directories and files receive explicit private modes
-  inside the gateway, independent of Docker, Apple Container, or SSH copy
-  semantics.
 - Unit tests cover PID/path reuse, partial backend enumeration, transient
   policy resolution failure, slow bodies, concurrency saturation, and stream
   closure races.
@@ -163,17 +135,6 @@ The gateway output pump catches only stream-closure exceptions expected after
 the supervisor closes child pipes. Other I/O failures remain visible and are
 reported through the supervisor's normal diagnostic channel.
 
-### Shared filesystem security
-
-The common SQLite store owns database creation for every backend. It creates
-the parent directory and an empty database with private modes before opening
-SQLite, repairs existing modes, verifies the resulting state, and propagates
-every enforcement failure. Backend launchers do not duplicate this policy.
-
-The backend-neutral gateway provisioner likewise applies directory and file
-modes after transport copies complete. This avoids relying on copy behavior
-that differs among Docker, Apple Container, and Firecracker's SSH transport.
-
 ## Implementation chunks
 
 1. Existing fail-closed security and backend enumeration fixes.
@@ -185,12 +146,6 @@ that differs among Docker, Apple Container, and Firecracker's SSH transport.
 7. Gateway shutdown log-pump closure handling.
 8. Lossless Firecracker process identities, authoritative Docker cleanup
    queries, and bounded Firecracker artifact transfers.
-9. Mandatory authenticated secret storage, shared cleanup-plan intersection
-   and mutation accounting, and contained Git backend process failures.
-10. Disk-spooled and separately bounded Git bodies, cleanup command deadlines,
-    and classified Firecracker signalling failures.
-11. Fail-closed shared database creation and backend-neutral gateway credential
-    permissions.
 
 ## Open questions
 

tests/unit/test_docker_cleanup.py

@@ -14,12 +14,9 @@ from __future__ import annotations
 import tempfile
 import unittest
 from pathlib import Path
-from unittest.mock import patch
 
 from tests.unit import use_bottle_root
 from bot_bottle import bottle_state
-from bot_bottle.backend import EnumerationError
-from bot_bottle.backend.docker import cleanup
 from bot_bottle.backend.docker.cleanup import _list_orphan_state_dirs
 
 
@@ -119,37 +116,5 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         )
 
 
-class TestAuthoritativeDiscovery(unittest.TestCase):
-    def test_prepare_requires_authoritative_compose_projects(self):
-        with patch.object(cleanup.docker_mod, "require_docker"), \
-             patch.object(
-                 cleanup, "list_compose_projects",
-                 side_effect=EnumerationError("compose unavailable"),
-             ) as projects, self.assertRaisesRegex(
-                 EnumerationError, "compose unavailable",
-             ):
-            cleanup.prepare_cleanup()
-        projects.assert_called_once_with(
-            warn_on_error=False,
-            raise_on_error=True,
-        )
-
-    def test_container_query_failure_raises(self):
-        failed = cleanup.subprocess.CompletedProcess(
-            [], 1, stdout="", stderr="daemon unavailable",
-        )
-        with patch.object(cleanup.subprocess, "run", return_value=failed), \
-             self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
-            cleanup._list_prefixed_containers()
-
-    def test_network_query_failure_raises(self):
-        failed = cleanup.subprocess.CompletedProcess(
-            [], 1, stdout="", stderr="daemon unavailable",
-        )
-        with patch.object(cleanup.subprocess, "run", return_value=failed), \
-             self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
-            cleanup._list_prefixed_networks()
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_firecracker_cleanup.py

@@ -32,19 +32,15 @@ class TestProcessScan(unittest.TestCase):
         self.assertEqual(
             Path("/cache/run/dev-a"),
             fc_cleanup._run_dir_of(
-                ("firecracker", "--config-file",
-                 f"{run_root}/dev-a/config.json"), run_root
+                f"firecracker --config-file {run_root}/dev-a/config.json", run_root
             ),
         )
         # infra/builder VMs elsewhere, or nested paths, are not ours.
         self.assertIsNone(
-            fc_cleanup._run_dir_of(
-                ("firecracker", "--config-file", "/elsewhere/config.json"),
-                run_root,
-            )
+            fc_cleanup._run_dir_of("firecracker --config-file /elsewhere/config.json", run_root)
         )
         self.assertIsNone(
-            fc_cleanup._run_dir_of(("firecracker", "--no-config"), run_root)
+            fc_cleanup._run_dir_of("firecracker --no-config", run_root)
         )
 
     def test_scan_splits_live_dirs_from_orphan_pids(self):
@@ -52,40 +48,17 @@ class TestProcessScan(unittest.TestCase):
             run_root = Path(tmp)
             (run_root / "live-a").mkdir()  # dir present -> live VM, protected
             # "gone-b" dir intentionally absent -> lingering VMM, orphan pid
-            args = {
-                111: ("firecracker", "--config-file",
-                      f"{run_root}/live-a/config.json"),
-                222: ("firecracker", "--config-file",
-                      f"{run_root}/gone-b/config.json"),
-                333: ("firecracker", "--config-file", "/elsewhere/config.json"),
-            }
-            with patch.object(
-                fc_cleanup.subprocess, "run",
-                return_value=_proc("111\n222\n333\nnotanint\n"),
-            ), patch.object(
-                fc_cleanup, "_process_args", side_effect=args.get,
-            ):
+            out = (
+                f"111 firecracker --config-file {run_root}/live-a/config.json\n"
+                f"222 firecracker --config-file {run_root}/gone-b/config.json\n"
+                "333 firecracker --config-file /elsewhere/config.json\n"
+                "notanint firecracker --config-file x\n"
+            )
+            with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)):
                 live, orphan_pids = fc_cleanup._scan_processes(run_root)
         self.assertEqual({str(run_root / "live-a")}, live)
         self.assertEqual([222], orphan_pids)
 
-    def test_scan_preserves_spaces_in_config_path(self):
-        with tempfile.TemporaryDirectory(prefix="fc cache ") as tmp:
-            run_root = Path(tmp)
-            live = run_root / "live bottle"
-            live.mkdir()
-            with patch.object(
-                fc_cleanup.subprocess, "run", return_value=_proc("111\n"),
-            ), patch.object(
-                fc_cleanup, "_process_args",
-                return_value=("firecracker", "--config-file",
-                              str(live / "config.json")),
-            ):
-                self.assertEqual(
-                    ({str(live)}, []),
-                    fc_cleanup._scan_processes(run_root),
-                )
-
     def test_scan_empty_when_pgrep_finds_no_processes(self):
         with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)):
             self.assertEqual((set(), []), fc_cleanup._scan_processes(Path("/x")))
@@ -144,14 +117,9 @@ class TestProcessScan(unittest.TestCase):
             run_root = Path(tmp)
             (run_root / "live-a").mkdir()
             (run_root / "dead-b").mkdir()
+            out = f"111 firecracker --config-file {run_root}/live-a/config.json\n"
             with patch.object(fc_cleanup, "_run_root", return_value=run_root), \
-                 patch.object(fc_cleanup.subprocess, "run",
-                              return_value=_proc("111\n")), \
-                 patch.object(
-                     fc_cleanup, "_process_args",
-                     return_value=("firecracker", "--config-file",
-                                   str(run_root / "live-a/config.json")),
-                 ):
+                 patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)):
                 plan = fc_cleanup.prepare_cleanup()
         self.assertEqual((), plan.vm_pids)
         self.assertEqual((str(run_root / "dead-b"),), plan.run_dirs)

tests/unit/test_infra_artifact.py

@@ -284,18 +284,6 @@ class TestEnsureArtifact(_CacheMixin):
             ia.ensure_artifact_gz(version, role=_ROLE)
             self.assertEqual(first_calls, len(net.calls))
 
-    def test_download_uses_network_deadline(self) -> None:
-        response = mock.MagicMock()
-        response.__enter__.return_value = io.BytesIO(b"payload")
-        with tempfile.TemporaryDirectory() as d, mock.patch.object(
-            ia.urllib.request, "urlopen", return_value=response,
-        ) as urlopen:
-            ia._download("https://registry/artifact", Path(d) / "artifact")
-        self.assertEqual(
-            ia.ARTIFACT_HTTP_TIMEOUT_SECONDS,
-            urlopen.call_args.kwargs["timeout"],
-        )
-
     def test_checksum_mismatch_fails_closed(self) -> None:
         version = "beefbeefbeefbeef"
         gz = _gz(b"payload")

tests/unit/test_publish_infra.py

@@ -49,16 +49,6 @@ class TestPut(unittest.TestCase):
         self.assertNotIsInstance(req.data, (bytes, bytearray))
         self.assertEqual(str(len(payload)), req.get_header("Content-length"))
 
-    def test_put_uses_network_deadline(self) -> None:
-        with mock.patch.object(
-            pub.urllib.request, "urlopen", return_value=_Resp(),
-        ) as urlopen:
-            pub._put("https://reg/pkg", b"payload", token="")
-        self.assertEqual(
-            pub._REGISTRY_HTTP_TIMEOUT_SECONDS,
-            urlopen.call_args.kwargs["timeout"],
-        )
-
     def test_small_bytes_body_still_works(self) -> None:
         captured: list[urllib.request.Request] = []