diff --git a/app/worker_api.py b/app/worker_api.py index 5cea9c5d..2b3ad9d8 100644 --- a/app/worker_api.py +++ b/app/worker_api.py @@ -944,13 +944,28 @@ def _seconds_to_next_window(now: float | None = None) -> float: return activity.WINDOW_SECONDS - (now % activity.WINDOW_SECONDS) + 1 +def _sweep_probes() -> None: + """Open a client, sweep, close it, all in one thread: a cancelled await cannot drop the client.""" + client = activity._client() + try: + activity.sweep(client) + finally: + with contextlib.suppress(Exception): + client.close() + + async def _activity_cycle() -> None: - """One round, if the owner switched Activity on and Docker answers.""" + """One round, if the owner switched Activity on and Docker answers. + + With Activity off there is no round, but probes other workers left behind + are still swept: their logs hold raw addresses and nothing else removes them. + """ global _pending_activity, _activity_seq - if not _activity_enabled: - return if not await asyncio.to_thread(orchestrator.docker_available): return + if not _activity_enabled: + await asyncio.to_thread(_sweep_probes) + return switches = _activity_switches result = await asyncio.to_thread(activity.run_round) if not _activity_enabled or _activity_switches != switches: @@ -973,12 +988,7 @@ async def _activity_loop() -> None: """ with contextlib.suppress(Exception): if await asyncio.to_thread(orchestrator.docker_available): - client = await asyncio.to_thread(activity._client) - try: - await asyncio.to_thread(activity.sweep, client) - finally: - with contextlib.suppress(Exception): - client.close() + await asyncio.to_thread(_sweep_probes) while True: await asyncio.sleep(_seconds_to_next_window()) try: diff --git a/tests/test_activity_worker.py b/tests/test_activity_worker.py index 2c26b2ae..74219066 100644 --- a/tests/test_activity_worker.py +++ b/tests/test_activity_worker.py @@ -484,9 +484,11 @@ def wapi(): with ( patch("app.worker_api.orchestrator.docker_available", return_value=True) as docker_ok, patch("app.worker_api.activity.run_round", return_value=None) as run_round, + patch("app.worker_api.activity._client", return_value=MagicMock()), + patch("app.worker_api.activity.sweep", return_value=[]), ): asyncio.run(worker_api._activity_cycle()) - assert not docker_ok.called and not run_round.called, "a round ran before any reply" + assert not run_round.called, "a round ran before any reply" reply = MagicMock() reply.json.return_value = {"activity": {"enabled": True}} worker_api._apply_activity_reply(reply, None) @@ -523,10 +525,39 @@ def test_disabled_never_runs_a_round(self, wapi): with ( patch("app.worker_api.orchestrator.docker_available", return_value=True), patch("app.worker_api.activity.run_round") as run_round, + patch("app.worker_api.activity._client", return_value=MagicMock()), + patch("app.worker_api.activity.sweep", return_value=[]), ): asyncio.run(wapi._activity_cycle()) assert not run_round.called + def test_disabled_still_sweeps_probes_other_workers_left(self, wapi): + """With Activity off the sweep at worker start is not the last one: a probe another + worker leaves later (it died and came back under a new id) holds raw addresses in + its log, and only a sweep removes it.""" + wapi._activity_enabled = False + client = MagicMock() + with ( + patch("app.worker_api.orchestrator.docker_available", return_value=True), + patch("app.worker_api.activity.run_round") as run_round, + patch("app.worker_api.activity._client", return_value=client), + patch("app.worker_api.activity.sweep", return_value=[]) as sweep, + ): + asyncio.run(wapi._activity_cycle()) + sweep.assert_called_once_with(client) + client.close.assert_called_once() + assert not run_round.called + + def test_disabled_without_docker_does_not_sweep(self, wapi): + wapi._activity_enabled = False + with ( + patch("app.worker_api.orchestrator.docker_available", return_value=False), + patch("app.worker_api.activity._client") as make_client, + patch("app.worker_api.activity.sweep") as sweep, + ): + asyncio.run(wapi._activity_cycle()) + assert not make_client.called and not sweep.called + def test_enabled_runs_a_round_negative_control(self, wapi): """The control for the test above: the same call with the switch on runs.""" wapi._activity_enabled = True