Enforce thread safety and harden close() against races

Add thread-identity check like sqlite3 stdlib: store the creator
thread ID and raise ProgrammingError if the connection or cursor is
used from a different thread. This turns silent data corruption into
a clear error message.

Harden close(): set _closed=True immediately and return early on
double-close, preventing races where close() tears down resources
while operations are in flight. Apply to both sync and async paths.

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

Author: antoineleclair

src/dqlitedbapi/aio/connection.py

@@ -66,10 +66,12 @@ async def connect(self) -> None:
 
     async def close(self) -> None:
         """Close the connection."""
+        if self._closed:
+            return
+        self._closed = True
         if self._async_conn is not None:
             await self._async_conn.close()
             self._async_conn = None
-        self._closed = True
 
     async def commit(self) -> None:
         """Commit any pending transaction."""

src/dqlitedbapi/connection.py

@@ -7,7 +7,7 @@
 
 from dqliteclient import DqliteConnection
 from dqlitedbapi.cursor import Cursor
-from dqlitedbapi.exceptions import InterfaceError, OperationalError
+from dqlitedbapi.exceptions import InterfaceError, OperationalError, ProgrammingError
 
 
 class Connection:
@@ -37,6 +37,17 @@ def __init__(
         self._loop_lock = threading.Lock()
         self._op_lock = threading.Lock()
         self._connect_lock: asyncio.Lock | None = None
+        self._creator_thread = threading.get_ident()
+
+    def _check_thread(self) -> None:
+        """Raise ProgrammingError if called from a different thread than the creator."""
+        current = threading.get_ident()
+        if current != self._creator_thread:
+            raise ProgrammingError(
+                f"Connection objects created in a thread can only be used in that "
+                f"same thread. The object was created in thread id "
+                f"{self._creator_thread} and this is thread id {current}."
+            )
 
     def _ensure_loop(self) -> asyncio.AbstractEventLoop:
         """Ensure a dedicated event loop is running in a background thread.
@@ -102,6 +113,10 @@ async def _get_async_connection(self) -> DqliteConnection:
 
     def close(self) -> None:
         """Close the connection."""
+        self._check_thread()
+        if self._closed:
+            return
+        self._closed = True
         try:
             if self._async_conn is not None:
                 with contextlib.suppress(Exception):
@@ -115,10 +130,10 @@ def close(self) -> None:
                 self._loop.close()
                 self._loop = None
                 self._thread = None
-            self._closed = True
 
     def commit(self) -> None:
         """Commit any pending transaction."""
+        self._check_thread()
         if self._closed:
             raise InterfaceError("Connection is closed")
 
@@ -132,6 +147,7 @@ async def _commit_async(self) -> None:
 
     def rollback(self) -> None:
         """Roll back any pending transaction."""
+        self._check_thread()
         if self._closed:
             raise InterfaceError("Connection is closed")
 
@@ -145,6 +161,7 @@ async def _rollback_async(self) -> None:
 
     def cursor(self) -> Cursor:
         """Return a new Cursor object."""
+        self._check_thread()
         if self._closed:
             raise InterfaceError("Connection is closed")
         return Cursor(self)

src/dqlitedbapi/cursor.py

@@ -82,6 +82,7 @@ def _check_closed(self) -> None:
 
     def execute(self, operation: str, parameters: Sequence[Any] | None = None) -> "Cursor":
         """Execute a database operation (query or command)."""
+        self._connection._check_thread()
         self._check_closed()
 
         self._connection._run_sync(self._execute_async(operation, parameters))
@@ -123,6 +124,7 @@ async def _execute_async(self, operation: str, parameters: Sequence[Any] | None
 
     def executemany(self, operation: str, seq_of_parameters: Sequence[Sequence[Any]]) -> "Cursor":
         """Execute a database operation multiple times."""
+        self._connection._check_thread()
         self._check_closed()
 
         self._connection._run_sync(self._executemany_async(operation, seq_of_parameters))
@@ -145,6 +147,7 @@ def _check_result_set(self) -> None:
 
     def fetchone(self) -> tuple[Any, ...] | None:
         """Fetch the next row of a query result set."""
+        self._connection._check_thread()
         self._check_closed()
         self._check_result_set()
 
@@ -157,6 +160,7 @@ def fetchone(self) -> tuple[Any, ...] | None:
 
     def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
         """Fetch the next set of rows of a query result."""
+        self._connection._check_thread()
         self._check_closed()
         self._check_result_set()
 
@@ -174,6 +178,7 @@ def fetchmany(self, size: int | None = None) -> list[tuple[Any, ...]]:
 
     def fetchall(self) -> list[tuple[Any, ...]]:
         """Fetch all remaining rows of a query result."""
+        self._connection._check_thread()
         self._check_closed()
         self._check_result_set()
 
@@ -183,6 +188,7 @@ def fetchall(self) -> list[tuple[Any, ...]]:
 
     def close(self) -> None:
         """Close the cursor."""
+        self._connection._check_thread()
         self._closed = True
         self._rows = []
         self._description = None

tests/test_protocol_serialization.py

@@ -74,122 +74,31 @@ async def mock_exec_sql(db_id: int, sql: str, params: object) -> tuple:
 
 
 class TestSyncProtocolSerialization:
-    """Test that concurrent sync operations are serialized."""
+    """Test that concurrent sync operations from wrong threads are rejected."""
 
-    def test_concurrent_run_sync_is_serialized(self) -> None:
-        """Two threads calling _run_sync must not overlap on the event loop.
+    def test_cross_thread_execute_raises_programming_error(self) -> None:
+        """Threads sharing a connection must get ProgrammingError.
 
-        Without serialization, both threads submit coroutines concurrently
-        to the same event loop, where they interleave at await points.
+        The thread-identity check (like sqlite3) prevents cross-thread
+        access before it reaches the protocol layer.
         """
-        conn = Connection("localhost:9001", timeout=5.0)
-
-        call_log: list[tuple[str, str]] = []
-        log_lock = threading.Lock()
-
-        async def mock_query_sql(db_id: int, sql: str, params: object) -> tuple:
-            with log_lock:
-                call_log.append((sql, "start"))
-            await asyncio.sleep(0.05)
-            with log_lock:
-                call_log.append((sql, "end"))
-            return (["id"], [[1]])
-
-        async def mock_exec_sql(db_id: int, sql: str, params: object) -> tuple:
-            with log_lock:
-                call_log.append((sql, "start"))
-            await asyncio.sleep(0.05)
-            with log_lock:
-                call_log.append((sql, "end"))
-            return (0, 1)
-
-        with patch("dqlitedbapi.connection.DqliteConnection") as MockDqliteConn:
-            mock_instance = AsyncMock()
-            mock_instance.connect = AsyncMock()
-            mock_instance._protocol = MagicMock()
-            mock_instance._protocol.query_sql = mock_query_sql
-            mock_instance._protocol.exec_sql = mock_exec_sql
-            mock_instance._db_id = 0
-            MockDqliteConn.return_value = mock_instance
-
-            cursor1 = Cursor(conn)
-            cursor2 = Cursor(conn)
+        from dqlitedbapi.exceptions import ProgrammingError
 
-            barrier = threading.Barrier(2)
-            errors: list[Exception] = []
-
-            def thread_work(cursor: Cursor, sql: str) -> None:
-                try:
-                    barrier.wait(timeout=5)
-                    cursor.execute(sql)
-                except Exception as e:
-                    errors.append(e)
-
-            t1 = threading.Thread(target=thread_work, args=(cursor1, "SELECT 1"))
-            t2 = threading.Thread(target=thread_work, args=(cursor2, "INSERT INTO t VALUES (1)"))
-            t1.start()
-            t2.start()
-            t1.join(timeout=10)
-            t2.join(timeout=10)
-
-            assert not errors, f"Threads raised errors: {errors}"
-
-            # Operations must be serialized: one completes before the other starts
-            assert len(call_log) == 4
-            assert call_log[0][1] == "start"
-            assert call_log[1][1] == "end"
-            assert call_log[2][1] == "start"
-            assert call_log[3][1] == "end"
-
-            conn.close()
-
-
-class TestSyncConnectionLazyInitRace:
-    """Test that _get_async_connection doesn't create duplicate connections."""
-
-    def test_concurrent_first_use_creates_single_connection(self) -> None:
-        """Two threads using a connection for the first time must not
-        create two underlying DqliteConnection instances."""
         conn = Connection("localhost:9001", timeout=5.0)
+        cursor = Cursor(conn)
 
-        connect_count = 0
-        count_lock = threading.Lock()
+        errors: list[Exception] = []
 
-        async def slow_connect() -> None:
-            nonlocal connect_count
-            with count_lock:
-                connect_count += 1
-            await asyncio.sleep(0.05)
+        def thread_work() -> None:
+            try:
+                cursor.execute("SELECT 1")
+            except Exception as e:
+                errors.append(e)
 
-        with patch("dqlitedbapi.connection.DqliteConnection") as MockDqliteConn:
-            mock_instance = AsyncMock()
-            mock_instance.connect = slow_connect
-            mock_instance._protocol = MagicMock()
-            mock_instance._protocol.query_sql = AsyncMock(return_value=(["id"], [[1]]))
-            mock_instance._db_id = 0
-            MockDqliteConn.return_value = mock_instance
+        t = threading.Thread(target=thread_work)
+        t.start()
+        t.join(timeout=5)
 
-            barrier = threading.Barrier(2)
-            errors: list[Exception] = []
-
-            def thread_work() -> None:
-                try:
-                    barrier.wait(timeout=5)
-                    cursor = conn.cursor()
-                    cursor.execute("SELECT 1")
-                except Exception as e:
-                    errors.append(e)
-
-            t1 = threading.Thread(target=thread_work)
-            t2 = threading.Thread(target=thread_work)
-            t1.start()
-            t2.start()
-            t1.join(timeout=10)
-            t2.join(timeout=10)
-
-            assert not errors, f"Threads raised errors: {errors}"
-            # Only one DqliteConnection should have been created
-            assert MockDqliteConn.call_count == 1
-            assert connect_count == 1
-
-            conn.close()
+        assert len(errors) == 1
+        assert isinstance(errors[0], ProgrammingError)
+        conn.close()