Remap loop-closed RuntimeError in _run_sync to OperationalError

``asyncio.run_coroutine_threadsafe`` raises bare ``RuntimeError("Event
loop is closed")`` when the loop is closed between ``_ensure_loop()``
returning and the schedule call landing — the canonical race shape
under SA's ``do_terminate`` from a finalizer thread, manual
``loop.close()`` from a test fixture, and SIGTERM-with-budget shutdown.

Two consequences before this fix:

- The bare RuntimeError escapes the PEP 249 ``Error`` hierarchy. SA's
  ``is_disconnect`` is gated on ``DatabaseError``; the connection is
  not invalidated correctly and the pool retains it.
- The unscheduled coroutine emits ``RuntimeWarning("coroutine was
  never awaited")`` at GC — a warning whose traceback does not point
  at dqlite, sending operators chasing the wrong layer.

The fix wraps the schedule call in ``try/except RuntimeError`` that
calls ``coro.close()`` (suppresses the warning unconditionally on
every RuntimeError arm) and remaps the closed-loop substring to
``OperationalError`` chained from the original RuntimeError. Other
RuntimeErrors (notably "Non-thread-safe operation invoked on an event
loop other than the current one" — a programmer-bug shape) propagate
unchanged so they are not silently classified as database connection
failures.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/connection.py

@@ -987,7 +987,40 @@ def _run_sync[T](self, coro: Coroutine[Any, Any, T]) -> T:
                     "use from another thread)"
                 )
             loop = self._ensure_loop()
-            future = asyncio.run_coroutine_threadsafe(coro, loop)
+            try:
+                future = asyncio.run_coroutine_threadsafe(coro, loop)
+            except RuntimeError as e:
+                # ``asyncio.run_coroutine_threadsafe`` raises bare
+                # ``RuntimeError("Event loop is closed")`` when the
+                # loop is closed between ``_ensure_loop()`` returning
+                # and the schedule call landing — the canonical race
+                # is a sibling thread (``do_terminate`` from a
+                # finalizer thread, manual ``loop.close()``, SIGTERM-
+                # with-budget shutdown). Without this catch the bare
+                # RuntimeError escapes the PEP 249 ``Error`` hierarchy
+                # (SA's ``is_disconnect`` is gated on ``DatabaseError``
+                # so it cannot classify the failure correctly), AND
+                # the unscheduled coroutine emits
+                # ``RuntimeWarning("coroutine was never awaited")`` at
+                # GC — a warning whose traceback does not point at
+                # dqlite, sending operators chasing the wrong layer.
+                # Close the coroutine and remap to ``OperationalError``
+                # (a ``DatabaseError`` subclass) with the original
+                # RuntimeError chained for diagnostics. Narrow the
+                # remap to the closed-loop substring so unrelated
+                # RuntimeErrors ("Non-thread-safe operation invoked on
+                # an event loop other than the current one" — a
+                # programmer-bug shape) propagate as themselves
+                # rather than being silently classified as a database
+                # connection failure. Close the coroutine on every
+                # arm so neither path leaks the unawaited-coroutine
+                # warning.
+                coro.close()
+                if "Event loop is closed" not in str(e):
+                    raise
+                raise OperationalError(
+                    f"event loop closed before coroutine could be scheduled: {e}"
+                ) from e
             try:
                 # Future.result() provides a happens-before memory barrier,
                 # ensuring all writes by the event loop thread are visible here.

tests/test_run_sync_loop_closed_during_schedule.py

@@ -0,0 +1,143 @@
+"""Pin: ``_run_sync`` does not let a bare ``RuntimeError`` from
+``asyncio.run_coroutine_threadsafe`` escape the PEP 249 ``Error``
+hierarchy, and does not leak the unawaited coroutine.
+
+The race shape: ``_run_sync`` calls ``self._ensure_loop()`` to obtain
+a live loop reference, then ``asyncio.run_coroutine_threadsafe(coro,
+loop)``. A sibling thread closing the loop between those two calls
+(typical: ``do_terminate`` from a finalizer thread, SIGTERM-with-
+budget shutdown, manual ``loop.close()`` from a test fixture)
+triggers a bare ``RuntimeError`` from ``run_coroutine_threadsafe``.
+
+Bare RuntimeError consequences:
+
+1. Escapes the PEP 249 ``Error`` hierarchy. SA's ``is_disconnect`` is
+   gated on ``DatabaseError``; a bare RuntimeError here breaks
+   classification and the connection is not retried / invalidated
+   correctly.
+2. The coroutine's lifecycle is broken — it was never scheduled, and
+   nothing called ``coro.close()``. At GC, asyncio emits
+   ``RuntimeWarning("coroutine was never awaited")`` that does not
+   point at dqlite, sending operators on a wild goose chase.
+
+The fix wraps ``run_coroutine_threadsafe`` in ``try/except
+RuntimeError``, calls ``coro.close()`` to suppress the warning, and
+raises an ``OperationalError`` (a ``DatabaseError`` subclass that SA's
+``is_disconnect`` can classify) chained from the underlying
+RuntimeError so the diagnostic is preserved.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import warnings
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+import dqlitedbapi.exceptions as _dbapi_exc
+from dqlitedbapi.connection import Connection
+
+
+def _make_connection() -> Connection:
+    """Build a minimal sync Connection bypassing the constructor's
+    cluster machinery; only the state needed for ``_run_sync`` is
+    populated.
+    """
+    conn = Connection.__new__(Connection)
+    # Connection's __init__ would normally do this; we only need
+    # _op_lock and _timeout for _run_sync's lock-acquire path.
+    import threading
+
+    conn._op_lock = threading.Lock()
+    conn._timeout = 5.0
+    conn._closed_flag = [False]
+    conn._async_conn = None  # not needed for the schedule-time race
+    conn._creator_pid = 0
+    return conn
+
+
+async def _trivial_coro() -> int:
+    return 42
+
+
+def test_run_sync_raises_pep249_error_when_loop_closed_at_schedule() -> None:
+    """A loop closed between ``_ensure_loop`` returning and
+    ``run_coroutine_threadsafe`` running must surface as a PEP 249
+    ``OperationalError`` (a ``DatabaseError`` subclass), NOT a bare
+    ``RuntimeError``.
+    """
+    conn = _make_connection()
+    closed_loop = asyncio.new_event_loop()
+    closed_loop.close()  # simulate the race outcome
+
+    with patch.object(conn, "_ensure_loop", return_value=closed_loop):
+        with pytest.raises(_dbapi_exc.Error) as exc_info:
+            conn._run_sync(_trivial_coro())
+        # Specifically OperationalError so SA's is_disconnect can
+        # classify it, with the underlying RuntimeError chained on
+        # __cause__ for diagnostics.
+        assert isinstance(exc_info.value, _dbapi_exc.OperationalError), (
+            f"expected OperationalError, got {type(exc_info.value).__name__}"
+        )
+        assert isinstance(exc_info.value.__cause__, RuntimeError), (
+            f"original RuntimeError must be chained; got {type(exc_info.value.__cause__).__name__}"
+        )
+
+
+def test_run_sync_does_not_leak_unawaited_coroutine_warning() -> None:
+    """The fix must call ``coro.close()`` on the schedule-failure
+    path so asyncio's ``RuntimeWarning("coroutine was never awaited")``
+    does NOT fire at GC. Without ``coro.close()`` the warning surfaces
+    in caller code with no dqlite frame in the traceback, sending
+    operators chasing the wrong layer.
+    """
+    conn = _make_connection()
+    closed_loop = asyncio.new_event_loop()
+    closed_loop.close()
+
+    coro = _trivial_coro()
+    with patch.object(conn, "_ensure_loop", return_value=closed_loop):
+        with warnings.catch_warnings(record=True) as captured:
+            warnings.simplefilter("always")
+            with pytest.raises(_dbapi_exc.OperationalError):
+                conn._run_sync(coro)
+            # Force coroutine GC to surface the warning if it would
+            # have fired.
+            del coro
+            import gc
+
+            gc.collect()
+        unawaited = [
+            w
+            for w in captured
+            if issubclass(w.category, RuntimeWarning)
+            and "coroutine was never awaited" in str(w.message)
+        ]
+        assert not unawaited, (
+            f"unawaited-coroutine warning leaked from _run_sync's "
+            f"schedule-failure path: {[str(w.message) for w in unawaited]}"
+        )
+
+
+def test_run_sync_propagates_runtimeerror_message_in_cause() -> None:
+    """The OperationalError raised from the schedule-failure path
+    must carry the underlying RuntimeError as ``__cause__`` so
+    operators reading the traceback see the actual loop-closed signal,
+    not just the dqlite-level remap.
+    """
+    conn = _make_connection()
+    closed_loop = asyncio.new_event_loop()
+    closed_loop.close()
+
+    with (
+        patch.object(conn, "_ensure_loop", return_value=closed_loop),
+        pytest.raises(_dbapi_exc.OperationalError) as exc_info,
+    ):
+        conn._run_sync(_trivial_coro())
+    cause: Any = exc_info.value.__cause__
+    assert isinstance(cause, RuntimeError)
+    # asyncio's wording is "Event loop is closed" — pin a non-empty
+    # str(cause) so the diagnostic carries through.
+    assert str(cause), "RuntimeError cause must carry a non-empty message"