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

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

@@ -6,10 +6,12 @@
 import atexit
 import builtins
 from collections.abc import Iterator
+from concurrent.futures import TimeoutError as FuturesTimeoutError
 from dataclasses import dataclass
 from datetime import datetime, timezone
 import hashlib
 import logging
+import threading
 from typing import TYPE_CHECKING, Any
 
 from aiocache import Cache  # type: ignore[import-untyped]
@@ -24,6 +26,31 @@
 
 logger = logging.getLogger(__name__)
 
+# How long a sync wrapper waits for its coroutine when it had to hand the work
+# to a worker loop. These are cache reads and writes, so the bound exists to
+# turn a hang into an error rather than to police a plausible duration.
+_RUN_SYNC_TIMEOUT_SECONDS = 30
+# How long to wait for the worker loop to stop after the task is done or
+# cancelled. Reached only if the coroutine ignores cancellation.
+_RUN_SYNC_JOIN_TIMEOUT_SECONDS = 5
+
+
+async def _settling(coro: Any, done: threading.Event) -> Any:
+    """Run ``coro``, signalling ``done`` once it has fully unwound.
+
+    The concurrent future that :func:`asyncio.run_coroutine_threadsafe` returns
+    is not a reliable "the coroutine has stopped" signal after a cancellation:
+    it is completed when the cancellation is *chained* to it, while the task may
+    still be running its own ``finally`` blocks. Stopping the loop on that signal
+    destroys the task mid-unwind ("Task was destroyed but it is pending").
+    Setting the flag from inside the coroutine's own ``finally`` cannot be early
+    by construction.
+    """
+    try:
+        return await coro
+    finally:
+        done.set()
+
 
 @dataclass
 class CachedUpload:
@@ -401,16 +428,77 @@ async def aget_all_for_provider(self, provider: ProviderType) -> list[CachedUplo
 
     @staticmethod
     def _run_sync(coro: Any) -> Any:
-        """Run an async coroutine from sync context without blocking event loop."""
+        """Run an async coroutine from a synchronous caller.
+
+        When there is already a running loop in this thread, the coroutine
+        cannot be scheduled on it: this function blocks the very thread that
+        loop runs on, so the loop could never advance the coroutine and the
+        wait would time out. The coroutine is driven on a worker thread with
+        its own loop instead.
+
+        The loop is owned here rather than handed to ``asyncio.run`` on a
+        pooled thread, because the timeout has to be able to *stop* the work.
+        ``asyncio.run`` is opaque from the outside: the only handle on it is
+        the thread, and threads cannot be interrupted, so a coroutine that
+        overran the bound would keep running -- and, being a non-daemon pool
+        thread, hold up interpreter exit until it finished. Scheduling through
+        :func:`asyncio.run_coroutine_threadsafe` keeps a handle on the task
+        itself, so a timeout cancels it and the loop shuts down for real.
+        """
         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)
+        if loop is None or not loop.is_running():
+            return asyncio.run(coro)
+
+        worker_loop = asyncio.new_event_loop()
+        # Daemon so that a worker wedged in uncancellable work (a blocking
+        # call inside the coroutine) cannot keep the process alive; the
+        # cancellation below is what handles the ordinary case.
+        thread = threading.Thread(
+            target=worker_loop.run_forever,
+            name="crewai-files-upload-cache",
+            daemon=True,
+        )
+        thread.start()
+        unwound = threading.Event()
+        try:
+            future = asyncio.run_coroutine_threadsafe(
+                _settling(coro, unwound), worker_loop
+            )
+            try:
+                return future.result(timeout=_RUN_SYNC_TIMEOUT_SECONDS)
+            except FuturesTimeoutError:
+                # Cancelling the task, not just abandoning the thread: the
+                # coroutine touches the shared cache, so letting it run on
+                # after the caller gave up would mutate state nobody is
+                # waiting for.
+                future.cancel()
+                # ``cancel()`` only *requests* it -- the exception is thrown in
+                # on the worker loop's next pass. Stopping the loop right away
+                # would race that, leaving the coroutine suspended with its
+                # cleanup never run. The bound keeps a coroutine that swallows
+                # cancellation from turning this into a second hang.
+                unwound.wait(timeout=_RUN_SYNC_JOIN_TIMEOUT_SECONDS)
+                raise
+        finally:
+            worker_loop.call_soon_threadsafe(worker_loop.stop)
+            thread.join(timeout=_RUN_SYNC_JOIN_TIMEOUT_SECONDS)
+            if thread.is_alive():
+                # Closing a loop whose thread is still in ``run_forever``
+                # would raise, and would leave the loop's internals being
+                # torn down underneath it. The join timeout is only reached
+                # when the coroutine ignores cancellation, so the loop is
+                # leaked deliberately in preference to that.
+                logger.warning(
+                    "Upload cache worker did not stop within "
+                    f"{_RUN_SYNC_JOIN_TIMEOUT_SECONDS}s; its event loop is "
+                    "left open because closing a running loop is unsafe."
+                )
+            else:
+                worker_loop.close()
 
     def get(self, file: FileInput, provider: ProviderType) -> CachedUpload | None:
         """Sync wrapper for aget."""

lib/crewai-files/tests/test_upload_cache.py

@@ -1,9 +1,19 @@
 """Tests for upload cache."""
 
+import asyncio
+from concurrent.futures import TimeoutError as FuturesTimeoutError
 from datetime import datetime, timedelta, timezone
+import threading
+import time
 
 from crewai_files import FileBytes, ImageFile
-from crewai_files.cache.upload_cache import CachedUpload, UploadCache
+import crewai_files.cache.upload_cache as upload_cache_mod
+from crewai_files.cache.upload_cache import (
+    CachedUpload,
+    UploadCache,
+    _compute_file_hash,
+)
+import pytest
 
 
 # Minimal valid PNG
@@ -207,3 +217,192 @@ def test_get_all_for_provider(self):
 
         assert len(gemini_uploads) == 2
         assert len(anthropic_uploads) == 1
+
+
+class TestRunSyncFromInsideAnEventLoop:
+    """The sync wrappers must work when a loop is already running.
+
+    ``UploadCache`` exposes ten synchronous wrappers that all funnel through
+    ``_run_sync``. Its in-loop branch used to schedule the coroutine on the
+    caller's own running loop and then block waiting for it, which the loop
+    could never satisfy because ``_run_sync`` occupies its thread. Every such
+    call stalled for the full 30s timeout and then raised ``TimeoutError``.
+    """
+
+    @staticmethod
+    def _make_file(suffix: bytes = b"") -> ImageFile:
+        return ImageFile(
+            source=FileBytes(data=MINIMAL_PNG + suffix, filename="test.png")
+        )
+
+    def test_set_and_get_outside_event_loop(self):
+        """Baseline: with no running loop the wrappers already worked."""
+        cache = UploadCache()
+        file = self._make_file()
+
+        cache.set(file=file, provider="gemini", file_id="file-123")
+
+        cached = cache.get(file=file, provider="gemini")
+        assert cached is not None
+        assert cached.file_id == "file-123"
+
+    @pytest.mark.asyncio
+    async def test_set_and_get_inside_running_event_loop(self):
+        """The regression: sync wrappers called from async code.
+
+        A stalled call would fail this by timing out rather than by
+        returning a wrong value, so the assertion is preceded by a wall
+        clock bound: the whole exchange must finish well inside the 30s
+        timeout the old code always spent.
+        """
+        cache = UploadCache()
+        file = self._make_file()
+
+        started = time.monotonic()
+        cache.set(file=file, provider="gemini", file_id="file-123")
+        cached = cache.get(file=file, provider="gemini")
+        elapsed = time.monotonic() - started
+
+        assert cached is not None
+        assert cached.file_id == "file-123"
+        assert elapsed < 10, (
+            f"took {elapsed:.1f}s -- the coroutine is being scheduled on the "
+            "caller's own loop, which _run_sync blocks while waiting"
+        )
+
+    @pytest.mark.asyncio
+    async def test_sync_wrappers_do_not_use_the_callers_loop(self):
+        """The coroutine must not be driven by the loop ``_run_sync`` blocks.
+
+        Scheduling it there is the cause of the stall, so the property is
+        asserted directly rather than inferred from the absence of a
+        timeout.
+        """
+        caller_loop = asyncio.get_running_loop()
+        seen: list[asyncio.AbstractEventLoop] = []
+
+        async def probe() -> str:
+            seen.append(asyncio.get_running_loop())
+            return "done"
+
+        assert UploadCache._run_sync(probe()) == "done"
+        assert seen and seen[0] is not caller_loop
+
+    @pytest.mark.asyncio
+    async def test_remaining_sync_wrappers_inside_running_event_loop(self):
+        """The other wrappers share ``_run_sync``, so they are covered too."""
+        cache = UploadCache()
+        file = self._make_file()
+        other = self._make_file(suffix=b"x")
+
+        cache.set(file=file, provider="gemini", file_id="file-1")
+        # set_by_hash rather than a second set(): it is a wrapper in its own
+        # right, and calling set twice would leave it uncovered.
+        cache.set_by_hash(
+            file_hash=_compute_file_hash(other),
+            content_type=other.content_type,
+            provider="gemini",
+            file_id="file-2",
+        )
+
+        assert len(cache.get_all_for_provider("gemini")) == 2
+        assert cache.get_by_hash(_compute_file_hash(file), "gemini") is not None
+        assert cache.remove(file=file, provider="gemini") is True
+        assert cache.remove_by_file_id("file-2", "gemini") is True
+        assert cache.clear_expired() == 0
+        assert cache.clear() == 0
+
+    @pytest.mark.asyncio
+    async def test_exceptions_propagate_from_inside_running_event_loop(self):
+        """A failure inside the coroutine must reach the sync caller."""
+
+        async def boom() -> None:
+            raise RuntimeError("coroutine failed")
+
+        with pytest.raises(RuntimeError, match="coroutine failed"):
+            UploadCache._run_sync(boom())
+
+    @pytest.mark.asyncio
+    async def test_timeout_cancels_the_coroutine_instead_of_abandoning_it(
+        self, monkeypatch
+    ):
+        """A coroutine that overruns the bound must be stopped, not left running.
+
+        The timeout unblocks the caller either way, so waiting on the caller
+        proves nothing: the question is what happens to the work afterwards.
+        These coroutines mutate the shared cache, so one that runs on past the
+        point where its caller gave up writes to state nobody is waiting for.
+        """
+        # Long enough that the coroutine is certainly running when the bound
+        # expires. With a very short one the future can be cancelled before the
+        # worker loop has picked it up -- also a correct outcome, but it would
+        # leave this test passing without a running task ever being interrupted,
+        # which is the thing under test. ``started`` pins that premise.
+        monkeypatch.setattr(upload_cache_mod, "_RUN_SYNC_TIMEOUT_SECONDS", 2.0)
+        outcome: list[str] = []
+        started = threading.Event()
+        finished = threading.Event()
+
+        async def overruns() -> None:
+            started.set()
+            try:
+                await asyncio.sleep(30)
+                outcome.append("completed")
+            except asyncio.CancelledError:
+                outcome.append("cancelled")
+                raise
+            finally:
+                finished.set()
+
+        with pytest.raises(FuturesTimeoutError):
+            UploadCache._run_sync(overruns())
+
+        assert started.is_set(), "coroutine never ran; nothing was interrupted"
+        assert finished.wait(timeout=5), "coroutine neither finished nor was stopped"
+        assert outcome == ["cancelled"]
+
+    @pytest.mark.asyncio
+    async def test_timed_out_worker_is_cleaned_up_before_returning(self, monkeypatch):
+        """The worker thread and its loop must be gone by the time we return.
+
+        Otherwise the leak is only moved: a non-daemon pool thread still
+        running at exit delays interpreter shutdown until its coroutine ends,
+        so a 30s upload check would add 30s to the exit of a process that had
+        already given up on it.
+        """
+        monkeypatch.setattr(upload_cache_mod, "_RUN_SYNC_TIMEOUT_SECONDS", 2.0)
+        before = set(threading.enumerate())
+        started = threading.Event()
+
+        async def overruns() -> None:
+            started.set()
+            await asyncio.sleep(30)
+
+        with pytest.raises(FuturesTimeoutError):
+            UploadCache._run_sync(overruns())
+
+        assert started.is_set(), "coroutine never ran; nothing was left behind"
+        # Compared by identity over every thread rather than by a name prefix:
+        # a name filter would only ever catch the threads this implementation
+        # happens to create, so an implementation that leaked a differently
+        # named one would pass.
+        leaked = [t for t in threading.enumerate() if t not in before and t.is_alive()]
+        assert leaked == [], f"worker thread outlived the call: {leaked}"
+
+    @pytest.mark.asyncio
+    async def test_successful_call_leaves_no_worker_thread_behind(self):
+        """The ordinary path must not accumulate a thread per wrapper call.
+
+        Every sync wrapper called from async code goes through here, so a
+        thread that outlives its call would leak once per cache lookup rather
+        than once per timeout.
+        """
+        before = set(threading.enumerate())
+        cache = UploadCache()
+        file = self._make_file()
+
+        cache.set(file=file, provider="gemini", file_id="file-123")
+        assert cache.get(file=file, provider="gemini") is not None
+
+        leaked = [t for t in threading.enumerate() if t not in before and t.is_alive()]
+        assert leaked == [], f"worker threads outlived their calls: {leaked}"