From 97294074309360949a42a7b49d88b87a856eb278 Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 26 Jul 2026 17:52:44 +0000 Subject: [PATCH] feat(orchestrator): list_live broker op + reconcile via broker (#468) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Chunk 3 of the host-control-server stack: grow the broker op vocabulary (PRD gap 3), starting with `list_live`, and invert `reconcile` onto it. - broker: split the closed op vocabulary into mutation (`launch`/`teardown`, carry a bottle id + static flags) and query (`list_live`, carries nothing but its op name) kinds. `verify_request` now enforces a **strict schema** (open question 1, resolved yes): unknown claim keys are rejected, a mutation must name its bottle, and a query that smuggles any id/flag is refused. - broker verb: `LaunchBroker.list_live` / `SubmitBroker.list_live` return the backend's live source IPs; a backend enumeration failure is converted to the single `BrokerUnavailableError` "live set unknown" signal. `DockerBroker` enumerates its labelled containers; `StubBroker` derives from launches (or a test override). - host controller: `POST /broker/live` verifies a signed `list_live` token and returns `{source_ips}`; `BrokerClient.list_live` is its drop-in client. - reconcile: `OrchestratorCore.reconcile()` drops the `live_source_ips` parameter and pulls the live set from the broker itself — the tell that the orchestrator couldn't see the backend goes away. **Fail-safe**: if the broker can't return an authoritative set the sweep is skipped, never run against an empty/partial set (which would reap healthy rows). The `/reconcile` HTTP contract + `OrchestratorClient.reconcile` become a bare trigger. The macOS launcher's Apple-container enumeration stays for now; it becomes the host controller's `list_live` when launch itself moves behind the broker (the pulled-forward chunk 5, next in the stack). Tests + pyright clean; pylint 10.0 on broker. Co-Authored-By: Claude Opus 4.8 --- .../macos_container/consolidated_launch.py | 12 +- bot_bottle/orchestrator/broker.py | 128 +++++++++++++++--- bot_bottle/orchestrator/broker_client.py | 42 ++++++ bot_bottle/orchestrator/client.py | 16 +-- bot_bottle/orchestrator/docker_broker.py | 25 ++++ bot_bottle/orchestrator/host_server.py | 37 ++++- bot_bottle/orchestrator/server.py | 19 ++- bot_bottle/orchestrator/service.py | 39 ++++-- tests/unit/test_macos_consolidated_launch.py | 22 +-- tests/unit/test_orchestrator_broker.py | 85 ++++++++++++ tests/unit/test_orchestrator_broker_client.py | 43 ++++++ tests/unit/test_orchestrator_client.py | 15 +- tests/unit/test_orchestrator_docker_broker.py | 46 +++++++ tests/unit/test_orchestrator_host_server.py | 52 ++++++- tests/unit/test_orchestrator_server.py | 39 +++--- tests/unit/test_orchestrator_service.py | 38 +++++- 16 files changed, 552 insertions(+), 106 deletions(-) diff --git a/bot_bottle/backend/macos_container/consolidated_launch.py b/bot_bottle/backend/macos_container/consolidated_launch.py index c8b8b6c4..1f75094e 100644 --- a/bot_bottle/backend/macos_container/consolidated_launch.py +++ b/bot_bottle/backend/macos_container/consolidated_launch.py @@ -166,9 +166,17 @@ def register_agent( # collide with this bottle's — and `by_source_ip` fail-closes on ambiguity, # which would resolve no policy at all and deny every host. Best-effort: a # reconciliation failure must not block an otherwise-fine launch. + # + # The orchestrator now pulls the live set from the broker itself, so this is + # a bare trigger. Until macOS moves onto the host controller (the follow-up + # that routes launch through the broker), the orchestrator's broker is the + # stub, whose live set is the bottles it recorded launches for — so the sweep + # only under-reaps (leaves an orphan a cycle longer), never reaps a healthy + # bottle. The Apple-container enumeration below (`live_source_ips`) becomes + # that host controller's `list_live` there. try: - client.reconcile(live_source_ips(endpoint.network)) - except (OrchestratorClientError, EnumerationError) as e: + client.reconcile() + except OrchestratorClientError as e: info(f"registry reconciliation skipped: {e}") reg = provision_bottle( client, source_ip, egress_plan, git_gate_plan, MacosGatewayTransport(), diff --git a/bot_bottle/orchestrator/broker.py b/bot_bottle/orchestrator/broker.py index 6d5c567f..4d64f334 100644 --- a/bot_bottle/orchestrator/broker.py +++ b/bot_bottle/orchestrator/broker.py @@ -32,7 +32,23 @@ from dataclasses import dataclass from typing import Protocol _JWT_HEADER = {"alg": "HS256", "typ": "JWT"} -_ALLOWED_OPS = ("launch", "teardown") + +# The closed op vocabulary, split by kind (PRD "ids + static flags" rule): +# * mutation ops act on ONE bottle — they carry its id (+ static launch flags); +# * query ops enumerate host state — they carry NO ids or flags at all. +# Each new host-privileged op is added here deliberately; `verify_request` +# enforces the per-kind shape so the privileged surface can't drift. +_MUTATION_OPS = ("launch", "teardown") +_QUERY_OPS = ("list_live",) +_ALLOWED_OPS = _MUTATION_OPS + _QUERY_OPS + +# Every claim key a signed request may carry — the request fields plus the two +# per-signature envelope claims. A strict allow-list (open question 1, resolved +# "yes"): an unknown key is rejected outright, so the schema can't silently widen +# as ops are added. +_ALLOWED_CLAIMS = frozenset( + {"op", "bottle_id", "source_ip", "image_ref", "slot", "jti", "iat"} +) class BrokerAuthError(Exception): @@ -56,17 +72,28 @@ class BrokerUnavailableError(Exception): @dataclass(frozen=True) -class LaunchRequest: - """The structured, un-coercible launch/teardown request. Ids + flags - only — `image_ref` is a content-addressed id, never a path/argv.""" +class BrokerRequest: + """A structured, un-coercible broker request. Ids + static flags only — + `image_ref` is a content-addressed id, never a path/argv. + + A **mutation** op (`launch`/`teardown`) names the bottle it acts on in + `bottle_id` (plus launch's static flags). A **query** op (`list_live`) acts + on no single bottle, so it carries nothing but its `op` name — `bottle_id` + and every flag stay empty, and `verify_request` rejects a query that smuggles + any in. `LaunchRequest` is kept as an alias for the launch/teardown callers.""" op: str # one of _ALLOWED_OPS - bottle_id: str + bottle_id: str = "" source_ip: str = "" image_ref: str = "" slot: int | None = None +# Historical name — the request type predates the query ops. Kept so the launch +# path (`OrchestratorCore`, `DockerBroker`) reads unchanged. +LaunchRequest = BrokerRequest + + # --- minimal JWS/JWT (HS256), stdlib only ---------------------------------- def _b64url(data: bytes) -> str: @@ -118,20 +145,35 @@ def verify_request(token: str, secret: bytes) -> LaunchRequest: raise BrokerAuthError("unexpected header/alg") if not isinstance(claims, dict): raise BrokerAuthError("claims are not an object") + # Strict schema (open question 1): a signed request may carry ONLY the known + # claim keys, so the privileged surface can't widen by smuggling an extra + # claim past the fixed field checks below. + unknown = set(claims) - _ALLOWED_CLAIMS + if unknown: + raise BrokerAuthError(f"unknown claim(s): {', '.join(sorted(unknown))}") op = claims.get("op") - bottle_id = claims.get("bottle_id") + if not isinstance(op, str) or op not in _ALLOWED_OPS: + raise BrokerAuthError("request does not match the broker-request schema") + bottle_id = claims.get("bottle_id", "") source_ip = claims.get("source_ip", "") image_ref = claims.get("image_ref", "") slot = claims.get("slot") if ( - not isinstance(op, str) or op not in _ALLOWED_OPS - or not isinstance(bottle_id, str) or not bottle_id + not isinstance(bottle_id, str) or not isinstance(source_ip, str) or not isinstance(image_ref, str) or not (slot is None or isinstance(slot, int)) ): - raise BrokerAuthError("request does not match the launch-request schema") - return LaunchRequest( + raise BrokerAuthError("request does not match the broker-request schema") + # Per-kind shape: a mutation names exactly one bottle; a query names none and + # carries no flags (the "no arguments" rule for `list_live`). Enforcing both + # halves keeps `bottle_id`/flags from being a coercible field on a query op. + if op in _MUTATION_OPS: + if not bottle_id: + raise BrokerAuthError(f"{op} requires a bottle_id") + elif bottle_id or source_ip or image_ref or slot is not None: + raise BrokerAuthError(f"{op} is a query op and takes no ids or flags") + return BrokerRequest( op=op, bottle_id=bottle_id, source_ip=source_ip, image_ref=image_ref, slot=slot ) @@ -139,14 +181,18 @@ def verify_request(token: str, secret: bytes) -> LaunchRequest: # --- the broker itself ------------------------------------------------------ class SubmitBroker(Protocol): - """The single method `OrchestratorCore` depends on: verify a signed token and - perform its op, returning the verified request. Both the in-process - `LaunchBroker` and the out-of-process `BrokerClient` (which relays the token - to the host control server) satisfy it structurally, so the core is unchanged - whether the backend is local or a real host service.""" + """The broker surface `OrchestratorCore` depends on. `submit` verifies a + signed *mutation* token and performs its op, returning the verified request; + `list_live` verifies a signed `list_live` token and returns the source IPs of + the bottles the backend currently has running (reconcile's live set). Both the + in-process `LaunchBroker` and the out-of-process `BrokerClient` (which relays + the token to the host control server) satisfy it structurally, so the core is + unchanged whether the backend is local or a real host service.""" def submit(self, token: str) -> LaunchRequest: ... + def list_live(self, token: str) -> list[str]: ... + class LaunchBroker(abc.ABC): """Verifies a signed request came from the orchestrator, then performs @@ -157,15 +203,35 @@ class LaunchBroker(abc.ABC): self._secret = secret def submit(self, token: str) -> LaunchRequest: - """Verify `token` and perform its op. Returns the verified request; - raises `BrokerAuthError` if provenance/shape fails.""" + """Verify `token` and perform its mutation op. Returns the verified + request; raises `BrokerAuthError` if provenance/shape fails, or if the + token is a non-mutation op (a `list_live` token routed here by mistake).""" req = verify_request(token, self._secret) if req.op == "launch": self._launch(req) - else: + elif req.op == "teardown": self._teardown(req) + else: + raise BrokerAuthError(f"{req.op} is not a submit op") return req + def list_live(self, token: str) -> list[str]: + """Verify a `list_live` `token` and return the source IPs of the bottles + the backend currently has running. Raises `BrokerAuthError` on bad + provenance/shape or a non-`list_live` op; a backend enumeration failure + is converted to `BrokerUnavailableError` — the single "live set could not + be determined" signal reconcile catches to skip the sweep rather than + reaping healthy rows against a partial set (the out-of-process + `BrokerClient` raises the same on an enumeration 502 / no response).""" + req = verify_request(token, self._secret) + if req.op != "list_live": + raise BrokerAuthError(f"{req.op} is not a list_live op") + try: + return list(self._list_live()) + except Exception as e: # noqa: BLE001 — any enumeration failure means the + # live set is unknown; reconcile must skip, never reap against it. + raise BrokerUnavailableError(f"list_live enumeration failed: {e}") from e + @abc.abstractmethod def _launch(self, req: LaunchRequest) -> None: ... @@ -174,15 +240,31 @@ class LaunchBroker(abc.ABC): def _teardown(self, req: LaunchRequest) -> None: ... + @abc.abstractmethod + def _list_live(self) -> list[str]: + """Every running bottle's source IP, backend-native. Must raise (not + return a partial list) if the live set can't be determined + authoritatively, so reconcile skips rather than reaping healthy rows.""" + ... + class StubBroker(LaunchBroker): """Dev-harness broker: records verified requests without launching - anything. Exercises the full sign -> verify -> act contract in-process.""" + anything. Exercises the full sign -> verify -> act contract in-process. + + Its live set (`_list_live`) is, by default, every bottle it recorded a + launch for and no teardown since — the in-process analogue of enumerating + the backend. Tests that need to simulate a bottle dying out from under the + registry (the case reconcile exists for) set `live_source_ips` explicitly to + override that derived set.""" def __init__(self, secret: bytes) -> None: super().__init__(secret) self.launched: list[LaunchRequest] = [] self.torn_down: list[LaunchRequest] = [] + # When not None, the exact live set `_list_live` reports — lets a test + # say "only these IPs are still up" regardless of what was launched. + self.live_source_ips: list[str] | None = None def _launch(self, req: LaunchRequest) -> None: self.launched.append(req) @@ -190,10 +272,18 @@ class StubBroker(LaunchBroker): def _teardown(self, req: LaunchRequest) -> None: self.torn_down.append(req) + def _list_live(self) -> list[str]: + if self.live_source_ips is not None: + return list(self.live_source_ips) + torn = {r.bottle_id for r in self.torn_down} + return [r.source_ip for r in self.launched + if r.bottle_id not in torn and r.source_ip] + __all__ = [ "BrokerAuthError", "BrokerUnavailableError", + "BrokerRequest", "LaunchRequest", "SubmitBroker", "LaunchBroker", diff --git a/bot_bottle/orchestrator/broker_client.py b/bot_bottle/orchestrator/broker_client.py index 62783418..6eb867e8 100644 --- a/bot_bottle/orchestrator/broker_client.py +++ b/bot_bottle/orchestrator/broker_client.py @@ -75,6 +75,37 @@ class BrokerClient: # launched the bottle, so this is NOT a definite failure. raise BrokerUnavailableError(f"POST /broker: {e}") from e + def list_live(self, token: str) -> list[str]: + """POST the signed `list_live` token to the host controller and return + the source IPs of the bottles it enumerated as running. + + Raises `BrokerAuthError` on a fail-closed 401 (bad provenance/schema); + `BrokerClientError` on any other non-success status or a malformed body + (a backend-enumeration 502 included); or `BrokerUnavailableError` if no + response is obtained. A query has no backend side effect, so — unlike + `submit` — every one of these is a *definite* "no live set"; reconcile + catches all three and skips the sweep rather than reaping against an + unknown or partial set.""" + data = json.dumps({"token": token}).encode() + req = urllib.request.Request( + f"{self._base}/broker/live", data=data, method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + return _source_ips_from(_json_object(resp.read())) + except urllib.error.HTTPError as e: + detail = _error_detail(e) + if e.code == 401: + raise BrokerAuthError( + detail or "host controller rejected the request" + ) from e + raise BrokerClientError( + f"POST /broker/live: HTTP {e.code} {detail}".rstrip() + ) from e + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise BrokerUnavailableError(f"POST /broker/live: {e}") from e + def _json_object(raw: bytes) -> dict[str, object]: """Parse a JSON object, tolerating an empty or malformed body (→ {}), like @@ -99,6 +130,17 @@ def _error_detail(e: urllib.error.HTTPError) -> str: return detail if isinstance(detail, str) else "" +def _source_ips_from(payload: dict[str, object]) -> list[str]: + """The `source_ips` list from a `/broker/live` response — every string + entry, ignoring any non-string the host controller should never send. A + missing/!list field is a malformed response (the query is meaningless + without it), so it fails rather than silently reconciling against [].""" + raw = payload.get("source_ips") + if not isinstance(raw, list): + raise BrokerClientError("host controller response missing source_ips") + return [ip for ip in raw if isinstance(ip, str) and ip] + + def _request_from(payload: dict[str, object]) -> LaunchRequest: """Reconstruct the verified `LaunchRequest` the controller echoed, so the returned value matches the in-process broker's (which returns the request it diff --git a/bot_bottle/orchestrator/client.py b/bot_bottle/orchestrator/client.py index 9d51a5a5..ed4d59a1 100644 --- a/bot_bottle/orchestrator/client.py +++ b/bot_bottle/orchestrator/client.py @@ -15,7 +15,6 @@ from __future__ import annotations import json import urllib.error import urllib.request -from collections.abc import Iterable from dataclasses import dataclass from ..orchestrator_auth import ROLE_CLI @@ -176,14 +175,13 @@ class OrchestratorClient: raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}") return True - def reconcile( - self, live_source_ips: Iterable[str], *, grace_seconds: float | None = None, - ) -> list[str]: - """Drop registry rows for bottles that are no longer running - (`POST /reconcile`), returning the reaped bottle ids. `live_source_ips` - is the caller's enumeration of its live bottles — the orchestrator - can't see the backend from inside the orchestrator container/VM.""" - body: dict[str, object] = {"live_source_ips": list(live_source_ips)} + def reconcile(self, *, grace_seconds: float | None = None) -> list[str]: + """Trigger the orchestrator's self-heal sweep (`POST /reconcile`), + returning the reaped bottle ids. The caller no longer enumerates the + live set: the orchestrator pulls it from the host controller + (`list_live`) itself, so this is a bare trigger with an optional + `grace_seconds`.""" + body: dict[str, object] = {} if grace_seconds is not None: body["grace_seconds"] = grace_seconds payload = self._ok("POST", "/reconcile", body) diff --git a/bot_bottle/orchestrator/docker_broker.py b/bot_bottle/orchestrator/docker_broker.py index 2e22845e..5254c758 100644 --- a/bot_bottle/orchestrator/docker_broker.py +++ b/bot_bottle/orchestrator/docker_broker.py @@ -71,6 +71,31 @@ class DockerBroker(LaunchBroker): f"docker rm failed for {req.bottle_id}: {proc.stderr.strip()}" ) + def _list_live(self) -> list[str]: + """Every running bottle container's network IP, from `docker ps` filtered + to this broker's label. Raises `DockerBrokerError` if the listing or any + inspect fails, so reconcile treats the live set as unknown and skips the + sweep rather than reaping healthy rows against a partial enumeration.""" + listing = self._docker([ + "docker", "ps", "--filter", f"label={BOTTLE_ID_LABEL}", + "--format", "{{.ID}}", + ]) + if listing.returncode != 0: + raise DockerBrokerError(f"docker ps failed: {listing.stderr.strip()}") + ips: list[str] = [] + for cid in (line for line in listing.stdout.splitlines() if line): + inspected = self._docker([ + "docker", "inspect", "--format", + "{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", cid, + ]) + if inspected.returncode != 0: + raise DockerBrokerError( + f"docker inspect {cid} failed; live set is not authoritative: " + f"{inspected.stderr.strip()}" + ) + ips.extend(ip for ip in inspected.stdout.split() if ip) + return ips + __all__ = [ "DockerBroker", diff --git a/bot_bottle/orchestrator/host_server.py b/bot_bottle/orchestrator/host_server.py index 20b948fb..cf90b8ca 100644 --- a/bot_bottle/orchestrator/host_server.py +++ b/bot_bottle/orchestrator/host_server.py @@ -8,10 +8,19 @@ This module is that service — the single privileged host component — reached plane's shape (`orchestrator/server.py`): a pure `dispatch()` for socket-free testing, wrapped by a thin stdlib `http.server` adapter. - GET /health -> 200 {"status": "ok"} - POST /broker -> 200 {"op", "bottle_id", "source_ip", "image_ref", "slot"} - 400 (bad body) | 401 (bad provenance/schema) | 502 (backend) - body: {"token": ""} + GET /health -> 200 {"status": "ok"} + POST /broker -> 200 {"op", "bottle_id", "source_ip", "image_ref", "slot"} + 400 (bad body) | 401 (bad provenance/schema) | 502 (backend) + body: {"token": ""} + POST /broker/live -> 200 {"source_ips": [...]} + 400 (bad body) | 401 (bad provenance/schema) | 502 (backend) + body: {"token": ""} + +The op vocabulary grows one signed op at a time (PRD gap 3); each verb reaches +the backend only through `verify_request`, so nothing free-form ever crosses the +wire. `/broker/live` is the first query op — reconcile's live-bottle enumeration, +which the orchestrator (blind to the backend from inside its container) now pulls +from the host controller instead of being handed by the CLI. Only the **signed token** crosses the wire; the server holds the shared HS256 secret and a real `LaunchBroker` (e.g. `DockerBroker`) and runs the existing @@ -150,6 +159,26 @@ def dispatch( # pylint: disable=too-many-return-statements "slot": req.slot, } + if method == "POST" and route == "/broker/live": + try: + data = _parse_json_object(body) + except ValueError as e: + return 400, {"error": f"invalid JSON: {e}"} + token = data.get("token") + if not isinstance(token, str) or not token: + return 400, {"error": "token (string) is required"} + try: + source_ips = broker.list_live(token) + except BrokerAuthError as e: + # Fail-closed: bad signature, malformed token, or a non-list_live op. + return 401, {"error": f"broker auth failed: {e}"} + except Exception as e: # noqa: BLE001 — a backend enumeration failure + # (docker down, inspect failed) is operational, not a control-plane + # bug; surface it as a 502 so the caller skips reconcile rather than + # reaping against a partial set, and keep the controller up. + return 502, {"error": f"backend enumeration failed: {e}"} + return 200, {"source_ips": source_ips} + return 404, {"error": "not found"} diff --git a/bot_bottle/orchestrator/server.py b/bot_bottle/orchestrator/server.py index a18dac3f..4896174b 100644 --- a/bot_bottle/orchestrator/server.py +++ b/bot_bottle/orchestrator/server.py @@ -18,8 +18,8 @@ vsock / unix-socket portability caveats): body: {"env_var_secret"} DELETE /bottles/ -> 200 {"torn_down": true} | 404 (teardown) POST /reconcile -> 200 {"reaped": [bottle_id, ...]} - body: {"live_source_ips": [...], - ["grace_seconds"]} + body: {["grace_seconds"]} (live set is + pulled from the host controller, not sent) POST /attribute -> 200 {"bottle_id"} | 403 POST /resolve -> 200 {"bottle_id","policy"} | 403 body: {"source_ip","identity_token"} @@ -206,25 +206,22 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches return 404, {"error": "no such bottle"} if method == "POST" and route == "/reconcile": - # Host-driven self-heal: the caller enumerates its live bottles (only - # the host can see the backend) and the orchestrator drops rows for - # every other active bottle. Trusted-caller only — an agent that could - # reach this would be able to unregister its neighbours. + # Host-driven self-heal trigger: the orchestrator pulls its own live set + # from the host controller (`list_live`) and drops rows for every other + # active bottle — the caller no longer supplies the live IPs. Trusted- + # caller only: an agent that could reach this would trigger a sweep that + # unregisters its neighbours. Body is optional (just `grace_seconds`). try: data = _parse_json_object(body) except ValueError as e: return 400, {"error": f"invalid JSON: {e}"} - raw_ips = data.get("live_source_ips") - if not isinstance(raw_ips, list): - return 400, {"error": "live_source_ips (list of strings) is required"} - live = [ip for ip in raw_ips if isinstance(ip, str) and ip] grace = data.get("grace_seconds") kwargs = ( {"grace_seconds": float(grace)} if isinstance(grace, (int, float)) and not isinstance(grace, bool) else {} ) - return 200, {"reaped": orch.reconcile(live, **kwargs)} + return 200, {"reaped": orch.reconcile(**kwargs)} if method == "POST" and route == "/attribute": try: diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index 70de69d7..1d60987e 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -22,10 +22,17 @@ Launch lifecycle: from __future__ import annotations import json -from collections.abc import Iterable from datetime import datetime, timezone -from .broker import BrokerUnavailableError, LaunchRequest, SubmitBroker, sign_request +from .. import log +from .broker import ( + BrokerAuthError, + BrokerUnavailableError, + LaunchRequest, + SubmitBroker, + sign_request, +) +from .broker_client import BrokerClientError from .store.registry_store import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore from .supervisor import ( AuditEntry, @@ -145,22 +152,36 @@ class OrchestratorCore: def reconcile( self, - live_source_ips: Iterable[str], *, grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS, ) -> list[str]: """Drop registry rows for bottles that are no longer running, and forget their in-memory egress tokens. Returns the reaped bottle ids. - The caller supplies the live set because only the host can enumerate - its own containers — the orchestrator runs *inside* the infra - container and has no view of the backend. Deliberately does not - broker a teardown: the container is already gone, so there is nothing - to stop, and a broker error must not stop the sweep from clearing - the row that would otherwise brick the next bottle at that address. + The live set is pulled from the **broker** (`list_live`), not passed in: + only the host can enumerate its own containers, and the host controller + is now that host component, so the orchestrator — blind to the backend + from inside its container — asks it over the same signed seam it launches + through. (Before the host control server this was a `live_source_ips` + argument the CLI handed in; the tell that the orchestrator couldn't see + the backend goes away with it — PRD gap 3.) + + Fail-safe: if the broker can't return an authoritative live set + (unreachable, timed out, enumeration failed), the sweep is **skipped**, + never run against an empty/partial set — reaping a healthy bottle is far + worse than leaving an orphan one more cycle. Deliberately does not broker + a teardown: a reaped container is already gone, so there is nothing to + stop. See `RegistryStore.reap_absent` for why orphans accumulate and why they are harmful rather than merely untidy.""" + token = sign_request(LaunchRequest(op="list_live"), self._secret) + try: + live_source_ips = self._broker.list_live(token) + except (BrokerUnavailableError, BrokerAuthError, BrokerClientError) as e: + log.info("reconcile skipped: live set unavailable", + context={"error": str(e)}) + return [] reaped = self.registry.reap_absent( live_source_ips, grace_seconds=grace_seconds) for rec in reaped: diff --git a/tests/unit/test_macos_consolidated_launch.py b/tests/unit/test_macos_consolidated_launch.py index c340f1b7..04bfab0c 100644 --- a/tests/unit/test_macos_consolidated_launch.py +++ b/tests/unit/test_macos_consolidated_launch.py @@ -190,8 +190,7 @@ class TestRegisterAgentReconciles(unittest.TestCase): def _register(self, client: Mock) -> None: with patch(f"{_MOD}.OrchestratorClient", return_value=client), \ - patch(f"{_UTIL}.provision_git_gate"), \ - patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]): + patch(f"{_UTIL}.provision_git_gate"): register_agent( _egress_plan(), _git_plan(), source_ip="10.0.0.7", endpoint=_endpoint(), @@ -213,7 +212,9 @@ class TestRegisterAgentReconciles(unittest.TestCase): client.register_bottle.side_effect = _register_bottle self._register(client) self.assertEqual(["reconcile", "register"], calls) - client.reconcile.assert_called_once_with(["10.0.0.7"]) + # A bare trigger now — the orchestrator enumerates the live set itself + # (from the host controller), so the CLI passes no IPs. + client.reconcile.assert_called_once_with() def test_a_reconcile_failure_does_not_block_the_launch(self) -> None: from bot_bottle.orchestrator.client import OrchestratorClientError @@ -221,18 +222,3 @@ class TestRegisterAgentReconciles(unittest.TestCase): client.reconcile.side_effect = OrchestratorClientError("unreachable") self._register(client) client.register_bottle.assert_called_once() - - def test_enumeration_error_does_not_block_the_launch(self) -> None: - """A partial container listing must not abort the launch — skip - reconciliation and proceed, just as with an unreachable orchestrator.""" - from bot_bottle.backend.macos_container.enumerate import EnumerationError - client = _client() - with patch(f"{_MOD}.OrchestratorClient", return_value=client), \ - patch(f"{_UTIL}.provision_git_gate"), \ - patch(f"{_MOD}.live_source_ips", - side_effect=EnumerationError("container list failed")): - register_agent( - _egress_plan(), _git_plan(), - source_ip="10.0.0.7", endpoint=_endpoint(), - ) - client.register_bottle.assert_called_once() diff --git a/tests/unit/test_orchestrator_broker.py b/tests/unit/test_orchestrator_broker.py index e7588b05..ea7cb4aa 100644 --- a/tests/unit/test_orchestrator_broker.py +++ b/tests/unit/test_orchestrator_broker.py @@ -2,6 +2,10 @@ from __future__ import annotations +import base64 +import hashlib +import hmac +import json import secrets import unittest @@ -14,6 +18,21 @@ from bot_bottle.orchestrator.broker import ( ) +def _sign_claims(claims: dict[str, object], secret: bytes) -> str: + """Sign an arbitrary claims dict (bypassing `sign_request`'s fixed fields) + so a test can craft off-schema payloads — e.g. an extra claim key, or a + query op carrying ids — with a *valid* signature, isolating the schema check + from the provenance check.""" + def b64(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}, sort_keys=True, + separators=(",", ":")).encode()) + payload = b64(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode()) + signing_input = f"{header}.{payload}" + sig = b64(hmac.new(secret, signing_input.encode("ascii"), hashlib.sha256).digest()) + return f"{signing_input}.{sig}" + + class TestSignVerify(unittest.TestCase): def setUp(self) -> None: self.secret = secrets.token_bytes(16) @@ -54,6 +73,41 @@ class TestSignVerify(unittest.TestCase): with self.assertRaises(BrokerAuthError): verify_request(token, self.secret) + def test_list_live_round_trip(self) -> None: + req = LaunchRequest(op="list_live") + got = verify_request(sign_request(req, self.secret), self.secret) + self.assertEqual("list_live", got.op) + self.assertEqual("", got.bottle_id) + + def test_mutation_without_bottle_id_rejected(self) -> None: + # A launch/teardown must name its bottle even with a valid signature. + for op in ("launch", "teardown"): + token = _sign_claims( + {"op": op, "bottle_id": "", "jti": "x", "iat": 1}, self.secret) + with self.assertRaises(BrokerAuthError): + verify_request(token, self.secret) + + def test_query_op_carrying_ids_or_flags_rejected(self) -> None: + # list_live is "no arguments" — a signed one that smuggles a bottle_id / + # source_ip / image_ref / slot is off-schema and refused. + for extra in ( + {"bottle_id": "b1"}, {"source_ip": "10.0.0.1"}, + {"image_ref": "img"}, {"slot": 2}, + ): + claims: dict[str, object] = {"op": "list_live", "jti": "x", "iat": 1, **extra} + with self.assertRaises(BrokerAuthError): + verify_request(_sign_claims(claims, self.secret), self.secret) + + def test_unknown_claim_key_rejected(self) -> None: + # Strict schema (open question 1): a well-signed token with an extra claim + # key is refused, so the surface can't widen past the fixed fields. + token = _sign_claims( + {"op": "launch", "bottle_id": "b1", "cmd": "rm -rf /", "jti": "x", "iat": 1}, + self.secret, + ) + with self.assertRaises(BrokerAuthError): + verify_request(token, self.secret) + class TestStubBroker(unittest.TestCase): def setUp(self) -> None: @@ -81,6 +135,37 @@ class TestStubBroker(unittest.TestCase): self.broker.submit(forged) self.assertEqual([], self.broker.launched) # nothing acted on + def test_list_live_derives_from_launches(self) -> None: + # Default live set: launched, minus torn-down, by source IP. + self.broker.submit( + sign_request(LaunchRequest(op="launch", bottle_id="b1", source_ip="10.0.0.1"), + self.secret)) + self.broker.submit( + sign_request(LaunchRequest(op="launch", bottle_id="b2", source_ip="10.0.0.2"), + self.secret)) + self.broker.submit( + sign_request(LaunchRequest(op="teardown", bottle_id="b1", source_ip="10.0.0.1"), + self.secret)) + live = self.broker.list_live( + sign_request(LaunchRequest(op="list_live"), self.secret)) + self.assertEqual(["10.0.0.2"], live) + + def test_list_live_override(self) -> None: + self.broker.live_source_ips = ["10.0.0.9"] + live = self.broker.list_live( + sign_request(LaunchRequest(op="list_live"), self.secret)) + self.assertEqual(["10.0.0.9"], live) + + def test_submit_rejects_a_list_live_token(self) -> None: + # The verbs don't cross: a query token routed to submit is fail-closed. + with self.assertRaises(BrokerAuthError): + self.broker.submit(sign_request(LaunchRequest(op="list_live"), self.secret)) + + def test_list_live_rejects_a_mutation_token(self) -> None: + with self.assertRaises(BrokerAuthError): + self.broker.list_live( + sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret)) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_broker_client.py b/tests/unit/test_orchestrator_broker_client.py index e0de4747..1bb7efee 100644 --- a/tests/unit/test_orchestrator_broker_client.py +++ b/tests/unit/test_orchestrator_broker_client.py @@ -113,5 +113,48 @@ class TestSubmit(unittest.TestCase): self.c.submit("tok") +class TestListLive(unittest.TestCase): + def setUp(self) -> None: + self.c = BrokerClient("http://host:8091") + + def test_returns_source_ips(self) -> None: + with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", "10.0.0.2"]})): + self.assertEqual(["10.0.0.1", "10.0.0.2"], self.c.list_live("tok")) + + def test_posts_token_to_live_endpoint(self) -> None: + with patch(_URLOPEN, return_value=_resp({"source_ips": []})) as m: + self.c.list_live("signed-token") + request = m.call_args.args[0] + self.assertEqual("POST", request.get_method()) + self.assertTrue(request.full_url.endswith("/broker/live")) + self.assertEqual({"token": "signed-token"}, json.loads(request.data)) + + def test_non_string_entries_are_filtered(self) -> None: + with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", 5, None, ""]})): + self.assertEqual(["10.0.0.1"], self.c.list_live("tok")) + + def test_missing_source_ips_raises(self) -> None: + # A response without the field is malformed — better to fail (reconcile + # skips) than to reconcile against a silent empty set. + with patch(_URLOPEN, return_value=_resp({})): + with self.assertRaises(BrokerClientError): + self.c.list_live("tok") + + def test_401_raises_broker_auth_error(self) -> None: + with patch(_URLOPEN, side_effect=_http_error(401, {"error": "bad signature"})): + with self.assertRaises(BrokerAuthError): + self.c.list_live("forged") + + def test_502_enumeration_failure_is_client_error(self) -> None: + with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})): + with self.assertRaises(BrokerClientError): + self.c.list_live("tok") + + def test_unreachable_is_unavailable(self) -> None: + with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): + with self.assertRaises(BrokerUnavailableError): + self.c.list_live("tok") + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_client.py b/tests/unit/test_orchestrator_client.py index aa04a3e4..08cbf6f7 100644 --- a/tests/unit/test_orchestrator_client.py +++ b/tests/unit/test_orchestrator_client.py @@ -148,26 +148,27 @@ class TestReconcile(unittest.TestCase): def setUp(self) -> None: self.c = OrchestratorClient("http://orch:8080") - def test_posts_live_ips_and_returns_reaped(self) -> None: + def test_triggers_sweep_and_returns_reaped(self) -> None: with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["b1", "b2"]})) as m: - got = self.c.reconcile(["10.0.0.2", "10.0.0.3"]) + got = self.c.reconcile() self.assertEqual(["b1", "b2"], got) sent = json.loads(m.call_args.args[0].data) - self.assertEqual(["10.0.0.2", "10.0.0.3"], sent["live_source_ips"]) + # The live set is no longer sent — the orchestrator enumerates it. + self.assertNotIn("live_source_ips", sent) self.assertNotIn("grace_seconds", sent) # omitted -> server default def test_grace_seconds_is_forwarded_when_given(self) -> None: with patch(_URLOPEN, return_value=_resp(200, {"reaped": []})) as m: - self.c.reconcile([], grace_seconds=30) + self.c.reconcile(grace_seconds=30) self.assertEqual(30, json.loads(m.call_args.args[0].data)["grace_seconds"]) def test_malformed_reaped_is_tolerated(self) -> None: with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["ok", 5, None]})): - self.assertEqual(["ok"], self.c.reconcile([])) + self.assertEqual(["ok"], self.c.reconcile()) with patch(_URLOPEN, return_value=_resp(200, {})): - self.assertEqual([], self.c.reconcile([])) + self.assertEqual([], self.c.reconcile()) def test_error_status_raises(self) -> None: with patch(_URLOPEN, side_effect=_http_error(500)): with self.assertRaises(OrchestratorClientError): - self.c.reconcile([]) + self.c.reconcile() diff --git a/tests/unit/test_orchestrator_docker_broker.py b/tests/unit/test_orchestrator_docker_broker.py index 909a0b59..ceb4fa02 100644 --- a/tests/unit/test_orchestrator_docker_broker.py +++ b/tests/unit/test_orchestrator_docker_broker.py @@ -8,6 +8,7 @@ from unittest.mock import Mock, patch from bot_bottle.orchestrator.broker import ( BrokerAuthError, + BrokerUnavailableError, LaunchRequest, sign_request, ) @@ -96,5 +97,50 @@ class TestDockerBroker(unittest.TestCase): m.assert_not_called() +class TestDockerBrokerListLive(unittest.TestCase): + def setUp(self) -> None: + self.secret = secrets.token_bytes(16) + self.broker = DockerBroker(self.secret) + + def test_enumerates_labeled_container_ips(self) -> None: + # `docker ps` lists two labeled containers; each inspect yields an IP. + ps = Mock(returncode=0, stdout="c1\nc2\n", stderr="") + i1 = Mock(returncode=0, stdout="10.0.0.1 \n", stderr="") + i2 = Mock(returncode=0, stdout="10.0.0.2 \n", stderr="") + with patch.object(self.broker, "_docker", side_effect=[ps, i1, i2]) as m: + got = self.broker._list_live() + self.assertEqual(["10.0.0.1", "10.0.0.2"], got) + # First call is the label-filtered `docker ps`. + ps_argv = m.call_args_list[0].args[0] + self.assertEqual(["docker", "ps", "--filter", f"label={BOTTLE_ID_LABEL}"], ps_argv[:4]) + + def test_no_containers_is_empty(self) -> None: + with patch.object(self.broker, "_docker", + return_value=Mock(returncode=0, stdout="", stderr="")): + self.assertEqual([], self.broker._list_live()) + + def test_ps_failure_raises(self) -> None: + with patch.object(self.broker, "_docker", + return_value=Mock(returncode=1, stdout="", stderr="daemon down")): + with self.assertRaises(DockerBrokerError): + self.broker._list_live() + + def test_inspect_failure_raises(self) -> None: + ps = Mock(returncode=0, stdout="c1\n", stderr="") + bad = Mock(returncode=1, stdout="", stderr="no such container") + with patch.object(self.broker, "_docker", side_effect=[ps, bad]): + with self.assertRaises(DockerBrokerError): + self.broker._list_live() + + def test_enumeration_failure_surfaces_as_unavailable_via_list_live(self) -> None: + # Through the verb, a backend failure is the single "live set unknown" + # signal (BrokerUnavailableError) reconcile catches to skip the sweep. + token = sign_request(LaunchRequest(op="list_live"), self.secret) + with patch.object(self.broker, "_docker", + return_value=Mock(returncode=1, stdout="", stderr="down")): + with self.assertRaises(BrokerUnavailableError): + self.broker.list_live(token) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_host_server.py b/tests/unit/test_orchestrator_host_server.py index 2e93444d..a63efd08 100644 --- a/tests/unit/test_orchestrator_host_server.py +++ b/tests/unit/test_orchestrator_host_server.py @@ -43,8 +43,9 @@ def _body(obj: object) -> bytes: class _RaisingBroker(LaunchBroker): - """A broker whose backend launch always fails — exercises the 502 path (an - operational backend failure, distinct from a fail-closed provenance 401).""" + """A broker whose backend launch/enumeration always fails — exercises the + 502 path (an operational backend failure, distinct from a fail-closed + provenance 401).""" def _launch(self, req: LaunchRequest) -> None: raise RuntimeError("docker down") @@ -52,6 +53,9 @@ class _RaisingBroker(LaunchBroker): def _teardown(self, req: LaunchRequest) -> None: raise RuntimeError("docker down") + def _list_live(self) -> list[str]: + raise RuntimeError("docker down") + class TestDispatch(unittest.TestCase): def setUp(self) -> None: @@ -100,6 +104,42 @@ class TestDispatch(unittest.TestCase): self.assertEqual(502, status) self.assertIn("backend launch failed", str(payload["error"])) + def test_broker_live_returns_source_ips(self) -> None: + # Two launched bottles -> the stub reports both as live. + self.broker.submit(self._token(op="launch", bottle_id="b1", source_ip="10.0.0.1")) + self.broker.submit(self._token(op="launch", bottle_id="b2", source_ip="10.0.0.2")) + token = self._token(op="list_live") + status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": token})) + self.assertEqual(200, status) + self.assertEqual( + ["10.0.0.1", "10.0.0.2"], + sorted(typing.cast("list[str]", payload["source_ips"])), + ) + + def test_broker_live_forged_token_is_401(self) -> None: + forged = sign_request(LaunchRequest(op="list_live"), secrets.token_bytes(16)) + status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": forged})) + self.assertEqual(401, status) + self.assertIn("broker auth failed", str(payload["error"])) + + def test_broker_live_rejects_a_mutation_token(self) -> None: + # A launch token routed to the query endpoint is a fail-closed 401 — the + # endpoints don't share a schema even though they share a secret. + token = self._token(op="launch", bottle_id="b1", image_ref="img") + status, _ = dispatch(self.broker, "POST", "/broker/live", _body({"token": token})) + self.assertEqual(401, status) + + def test_broker_live_backend_failure_is_502(self) -> None: + broker = _RaisingBroker(self.secret) + token = self._token(op="list_live") + status, payload = dispatch(broker, "POST", "/broker/live", _body({"token": token})) + self.assertEqual(502, status) + self.assertIn("backend enumeration failed", str(payload["error"])) + + def test_broker_live_missing_token_is_400(self) -> None: + status, _ = dispatch(self.broker, "POST", "/broker/live", _body({})) + self.assertEqual(400, status) + def test_missing_token_is_400(self) -> None: status, _ = dispatch(self.broker, "POST", "/broker", _body({})) self.assertEqual(400, status) @@ -190,6 +230,14 @@ class TestSeamRoundTrip(unittest.TestCase): client.submit(forged) self.assertEqual([], broker.launched) # fail-closed across the wire + def test_list_live_enumerates_over_http(self) -> None: + secret = secrets.token_bytes(16) + broker = StubBroker(secret) + broker.live_source_ips = ["10.0.0.7", "10.0.0.8"] + client = self._serve(broker) + got = client.list_live(sign_request(LaunchRequest(op="list_live"), secret)) + self.assertEqual(["10.0.0.7", "10.0.0.8"], got) + class TestRequestLimits(unittest.TestCase): """The privileged listener must not let a caller that can merely reach the diff --git a/tests/unit/test_orchestrator_server.py b/tests/unit/test_orchestrator_server.py index 495cc998..30e33e08 100644 --- a/tests/unit/test_orchestrator_server.py +++ b/tests/unit/test_orchestrator_server.py @@ -604,13 +604,18 @@ if __name__ == "__main__": class TestReconcileRoute(unittest.TestCase): - """`POST /reconcile` — the host tells the orchestrator which bottles are - actually up, since the orchestrator can't see the backend from inside the - infra container.""" + """`POST /reconcile` — a bare self-heal trigger. The orchestrator pulls its + own live set from the broker (`list_live`); the caller no longer supplies it, + so the request body carries at most `grace_seconds`.""" def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() - self.orch = _orchestrator(Path(self._tmp.name) / "r.db") + store = RegistryStore(Path(self._tmp.name) / "r.db") + store.migrate() + secret = secrets.token_bytes(16) + # Hold a typed StubBroker so tests can set its live set directly. + self.broker = StubBroker(secret) + self.orch = OrchestratorCore(store, self.broker, secret) def tearDown(self) -> None: self._tmp.cleanup() @@ -627,30 +632,26 @@ class TestReconcileRoute(unittest.TestCase): def test_reaps_absent_and_reports_ids(self) -> None: dead = self._old("10.0.0.1") alive = self._old("10.0.0.2") - status, payload = dispatch( - self.orch, "POST", "/reconcile", _body({"live_source_ips": ["10.0.0.2"]})) + self.broker.live_source_ips = ["10.0.0.2"] # broker: only .2 is up + status, payload = dispatch(self.orch, "POST", "/reconcile", _body({})) self.assertEqual(200, status) self.assertEqual([dead], payload["reaped"]) self.assertIsNone(self.orch.registry.get(dead)) self.assertIsNotNone(self.orch.registry.get(alive)) - def test_missing_live_source_ips_is_400(self) -> None: - status, _ = dispatch(self.orch, "POST", "/reconcile", _body({})) - self.assertEqual(400, status) + def test_empty_body_triggers_the_sweep(self) -> None: + """No body is fine — the live set comes from the broker, not the body.""" + dead = self._old("10.0.0.1") + self.broker.live_source_ips = [] # broker: nothing running + status, payload = dispatch(self.orch, "POST", "/reconcile", b"") + self.assertEqual(200, status) + self.assertEqual([dead], payload["reaped"]) def test_grace_seconds_is_honoured(self) -> None: """A grace window wide enough to cover the row protects it.""" self.orch.registry.register("10.0.0.3") + self.broker.live_source_ips = [] status, payload = dispatch( - self.orch, "POST", "/reconcile", - _body({"live_source_ips": [], "grace_seconds": 3600})) + self.orch, "POST", "/reconcile", _body({"grace_seconds": 3600})) self.assertEqual(200, status) self.assertEqual([], payload["reaped"]) - - def test_non_string_entries_are_ignored(self) -> None: - dead = self._old("10.0.0.4") - status, payload = dispatch( - self.orch, "POST", "/reconcile", - _body({"live_source_ips": [None, 7, "10.0.0.9"]})) - self.assertEqual(200, status) - self.assertEqual([dead], payload["reaped"]) diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index c9031209..1b2a4d5d 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -39,6 +39,9 @@ class _FailingBroker(LaunchBroker): def _teardown(self, req: LaunchRequest) -> None: pass + def _list_live(self) -> list[str]: + return [] + class _UnavailableBroker(LaunchBroker): """Verifies the token, then raises the *ambiguous* BrokerUnavailableError — @@ -51,6 +54,9 @@ class _UnavailableBroker(LaunchBroker): def _teardown(self, req: LaunchRequest) -> None: pass + def _list_live(self) -> list[str]: + return [] + class TestOrchestrator(unittest.TestCase): def setUp(self) -> None: @@ -357,8 +363,9 @@ class TestOrchestratorSupervise(unittest.TestCase): pid = self.orch.supervise_queue_proposal( rec.bottle_id, tool=TOOL_EGRESS_ALLOW, proposed_file="routes:\n - host: google.com\n", justification="j") - # No live source IPs -> the bottle is reaped (grace 0 so it's immediate). - self.assertEqual([rec.bottle_id], self.orch.reconcile([], grace_seconds=0)) + # No live source IPs (nothing launched through the stub) -> the bottle + # is reaped (grace 0 so it's immediate). + self.assertEqual([rec.bottle_id], self.orch.reconcile(grace_seconds=0)) self.assertEqual( {"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid)) @@ -403,7 +410,9 @@ class TestOrchestratorReconcile(unittest.TestCase): live = self.orch.launch_bottle("10.243.0.2", tokens={"EGRESS_TOKEN_0": "keep"}) self._age_all(600) - self.assertEqual([dead.bottle_id], self.orch.reconcile(["10.243.0.2"])) + # The broker reports only .2 as still running -> .1 is reaped. + self.broker.live_source_ips = ["10.243.0.2"] + self.assertEqual([dead.bottle_id], self.orch.reconcile()) self.assertIsNone(self.store.get(dead.bottle_id)) self.assertIsNotNone(self.store.get(live.bottle_id)) # The in-memory egress credential goes with the row. @@ -415,14 +424,31 @@ class TestOrchestratorReconcile(unittest.TestCase): broker error must not stop the sweep clearing the row.""" self.orch.launch_bottle("10.243.0.1") self._age_all(600) - self.broker.launched.clear() - self.orch.reconcile([]) + self.broker.live_source_ips = [] # broker reports nothing running + self.orch.reconcile() self.assertEqual([], self.broker.torn_down) def test_reconcile_keeps_everything_when_all_are_live(self) -> None: a = self.orch.launch_bottle("10.243.0.1") b = self.orch.launch_bottle("10.243.0.2") self._age_all(600) - self.assertEqual([], self.orch.reconcile(["10.243.0.1", "10.243.0.2"])) + self.broker.live_source_ips = ["10.243.0.1", "10.243.0.2"] + self.assertEqual([], self.orch.reconcile()) + self.assertIsNotNone(self.store.get(a.bottle_id)) + self.assertIsNotNone(self.store.get(b.bottle_id)) + + def test_reconcile_skipped_when_broker_cannot_enumerate(self) -> None: + """A broker that can't return an authoritative live set must NOT be + treated as "nothing is live" — that would reap every healthy bottle. + The sweep is skipped instead.""" + a = self.orch.launch_bottle("10.243.0.1") + b = self.orch.launch_bottle("10.243.0.2") + self._age_all(600) + + def _boom() -> list[str]: + raise RuntimeError("docker ps failed") + + self.broker._list_live = _boom # type: ignore[method-assign] + self.assertEqual([], self.orch.reconcile()) self.assertIsNotNone(self.store.get(a.bottle_id)) self.assertIsNotNone(self.store.get(b.bottle_id))