Make AsyncCursor.close sync and harden execute against close-race

Three coupled defects on AsyncCursor close + execute, resolved
together per the architect's hybrid plan:

* AsyncCursor.close was ``async def`` despite a body with zero await
  statements (pure attribute writes + weakref-proxy swap). A
  forgotten ``await`` produced a discarded coroutine, the cursor
  stayed open, and the only signal was a GC-time
  RuntimeWarning("coroutine was never awaited") pointing at asyncio
  internals. Convert to plain ``def`` matching stdlib
  sqlite3.Cursor.close and the project's executescript / interrupt /
  backup / tpc_* family.
* close did not acquire _op_lock and was racy with in-flight execute
  on a sibling task. Hybrid resolution: keep close sync but flip
  ``_closed = True`` FIRST (GIL-atomic write) so a sibling-task
  executor parked on the wire await observes the flag-flip on
  resume.
* _execute_unlocked populated _description / _rows / _rowcount after
  the wire await without re-checking _closed. Add a post-await
  ``if self._closed: return`` short-circuit (both query and non-
  query branches) so a concurrent close does not get its state
  scrubbing overwritten by the late wire response. Mirrors the same
  discipline already applied in _ExecuteManyAccumulator.apply.

Updated every ``await cur.close()`` call site in the dbapi/aio
package and in the test fakes that mocked the old async-def
contract.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -1683,7 +1683,7 @@ async def execute(
                 await cur.execute(operation, parameters)
         except BaseException:
             with contextlib.suppress(Exception):
-                await cur.close()
+                cur.close()
             raise
         return cur
 
@@ -1729,13 +1729,13 @@ async def executemany(
             _validate_executemany_seq_shape(seq_of_parameters)
         except ProgrammingError:
             with contextlib.suppress(Exception):
-                await cur.close()
+                cur.close()
             raise
         try:
             await cur.executemany(operation, seq_of_parameters)
         except BaseException:
             with contextlib.suppress(Exception):
-                await cur.close()
+                cur.close()
             raise
         return cur
 

src/dqlitedbapi/aio/cursor.py

@@ -306,6 +306,20 @@ async def _execute_unlocked(
             columns, column_types, row_types, rows = await _call_client(
                 conn.query_raw_typed(operation, params)
             )
+            # Post-await close-race guard: a sibling task may have
+            # called ``self.close()`` while we were parked on the
+            # wire. ``close()`` is synchronous and sets ``_closed =
+            # True`` GIL-atomically, plus clears ``_rows`` /
+            # ``_description``; without this re-check, the result-
+            # population below would re-populate state onto the now-
+            # closed cursor. Drop the result silently (close is a no-
+            # error termination — raising into the awaiter's frame
+            # would surprise a caller that just used a try/except for
+            # CancelledError). Mirrors the same discipline already
+            # applied in ``_ExecuteManyAccumulator.apply``'s post-
+            # await re-check arm.
+            if self._closed:
+                return
             if not columns:
                 # PRAGMA write-form dispatches through the row-
                 # returning branch but produces no columns; match
@@ -358,6 +372,9 @@ async def _execute_unlocked(
             self._rowcount = len(rows)
         else:
             last_id, affected = await _call_client(conn.execute(operation, params))
+            # Same post-await close-race guard as the query branch.
+            if self._closed:
+                return
             # stdlib-parity: lastrowid only updates on INSERT / REPLACE.
             # See ``_is_insert_or_replace`` in the sync cursor for
             # rationale — sync and async share the same contract.
@@ -873,10 +890,22 @@ def drain_rows(self) -> list[tuple[Any, ...]]:
         self._row_index = 0
         return rows
 
-    async def close(self) -> None:
+    def close(self) -> None:
         """Close the cursor.
 
-        Idempotent: a second call is a no-op.
+        Idempotent and **synchronous by design**. The body has zero
+        await statements (every operation is a GIL-atomic attribute
+        write plus the weakref-proxy swap); making it ``async def``
+        would invite a forgot-``await`` footgun where ``cur.close()``
+        silently produced a discarded coroutine and the cursor was
+        left undrained, with only a GC-time
+        ``RuntimeWarning("coroutine was never awaited")`` pointing at
+        asyncio internals rather than at dqlite. Mirrors stdlib
+        ``sqlite3.Cursor.close`` and the project's
+        ``executescript`` / ``interrupt`` / ``backup`` / ``tpc_*``
+        family (sync ``def`` stubs that surface forgot-call as
+        immediate ``NotSupportedError`` rather than a discarded
+        coroutine).
 
         Scrubs ``description`` / ``rowcount`` / ``lastrowid`` /
         ``_rows`` / ``_row_index`` symmetrically with the sync sibling
@@ -890,11 +919,21 @@ async def close(self) -> None:
         retain ``arraysize`` across ``close()``; this driver matches
         that parity. ``arraysize`` is therefore the single PEP 249
         §6.1.2 attribute outside the scrub set above — by design.
+
+        Cross-task safety: ``_closed = True`` is set FIRST (GIL-atomic
+        write) so a sibling task whose ``_execute_unlocked`` is parked
+        on the wire await observes the flag-flip on resume. The
+        executor's post-await ``if self._closed: return`` short-circuit
+        prevents the wire response from re-populating ``_rows`` /
+        ``_description`` onto a closed cursor.
         """
         # PEP 249 §6.1.2 messages-clear contract; see Cursor.close.
         del self.messages[:]
         if self._closed:
             return
+        # Set the flag FIRST so a sibling-task ``_execute_unlocked``
+        # that is parked on the wire await observes it on resume and
+        # short-circuits before repopulating state.
         self._closed = True
         self._rows = []
         self._description = None
@@ -918,7 +957,7 @@ async def close(self) -> None:
         ):  # pragma: no cover - AsyncConnection always supports weakref
             self._connection = weakref.proxy(self._connection)
 
-    def setinputsizes(self, sizes: Sequence[Any]) -> None:
+    def setinputsizes(self, sizes: Sequence[Any] | None) -> None:
         """Set input sizes (no-op for dqlite).
 
         PEP 249 §6.1.1 names ``setinputsizes`` among the methods that
@@ -1208,4 +1247,5 @@ async def __aexit__(
         exc_val: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        await self.close()
+        # ``close`` is now sync (see docstring) — no await needed.
+        self.close()

src/dqlitedbapi/cursor.py

@@ -2057,7 +2057,7 @@ def close(self) -> None:
         ):  # pragma: no cover - Connection always supports weakref
             self._connection = weakref.proxy(self._connection)
 
-    def setinputsizes(self, sizes: Sequence[Any]) -> None:
+    def setinputsizes(self, sizes: Sequence[Any] | None) -> None:
         """Set input sizes (no-op for dqlite).
 
         PEP 249 §6.1.1 names ``setinputsizes`` among the methods that

tests/aio/test_async_iter_protocol_invariants.py

@@ -32,7 +32,7 @@ async def test_aiter_on_closed_cursor_does_not_raise() -> None:
     to ``fetchone`` and raises ``InterfaceError("Cursor is closed")``)."""
     aconn = AsyncConnection("127.0.0.1:9999", database="x")
     cur = aconn.cursor()
-    await cur.close()
+    cur.close()
 
     # Must NOT raise. Pre-fix this raised InterfaceError because
     # ``_check_loop_binding`` ran a closed-state check before

tests/integration/test_description_after_ddl_pragma.py

@@ -102,6 +102,6 @@ async def test_description_is_none_after_ddl(self) -> None:
 
             await cur.execute("DROP TABLE t_ddl_pragma_async")
             assert cur.description is None
-            await cur.close()
+            cur.close()
         finally:
             await conn.close()

tests/integration/test_lastrowid_autoincrement.py

@@ -86,6 +86,6 @@ async def test_lastrowid_after_insert(self) -> None:
             assert cur.lastrowid == first + 1
 
             await cur.execute("DROP TABLE t_rowid_async")
-            await cur.close()
+            cur.close()
         finally:
             await aconn.close()

tests/integration/test_lastrowid_survives_cross_cursor_rollback.py

@@ -97,7 +97,7 @@ async def test_two_cursor_rollback_keeps_each_lastrowid(
             await setup.execute("DROP TABLE IF EXISTS t_xc_rollback_async")
             await setup.execute("CREATE TABLE t_xc_rollback_async (id INTEGER PRIMARY KEY, v TEXT)")
         finally:
-            await setup.close()
+            setup.close()
 
         cur_a = aconn.cursor()
         cur_b = aconn.cursor()
@@ -120,10 +120,10 @@ async def test_two_cursor_rollback_keeps_each_lastrowid(
             row = await cur_a.fetchone()
             assert row is not None and row[0] == 0
         finally:
-            await cur_a.close()
-            await cur_b.close()
+            cur_a.close()
+            cur_b.close()
             cleanup = aconn.cursor()
             try:
                 await cleanup.execute("DROP TABLE IF EXISTS t_xc_rollback_async")
             finally:
-                await cleanup.close()
+                cleanup.close()

tests/integration/test_no_transaction_error_wording.py

@@ -99,7 +99,7 @@ async def test_stray_commit_emits_known_error(
             msg = str(ei.value).lower()
             assert "no transaction is active" in msg, f"unexpected msg: {ei.value!s}"
         finally:
-            await cur.close()
+            cur.close()
 
     async def test_stray_rollback_emits_known_error(
         self, aconn: dqlitedbapi.aio.AsyncConnection
@@ -112,7 +112,7 @@ async def test_stray_rollback_emits_known_error(
             msg = str(ei.value).lower()
             assert "no transaction is active" in msg, f"unexpected msg: {ei.value!s}"
         finally:
-            await cur.close()
+            cur.close()
 
     async def test_commit_swallows_no_tx_via_connection_method(
         self, aconn: dqlitedbapi.aio.AsyncConnection

tests/test_async_cursor.py

@@ -37,14 +37,14 @@ def test_lastrowid_initially_none(self) -> None:
     async def test_close_marks_cursor_closed(self) -> None:
         conn = AsyncConnection("localhost:9001")
         cursor = AsyncCursor(conn)
-        await cursor.close()
+        cursor.close()
         assert cursor._closed
 
     async def test_close_is_idempotent(self) -> None:
         conn = AsyncConnection("localhost:9001")
         cursor = AsyncCursor(conn)
-        await cursor.close()
-        await cursor.close()  # must not raise
+        cursor.close()
+        cursor.close()  # must not raise
         assert cursor._closed
 
     def test_connection_property(self) -> None:
@@ -55,22 +55,22 @@ def test_connection_property(self) -> None:
     async def test_fetchone_on_closed_cursor_raises(self) -> None:
         conn = AsyncConnection("localhost:9001")
         cursor = AsyncCursor(conn)
-        await cursor.close()
+        cursor.close()
 
         with pytest.raises(InterfaceError, match="Cursor is closed"):
             await cursor.fetchone()
 
     async def test_fetchmany_on_closed_cursor_raises(self) -> None:
         conn = AsyncConnection("localhost:9001")
         cursor = AsyncCursor(conn)
-        await cursor.close()
+        cursor.close()
         with pytest.raises(InterfaceError, match="Cursor is closed"):
             await cursor.fetchmany(5)
 
     async def test_fetchall_on_closed_cursor_raises(self) -> None:
         conn = AsyncConnection("localhost:9001")
         cursor = AsyncCursor(conn)
-        await cursor.close()
+        cursor.close()
         with pytest.raises(InterfaceError, match="Cursor is closed"):
             await cursor.fetchall()
 
@@ -264,7 +264,7 @@ async def run_execute() -> None:
 
         task = asyncio.create_task(run_execute())
         await ensure_entered.wait()
-        await cursor.close()
+        cursor.close()
         close_allowed.set()
         with pytest.raises(InterfaceError, match="Cursor is closed"):
             await task

tests/test_async_cursor_aenter_gc_parent_pep249.py

@@ -22,7 +22,7 @@ async def test_aenter_on_closed_cursor_with_gc_parent_raises_interface_error() -
     ``ReferenceError`` out of ``__aenter__``."""
     conn = AsyncConnection("localhost:9001")
     cur = conn.cursor()
-    await cur.close()
+    cur.close()
     # Drop the connection ref and force GC so the proxy referent
     # disappears.
     del conn