diff --git a/lib/crewai-files/src/crewai_files/cache/upload_cache.py b/lib/crewai-files/src/crewai_files/cache/upload_cache.py index 48cebdfa14..8aab4ad4ef 100644 --- a/lib/crewai-files/src/crewai_files/cache/upload_cache.py +++ b/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.""" diff --git a/lib/crewai-files/tests/test_upload_cache.py b/lib/crewai-files/tests/test_upload_cache.py index f7d9f86a33..cd66f77bcb 100644 --- a/lib/crewai-files/tests/test_upload_cache.py +++ b/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}"