fix: stop UploadCache sync wrappers deadlocking inside a running event loop by LHMQ878 · Pull Request #6743 · crewAIInc/crewAI

LHMQ878 · 2026-07-30T20:48:08Z

Description

UploadCache._run_sync's branch for being called while an event loop is already running could not succeed. It scheduled the coroutine on the caller's own running loop with asyncio.run_coroutine_threadsafe, then blocked that same thread on .result(timeout=30). The blocked thread is the thread running the target loop, so the loop can never advance the coroutine it was just handed. Every such call stalled for the full 30 seconds and raised TimeoutError.

The docstring claimed the opposite of the behaviour: "Run an async coroutine from sync context without blocking event loop."

Fixes #6742

Root cause

lib/crewai-files/src/crewai_files/cache/upload_cache.py:402-413 on main (ebe0082):

@staticmethod
def _run_sync(coro: Any) -> Any:
    """Run an async coroutine from sync context without blocking event loop."""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop is not None and loop.is_running():
        future = asyncio.run_coroutine_threadsafe(coro, loop)
        return future.result(timeout=30)
    return asyncio.run(coro)

asyncio.run_coroutine_threadsafe is for submitting work to a loop running in a different thread. loop here came from asyncio.get_running_loop(), so it is this thread's loop — the precondition is inverted.

All ten synchronous wrappers funnel through _run_sync (get, get_by_hash, set, set_by_hash, remove, remove_by_file_id, clear_expired, clear, get_all_for_provider), and UploadCache / cleanup_expired_files are public exports of crewai_files. So any of them called from a FastAPI handler, a notebook cell, or an async crew callback hung for 30s and then raised.

The change

The coroutine runs on a worker thread with its own loop, so the thread that blocks is not the thread that has to make progress:

if loop is not None and loop.is_running():
    executor = ThreadPoolExecutor(max_workers=1)
    try:
        return executor.submit(asyncio.run, coro).result(timeout=30)
    finally:
        # ``shutdown(wait=True)`` -- what the context-manager form
        # does -- would block until the worker finishes, so a
        # coroutine that overran the timeout would still hang the
        # caller and make the bound meaningless.
        executor.shutdown(wait=False)

The wait=False shutdown is deliberate rather than stylistic. Writing this as with ThreadPoolExecutor(...) as executor: reads better but reintroduces the bug it is fixing on the timeout path: __exit__ calls shutdown(wait=True), which blocks until the worker finishes, so a coroutine that overran the 30s bound would still hang the caller indefinitely — the bound would only appear to be enforced.

Deliberately not changed:

The else branch is untouched. With no running loop, asyncio.run(coro) was already correct.
The 30s timeout value is kept, so this PR changes only whether the call can complete, not how long a caller may wait.
No persistent background loop. A per-call worker keeps the thread's lifetime tied to the call and adds no global state; these wrappers are coarse-grained and already paid for a fresh loop per call on the else path.

Loop affinity was checked before choosing this shape: the pre-existing else branch already ran every coroutine under a fresh asyncio.run, so the aiocache backend was already required to be loop-agnostic, and the default memory backend is Cache(serializer=PickleSerializer(), namespace=namespace) — a plain dict.

Tests

lib/crewai-files/tests/test_upload_cache.py had no async coverage at all, which is why a branch that always timed out went unnoticed. Five tests added in TestRunSyncFromInsideAnEventLoop:

test	asserts
`test_set_and_get_outside_event_loop`	baseline: with no running loop the wrappers already worked
`test_set_and_get_inside_running_event_loop`	the regression: `set`/`get` from async code complete, bounded by a wall clock assertion so a stall fails with a diagnosis rather than just a timeout
`test_sync_wrappers_do_not_use_the_callers_loop`	the coroutine runs on a loop that is not the caller's — the property whose absence causes the stall, checked directly rather than inferred from elapsed time
`test_remaining_sync_wrappers_inside_running_event_loop`	the other wrappers (`get_all_for_provider`, `get_by_hash`, `remove`, `remove_by_file_id`, `clear_expired`, `clear`) share `_run_sync` and are covered too
`test_exceptions_propagate_from_inside_running_event_loop`	a failure inside the coroutine surfaces to the sync caller instead of being swallowed by the bridge

Control experiment

Reverting only upload_cache.py and keeping the tests:

4 failed, 15 passed   (122.68s)

The 4 failures are exactly the four in-loop tests, and each fails at concurrent/futures/_base.py:458: TimeoutError — the reported defect, not an incidental assertion. The sync baseline and all 14 pre-existing tests pass unchanged.

Worth flagging for review: that control run takes ~2 minutes because the unfixed code stalls for its 30s timeout on each affected test rather than failing fast. The wall clock bound in test_set_and_get_inside_running_event_loop (elapsed < 10) exists so the headline regression reports the cause rather than only timing out.

Verification

suite	result
`lib/crewai-files/tests/test_upload_cache.py`	19 passed (14 pre-existing + 5 new)
`ruff check` / `ruff format --check` on both changed files	clean
`mypy lib/crewai-files/src/crewai_files/cache/upload_cache.py`	clean

On Windows these tests need --allowed-hosts=127.0.0.1,::1 (the ProactorEventLoop self-pipe uses socket.socketpair(), which pytest-recording blocks) and a working libmagic; both are pre-existing environment concerns unrelated to this change.

UploadCache._run_sync had a branch for being called while a loop is already running -- the case its docstring advertised as "without blocking event loop" -- that could not succeed. It scheduled the coroutine on that same loop with asyncio.run_coroutine_threadsafe and then blocked on .result(timeout=30). The blocked thread *is* the thread running the target loop, so the loop can never advance the coroutine it was just handed. Every such call stalled for the full 30 seconds and raised TimeoutError. All ten synchronous wrappers funnel through _run_sync -- get, get_by_hash, set, set_by_hash, remove, remove_by_file_id, clear_expired, clear, get_all_for_provider -- so calling any of them from async code (a FastAPI handler, a notebook cell, an async crew callback) hung. UploadCache and cleanup_expired_files are public exports of crewai_files. The coroutine now runs on a worker thread with its own loop, so the thread that blocks is not the thread making progress. The executor is shut down with wait=False rather than via the context-manager form, whose shutdown(wait=True) would block until the worker finished and so make the 30s bound meaningless on the very timeout it guards. Loop affinity is not a concern: the pre-existing else branch already ran each coroutine under a fresh asyncio.run, so the aiocache backend was already required to be loop-agnostic. The memory backend is a plain dict. Reverting only upload_cache.py fails 4 of the 5 new tests, each with concurrent.futures TimeoutError -- the reported defect, not an incidental assertion. 19 passed with the fix (14 pre-existing + 5 new).

coderabbitai · 2026-07-30T20:54:15Z

📝 Walkthrough

Walkthrough

Changes

UploadCache synchronous execution

Layer / File(s)	Summary
Worker-thread coroutine execution `lib/crewai-files/src/crewai_files/cache/upload_cache.py`	`_run_sync` runs coroutines on a dedicated worker event loop when called inside a running loop, cancels timed-out work, waits for settlement, and bounds loop shutdown.
Inside-loop wrapper coverage `lib/crewai-files/tests/test_upload_cache.py`	Tests cover synchronous wrappers, active-loop calls, exception propagation, timeout cancellation, and worker cleanup.

Sequence Diagram(s)

sequenceDiagram
  participant Caller as Calling thread
  participant Worker as Worker event-loop thread
  participant Coroutine as UploadCache coroutine
  Caller->>Worker: submit coroutine
  Worker->>Coroutine: execute
  Coroutine-->>Caller: return result or exception
  Caller->>Worker: cancel after timeout
  Worker-->>Caller: signal settlement
  Caller->>Worker: stop and join loop

Suggested reviewers: greysonlalonde

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly matches the main fix: preventing UploadCache sync wrappers from deadlocking in a running event loop.
Description check	✅ Passed	The description directly explains the deadlock root cause and the worker-thread fix, matching the changes.
Linked Issues check	✅ Passed	The code and tests address `#6742` by running _run_sync work off the caller loop, covering the listed sync wrappers and timeout cleanup.
Out of Scope Changes check	✅ Passed	The added tests and cleanup logic stay within the issue scope; no unrelated functional changes are indicated.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

🧪 Generate unit tests (beta)

Create PR with unit tests

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands.}

coderabbitai

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/crewai-files/src/crewai_files/cache/upload_cache.py`:
- Around line 419-427: Update the executor-based coroutine path around
executor.submit and asyncio.run so a timeout cancels the running asyncio task
instead of abandoning the worker via executor.shutdown(wait=False). Use a
lifecycle-managed worker that bounds execution, propagates cancellation through
nested coroutines, and completes cleanup before returning; update the associated
tests to verify timed-out tasks are cancelled and cleaned up.

In `@lib/crewai-files/tests/test_upload_cache.py`:
- Around line 295-303: Update the running-loop regression test setup to call
cache.set_by_hash for the second cache entry instead of cache.set, while
preserving the existing provider, file identity, and file ID assertions.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

Push a commit to this branch (recommended)
Create a new PR with the fixes

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 5444d5d5-7ec6-4e5a-9b9f-acb3b420be19

📥 Commits

Reviewing files that changed from the base of the PR and between ebe0082 and b262f4e.

📒 Files selected for processing (2)

lib/crewai-files/src/crewai_files/cache/upload_cache.py
lib/crewai-files/tests/test_upload_cache.py

CodeRabbit's major finding on this PR, confirmed and worse than stated. `executor.shutdown(wait=False)` unblocks the caller but cannot stop the work: `asyncio.run` on a pooled thread is opaque from the outside, the only handle on it is the thread, and threads cannot be interrupted. So a coroutine that overran the 30s bound kept running -- and because pool threads are non-daemon, it also held up interpreter exit until it finished. Measured on the previous head: caller unblocked after 0.50s: TimeoutError 1.0s after the timeout, marks = [] <- still running about to exit; marks = [] interpreter exit took 1.50s, marks = ['COMPLETED AFTER TIMEOUT'] That last line is the part worth fixing rather than documenting: the coroutine did not merely outlive its caller, it ran to completion during interpreter teardown and wrote to the shared cache there. A 30s timeout would have added 30s to the exit of a process that had already given up. Now the loop is owned here and the coroutine scheduled through `asyncio.run_coroutine_threadsafe`, which keeps a handle on the task itself, so the timeout cancels it. Two details the fix turns on: - The concurrent future is not a "the coroutine has stopped" signal. It completes when the cancellation is chained to it, while the task may still be in its own `finally`; stopping the loop then destroys the task mid-unwind ("Task was destroyed but it is pending"). The flag is set from inside the coroutine's own `finally` instead, which cannot be early by construction. - The thread is a daemon and the loop is left open rather than closed if the join times out, since closing a loop still in `run_forever` raises. That path is only reachable when the coroutine ignores cancellation. Three tests: cancellation of a *running* task (with `started` pinning that premise -- a short bound can cancel the future before the loop picks it up, which would pass without anything being interrupted), no thread outliving a timed-out call, and none outliving a successful one. Thread leaks are compared by identity across all threads rather than by name prefix, so an implementation that leaks a differently-named thread cannot pass. Swapping the old mechanism back in while keeping the constants fails all three. Also addresses the minor finding: the wrapper-coverage test now calls `set_by_hash` rather than `set` twice, so that wrapper is exercised. Signed-off-by: LHMQ878 <LHMQ878@users.noreply.github.com>

LHMQ878 · 2026-07-30T21:18:22Z

Both findings were right. Fixed in cf0182f.

The major one: `shutdown(wait=False)` cannot stop the work

Confirmed, and the reason matters for the shape of the fix. asyncio.run on a pooled thread is opaque from the outside — the only handle on it is the thread, and threads cannot be interrupted — so there was nothing to cancel with. Measured on b262f4e:

caller unblocked after 0.50s: TimeoutError
1.0s after the timeout, marks = []                        <- still running
about to exit;              marks = []
interpreter exit took 1.50s, marks = ['COMPLETED AFTER TIMEOUT']

That last line is worse than the finding states and is what decided me. The coroutine did not merely outlive its caller: pool threads are non-daemon, so it held up interpreter exit and then ran to completion during teardown, writing to the shared cache there. A 30s timeout would have added 30s to the exit of a process that had already given up on the call.

After the fix, same probe:

timed out after 0.22s
settled? True  outcome = ['cancelled']
threads: ['MainThread']
interpreter exit took 0.00s

Two things the fix turns on

I went with run_coroutine_threadsafe on an owned loop rather than a lifecycle-managed pool, since keeping a handle on the task is the whole point. Two details cost me a debugging pass each and are worth stating:

1. The concurrent future is not a "the coroutine has stopped" signal. It completes when the cancellation is chained to it, while the task may still be running its own finally. Waiting on future.exception() and then stopping the loop still produced:

Task was destroyed but it is pending!
task: <Task cancelling ... wait_for=<Future cancelled>>

So the flag is now set from inside the coroutine's own finally (_settling), which cannot be early by construction.

2. The loop is left open, not closed, if the join times out. Closing a loop whose thread is still in run_forever raises, and would tear down its internals underneath the running task. That path is only reachable when the coroutine ignores cancellation; the thread is a daemon so it still cannot hold the process open.

Tests

test_timeout_cancels_the_coroutine_instead_of_abandoning_it
test_timed_out_worker_is_cleaned_up_before_returning
test_successful_call_leaves_no_worker_thread_behind

Two things I had to correct in my own tests, since both would have passed for the wrong reason:

With a 0.2s bound the future gets cancelled before the worker loop picks it up. That is a correct outcome, but it means no running task was ever interrupted — which is the property under test. The bound is now 2.0s and a started event pins the premise.
Filtering leaked threads by the name prefix this implementation happens to use would pass for any implementation that leaks a differently-named thread. Comparison is now by identity across threading.enumerate().

Discrimination: swapping the old ThreadPoolExecutor + shutdown(wait=False) mechanism back in while keeping the new constants fails all three (19 passed / 3 failed); with the fix, 22 passed. Whole package: 175 passed / 4 skipped. ruff check, ruff format --check and mypy clean.

The minor one

Taken as suggested — test_remaining_sync_wrappers_inside_running_event_loop now calls set_by_hash instead of set twice, so that wrapper is exercised directly rather than being listed in scope and left uncovered.

coderabbitai

🧹 Nitpick comments (2)

lib/crewai-files/src/crewai_files/cache/upload_cache.py (2)
486-501: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value

Drain async generators before closing the worker loop.

After run_forever returns, worker_loop.close() skips loop.shutdown_asyncgens(), so any async generator left suspended by a cancelled coroutine is finalized by GC later with an "asynchronous generator was garbage collected" warning. A short bounded drain keeps teardown clean.
♻️ Suggested teardown
             else:
+                worker_loop.run_until_complete(worker_loop.shutdown_asyncgens())
                 worker_loop.close()
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/crewai-files/src/crewai_files/cache/upload_cache.py` around lines 486 -
501, Update the worker-loop teardown in the finally block to run a short,
bounded async-generator shutdown via worker_loop.shutdown_asyncgens() after the
thread has stopped and before worker_loop.close(). Keep the existing timeout and
running-thread leak behavior unchanged, and ensure the drain does not block
teardown indefinitely.
456-465: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoff

Consider a shared, lazily created worker loop instead of one per call.

Every in-loop wrapper call spins up a fresh OS thread, selector, and event loop; for hot cache reads this is a lot of setup per lookup. A module-level lazily started worker loop (with the same cancel-on-timeout handle) would amortize it. Note this would require relaxing test_successful_call_leaves_no_worker_thread_behind / test_timed_out_worker_is_cleaned_up_before_returning, which currently assert no surviving thread, so it's a deliberate tradeoff rather than a defect.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/crewai-files/src/crewai_files/cache/upload_cache.py` around lines 456 -
465, Refactor the in-loop wrapper’s worker management to use a module-level,
lazily initialized event loop and worker thread instead of creating them per
call. Preserve the existing cancellation-on-timeout handle and update the
affected thread-cleanup tests to expect the shared worker’s lifetime rather than
asserting no surviving thread after each call.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@lib/crewai-files/src/crewai_files/cache/upload_cache.py`:
- Around line 486-501: Update the worker-loop teardown in the finally block to
run a short, bounded async-generator shutdown via
worker_loop.shutdown_asyncgens() after the thread has stopped and before
worker_loop.close(). Keep the existing timeout and running-thread leak behavior
unchanged, and ensure the drain does not block teardown indefinitely.
- Around line 456-465: Refactor the in-loop wrapper’s worker management to use a
module-level, lazily initialized event loop and worker thread instead of
creating them per call. Preserve the existing cancellation-on-timeout handle and
update the affected thread-cleanup tests to expect the shared worker’s lifetime
rather than asserting no surviving thread after each call.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 9fbf449b-ddd7-4099-a6a6-da3d19880dac

📥 Commits

Reviewing files that changed from the base of the PR and between b262f4e and cf0182f.

📒 Files selected for processing (2)

lib/crewai-files/src/crewai_files/cache/upload_cache.py
lib/crewai-files/tests/test_upload_cache.py

coderabbitai Bot reviewed Jul 30, 2026

View reviewed changes

Comment thread lib/crewai-files/src/crewai_files/cache/upload_cache.py Outdated

Comment thread lib/crewai-files/tests/test_upload_cache.py

coderabbitai Bot reviewed Jul 30, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

fix: stop UploadCache sync wrappers deadlocking inside a running event loop - #6743

fix: stop UploadCache sync wrappers deadlocking inside a running event loop#6743
LHMQ878 wants to merge 2 commits into
crewAIInc:mainfrom
LHMQ878:fix/upload-cache-run-sync-deadlock

LHMQ878 commented Jul 30, 2026

Uh oh!

coderabbitai Bot commented Jul 30, 2026 •

edited

Loading

Walkthrough

Changes

Sequence Diagram(s)

Uh oh!

coderabbitai Bot left a comment

Uh oh!

Uh oh!

Uh oh!

LHMQ878 commented Jul 30, 2026

Uh oh!

coderabbitai Bot left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

LHMQ878 commented Jul 30, 2026

Description

Root cause

The change

Tests

Control experiment

Verification

Uh oh!

coderabbitai Bot commented Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Walkthrough

Changes

Sequence Diagram(s)

Uh oh!

coderabbitai Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

LHMQ878 commented Jul 30, 2026

The major one: shutdown(wait=False) cannot stop the work

Two things the fix turns on

Tests

The minor one

Uh oh!

coderabbitai Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

coderabbitai Bot commented Jul 30, 2026 •

edited

Loading

The major one: `shutdown(wait=False)` cannot stop the work