fix(portal): improve shutdown safety and add call timeout

- Only close event loop after thread has terminated to avoid undefined
  behavior when closing a running loop
- Add configurable timeout to call() method (default 300s) to prevent
  deadlocks during shutdown races
- Add test for timeout functionality

Author: cofin

sqlspec/utils/portal.py

@@ -113,6 +113,8 @@ def stop(self) -> None:
         """Stop the background thread and event loop.
 
         Gracefully shuts down the event loop and waits for thread to finish.
+        Only closes the loop after the thread has terminated to avoid
+        undefined behavior from closing a running loop.
         """
         if self._loop is None or self._loop_thread is None:
             logger.debug("Portal provider not running")
@@ -122,9 +124,10 @@ def stop(self) -> None:
         self._loop_thread.join(timeout=5)
 
         if self._loop_thread.is_alive():
-            logger.warning("Portal thread did not stop within 5 seconds")
+            logger.warning("Portal thread did not stop within 5 seconds, skipping loop.close()")
+        else:
+            self._loop.close()
 
-        self._loop.close()
         self._loop = None
         self._loop_thread = None
         self._ready_event.clear()
@@ -146,22 +149,25 @@ async def _async_caller(
         result: _R = await func(*args, **kwargs)
         return result
 
-    def call(self, func: "Callable[..., Coroutine[Any, Any, _R]]", *args: Any, **kwargs: Any) -> _R:
+    def call(
+        self, func: "Callable[..., Coroutine[Any, Any, _R]]", *args: Any, timeout: float = 300.0, **kwargs: Any
+    ) -> _R:
         """Call an async function from synchronous context.
 
         Executes the async function in the background event loop and blocks
-        until the result is available.
+        until the result is available or timeout is reached.
 
         Args:
             func: The async function to call.
             *args: Positional arguments to the function.
+            timeout: Maximum seconds to wait for result (default 300).
             **kwargs: Keyword arguments to the function.
 
         Returns:
             Result of the async function.
 
         Raises:
-            ImproperConfigurationError: If portal provider not started.
+            ImproperConfigurationError: If portal provider not started or timeout reached.
 
         """
         if self._loop is None or not self.is_running:
@@ -174,7 +180,11 @@ def call(self, func: "Callable[..., Coroutine[Any, Any, _R]]", *args: Any, **kwa
 
         self._loop.call_soon_threadsafe(self._process_request)
 
-        result, exception = local_result_queue.get()
+        try:
+            result, exception = local_result_queue.get(timeout=timeout)
+        except queue.Empty:
+            msg = f"Portal call timed out after {timeout} seconds"
+            raise ImproperConfigurationError(msg) from None
 
         if exception:
             raise exception
@@ -223,19 +233,22 @@ def __init__(self, provider: "PortalProvider") -> None:
         """
         self._provider = provider
 
-    def call(self, func: "Callable[..., Coroutine[Any, Any, _R]]", *args: Any, **kwargs: Any) -> _R:
+    def call(
+        self, func: "Callable[..., Coroutine[Any, Any, _R]]", *args: Any, timeout: float = 300.0, **kwargs: Any
+    ) -> _R:
         """Call an async function using the portal provider.
 
         Args:
             func: The async function to call.
             *args: Positional arguments to the function.
+            timeout: Maximum seconds to wait for result (default 300).
             **kwargs: Keyword arguments to the function.
 
         Returns:
             Result of the async function.
 
         """
-        return self._provider.call(func, *args, **kwargs)
+        return self._provider.call(func, *args, timeout=timeout, **kwargs)
 
 
 class PortalManager(metaclass=SingletonMeta):

tests/unit/test_utils/test_portal.py

@@ -381,3 +381,19 @@ def test_portal_manager_atexit_cleanup_noop_when_stopped() -> None:
     manager._atexit_cleanup()  # pyright: ignore[reportPrivateUsage]
 
     assert not manager.is_running
+
+
+def test_portal_call_timeout() -> None:
+    """PortalProvider.call raises error on timeout."""
+
+    async def slow_function() -> int:
+        await asyncio.sleep(10)
+        return 42
+
+    provider = PortalProvider()
+    provider.start()
+
+    with pytest.raises(ImproperConfigurationError, match="timed out after"):
+        provider.call(slow_function, timeout=0.1)
+
+    provider.stop()