Close the worker spawn races and the cancelled-request leak - #5572
Conversation
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
![griptapeops[bot]](https://avatars.githubusercontent.com/in/3186107?s=60&v=4){kind=small}
There was a problem hiding this comment.
Each of the nine items lands where its own failure originates, so the framing holds. Checks clean on the changed files; 216 unit tests pass for the touched modules; the two exec-dep e2e files give 28 passed, 1 failed, and I confirmed that failure reproduces on the base commit with none of this branch's changes.
I mutation-checked the new tests rather than trusting them — inverting the settled check, dropping the claim set from the duplicate guard, removing discard_request from the cancel path, removing the task.cancelled() guard, and shifting the retry's exhaustion boundary each turned a test red.
One correctness finding on the spawn claim, reproduced on this branch. One design and one LLMisms finding, all inline.
|
@griptapeops re-review |
|
|
||
| released.set() | ||
| with pytest.raises(OSError, match="stale spawn died"): | ||
| await stale |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
All 3 findings from the first pass are fixed and resolved, each in its own commit. I verified the claim fix by mutation rather than by reading it: reverting the release to an unconditional pop turns the new interleaving test red.
make check clean, 6113 unit tests pass, pyright clean on both changed source files.
One advisory finding on the retry restructure, inline. Nothing blocking.
cc4a67e to
f5b1fab
Compare
| async def _move_preview_into_place(self, temp_path: Path, destination_path: Path) -> None: | ||
| """Replace the previous preview with the finished one. | ||
|
|
||
| Same directory keeps this on one filesystem, so it is atomic: a reader sees either the | ||
| previous preview or this finished one, never a torn file. | ||
|
|
||
| On Windows a replace fails with access denied while any handle on the destination is open, | ||
| which serving the previous preview does. That handle closes when the response completes, so | ||
| retrying is what turns a discarded encode back into a preview. POSIX never reaches the retry. | ||
|
|
||
| Args: | ||
| temp_path: The finished encode, in the destination's directory | ||
| destination_path: Where the preview is served from | ||
| """ | ||
| # `while True` rather than a bounded `for`: the only ways out are a completed move and a | ||
| # raise, so exhaustion cannot fall through and hand the caller a None it reads as success. | ||
| attempts_left = _REPLACE_ATTEMPTS | ||
| while True: | ||
| attempts_left -= 1 | ||
| try: | ||
| await anyio.Path(temp_path).replace(destination_path) | ||
| except PermissionError: | ||
| if attempts_left == 0: | ||
| raise | ||
| logger.debug( | ||
| "Could not move the finished preview to %s yet; something still holds it open. Retrying.", | ||
| destination_path, | ||
| ) | ||
| await anyio.sleep(_REPLACE_RETRY_DELAY_S) | ||
| else: | ||
| return |
There was a problem hiding this comment.
I think I'd prefer this as its own PR outside the stack. Or is it related to workers?
There was a problem hiding this comment.
Nothing can keep an agent from scope creeping, nothing... (removed from this PR)
There was a problem hiding this comment.
Is this related to workers? Would prefer to have it split out.
|
|
||
| return self.destination_preview_file_name | ||
|
|
||
| async def _move_preview_into_place(self, temp_path: Path, destination_path: Path) -> None: |
There was a problem hiding this comment.
Is this the same class of bug James recently fixed? Can we generalize the solution?
There was a problem hiding this comment.
Deferring to James' approach
f5b1fab to
48d943a
Compare
cjkindel
changed the title
8e65cd4 to
91ed58f
Compare
518071e to
6f5e244
Compare
|
@griptapeops re-review |
There was a problem hiding this comment.
All 4 findings from earlier cycles are resolved. The branch rebased onto b1c2b36 and dropped the three non-worker items, so the two threads on files that left the PR are moot here.
I mutation-checked each new guard rather than reading it: an unconditional claim release, dropping _spawns_in_flight.clear() from reset_workers, dropping the task.cancelled() guard, dropping discard_request from the cancel path, and reverting the duplicate check each turn a test red. 6340 unit tests pass; make check clean. The deferred double-fork-through-the-registry item reproduces exactly as the description states it.
3 advisory findings inline, none blocking.
| # out of here means only one thing -- the caller was cancelled -- because a worker going | ||
| # away raises WorkerGoneError instead. | ||
| return await asyncio.wrap_future(future) | ||
| try: |
There was a problem hiding this comment.
The cancelled-run leak is closed on one of the two awaits it can land on. The try opens after the publish, so a cancellation delivered while forward_event_to_worker is parked on send_message unwinds with the entry still tracked — the same leak this commit exists to close, one per cancelled execution, each still carrying its worker's tag. I reproduced it: parking the publish and cancelling the task leaves the pending map holding the entry after CancelledError propagates.
The description defers "a raising publish in route_to_worker leaking the same entry from outside the try", and a CancelledError out of the publish is one instance of that, so this is the deferred item rather than a new one. Raising it only because the try that fixes it now sits directly below: moving the publish inside it and widening the except to BaseException closes both halves for no extra structure. I ran that shape against the suite plus the reproduction above — 129 passed.
| return await asyncio.wrap_future(future) | ||
| except asyncio.CancelledError: | ||
| # The other ways out of this await remove the request themselves: a response pops it in | ||
| # _try_match, eviction in cancel_requests_by_tag. A cancelled caller does not, so |
There was a problem hiding this comment.
cancel_requests_by_tag is not a method on RequestClient. It is fail_requests_by_tag, renamed in edc784d ("Fail a request with why its worker went away, rather than cancelling it"), which predates this branch's base — so this points at a name that exists on no commit the code can be read against. Same name again on line 555 and in the docstring of test_flow_cancellation_stops_tracking_the_request (test_app_worker.py:1421).
The rename is load-bearing to the sentence, not just cosmetic: fail_requests_by_tag raises at the awaiter instead of cancelling it, which is the reason a cancelled caller is the one exit that has to clean up after itself. Worth correcting all three while the wrong name still reads as a method someone could go look up.
| task = asyncio.create_task(_raise()) | ||
| await asyncio.gather(task, return_exceptions=True) | ||
|
|
||
| with patch.object(worker_manager, "note_worker_unavailable") as mock_refuse: |
There was a problem hiding this comment.
Five of the eight commits leave the suite red. 27dfa355 added this test patching _refuse_spawn, which exists neither on this branch nor its base, so patch.object raises AttributeError — and it stayed red until this line changed it to note_worker_unavailable in 3f2463a6, six commits later. 27dfa355, faabeffa, 30a3df56, bd43c497 and 62e08937 each fail test_a_failed_spawn_still_records_a_refusal.
That cuts against the description's "one commit per item so they can be read or dropped independently": dropping the last two commits leaves a failing suite, and a bisect landing in the middle of the stack hits a failure unrelated to whatever it is hunting. A fixup of the name into 27dfa355 restores it, since the fix is one identifier.
Verified per commit: 118 pass at 78401014, one failure from 27dfa355 through 62e08937, clean again at 3f2463a6 and 6f5e244.
`spawn_worker` checked `_managed_worker_processes` for a duplicate, then wrote its entry 68 lines later, with the static-URL await and the fork in between. Two spawns for one library both got past the check and both forked; only one could be recorded, leaving the other's process untracked and holding that library's dependencies in memory until its own heartbeat lapsed. Two `StartWorkerRequest`s for one library name overlap whenever `_library_file_path_to_info` carries duplicate entries for it (a FAILURE entry alongside the LOADED copy), or when `_start_workers` runs twice before the first worker registers. The key is now claimed in the same step as the check, and released in a `finally` so a failed fork does not leave the library permanently unspawnable.
`_log_spawn_error` asked `task.exception()` first, which raises `CancelledError` on a cancelled task. Raising from a done-callback surfaces as loop-level "Exception in callback" noise and skips the `note_worker_unavailable` below it, so nothing records that no worker is coming. Cancellation reaches this callback at loop teardown, where nothing is waiting on the worker, so the cost today is the noise rather than a stalled run. Note on A1 of the tracking issue, which covers the same function: the claim that the spawn task can be garbage collected while parked does not hold. `Event.wait()` appends a future to the event's `_waiters`, and the awaiting task installs `Task.__wakeup` as that future's done-callback, so the task stays reachable from the manager for as long as it is parked. No change made there.
`route_to_worker` registers a request then awaits it. Every other exit from that await removes the entry: a response pops it in `_try_match`, eviction in `cancel_requests_by_tag`. Flow cancellation re-raised without removing anything, so the entry stayed in `_pending_requests` for the life of the process -- one per cancelled node execution. Each leaked entry keeps its worker's tag, so a later `cancel_requests_by_tag` walks and re-settles long-dead requests, and `pending_count`, the only visibility into that map, reports a number that never comes down. Nothing fails loudly; it degrades over a long editing session. `_cancel_request` did exactly the removal needed but was async, and awaiting inside an except block while a cancellation is being delivered adds a suspension point in the one place it is least wanted. Its body never awaited, so it becomes the public synchronous `discard_request` and its five internal callers drop their `await`.
`pending_count` and `pending_request_ids` read `_pending_requests` while holding nothing, and it is mutated from more than one loop: `_try_match` on the transport loop, `track_request` and `cancel_requests_by_tag` from whichever loop issued the request. `len()` only risks a stale answer, which is fine for a diagnostic. Building the id list iterates, so a concurrent insert or pop raises "dictionary changed size during iteration" -- a diagnostic that can raise is worse than one that answers a moment out of date. Both are diagnostics with no production caller today, so nothing holds the lock across a call into them.
`_orchestrator_static_server_base_url` picks between a direct read and a thread hop, and between two warnings when no URL arrives. Nothing exercised either choice: inverting the settled check left the suite green, and so did swapping the two warnings. Both choices matter to whoever has to act on them. Taking the hop when the answer is already in parks an uncancellable default-executor thread that teardown joins, and blaming the settle timeout for a resolution that raised points an operator at slow startup instead of the bind failure that actually happened. Five tests: the direct read, the hop with its timeout, each warning, and a cloud backend staying quiet because there its URLs outlive the worker anyway. Each fails if the branch it covers is inverted.
The static-URL tests set `static_server_base_url_settled`, which is a property on the real manager, so pyright rejected the assignment even though the object is a MagicMock. Casting once per test says what the object actually is instead of suppressing each line.
A regression in the spawn claim: `reset_workers` cleared the process registry but not the new claim set, so a spawn suspended across a library reload kept its claim and the reload's own spawn for that library was refused. The refusal records nothing (a worker is normally on its way when a key is claimed) so the next run would wait out the whole startup grace and then blame the library load. Cleared with the registry, with a test that fails without it. `note_worker_unavailable` is deliberately still NOT called from the duplicate guard: it records a reason and settles the library, which would falsely fail a library that already has, or is about to have, a worker. Comments that were wrong or that answered a reviewer are gone rather than reworded again: a claim that `list(dict.keys())` can raise "dictionary changed size during iteration" (it cannot, the whole build is one C loop under the GIL) and a `discard_request` docstring citing a suspension point that never existed. `test_an_undecided_url_is_waited_for_off_the_loop` asserted its own name only loosely; it now records the executing thread and fails if the blocking wait runs on the event loop.
`reset_workers` drops the claims so a reload can spawn its libraries again, which leaves a spawn suspended between its claim and the registry write resuming to find the key claimed by the reload's spawn. Releasing by name alone freed that one, so the library was admitted again while a spawn was genuinely in flight -- and a second fork leaves one of the two processes untracked, which is the leak the claim exists to prevent. Claims are now per attempt: the map holds a token identifying the holder, and the release is conditional on the stored token still being the one this attempt took. The test drives the interleaving: park a spawn, reset, let the reload claim the key, then release the stale spawn with its fork failing so no registry entry shadows a wrongly-freed claim. It fails against an unconditional release. The comment on the claim also drops its comparison with the guard this replaced and states the constraint instead: the registry entry is written only once the subprocess exists, and the work in between suspends.
6f5e244 to
7047794
Compare
|
|
||
| task.cancel() | ||
| with pytest.raises(asyncio.CancelledError): | ||
| await task |
|
|
||
| task.cancel() | ||
| with pytest.raises(asyncio.CancelledError): | ||
| await task |
2 participants
Four worker defects found while reviewing this stack: two races, one leak, and one done-callback that raises. Stacked on #5472, one commit per item so they can be read or dropped independently.
Races and leaks
Claim a worker spawn before it suspends.
spawn_workerchecked_managed_worker_processesfor a duplicate, then wrote its entry 68 lines later, with the static-URL await and the fork in between. Two spawns for one library both got past the check and both forked; only one could be recorded, leaving the other's process untracked and holding that library's dependencies in memory until its own heartbeat lapsed. TwoStartWorkerRequests for one library overlap whenever_library_file_path_to_infocarries duplicate entries for it, which we have seen. The claim is a per-attempt token and is released only to the attempt that took it, so a spawn outlived by areset_workerscannot free the claim the reload's own spawn has since taken.Stop tracking a request whose run was cancelled.
route_to_workerregisters a request then awaits it. The other exits remove the entry (a response pops it in_try_match, eviction infail_requests_by_tag) but a cancelled caller did not, so the entry outlived the run, one per cancelled node execution. Each kept its worker's tag, so a laterfail_requests_by_tagwalked and re-settled long-dead requests, andpending_countreported a number that never came down._cancel_requestdid exactly the removal needed but was async and never awaited anything, so it becomes the public synchronousdiscard_requestand its five internal callers drop theirawait.Read a spawn task's cancellation before its exception.
_log_spawn_erroraskedtask.exception()first, which raisesCancelledErroron a cancelled task. Raising from a done-callback becomes loop-level "Exception in callback" noise and skips thenote_worker_unavailablebelow it, so nothing records that no worker is coming. Cancellation reaches this callback at loop teardown, so the cost today is the noise.Read the pending request map under its lock.
pending_countandpending_request_idsread_pending_requestswhile holding nothing, though it is mutated from more than one loop. Both are diagnostics with no production caller, so nothing holds the lock across a call into them.Coverage
Cover the spawn side of the static-URL handover.
_orchestrator_static_server_base_urlpicks between a direct read and a thread hop, and between two warnings when no URL arrives. Nothing exercised either choice: inverting the settled check left the suite green, and so did swapping the warnings. Both matter to whoever acts on them: taking the hop when the answer is in parks an uncancellable executor thread, and blaming the settle timeout for a resolution that raised points an operator at slow startup instead of the bind failure.Scope
Worker code only:
worker_manager.py,request_client.py, andtest_app_worker.py.Three non-worker items were previously on this branch and have been removed. They were found while reviewing this stack, which is not a reason to ship them with it:
Path.replaceinatomic_write_bytes. Filed as Windows: a finished video preview is discarded when its rename target is still held open #5579, referencing Systematize atomic file saving #3166.servers/static.pyresolving its managers throughcurrent_engine()on uvicorn's thread, andStaticFilesManager._resolve_static_serverrunning unserialized. Both latent while a process holds oneEngine. Filed as Static file server resolves its engine ambiently and starts unserialized #5580.State
make checkclean. Unit suite 6114 passed.Also on the punch list and deliberately not fixed here: a raising publish in
route_to_workerleaking the same entry from outside thetry;reset_workersnot cancelling a spawn suspended between claim and registry write, so a reload can still double-fork before the claim token is taken.A1 of the tracking issue, a scheduled spawn being garbage collected, is a false positive.
Event.wait()appends a future to the event's_waiters, and the awaiting task installsTask.__wakeupas that future's done-callback, so the task stays reachable from the manager for as long as it is parked. The cancellation half of that item is the done-callback commit above.