fix(sdk/python): resolve ResultCache cross-loop deadlock (#623) - #799
fix(sdk/python): resolve ResultCache cross-loop deadlock (#623)#7997vignesh wants to merge 8 commits into
Conversation
Performance
✓ No regressions detected |
📊 Coverage gateThresholds from
✅ Gate passedNo surface regressed past the allowed threshold and the aggregate stayed above the floor. |
📐 Patch coverage gateThreshold: 80% on lines this PR touches vs
✅ Patch gate passedEvery surface whose lines were touched by this PR has patch coverage at or above the threshold. |
AbirAbbas
left a comment
AbirAbbas
left a comment
There was a problem hiding this comment.
I took this for a spin rather than just reading the diff. On the merge-base I reproduced the failure with start() on a background thread's loop and stop() via asyncio.run() on the main thread: RuntimeError: got Future attached to a different loop, and since stop() aborts mid-way the cache is left populated. On this branch the same scenario (and the owning-loop-closed variant) completes cleanly, and I checked that the stale cleanup task really does get cancelled on its owning loop rather than leaked. I also ran the new test file against the merge-base — 3 of the 8 genuinely fail pre-fix, so these are real regression tests rather than mirrors of the implementation. Full suite on the branch is green for me locally (1767 passed, 4 skipped) and ruff is clean. Nice touch that stop() now clears the cache even on the cross-loop path — before, the failed stop left the contents behind.
The one thing I'd like sorted before this closes #623: the same foreign-loop task.cancel(); await task pattern still lives in AsyncExecutionManager.stop() (async_execution_manager.py:337-340) and ConnectionManager.close() (connection_manager.py:99-102). I verified on this branch that starting the manager on one loop and stopping it from another still raises the identical RuntimeError at async_execution_manager.py:339 — before result_cache.stop() is ever reached. Since client.aclose() goes through manager.stop(), the end-to-end sync/async mixing case from #620/#623 still fails one level up. Happy with either extending the same loop-aware treatment to those two sites in this PR or a follow-up, but either way I'd change "Fixes #623" to "Partially addresses #623" so the issue doesn't auto-close on merge.
| # Cross-loop stop: cancel on the owning loop, never await here. | ||
| self._discard_stale_cleanup_task() | ||
|
|
||
| self._cleanup_task = None |
There was a problem hiding this comment.
One residual race here: stop() snapshots these fields at the top, awaits the task in between, and then unconditionally nulls all three — with no lock on either side. A start() racing in from another thread (or even on the same loop during the await task) can install a fresh task that gets clobbered here and leaks, still running on its loop. The serialized cross-loop case — the #623 scenario — is handled, and the old code was just as unguarded, so not blocking. But since the instance already has self._lock, it'd be cheap to take it around the snapshot and around this mutation block (never across the await).
…ool (Agent-Field#623) Addresses review feedback on Agent-Field#799: the same foreign-loop 'task.cancel(); await task' hazard that affected ResultCache also lived in AsyncExecutionManager.stop() and http_connection_manager ConnectionManager.close(). Since client.aclose() flows through manager.stop() -> connection_manager.close(), the end-to-end sync/async mixing case (Agent-Field#620/Agent-Field#623) still raised 'got Future attached to a different loop' one level up, before result_cache.stop() was reached. Changes: - New agentfield/async_lifecycle.py with shared loop-aware teardown helpers: current_running_loop(), cancel_task_cross_loop(), cancel_and_await_if_same_loop(). Single source of truth for the safe pattern. - ResultCache refactored to use the shared helpers (behaviour unchanged). - AsyncExecutionManager records its owning loop at start(); stop() only awaits background tasks / sets the shutdown Event on that loop, and cancels cross-loop without awaiting otherwise. - http_connection_manager ConnectionManager records its owning loop at start(); close() takes the async lock + awaits the session only on the owning loop, and on a foreign loop schedules session/connector close on the owning loop via call_soon_threadsafe without awaiting. Tests: new tests/test_async_lifecycle_deadlock.py covers cross-loop teardown of both components (3 of its 4 checks fail pre-fix on the merge-base). Full result_cache + manager + connection suite green (110 passed); result_cache.py coverage 98%.
|
Good catch you're right that the same foreign-loop What changed in the latest push:
I verified the end-to-end path: starting the manager on a background thread's loop and stopping it from another no longer raises at async_execution_manager.py:339, and New Since the whole |
|
@santoshkumarradha pls review |
|
Thanks for the thorough review @AbirAbbas! Just pushed the two in-scope fixes:
For the other two (RuntimeError catch breadth in async_lifecycle.py and the connection manager's lockless mutation on cross-loop close), I'll address those in a follow-up once this lands. They're on my radar. Also already resolved the merge conflict from the ruff pin in the previous push. |
…#623) ResultCache mixed a threading.RLock (for its data) with loop-bound asyncio primitives (asyncio.Event for shutdown, asyncio.Task for the cleanup loop). When start() and stop() ran on different event loops — which happens when the AgentFieldClient's sync and async execution paths are mixed (Agent-Field#620) — stop() would raise 'got Future attached to a different loop' and could wedge the process waiting on a task it can never await. Fix: make the cache lifecycle loop-aware. - Record the event loop the cleanup task/shutdown event are bound to. - start() is now idempotent on the same loop and rebinds cleanly when called on a new loop, discarding the stale task via call_soon_threadsafe(task.cancel) on its owning loop — never a cross-loop await. - stop() only awaits the cleanup task when on its owning loop; from a different loop it cancels without awaiting. The cache is always cleared regardless (that path only needs the thread lock). - Shrink the cleanup loop's critical section so stats logging no longer runs while holding the lock, reducing contention with sync callers. Adds tests/test_result_cache_deadlock.py covering cross-loop stop, idempotent/rebinding start, concurrent sync access during cleanup, and the disabled-cache no-op path. result_cache.py coverage: 88% -> 95%.
…ool (Agent-Field#623) Addresses review feedback on Agent-Field#799: the same foreign-loop 'task.cancel(); await task' hazard that affected ResultCache also lived in AsyncExecutionManager.stop() and http_connection_manager ConnectionManager.close(). Since client.aclose() flows through manager.stop() -> connection_manager.close(), the end-to-end sync/async mixing case (Agent-Field#620/Agent-Field#623) still raised 'got Future attached to a different loop' one level up, before result_cache.stop() was reached. Changes: - New agentfield/async_lifecycle.py with shared loop-aware teardown helpers: current_running_loop(), cancel_task_cross_loop(), cancel_and_await_if_same_loop(). Single source of truth for the safe pattern. - ResultCache refactored to use the shared helpers (behaviour unchanged). - AsyncExecutionManager records its owning loop at start(); stop() only awaits background tasks / sets the shutdown Event on that loop, and cancels cross-loop without awaiting otherwise. - http_connection_manager ConnectionManager records its owning loop at start(); close() takes the async lock + awaits the session only on the owning loop, and on a foreign loop schedules session/connector close on the owning loop via call_soon_threadsafe without awaiting. Tests: new tests/test_async_lifecycle_deadlock.py covers cross-loop teardown of both components (3 of its 4 checks fail pre-fix on the merge-base). Full result_cache + manager + connection suite green (110 passed); result_cache.py coverage 98%.
…op lock (Agent-Field#623) 1. ResultCache.stop(): compare-before-clear so a concurrent start() that installs fresh references between the await and the cleanup doesn't get its new task orphaned. 2. AsyncExecutionManager.stop(): skip the _execution_lock section on cross-loop stop — the lock is bound to the owning loop, taking it from a foreign loop raises the same 'got Future attached to a different loop' error one line below the fix.
78641ff to
70078b9
Compare
|
Hey @AbirAbbas, resolved the merge conflict. Rebased the branch cleanly on top of current main so the diff only shows the actual fix commits now, no release-bot noise. The two review fixes you asked for are still there (compare-before-clear + skip execution lock cross-loop). Should be good to go once CI passes. |
There was a problem hiding this comment.
🟡 Not ready to approve
The new regression tests can be flaky because they close event loops immediately after scheduling cross-loop cancellations, risking pending-task warnings or RuntimeError during loop shutdown.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Review details
Suppressed comments (3)
sdk/python/tests/test_result_cache_deadlock.py:60
- The loop thread is stopped and the event loop is closed immediately after a cross-loop stop. Since stop() only schedules cancellation on the owning loop (it doesn’t await it), closing the loop without giving it a chance to process the scheduled cancel can intermittently produce "Task was destroyed but it is pending!" warnings (and make the test flaky).
finally:
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
loop1.close()
sdk/python/tests/test_result_cache_deadlock.py:130
- Same as above: this loop is closed immediately after scheduling cross-loop cancellation of the stale cleanup task. Without allowing loop1 to run briefly, this can leave the stale task pending at loop shutdown and cause intermittent pending-task warnings/flakes.
finally:
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
loop1.close()
sdk/python/tests/test_async_lifecycle_deadlock.py:52
- _shutdown_loop() closes the event loop even if the thread hasn’t actually stopped yet (join(timeout=2) can time out). Closing a still-running loop raises RuntimeError and would obscure the real failure mode this regression test is trying to detect.
loop.call_soon_threadsafe(loop.stop)
thread.join(timeout=2)
if not loop.is_closed():
loop.close()
- Files reviewed: 6/6 changed files
- Comments generated: 0 new
- Review effort level: Lite
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
Add time.sleep(0.2) before stopping loops to let scheduled cross-loop cancels process, and guard loop.close() against still-running threads. Addresses Copilot review feedback on test reliability.
81c1822 to
af01d7a
Compare
There was a problem hiding this comment.
🟡 Not ready to approve
The new test_result_cache_deadlock.py thread/loop shutdown code can attempt to loop.close() while the loop thread is still running (after a timed join), which can raise RuntimeError and cause flaky CI failures.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Review details
Suppressed comments (2)
sdk/python/tests/test_result_cache_deadlock.py:134
- Same issue as the earlier finally block:
loop1.close()is called even if the loop thread is still alive afterjoin(timeout=2), which can raiseRuntimeErrorand make this test flaky. Only close the loop once the thread has actually stopped.
time.sleep(0.2) # let the loop process scheduled cross-loop cancels
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
if not loop1.is_closed():
loop1.close()
sdk/python/tests/test_result_cache_deadlock.py:62
- The test cleanup unconditionally closes
loop1even if the loop thread didn’t stop within the join timeout. Closing a running event loop can raiseRuntimeError: Cannot close a running event loop, which makes this regression test flaky and can obscure the real failure. Guard theloop.close()call onnot t.is_alive()(similar to the helper used intest_async_lifecycle_deadlock.py).
This issue also appears on line 130 of the same file.
time.sleep(0.2) # let the loop process scheduled cross-loop cancels
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
if not loop1.is_closed():
loop1.close()
- Files reviewed: 6/6 changed files
- Comments generated: 0 new
- Review effort level: Lite
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
There was a problem hiding this comment.
🟡 Changes recommended
Cross-loop AsyncExecutionManager.stop() can return while leaving active executions running (and capacity unreleased), which can violate expected stop semantics and cause lingering work.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
This review doesn't count toward merge requirements. Sign up for the private preview to control whether Copilot approvals count.
Review details
Suppressed comments (5)
sdk/python/tests/test_result_cache_deadlock.py:134
- Same teardown concern as above: loop1 is closed without checking whether the thread actually stopped. This can produce sporadic RuntimeError/“pending task destroyed” warnings depending on timing.
Assert the thread stopped before closing loop1 to keep the test suite stable.
time.sleep(0.2) # let the loop process scheduled cross-loop cancels
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
if not loop1.is_closed():
loop1.close()
sdk/python/agentfield/async_execution_manager.py:378
- In the cross-loop stop() case, active executions are intentionally left running because _execution_lock can’t be acquired from a foreign loop. This changes the semantics of stop(): callers may assume it cancels in-flight executions and releases capacity, but on cross-loop teardown the manager may keep executing work indefinitely on the old loop.
Consider scheduling a best-effort cancellation coroutine onto the owning loop (without awaiting it) so cross-loop stop() still triggers execution cancellation and capacity release, while preserving the “never await foreign loop primitives” guarantee.
# Cancel all active executions. The _execution_lock is an asyncio.Lock
# bound to the owning loop — taking it from a foreign loop would raise
# "got Future attached to a different loop". On a cross-loop stop we
# skip this section: the executions will be cancelled when the owning
# loop tears down or the manager is restarted (#623 review feedback).
sdk/python/agentfield/http_connection_manager.py:254
- _close_cross_loop mutates _closed/_session/_connector and drops task refs without coordinating with the owning loop’s async lock. If close() is invoked cross-loop while start() is mid-flight (holding the lock), this can leak a newly-created session/connector and leave the manager in an inconsistent “closed but started” state.
A safer pattern is to schedule an owning-loop coroutine that acquires self._lock and performs teardown atomically (still without awaiting it from the foreign loop), so state transitions remain serialized.
if self._closed:
return
self._closed = True
cancel_task_cross_loop(self._health_check_task, owning_loop)
sdk/python/tests/test_async_lifecycle_deadlock.py:55
- _shutdown_loop silently returns if the loop thread doesn’t stop within the timeout. That can leave a daemon thread + running event loop alive for the rest of the test session, making failures flaky/non-local (later tests can be impacted) and masking cleanup problems.
It’s better to assert the thread stopped, so the test fails deterministically instead of leaking background loops.
thread.join(timeout=2)
if thread.is_alive():
# Thread didn't stop in time — don't close the loop while it's
# still running (would raise RuntimeError and obscure the real failure).
return
sdk/python/tests/test_result_cache_deadlock.py:62
- Test teardown closes loop1 even if the loop thread hasn’t actually stopped yet. If the thread is still running, closing the loop can raise or leave tasks pending, making failures flaky and harder to diagnose.
Assert that the thread has stopped before closing the loop (or fail), so cleanup is deterministic.
This issue also appears on line 130 of the same file.
time.sleep(0.2) # let the loop process scheduled cross-loop cancels
loop1.call_soon_threadsafe(loop1.stop)
t.join(timeout=2)
if not loop1.is_closed():
loop1.close()
- Files reviewed: 6/6 changed files
- Comments generated: 0 new
- Review effort level: Lite
We're testing this review assessment. Please use 👍 or 👎 to tell us if it's correct.
|
@santoshkumarradha @AbirAbbas ready for review when you get a chance. CI is green and all feedback has been addressed. |
santoshkumarradha
left a comment
santoshkumarradha
left a comment
There was a problem hiding this comment.
Checked the latest revision and I agree the original cross-loop crash is fixed, but I’m not ready to queue this one yet. Two semantics still feel too loose for a teardown path that callers will read as definitive stop behavior: on a foreign-loop stop, AsyncExecutionManager.stop() still leaves active executions running until the old loop happens to tear down, and ConnectionManager.close() still mutates shared session/connector state without serializing that transition on the owning loop’s lock. If you tighten those two paths so cross-loop stop/close still schedules the real teardown work on the owning loop, I’m happy to take another pass quickly.
Summary
ResultCachemixed athreading.RLockwith loop-bound asyncio primitives (anasyncio.Eventfor shutdown and anasyncio.Taskfor the cleanup loop). Whenstart()andstop()ran on different event loops which happens when the client's sync and async execution paths are mixed (#620)stop()raisedgot Future attached to a different loopand could wedge the process. This makes the cache lifecycle loop-aware so it never awaits a task from the wrong loop.Type of change
Test plan
cd sdk/python && python -m pytest tests/test_result_cache_deadlock.py -vcd sdk/python && python -m pytest tests/test_result_cache.py tests/test_result_cache_bigfiles_coverage.pyRuntimeError: got Future attached to a different loop) and confirmed cross-loopstop()now completes cleanly with no lingering pending-task warnings.Test coverage
coverage-baseline.jsonin this PR only if the removal caused a legitimate regression and I called it out in the summary above.result_cache.pycoverage went from 88% to 95%; the sdk-python aggregate stays at 94% (baseline 93.73%). Newtests/test_result_cache_deadlock.pycovers cross-loop stop, idempotent/rebinding start, concurrent sync access during cleanup, stop-without-start, and the disabled-cache no-op path.Checklist
Related issues / PRs
Fixes #623
Related to #620 (mixing threads and async code)