Release the op-lock when a signal interrupts the owner-stamp in _run_sync

A KeyboardInterrupt or SystemExit delivered after the op-lock was acquired
but before the body try/finally was entered — landing on the
_op_lock_owner stamp — escaped with the lock held by no thread,
permanently wedging the connection: every later _run_sync then timed out
on the bounded acquire.

Place the acquire, the owner-stamp, and the body under a single
try/finally so the release always runs once the lock is held. The inner
try/except around acquire() is kept for the acquire-return-before-store
window (it best-effort releases with acquired still False, so the outer
finally does not double-release), and acquired is bound before the outer
try so the finally never references an unbound local.

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

Author: antoineleclair

src/dqlitedbapi/connection.py

@@ -1100,41 +1100,48 @@ def _run_sync[T](self, coro: Coroutine[Any, Any, T]) -> T:
         ``self._timeout`` so a same-thread signal-handler re-entry
         raises cleanly instead of deadlocking the non-reentrant lock.
         """
-        # acquire(timeout=...) is SIGINT-interruptible: a KI/SystemExit
-        # escapes BEFORE the try block, skipping its cleanup arm. If a
-        # prior op is in flight it owns _in_use, wedging the connection;
-        # schedule a defensive _invalidate. Gated on _in_use so a KI on
-        # a quiet acquire doesn't invalidate gratuitously.
+        # The acquire, the owner-stamp, and the body all sit under one
+        # try/finally so a SIGINT-delivered KeyboardInterrupt/SystemExit
+        # raised anywhere after the lock is held — including in the
+        # owner-stamp gap — cannot escape with the lock latched. `acquired`
+        # is bound first so the finally never NameErrors on the inner arm
+        # below (which re-raises before `acquired` is assigned).
+        acquired = False
         try:
-            acquired = self._op_lock.acquire(timeout=self._timeout)
-        except (KeyboardInterrupt, SystemExit):
-            # The lock may be held if KI landed in the gap between
-            # acquire returning True and the STORE_FAST; best-effort
-            # release (suppress the unlocked-lock RuntimeError).
-            with contextlib.suppress(RuntimeError):
-                self._op_lock_owner = None
-                self._op_lock.release()
-            # Close the never-scheduled coro to avoid a warning.
-            coro.close()
-            # Synchronously null _async_conn so a retry from the signal
-            # handler doesn't hit "another operation is in progress" on
-            # the stale in-use conn (the queued _invalidate only lands
-            # when the slow read yields). GIL-atomic STORE_ATTR; the
-            # loop coro keeps its own ref and reaps via _invalidate.
-            dying = self._async_conn
-            if dying is not None and self._loop is not None and dying._in_use:
-                self._async_conn = None
+            # acquire(timeout=...) is SIGINT-interruptible: a KI/SystemExit
+            # can land between acquire returning True and the STORE_FAST,
+            # leaving the lock held with `acquired` still False — so the
+            # inner arm best-effort releases it (the outer finally skips
+            # release when `acquired` is False). A prior op in flight owns
+            # _in_use, wedging the connection; schedule a defensive
+            # _invalidate, gated on _in_use so a KI on a quiet acquire
+            # doesn't invalidate gratuitously.
+            try:
+                acquired = self._op_lock.acquire(timeout=self._timeout)
+            except (KeyboardInterrupt, SystemExit):
                 with contextlib.suppress(RuntimeError):
-                    self._loop.call_soon_threadsafe(
-                        dying._invalidate,
-                        InterfaceError("operation interrupted during op-lock acquire"),
-                    )
-            raise
-        # Stamp owner before the try/finally so the close() bypass probe
-        # can read it (only on the acquired arm). GIL-atomic.
-        if acquired:
-            self._op_lock_owner = threading.get_ident()
-        try:
+                    self._op_lock_owner = None
+                    self._op_lock.release()
+                # Close the never-scheduled coro to avoid a warning.
+                coro.close()
+                # Synchronously null _async_conn so a retry from the signal
+                # handler doesn't hit "another operation is in progress" on
+                # the stale in-use conn (the queued _invalidate only lands
+                # when the slow read yields). GIL-atomic STORE_ATTR; the
+                # loop coro keeps its own ref and reaps via _invalidate.
+                dying = self._async_conn
+                if dying is not None and self._loop is not None and dying._in_use:
+                    self._async_conn = None
+                    with contextlib.suppress(RuntimeError):
+                        self._loop.call_soon_threadsafe(
+                            dying._invalidate,
+                            InterfaceError("operation interrupted during op-lock acquire"),
+                        )
+                raise
+            # Stamp owner on the acquired arm so the close() bypass probe
+            # can read it. GIL-atomic.
+            if acquired:
+                self._op_lock_owner = threading.get_ident()
             if not acquired:
                 coro.close()
                 # OperationalError (not InterfaceError) so SA's

tests/test_run_sync_owner_stamp_window.py

@@ -0,0 +1,82 @@
+"""Pin: a KeyboardInterrupt/SystemExit raised in the op-lock owner-stamp gap
+must not leak ``_op_lock``.
+
+The owner-stamp (``self._op_lock_owner = threading.get_ident()``) runs on the
+acquired arm of ``_run_sync``. If a signal lands there while the lock is held,
+the release path must still run; otherwise the lock is held by no thread and
+every later ``_run_sync`` wedges on the bounded acquire.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from dqlitedbapi import Connection
+
+
+def _make_with_loop_thread() -> Connection:
+    """Sync Connection with the loop thread up and a mock async connection."""
+    conn = Connection("localhost:9001")
+    conn._ensure_loop()
+    fake = MagicMock()
+    fake.execute = AsyncMock(return_value=(0, 0))
+    fake.close = AsyncMock()
+    fake._invalidate = MagicMock()
+    fake._in_use = False
+    fake._bound_loop = None
+    conn._async_conn = fake
+    return conn
+
+
+async def _noop() -> None:
+    return None
+
+
+def _ident_raising_in_stamp_gap(
+    conn: Connection, exc: type[BaseException], state: dict[str, bool]
+) -> Callable[[], int]:
+    """A ``threading.get_ident`` replacement that raises once, only while
+    ``_op_lock`` is held — i.e. at the owner-stamp, never at ``_check_thread``
+    or ``close`` (which also call ``get_ident`` but with the lock unheld)."""
+    real = threading.get_ident
+
+    def fake() -> int:
+        if not state["fired"] and conn._op_lock.locked():
+            state["fired"] = True
+            raise exc
+        return real()
+
+    return fake
+
+
+@pytest.mark.parametrize("exc", [KeyboardInterrupt, SystemExit])
+def test_run_sync_signal_in_owner_stamp_gap_releases_op_lock(
+    exc: type[BaseException],
+) -> None:
+    conn = _make_with_loop_thread()
+    coro = _noop()
+    try:
+        state = {"fired": False}
+        with (
+            patch(
+                "dqlitedbapi.connection.threading.get_ident",
+                side_effect=_ident_raising_in_stamp_gap(conn, exc, state),
+            ),
+            pytest.raises(exc),
+        ):
+            conn._run_sync(coro)
+
+        assert state["fired"], "test did not exercise the owner-stamp gap"
+        # The signal escaped before the body try was entered, but the lock was
+        # held — it must be released, else the connection wedges for life.
+        assert not conn._op_lock.locked()
+        assert conn._op_lock.acquire(timeout=0) is True
+        conn._op_lock.release()
+        assert conn._op_lock_owner is None
+    finally:
+        coro.close()
+        conn._closed = True