From 7ab85e9ea639149ccef46aeb93009f9d44e8a22f Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 26 Jul 2026 08:43:00 +0000 Subject: [PATCH] fix(orchestrator): address review on host control server transport (#468) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex review on #496: - **High — ambiguous delivery no longer orphans a launched bottle.** A timeout / dropped response from the host controller is now the ambiguous BrokerUnavailableError (distinct from the definite BrokerAuthError / BrokerClientError). OrchestratorCore.launch_bottle keeps the registry row on the ambiguous case instead of deregistering — deregistering would orphan a running container with no record (reconcile reaps rows, never containers). The row is left for reconcile to reap iff the bottle is not actually live. Definite failures still roll back, so a real failure leaves no orphan row. - **Medium — the privileged endpoint bounds request bodies.** The host server rejects an oversized Content-Length with 413 before reading it, and sets a per-request socket timeout, so a caller that can merely reach the socket (no signed token) can't exhaust memory or a handler thread. Tests: ambiguous-keep vs definite-rollback in the launch path; the BrokerUnavailableError/BrokerClientError split in BrokerClient; the 413 body cap + handler error paths (driven in-thread, since daemon request threads lose coverage); and the __main__ entrypoint broker selection. Diff-coverage 98%; pyright clean; pylint 9.88. Co-Authored-By: Claude Opus 4.8 --- bot_bottle/orchestrator/broker.py | 17 ++- bot_bottle/orchestrator/broker_client.py | 25 ++-- bot_bottle/orchestrator/host_server.py | 35 ++++- bot_bottle/orchestrator/service.py | 23 +++- tests/unit/test_orchestrator_broker_client.py | 41 +++++- tests/unit/test_orchestrator_host_server.py | 123 ++++++++++++++++++ tests/unit/test_orchestrator_main.py | 64 +++++++++ tests/unit/test_orchestrator_service.py | 36 ++++- 8 files changed, 334 insertions(+), 30 deletions(-) create mode 100644 tests/unit/test_orchestrator_main.py diff --git a/bot_bottle/orchestrator/broker.py b/bot_bottle/orchestrator/broker.py index 1cd91687..6d5c567f 100644 --- a/bot_bottle/orchestrator/broker.py +++ b/bot_bottle/orchestrator/broker.py @@ -38,7 +38,21 @@ _ALLOWED_OPS = ("launch", "teardown") class BrokerAuthError(Exception): """A broker request failed provenance or schema verification — bad/absent signature, malformed token, or a payload that doesn't match - the fixed launch-request shape. Fail-closed: the broker must not act.""" + the fixed launch-request shape. Fail-closed: the broker must not act. + + A **definite** negative: nothing was launched, so a caller may safely roll + back as if the op never happened.""" + + +class BrokerUnavailableError(Exception): + """A brokered request could not be carried to a verdict: the broker (or the + wire to it) was unreachable, timed out, or dropped the response. + + Crucially **ambiguous** — unlike `BrokerAuthError`, the op MAY already have + taken effect on the backend before the response was lost, so a caller must + NOT assume it did nothing (e.g. must not roll a registry row back as if no + launch happened, which would orphan a running container). Only the in-process + brokers never raise this; the out-of-process `BrokerClient` does.""" @dataclass(frozen=True) @@ -179,6 +193,7 @@ class StubBroker(LaunchBroker): __all__ = [ "BrokerAuthError", + "BrokerUnavailableError", "LaunchRequest", "SubmitBroker", "LaunchBroker", diff --git a/bot_bottle/orchestrator/broker_client.py b/bot_bottle/orchestrator/broker_client.py index 79879def..62783418 100644 --- a/bot_bottle/orchestrator/broker_client.py +++ b/bot_bottle/orchestrator/broker_client.py @@ -20,15 +20,17 @@ import json import urllib.error import urllib.request -from .broker import BrokerAuthError, LaunchRequest +from .broker import BrokerAuthError, BrokerUnavailableError, LaunchRequest DEFAULT_TIMEOUT_SECONDS = 5.0 class BrokerClientError(RuntimeError): - """A brokered launch/teardown could not be delivered to the host control - server: unreachable, or an unexpected status other than the fail-closed 401 - (which surfaces as `BrokerAuthError`, matching the in-process broker).""" + """The host control server *responded*, but with an unexpected status other + than the fail-closed 401 (which surfaces as `BrokerAuthError`) — e.g. a 502 + backend failure or a malformed body. A definite negative: the host processed + the request and it did not launch. (A *no-response* failure — unreachable / + timeout / dropped — is the ambiguous `BrokerUnavailableError` instead.)""" class BrokerClient: @@ -45,9 +47,11 @@ class BrokerClient: verified and acted on. Raises `BrokerAuthError` on a fail-closed 401 (bad provenance/schema — - the same exception the in-process broker raises), or `BrokerClientError` - if the controller is unreachable, times out, or returns any other - non-success status.""" + the same exception the in-process broker raises); `BrokerClientError` if + the host *responds* with any other non-success status or a malformed + body (a definite negative); or `BrokerUnavailableError` if no response is + obtained (unreachable / timeout / dropped) — the **ambiguous** case, where + the host may already have acted, so the caller must not roll back.""" data = json.dumps({"token": token}).encode() req = urllib.request.Request( f"{self._base}/broker", data=data, method="POST", @@ -65,8 +69,11 @@ class BrokerClient: raise BrokerClientError( f"POST /broker: HTTP {e.code} {detail}".rstrip() ) from e - except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e: - raise BrokerClientError(f"POST /broker: {e}") from e + except (urllib.error.URLError, TimeoutError, OSError) as e: + # No usable response — unreachable, timed out, or the connection + # dropped mid-exchange. Ambiguous: the request may already have + # launched the bottle, so this is NOT a definite failure. + raise BrokerUnavailableError(f"POST /broker: {e}") from e def _json_object(raw: bytes) -> dict[str, object]: diff --git a/bot_bottle/orchestrator/host_server.py b/bot_bottle/orchestrator/host_server.py index e974f887..fe8130a0 100644 --- a/bot_bottle/orchestrator/host_server.py +++ b/bot_bottle/orchestrator/host_server.py @@ -57,6 +57,16 @@ BROKER_SECRET_ENV = "BOT_BOTTLE_BROKER_SECRET" # (8099) — a separate privileged component listening on its own socket. DEFAULT_PORT = 8091 +# Cap on the request body. A signed broker request is tiny, so rejecting anything +# larger *before reading it* keeps a caller that can merely reach the socket (no +# signed token needed) from exhausting memory or a handler thread with a huge +# Content-Length — the signed token, not mere reachability, is the authority. +MAX_BODY_BYTES = 64 * 1024 + +# Per-request socket timeout, bounding how long a stalled / slow-loris caller can +# hold a handler thread on this privileged listener. +REQUEST_TIMEOUT_SECONDS = 15 + def _parse_json_object(body: bytes) -> Json: """Parse a JSON object body. Raises ValueError for non-objects / bad JSON.""" @@ -130,6 +140,10 @@ def dispatch( # pylint: disable=too-many-return-statements class Handler(http.server.BaseHTTPRequestHandler): """Thin stdlib adapter: read the body, call `dispatch`, write JSON.""" + # Socket timeout per request (applied by StreamRequestHandler.setup) so a + # stalled caller can't pin a handler thread on this privileged listener. + timeout = REQUEST_TIMEOUT_SECONDS + # Quiet by default; opt back into stdlib access logging with # BOT_BOTTLE_HOST_CONTROLLER_DEBUG (the controller has its own logging). def log_message(self, format: str, *args: typing.Any) -> None: # noqa: A002 @@ -137,12 +151,21 @@ class Handler(http.server.BaseHTTPRequestHandler): super().log_message(format, *args) def _serve(self, method: str) -> None: - """Read the request body, dispatch it, and write the JSON reply. A - dispatch that raises (it shouldn't — dispatch is total) still returns a - 500 rather than dropping the connection.""" + """Read the request body (bounded), dispatch it, and write the JSON + reply. A dispatch that raises (it shouldn't — dispatch is total) still + returns a 500 rather than dropping the connection.""" server = self.server assert isinstance(server, HostControlServer) - length = int(self.headers.get("Content-Length") or 0) + try: + length = int(self.headers.get("Content-Length") or 0) + except ValueError: + self._reply(400, {"error": "invalid Content-Length"}) + return + if length < 0 or length > MAX_BODY_BYTES: + # Reject before reading: nothing legitimate is this big, so an + # oversized declared length is a bug or a resource-exhaustion attempt. + self._reply(413, {"error": "request body too large"}) + return body = self.rfile.read(length) if length > 0 else b"" try: status, payload = dispatch(server.broker, method, self.path, body) @@ -150,6 +173,10 @@ class Handler(http.server.BaseHTTPRequestHandler): sys.stderr.write(f"host controller: {method} {self.path} failed: {e!r}\n") sys.stderr.flush() status, payload = 500, {"error": f"internal error: {e}"} + self._reply(status, payload) + + def _reply(self, status: int, payload: typing.Mapping[str, object]) -> None: + """Write one JSON response with an explicit Content-Length.""" data = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index 96ca9f19..70de69d7 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -25,7 +25,7 @@ import json from collections.abc import Iterable from datetime import datetime, timezone -from .broker import LaunchRequest, SubmitBroker, sign_request +from .broker import BrokerUnavailableError, LaunchRequest, SubmitBroker, sign_request from .store.registry_store import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore from .supervisor import ( AuditEntry, @@ -111,14 +111,23 @@ class OrchestratorCore: image_ref=image_ref, slot=slot, ) - launched = False try: self._broker.submit(sign_request(req, self._secret)) - launched = True - finally: - if not launched: - self.registry.deregister(rec.bottle_id) - self._tokens.pop(rec.bottle_id, None) + except BrokerUnavailableError: + # Ambiguous delivery failure (timeout / dropped response): the broker + # may already have launched the bottle before the response was lost. + # Do NOT deregister — that would orphan a running container with no + # registry row (reconcile reaps rows, never containers). Keep the row + # so reconcile reaps it iff the bottle is not actually live; surface + # the error so the caller knows the launch is unconfirmed. + raise + except Exception: + # A definite failure — a fail-closed rejection, a backend launch + # error, or the host reporting it did not launch: nothing is running, + # so roll the registry entry back to leave no orphan. + self.registry.deregister(rec.bottle_id) + self._tokens.pop(rec.bottle_id, None) + raise return rec def teardown_bottle(self, bottle_id: str) -> bool: diff --git a/tests/unit/test_orchestrator_broker_client.py b/tests/unit/test_orchestrator_broker_client.py index e85d51d2..e0de4747 100644 --- a/tests/unit/test_orchestrator_broker_client.py +++ b/tests/unit/test_orchestrator_broker_client.py @@ -8,7 +8,11 @@ import unittest import urllib.error from unittest.mock import MagicMock, patch -from bot_bottle.orchestrator.broker import BrokerAuthError, LaunchRequest +from bot_bottle.orchestrator.broker import ( + BrokerAuthError, + BrokerUnavailableError, + LaunchRequest, +) from bot_bottle.orchestrator.broker_client import BrokerClient, BrokerClientError _URLOPEN = "bot_bottle.orchestrator.broker_client.urllib.request.urlopen" @@ -58,14 +62,25 @@ class TestSubmit(unittest.TestCase): with self.assertRaises(BrokerAuthError): self.c.submit("forged") - def test_502_raises_broker_client_error(self) -> None: + def test_502_is_a_definite_client_error(self) -> None: + # The host responded — it processed the request and did not launch, so a + # definite BrokerClientError (the caller may safely roll back). with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})): with self.assertRaises(BrokerClientError): self.c.submit("tok") - def test_unreachable_raises_broker_client_error(self) -> None: + def test_unreachable_is_ambiguous_unavailable(self) -> None: + # No response at all — the request may already have launched, so the + # AMBIGUOUS BrokerUnavailableError (the caller must NOT roll back). with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): - with self.assertRaises(BrokerClientError): + with self.assertRaises(BrokerUnavailableError): + self.c.submit("tok") + + def test_timeout_is_ambiguous_unavailable(self) -> None: + # A dropped/late response after the request was sent is the exact orphan + # risk: the host may have launched. Must be ambiguous, not a definite fail. + with patch(_URLOPEN, side_effect=TimeoutError("read timed out")): + with self.assertRaises(BrokerUnavailableError): self.c.submit("tok") def test_malformed_success_body_raises(self) -> None: @@ -79,6 +94,24 @@ class TestSubmit(unittest.TestCase): with self.assertRaises(BrokerAuthError): self.c.submit("forged") + def test_non_json_success_body_raises(self) -> None: + # A 200 whose body isn't JSON is tolerated into {} then fails the + # missing-field check — a definite client error, not a crash. + m = MagicMock() + m.__enter__.return_value.read.return_value = b"not json at all" + with patch(_URLOPEN, return_value=m): + with self.assertRaises(BrokerClientError): + self.c.submit("tok") + + def test_unreadable_error_body_is_tolerated(self) -> None: + # An HTTPError whose body can't be read (fp=None) still classifies by + # status — the error detail is best-effort. + err = urllib.error.HTTPError( + "http://host/broker", 502, "err", {}, None) # type: ignore[arg-type] + with patch(_URLOPEN, side_effect=err): + with self.assertRaises(BrokerClientError): + self.c.submit("tok") + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_host_server.py b/tests/unit/test_orchestrator_host_server.py index 3428280a..11d2f002 100644 --- a/tests/unit/test_orchestrator_host_server.py +++ b/tests/unit/test_orchestrator_host_server.py @@ -7,10 +7,14 @@ the full sign -> POST -> verify -> act seam over HTTP. from __future__ import annotations +import io import json import secrets import threading import unittest +import urllib.error +import urllib.request +from unittest.mock import MagicMock, patch from bot_bottle.orchestrator.broker import ( BrokerAuthError, @@ -21,8 +25,12 @@ from bot_bottle.orchestrator.broker import ( ) from bot_bottle.orchestrator.broker_client import BrokerClient from bot_bottle.orchestrator.host_server import ( + MAX_BODY_BYTES, + Handler, + HostControlServer, broker_secret_from_env, dispatch, + main, make_host_server, ) @@ -97,6 +105,16 @@ class TestDispatch(unittest.TestCase): status, _ = dispatch(self.broker, "POST", "/broker", b"{not json") self.assertEqual(400, status) + def test_empty_body_is_missing_token_400(self) -> None: + # Empty body parses to {} (no token) → 400, never reaching the broker. + status, _ = dispatch(self.broker, "POST", "/broker", b"") + self.assertEqual(400, status) + self.assertEqual([], self.broker.launched) + + def test_non_object_body_is_400(self) -> None: + status, _ = dispatch(self.broker, "POST", "/broker", b"[1, 2]") + self.assertEqual(400, status) + def test_unknown_route_404(self) -> None: status, _ = dispatch(self.broker, "GET", "/nope", b"") self.assertEqual(404, status) @@ -152,5 +170,110 @@ class TestSeamRoundTrip(unittest.TestCase): self.assertEqual([], broker.launched) # fail-closed across the wire +class TestRequestLimits(unittest.TestCase): + """The privileged listener must not let a caller that can merely reach the + socket (no signed token) exhaust it via an oversized declared body.""" + + def _base(self) -> str: + self.broker = StubBroker(secrets.token_bytes(16)) + server = make_host_server(self.broker, "127.0.0.1", 0) + self.addCleanup(server.server_close) + threading.Thread(target=server.serve_forever, daemon=True).start() + self.addCleanup(server.shutdown) + host, port = server.server_address[0], server.server_address[1] + return f"http://{host}:{port}" + + def test_oversized_body_is_rejected_before_acting(self) -> None: + base = self._base() + big = b"x" * (MAX_BODY_BYTES + 1) + req = urllib.request.Request( + f"{base}/broker", data=big, method="POST", + headers={"Content-Type": "application/json"}) + with self.assertRaises(urllib.error.HTTPError) as cm: + urllib.request.urlopen(req, timeout=5) + self.assertEqual(413, cm.exception.code) + self.assertEqual([], self.broker.launched) # never reached the broker + + +class TestServeUnit(unittest.TestCase): + """Drive `Handler._serve` directly (no socket). The real per-request handler + runs in a daemon thread whose coverage/trace data is lost, so the + bounded-body and error paths are exercised here in the main thread instead.""" + + def _handler(self, broker: LaunchBroker, headers: dict[str, str], + body: bytes = b"") -> tuple[Handler, MagicMock]: + server = HostControlServer.__new__(HostControlServer) + server.broker = broker + h = Handler.__new__(Handler) + h.server = server + h.headers = headers # type: ignore[assignment] — dict is a valid .get() stand-in + h.path = "/broker" + h.rfile = io.BytesIO(body) + h.wfile = io.BytesIO() + send_response = MagicMock() + h.send_response = send_response # type: ignore[method-assign] + h.send_header = MagicMock() # type: ignore[method-assign] + h.end_headers = MagicMock() # type: ignore[method-assign] + return h, send_response + + def test_oversized_content_length_is_413(self) -> None: + broker = StubBroker(secrets.token_bytes(16)) + h, send_response = self._handler(broker, {"Content-Length": str(MAX_BODY_BYTES + 1)}) + h.do_POST() # exercises do_POST -> _serve + send_response.assert_called_once_with(413) + self.assertEqual([], broker.launched) # rejected before the broker + + def test_invalid_content_length_is_400(self) -> None: + h, send_response = self._handler(StubBroker(secrets.token_bytes(16)), + {"Content-Length": "not-a-number"}) + h._serve("POST") + send_response.assert_called_once_with(400) + + def test_valid_request_dispatches_200(self) -> None: + secret = secrets.token_bytes(16) + broker = StubBroker(secret) + body = _body({"token": sign_request( + LaunchRequest(op="teardown", bottle_id="b1"), secret)}) + h, send_response = self._handler(broker, {"Content-Length": str(len(body))}, body) + h._serve("POST") + send_response.assert_called_once_with(200) + self.assertEqual(["b1"], [r.bottle_id for r in broker.torn_down]) + + def test_dispatch_exception_becomes_500(self) -> None: + # dispatch is total, but the handler still guards it: a raised dispatch + # returns 500 rather than dropping the connection. + h, send_response = self._handler( + StubBroker(secrets.token_bytes(16)), {"Content-Length": "0"}) + with patch("bot_bottle.orchestrator.host_server.dispatch", + side_effect=RuntimeError("boom")): + h._serve("POST") + send_response.assert_called_once_with(500) + + def test_health_over_do_get(self) -> None: + h, send_response = self._handler(StubBroker(secrets.token_bytes(16)), {}) + h.path = "/health" + h.do_GET() + send_response.assert_called_once_with(200) + + +class TestMain(unittest.TestCase): + def test_fail_closed_without_secret(self) -> None: + with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env", + return_value=None): + self.assertEqual(2, main(["--port", "0"])) + + def test_serves_then_shuts_down_cleanly(self) -> None: + fake = MagicMock() + fake.server_address = ("127.0.0.1", 0) + fake.serve_forever.side_effect = KeyboardInterrupt + with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env", + return_value=b"k"), \ + patch("bot_bottle.orchestrator.host_server.make_host_server", + return_value=fake): + self.assertEqual(0, main(["--port", "0"])) + fake.serve_forever.assert_called_once() + fake.server_close.assert_called_once() + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_main.py b/tests/unit/test_orchestrator_main.py new file mode 100644 index 00000000..366eaf12 --- /dev/null +++ b/tests/unit/test_orchestrator_main.py @@ -0,0 +1,64 @@ +"""Unit: the orchestrator dev-harness entrypoint (`python -m bot_bottle.orchestrator`). + +Exercises broker selection (stub / docker / http) and the fail-closed http path, +patching `make_server` so the serve loop returns instead of blocking. +""" + +from __future__ import annotations + +import os +import secrets +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from bot_bottle.orchestrator.__main__ import main + + +def _fake_server() -> MagicMock: + fake = MagicMock() + fake.server_address = ("127.0.0.1", 0) + # Break out of serve_forever immediately, exercising the try/finally. + fake.serve_forever.side_effect = KeyboardInterrupt + return fake + + +class TestMain(unittest.TestCase): + def _run(self, broker: str, env: dict[str, str] | None = None) -> tuple[int, MagicMock]: + fake = _fake_server() + with tempfile.TemporaryDirectory() as d: + argv = ["--db", str(Path(d) / "r.db"), "--port", "0", "--broker", broker] + with patch("bot_bottle.orchestrator.__main__.make_server", return_value=fake), \ + patch.dict("os.environ", env or {}, clear=False): + if env is None: + os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None) + rc = main(argv) + return rc, fake + + def test_stub_broker_serves_and_closes(self) -> None: + rc, fake = self._run("stub") + self.assertEqual(0, rc) + fake.serve_forever.assert_called_once() + fake.server_close.assert_called_once() + + def test_docker_broker_serves(self) -> None: + rc, _ = self._run("docker") + self.assertEqual(0, rc) + + def test_http_broker_with_secret_serves(self) -> None: + rc, _ = self._run( + "http", env={"BOT_BOTTLE_BROKER_SECRET": secrets.token_bytes(16).hex()}) + self.assertEqual(0, rc) + + def test_http_broker_without_secret_exits(self) -> None: + # Fail-closed: --broker http with no shared secret is a usage error. + with tempfile.TemporaryDirectory() as d: + with patch.dict("os.environ", {}, clear=False): + os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None) + with self.assertRaises(SystemExit): + main(["--db", str(Path(d) / "r.db"), "--broker", "http"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index 7f9b12f8..c9031209 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -11,7 +11,12 @@ from contextlib import closing from pathlib import Path from unittest.mock import patch -from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker +from bot_bottle.orchestrator.broker import ( + BrokerUnavailableError, + LaunchBroker, + LaunchRequest, + StubBroker, +) from bot_bottle.orchestrator.store.registry_store import RegistryStore from bot_bottle.orchestrator.service import OrchestratorCore from bot_bottle.orchestrator.store.secret_store import new_env_var_secret @@ -25,8 +30,8 @@ from bot_bottle.orchestrator.supervisor import ( class _FailingBroker(LaunchBroker): - """Verifies the token like any broker, then fails the launch — to - exercise the orchestrator's registry rollback.""" + """Verifies the token like any broker, then fails the launch *definitely* — + to exercise the orchestrator's registry rollback.""" def _launch(self, req: LaunchRequest) -> None: raise RuntimeError("launch failed") @@ -35,6 +40,18 @@ class _FailingBroker(LaunchBroker): pass +class _UnavailableBroker(LaunchBroker): + """Verifies the token, then raises the *ambiguous* BrokerUnavailableError — + the host may already have launched — so the orchestrator must KEEP the + registry row rather than orphan a running container.""" + + def _launch(self, req: LaunchRequest) -> None: + raise BrokerUnavailableError("delivery dropped after send") + + def _teardown(self, req: LaunchRequest) -> None: + pass + + class TestOrchestrator(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() @@ -136,11 +153,20 @@ class TestOrchestrator(unittest.TestCase): self.assertIsNotNone(self.orch.resolve("10.243.0.3", rec.identity_token)) self.assertIsNone(self.orch.resolve("10.243.0.3", "wrong-token")) - def test_launch_rolls_back_registry_on_broker_failure(self) -> None: + def test_launch_rolls_back_registry_on_definite_broker_failure(self) -> None: orch = OrchestratorCore(self.store, _FailingBroker(self.secret), self.secret) with self.assertRaises(RuntimeError): orch.launch_bottle("10.243.0.9") - self.assertEqual([], self.store.all()) # no orphan + self.assertEqual([], self.store.all()) # no orphan row + + def test_launch_keeps_registry_on_ambiguous_broker_failure(self) -> None: + # The host may already have launched the bottle before the response was + # lost, so deregistering would orphan a running container with no row. + # The row is kept for reconcile to reap iff the bottle is not live. + orch = OrchestratorCore(self.store, _UnavailableBroker(self.secret), self.secret) + with self.assertRaises(BrokerUnavailableError): + orch.launch_bottle("10.243.0.9") + self.assertEqual(1, len(self.store.all())) # row survives — no orphan container def test_gateway_status_reports_unconfigured(self) -> None: # The orchestrator no longer owns a standalone gateway lifecycle; the