From cd8831cfd3000cdfe447d5c74131c0df450aca88 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 11 Nov 2026 09:41:05 +0100 Subject: [PATCH] The worker agent lets a call it was just sent register before it reads the next frame, so a cancel that arrives in the same read as its call is no longer ignored --- worker/fluksio_worker/agent.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py index f76e6d7..e90c6b0 100644 --- a/worker/fluksio_worker/agent.py +++ b/worker/fluksio_worker/agent.py @@ -226,10 +226,10 @@ class Agent: # this import" is a question about *this* machine's # packages. log.info("%s %s", op, message.get("call_id")) - self._spawn(self._run(socket, message)) + await self._spawn(self._run(socket, message)) elif op == "serve": log.info("serve %s", message.get("call_id")) - self._spawn(self._serve(socket, message)) + await self._spawn(self._serve(socket, message)) elif op == "cancel": self._cancel(str(message.get("call_id") or "")) except Exception: @@ -245,13 +245,17 @@ class Agent: if self._idled: raise Idle - def _spawn(self, call: Any) -> None: + async def _spawn(self, call: Any) -> None: task = asyncio.create_task(call) self._calls.add(task) task.add_done_callback(self._calls.discard) # Without this a failure in the task is only noticed when it is # garbage collected, which reads as a call that vanished. task.add_done_callback(_report_failure) + # Its first step registers the call, and has to run before the next + # frame is read: frames that arrived together are read without a + # pause, and a cancel among them would find nothing to end. + await asyncio.sleep(0) async def _end_calls(self) -> None: calls = list(self._calls)