From 4efbbbc9771b60524a147ab6aeea65667e63d72c Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:38:05 +0100 Subject: [PATCH 1/2] fix(worker): keep removing stale activity probes while Activity is off With Activity off, _activity_cycle returned before its sweep, so the only sweep was the one at worker start. A finished probe another worker left after that (one that died and came back under a new id) stayed on the host, and its Docker log keeps raw addresses. The cycle now checks Docker first and, when Activity is off, runs the sweep without a round. --- app/worker_api.py | 16 +++++++++++++--- tests/test_activity_worker.py | 33 ++++++++++++++++++++++++++++++++- 2 files changed, 45 insertions(+), 4 deletions(-) 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 From bbf9a95f4562b21ddb43e164532297783a8361fb Mon Sep 17 00:00:00 2001 From: GeiserX <9169332+GeiserX@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:54:07 +0100 Subject: [PATCH 2/2] fix(worker): open, sweep and close the probe client in one thread A cancelled await on the client factory could drop a client created after the cancel, leaving its connection open. _sweep_probes does all three in one thread; the start-up sweep uses it too. --- app/worker_api.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/app/worker_api.py b/app/worker_api.py index 4ab49742..2b3ad9d8 100644 --- a/app/worker_api.py +++ b/app/worker_api.py @@ -944,6 +944,16 @@ 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. @@ -954,12 +964,7 @@ async def _activity_cycle() -> None: 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() + await asyncio.to_thread(_sweep_probes) return switches = _activity_switches result = await asyncio.to_thread(activity.run_round) @@ -983,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: