Re-check closed state under op_lock in async commit/rollback

AsyncConnection.commit / rollback checked _closed and _async_conn
before taking op_lock and never re-checked under it. A concurrent
close() that won the lock first would null _async_conn and release,
after which the parked commit/rollback dereferenced NoneType.execute
and raised AttributeError instead of the correct InterfaceError.
Add a post-lock re-check so a lost race surfaces a clean
"Connection is closed" error.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -169,7 +169,7 @@ async def close(self) -> None:
         # re-check before it touches the now-None primitive.
         self._connect_lock = None
         self._op_lock = None
-        self._loop_id = None
+        self._loop_ref = None
 
     async def commit(self) -> None:
         """Commit any pending transaction.
@@ -184,6 +184,12 @@ async def commit(self) -> None:
             return
         _, op_lock = self._ensure_locks()
         async with op_lock:
+            # Re-check under the lock: a concurrent close() may have
+            # acquired op_lock before us, closed the connection, and
+            # released. Without this second check we would dereference
+            # ``self._async_conn.execute`` on ``None``.
+            if self._closed or self._async_conn is None:
+                raise InterfaceError("Connection is closed")
             try:
                 await self._async_conn.execute("COMMIT")
             except (OperationalError, _client_exc.OperationalError) as e:
@@ -198,6 +204,9 @@ async def rollback(self) -> None:
             return
         _, op_lock = self._ensure_locks()
         async with op_lock:
+            # Re-check under the lock for the same race as commit().
+            if self._closed or self._async_conn is None:
+                raise InterfaceError("Connection is closed")
             try:
                 await self._async_conn.execute("ROLLBACK")
             except (OperationalError, _client_exc.OperationalError) as e:

tests/test_async_close_race.py

@@ -63,3 +63,55 @@ async def run_close() -> None:
         "close:start",
         "close:end",
     ], f"actual order: {order}"
+
+
+class TestCommitRollbackCloseRace:
+    """A race where ``close()`` wins the op_lock first must make the
+    parked ``commit()`` / ``rollback()`` see ``_closed=True`` and raise
+    ``InterfaceError`` instead of dereferencing ``_async_conn=None``.
+    """
+
+    async def test_commit_parked_on_lock_sees_close_first(self) -> None:
+        import asyncio
+        from unittest.mock import AsyncMock, MagicMock
+
+        import pytest
+
+        from dqlitedbapi.aio.connection import AsyncConnection
+        from dqlitedbapi.exceptions import InterfaceError
+
+        conn = AsyncConnection("localhost:19001", database="x")
+
+        # Prime _async_conn and the locks from within the running loop
+        # so close() / commit() both take the lock path.
+        conn._ensure_locks()
+        inner = MagicMock()
+        inner.close = AsyncMock()
+        inner.execute = AsyncMock()
+        conn._async_conn = inner
+
+        # Gate close() inside the lock so commit() can park.
+        assert conn._op_lock is not None
+        close_release = asyncio.Event()
+
+        real_close = inner.close
+
+        async def slow_close(*args: object, **kwargs: object) -> None:
+            await close_release.wait()
+            await real_close(*args, **kwargs)
+
+        inner.close = slow_close  # type: ignore[assignment]
+
+        close_task = asyncio.create_task(conn.close())
+        # Yield so close() acquires op_lock.
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+
+        commit_task = asyncio.create_task(conn.commit())
+        await asyncio.sleep(0)
+        # Let close complete.
+        close_release.set()
+
+        await close_task
+        with pytest.raises(InterfaceError, match="Connection is closed"):
+            await commit_task