diff --git a/backend/worker/service.py b/backend/worker/service.py index 3ddfcb50..2210493b 100644 --- a/backend/worker/service.py +++ b/backend/worker/service.py @@ -93,6 +93,10 @@ class ControlPlane: self._server = None self._tasks: list[asyncio.Task] = [] self._started = False + # The port we actually bound, which is not necessarily the configured + # one — an enrollment token carries this, so advertising the config + # value instead hands workers an endpoint nothing is listening on. + self._port: Optional[int] = None @property def running(self) -> bool: @@ -129,9 +133,10 @@ class ControlPlane: artifact_dir=locations["artifacts"], cert_fingerprint=self.credentials.fingerprint, ) + self._port = port or control_port() self._server = await serve( self.servicer, - port=port or control_port(), + port=self._port, certificate_pem=self.credentials.certificate_pem, private_key_pem=self.credentials.private_key_pem, ) @@ -140,7 +145,7 @@ class ControlPlane: asyncio.create_task(self._dispatch_loop(), name="worker-dispatch"), ] self._started = True - logger.info("Remote worker control plane started on port %d", port or control_port()) + logger.info("Remote worker control plane started on port %d", self._port) async def stop(self) -> None: for task in self._tasks: @@ -153,6 +158,7 @@ class ControlPlane: # would delay app shutdown for work that survives anyway. await self._server.stop(grace=2.0) self._server = None + self._port = None self._started = False async def _sweep_loop(self) -> None: @@ -226,7 +232,7 @@ class ControlPlane: or tls.primary_ip() or "127.0.0.1" ) - return f"{host}:{control_port()}" + return f"{host}:{self._port or control_port()}" def snapshot(self, *, now: Optional[float] = None) -> dict: """Everything the workers UI needs in one call.""" diff --git a/tests/test_worker_service_api.py b/tests/test_worker_service_api.py index f884de3d..b1540842 100644 --- a/tests/test_worker_service_api.py +++ b/tests/test_worker_service_api.py @@ -301,3 +301,37 @@ def test_consent_is_recorded_explicitly(client, db): def test_task_listing_is_empty_when_stopped(client): body = client.get("/workers/tasks").json() assert body == {"tasks": [], "queue_depth": 0} + + +@pytest.mark.asyncio +async def test_enrollment_advertises_the_port_actually_bound(db, monkeypatch, tmp_path): + """A token carries the endpoint a worker will dial. Advertising the + configured port while listening on another hands workers an address + nothing answers on — found by running the thing on a non-default port. + """ + monkeypatch.setattr( + service, + "paths", + lambda: { + "root": str(tmp_path), + "certificate": str(tmp_path / "cp.crt"), + "private_key": str(tmp_path / "cp.key"), + "worker_key": str(tmp_path / "w.key"), + "artifacts": str(tmp_path / "artifacts"), + }, + ) + monkeypatch.delenv("OMNIVOICE_WORKER_PORT", raising=False) + monkeypatch.delenv("OMNIVOICE_WORKER_ENDPOINT_HOST", raising=False) + + plane = service.ControlPlane() + await plane.start(port=7601) + try: + assert plane.default_endpoint().endswith(":7601") + assert plane.create_enrollment().endpoint.endswith(":7601") + finally: + await plane.stop() + + +def test_endpoint_falls_back_to_the_configured_port_when_stopped(monkeypatch): + monkeypatch.delenv("OMNIVOICE_WORKER_PORT", raising=False) + assert service.ControlPlane().default_endpoint().endswith(f":{service.DEFAULT_PORT}")