fix: give each thread its own SQLite connection

hivemind_sqlite_database/__init__.py

@@ -1,8 +1,8 @@
 import json
 import os.path
 import sqlite3
 import threading
-from typing import List, Optional, Union, Iterable
+from typing import ClassVar, List, Optional, Union, Iterable
 
 from ovos_utils.log import LOG
 from ovos_utils.xdg_utils import xdg_data_home
@@ -25,8 +25,12 @@
     """Database implementation using SQLite."""
     name: str = "clients"
     subfolder: str = "hivemind-core"
     password: Optional[str] = None
 
+    # How long SQLite waits for a file lock held by another connection
+    # before giving up with SQLITE_BUSY, in milliseconds.
+    BUSY_TIMEOUT_MS: ClassVar[int] = 10000
+
     def __post_init__(self):
         """
         Initialize the SQLiteDB connection.
@@ -38,14 +42,31 @@
 
         When *password* is ``None`` (default) the standard ``sqlite3`` module
         is used and the database file is unencrypted.
+
+        Each thread gets its own connection (see :attr:`conn`). A single
+        shared connection is not usable from several threads at once: the
+        transaction state belongs to the connection, so one thread's
+        ``COMMIT`` ends another thread's transaction and resets its
+        in-flight statements. Under a threaded network backend that shows
+        up as ``sqlite3.ProgrammingError: bad parameter or other API
+        misuse`` and as writes that land with corrupted bindings.
         """
-        db_path = os.path.join(xdg_data_home(), self.subfolder, self.name + ".db")
-        LOG.debug(f"sqlite database path: {db_path}")
-        os.makedirs(os.path.dirname(db_path), exist_ok=True)
+        self._db_path = os.path.join(xdg_data_home(), self.subfolder, self.name + ".db")
+        LOG.debug(f"sqlite database path: {self._db_path}")
+        os.makedirs(os.path.dirname(self._db_path), exist_ok=True)
+
+        if self.password is not None and self.password == "":
+            raise ValueError("password must be non-empty when encryption is enabled")
+
+        self._write_lock = threading.Lock()
+        # opening the first connection also applies the WAL pragma
+        self.conn.execute("PRAGMA journal_mode=WAL")
+        self._initialize_database()
+        self._maybe_migrate()
 
+    def _connect(self):
+        """Open one new connection to the backing file."""
         if self.password is not None:
-            if self.password == "":
-                raise ValueError("password must be non-empty when encryption is enabled")
             try:
                 import sqlcipher3 as _sqlcipher
             except ImportError:
@@ -54,18 +75,56 @@
                     "Install the system library (e.g. 'apt install libsqlcipher-dev') "
                     "then: pip install hivemind-sqlite-database[cipher]"
                 )
-            self.conn = _sqlcipher.connect(db_path, check_same_thread=False)
-            self.conn.row_factory = _sqlcipher.Row
+            conn = _sqlcipher.connect(self._db_path, check_same_thread=False)
+            conn.row_factory = _sqlcipher.Row
             escaped_password = self.password.replace("'", "''")
-            self.conn.execute(f"PRAGMA key='{escaped_password}'")
+            conn.execute(f"PRAGMA key='{escaped_password}'")
         else:
-            self.conn = sqlite3.connect(db_path, check_same_thread=False)
-            self.conn.row_factory = sqlite3.Row
-
-        self.conn.execute("PRAGMA journal_mode=WAL")
-        self._write_lock = threading.Lock()
-        self._initialize_database()
-        self._maybe_migrate()
+            conn = sqlite3.connect(self._db_path, check_same_thread=False)
+            conn.row_factory = sqlite3.Row
+        conn.execute(f"PRAGMA busy_timeout={int(self.BUSY_TIMEOUT_MS)}")
+        return conn
+
+    @property
+    def conn(self):
+        """The calling thread's own connection, opened on first use.
+
+        Connections are cheap and WAL lets many readers run beside one
+        writer, so a per-thread connection costs a file handle and buys
+        real thread safety. ``_write_lock`` still serialises writers
+        in-process so they do not fight over the file lock.
+        """
+        local = self._thread_state()
+        conn = getattr(local, "conn", None)
+        if conn is None:
+            conn = local.conn = self._connect()
+        return conn
+
+    @conn.setter
+    def conn(self, value) -> None:
+        """Adopt an already-open connection for the calling thread.
+
+        Only the calling thread sees it; other threads still open their
+        own. Tests use this to inject an in-memory database.
+        """
+        self._thread_state().conn = value
+
+    def _thread_state(self) -> threading.local:
+        local = self.__dict__.get("_local")
+        if local is None:
+            local = self.__dict__["_local"] = threading.local()
+        return local
+
+    def close(self) -> None:
+        """Close the calling thread's connection, if it has one."""
+        local = self._thread_state()
+        conn = getattr(local, "conn", None)
+        if conn is not None:
+            local.conn = None
+            try:
+                conn.close()
+            except sqlite3.Error as e:
+                LOG.error(f"Failed to close SQLite connection: {e}")
 
     def _initialize_database(self):
         """Initialize the database schema."""
@@ -220,7 +279,7 @@
             LOG.error(f"Failed to add client to SQLite: {e}")
             return False
 
     def search_by_value(self, key: str, val: Union[str, bool, int, float]) -> List[Client]:
         """
         Search for clients by a specific key-value pair in the SQLite database.
 
@@ -235,15 +294,17 @@
             LOG.error(f"Invalid search key: {key!r}")
             return []
         try:
-            with self.conn:
-                cur = self.conn.execute(f"SELECT * FROM clients WHERE {key} = ?", (val,))
-                rows = cur.fetchall()
-                return [self._row_to_client(row) for row in rows]
+            # deliberately NOT wrapped in ``with self.conn`` — that context
+            # manager commits the connection's transaction on exit, which a
+            # read has no business doing.
+            cur = self.conn.execute(f"SELECT * FROM clients WHERE {key} = ?", (val,))
+            rows = cur.fetchall()
+            return [self._row_to_client(row) for row in rows]
         except sqlite3.Error as e:
             LOG.error(f"Failed to search clients in SQLite: {e}")
             return []
 
     def get_client_by_id(self, client_id: int) -> Optional[Client]:
         """Fetch a single client row by primary key.
 
         Targeted lookup used by :meth:`refresh` on the admission hot
@@ -273,7 +334,7 @@
             LOG.error(f"Failed to count clients in SQLite: {e}")
             return 0
 
     def __iter__(self) -> Iterable['Client']:
         """
         Iterate over all clients in the SQLite database.
 
@@ -310,7 +371,7 @@
         # any residual metadata key from earlier migrations.
         metadata.pop("message_blacklist", None)
         for key in ("intent_blacklist", "skill_blacklist"):
             raw = row[key] if key in row.keys() else None
             if not raw or key in metadata:
                 continue
             try:

tests/test_sqlitedb.py

@@ -883,3 +883,66 @@ def test_refresh_returns_v2_fields(self):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TestSQLiteDBThreadSafety(unittest.TestCase):
+    """A real file-backed DB hammered from many threads at once.
+
+    Reproduces the failure seen under the threaded (webrockets) network
+    backend: concurrent reads and writes through ONE shared connection
+    raise ``bad parameter or other API misuse`` and write rows with
+    corrupted bindings (``NOT NULL constraint failed: clients.api_key``),
+    because one thread's implicit COMMIT ends another thread's
+    transaction.
+    """
+
+    def test_concurrent_read_write_is_clean(self):
+        import tempfile
+        from unittest.mock import patch
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch("hivemind_sqlite_database.xdg_data_home", return_value=tmp):
+                db = SQLiteDB()
+                n = 60
+                for i in range(n):
+                    self.assertTrue(db.add_item(make_client(i, f"key-{i}")))
+
+                errors = []
+                misses = []
+                barrier = threading.Barrier(24)
+
+                def reader(i):
+                    barrier.wait()
+                    try:
+                        for _ in range(40):
+                            got = db.search_by_value("api_key", f"key-{i % n}")
+                            if len(got) != 1:
+                                misses.append((i, len(got)))
+                    except Exception as e:  # noqa: BLE001
+                        errors.append(repr(e))
+
+                def writer(i):
+                    barrier.wait()
+                    try:
+                        for k in range(40):
+                            if not db.add_item(
+                                make_client(i % n, f"key-{i % n}",
+                                            name=f"w{k}")):
+                                errors.append(f"add_item returned False ({i},{k})")
+                    except Exception as e:  # noqa: BLE001
+                        errors.append(repr(e))
+
+                threads = [threading.Thread(target=reader, args=(i,))
+                           for i in range(16)]
+                threads += [threading.Thread(target=writer, args=(i,))
+                            for i in range(8)]
+                for t in threads:
+                    t.start()
+                for t in threads:
+                    t.join(timeout=120)
+
+                self.assertEqual(errors, [])
+                self.assertEqual(misses, [])
+                # every api_key still resolves to exactly one live row
+                for i in range(n):
+                    self.assertEqual(len(db.search_by_value("api_key", f"key-{i}")), 1)