diff --git a/openhands-agent-server/openhands/agent_server/conversation_service.py b/openhands-agent-server/openhands/agent_server/conversation_service.py index f641d7e3dd..6327258fa5 100644 --- a/openhands-agent-server/openhands/agent_server/conversation_service.py +++ b/openhands-agent-server/openhands/agent_server/conversation_service.py @@ -1029,6 +1029,20 @@ def _prepare_persisted_runtime(self, stored: StoredConversation) -> None: async def _get_or_load_event_service( self, conversation_id: UUID ) -> EventService | None: + # Fast path: check the cache *without* acquiring the lifecycle lock. + # This ensures that already-loaded conversations can be retrieved even + # when the lock is held by a slow ``asyncio.to_thread`` call for a + # different conversation. The dict lookup is safe because + # ``_event_services`` is only mutated under the lock (insert / delete); + # a fast-path miss falls through to the locked slow path, which + # re-checks the cache and handles any state changes correctly. + event_services = self._event_services + if event_services is not None: + cached = event_services.get(conversation_id) + if cached is not None and cached.is_open(): + cached.touch() + return cached + async with self._lifecycle_lock: return await self._get_or_load_event_service_locked(conversation_id) diff --git a/tests/agent_server/test_event_service_thread_pool_exhaustion.py b/tests/agent_server/test_event_service_thread_pool_exhaustion.py new file mode 100644 index 0000000000..c0febbfb85 --- /dev/null +++ b/tests/agent_server/test_event_service_thread_pool_exhaustion.py @@ -0,0 +1,144 @@ +"""Test: thread-pool exhaustion must not deadlock the lifecycle lock. + +``_get_or_load_event_service`` acquires ``lifecycle_lock`` and then calls +``asyncio.to_thread(_prepare_persisted_runtime)`` inside the lock. If the +default thread pool is exhausted, the ``to_thread`` call queues indefinitely +while still holding the lock. Every subsequent ``get_event_service`` call — +including the WebSocket event-stream path and the REST ``/events/search`` +endpoint — blocks waiting for the lock, making the entire agent-server appear +wedged even though simple endpoints (``/ready``, ``/api/settings``) still +respond. + +This test asserts the **correct** behavior: + +* Loading an already-cached conversation (no ``to_thread`` needed) must + succeed immediately even when the thread pool is full, because the + lifecycle lock must not be held while waiting for a thread. + +The test currently **fails** on unpatched code, demonstrating the bug. +""" + +import asyncio +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from uuid import UUID + +import pytest + +from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.event_service import EventService +from openhands.agent_server.models import StartConversationRequest +from openhands.sdk import LLM, Agent +from openhands.sdk.security.confirmation_policy import NeverConfirm +from openhands.sdk.workspace import LocalWorkspace + + +async def _create_persisted_conversation( + conversations_dir: Path, workspace_dir: Path +) -> UUID: + """Create a conversation on disk so it can be loaded later.""" + workspace_dir.mkdir(parents=True, exist_ok=True) + request = StartConversationRequest( + agent=Agent(llm=LLM(model="gpt-4o", usage_id="test-llm"), tools=[]), + workspace=LocalWorkspace(working_dir=str(workspace_dir)), + confirmation_policy=NeverConfirm(), + ) + async with ConversationService(conversations_dir=conversations_dir) as service: + info, _ = await service.start_conversation(request) + return info.id + + +@pytest.mark.asyncio +async def test_thread_pool_exhaustion_does_not_block_cached_conversation( + tmp_path, +): + """Loading an already-cached conversation must not deadlock when the + thread pool is exhausted. + + The lifecycle lock is held while ``asyncio.to_thread`` waits for a free + thread. If the pool is full, the lock is held indefinitely, blocking + every subsequent ``get_event_service`` call — even for conversations that + are already in the cache and need no thread work at all. + + This is the root cause of the "events don't load" production incident. + """ + conversations_dir = tmp_path / "conversations" + workspace_dir = tmp_path / "workspace" + + # 1. Create two persisted conversations on disk. + conv_a = await _create_persisted_conversation( + conversations_dir, workspace_dir / "a" + ) + conv_b = await _create_persisted_conversation( + conversations_dir, workspace_dir / "b" + ) + + # 2. Replace the event-loop's default executor with a 1-worker pool. + loop = asyncio.get_running_loop() + tiny_pool = ThreadPoolExecutor(max_workers=1) + loop.set_default_executor(tiny_pool) + + service = ConversationService(conversations_dir=conversations_dir) + await service.__aenter__() + block_event: threading.Event | None = None + block_task: asyncio.Task[None] | None = None + stuck_task: asyncio.Task[EventService | None] | None = None + try: + # Pre-load both conversations into the cache. + es_a = await service.get_event_service(conv_a) + assert es_a is not None + es_b = await service.get_event_service(conv_b) + assert es_b is not None + + # 3. Saturate the 1-thread pool with a blocking call that never + # returns on its own. + block_event = threading.Event() + + async def _block_pool() -> None: + await loop.run_in_executor(None, block_event.wait) + + block_task = asyncio.create_task(_block_pool()) + await asyncio.sleep(0.3) + + # 4. Evict conv_a from the cache so the next load needs + # asyncio.to_thread(_prepare_persisted_runtime) — this call will + # hang waiting for a thread. + if service._event_services is not None: + service._event_services.pop(conv_a, None) + + # Kick off the stuck load in the background. Give it time to + # acquire the lifecycle lock and enter asyncio.to_thread before we + # try the cached load below. + stuck_task = asyncio.create_task(service.get_event_service(conv_a)) + await asyncio.sleep(0.3) + + # 5. Loading conv_b — which IS in the cache and needs no thread work — + # must succeed immediately. The lifecycle lock must not be held + # while waiting for a thread. + es_b2 = await asyncio.wait_for(service.get_event_service(conv_b), timeout=3.0) + assert es_b2 is not None + assert es_b2 is es_b + + # Clean up the stuck task. + block_event.set() + assert stuck_task is not None + try: + await asyncio.wait_for(stuck_task, timeout=10.0) + except (TimeoutError, asyncio.CancelledError): + stuck_task.cancel() + finally: + if block_event is not None: + block_event.set() + if block_task is not None: + try: + await asyncio.wait_for(block_task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError): + block_task.cancel() + if stuck_task is not None: + try: + await asyncio.wait_for(stuck_task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError): + stuck_task.cancel() + await service.__aexit__(None, None, None) + tiny_pool.shutdown(wait=False)