diff --git a/app/worker_api.py b/app/worker_api.py index 5cea9c5d..4ab49742 100644 --- a/app/worker_api.py +++ b/app/worker_api.py @@ -945,12 +945,22 @@ def _seconds_to_next_window(now: float | None = None) -> float: 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: + client = await asyncio.to_thread(activity._client) + try: + await asyncio.to_thread(activity.sweep, client) + finally: + with contextlib.suppress(Exception): + client.close() + return switches = _activity_switches result = await asyncio.to_thread(activity.run_round) if not _activity_enabled or _activity_switches != switches: 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