Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
stroblme
2026-11-11 09:41:05 +01:00
parent a3d532f3a5
commit cd8831cfd3
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
worker/fluksio_worker/agent.py
+7 -3
View File
@@ -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)