fix(sqlite): make startup-owned stores safe for FastAPI worker threads

services/account_unification/app/audit.py

@@ -7,6 +7,8 @@
 from __future__ import annotations
 
 import json
+import sqlite3
+import threading
 import time
 import uuid
 from dataclasses import dataclass, field
@@ -58,7 +60,13 @@ def events_for(self, audit_id: str) -> list[AuditEvent]:
 
 
 class SqliteAuditSink:
-    """Durable append-only sink backed by the ``account_merge_audit`` table."""
+    """Durable append-only sink backed by the ``account_merge_audit`` table.
+
+    The sink is constructed during FastAPI startup and subsequently called from
+    synchronous request-worker threads. SQLite cross-thread access is therefore
+    enabled explicitly and guarded by a re-entrant lock so one transaction can
+    never interleave with another on the shared connection.
+    """
 
     _SCHEMA = """
     CREATE TABLE IF NOT EXISTS account_merge_audit (
@@ -75,39 +83,45 @@ class SqliteAuditSink:
 
     def __init__(self, database_path: str) -> None:
         """Open the audit database and ensure the audit table exists."""
-        import sqlite3
-
-        self._connection = sqlite3.connect(database_path)
+        self._connection_lock = threading.RLock()
+        self._connection = sqlite3.connect(
+            database_path,
+            timeout=5.0,
+            check_same_thread=False,
+        )
+        self._connection.execute("PRAGMA busy_timeout = 5000")
         self._connection.execute(self._SCHEMA)
         self._connection.commit()
 
     def record(self, event: AuditEvent) -> None:
         """Persist one event row and commit immediately."""
-        self._connection.execute(
-            "INSERT INTO account_merge_audit "
-            "(audit_id, event_type, actor_name, survivor_user_id, "
-            " duplicate_user_id, payload_json, created_at) "
-            "VALUES (?, ?, ?, ?, ?, ?, ?)",
-            (
-                event.audit_id,
-                event.event_type,
-                event.actor,
-                event.survivor_user_id,
-                event.duplicate_user_id,
-                event.payload_json,
-                event.created_at,
-            ),
-        )
-        self._connection.commit()
+        with self._connection_lock:
+            self._connection.execute(
+                "INSERT INTO account_merge_audit "
+                "(audit_id, event_type, actor_name, survivor_user_id, "
+                " duplicate_user_id, payload_json, created_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (
+                    event.audit_id,
+                    event.event_type,
+                    event.actor,
+                    event.survivor_user_id,
+                    event.duplicate_user_id,
+                    event.payload_json,
+                    event.created_at,
+                ),
+            )
+            self._connection.commit()
 
     def events_for(self, audit_id: str) -> list[AuditEvent]:
         """Load events for one correlation id in event sequence order."""
-        rows = self._connection.execute(
-            "SELECT audit_id, event_type, actor_name, survivor_user_id, "
-            "duplicate_user_id, payload_json, created_at "
-            "FROM account_merge_audit WHERE audit_id = ? ORDER BY event_sequence",
-            (audit_id,),
-        ).fetchall()
+        with self._connection_lock:
+            rows = self._connection.execute(
+                "SELECT audit_id, event_type, actor_name, survivor_user_id, "
+                "duplicate_user_id, payload_json, created_at "
+                "FROM account_merge_audit WHERE audit_id = ? ORDER BY event_sequence",
+                (audit_id,),
+            ).fetchall()
         return [
             AuditEvent(
                 audit_id=row[0],
@@ -122,8 +136,9 @@ def events_for(self, audit_id: str) -> list[AuditEvent]:
         ]
 
     def close(self) -> None:
-        """Close the SQLite connection."""
-        self._connection.close()
+        """Close the SQLite connection after all in-flight operations finish."""
+        with self._connection_lock:
+            self._connection.close()
 
 
 class AuditLogger:

services/account_unification/app/kv_store.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import sqlite3
+import threading
 from typing import Protocol
 
 
@@ -57,6 +58,12 @@ class SqliteKvStore:
     Table ``idp_config_entries`` is keyed by (``config_namespace``,
     ``entry_key``). Values are stored as text; secret handling (encryption at
     rest, rotation) is delegated to the platform for the postgres backend.
+
+    FastAPI executes synchronous handlers in worker threads. The store is built
+    during application startup and then reused by those handlers, so its SQLite
+    connection explicitly permits cross-thread use and every operation is
+    serialized with a re-entrant lock. This also preserves ``:memory:`` support,
+    which a connection-per-operation implementation would lose.
     """
 
     _SCHEMA = """
@@ -71,38 +78,49 @@ class SqliteKvStore:
     def __init__(self, database_path: str) -> None:
         """Open the SQLite store and ensure the config table exists."""
         self._database_path = database_path
-        self._connection = sqlite3.connect(database_path)
+        self._connection_lock = threading.RLock()
+        self._connection = sqlite3.connect(
+            database_path,
+            timeout=5.0,
+            check_same_thread=False,
+        )
+        self._connection.execute("PRAGMA busy_timeout = 5000")
         self._connection.execute(self._SCHEMA)
         self._connection.commit()
 
     def put(self, namespace: str, entry_key: str, entry_value: str) -> None:
         """Upsert one config value."""
-        self._connection.execute(
-            "INSERT INTO idp_config_entries (config_namespace, entry_key, entry_value) "
-            "VALUES (?, ?, ?) ON CONFLICT(config_namespace, entry_key) "
-            "DO UPDATE SET entry_value = excluded.entry_value",
-            (namespace, entry_key, entry_value),
-        )
-        self._connection.commit()
+        with self._connection_lock:
+            self._connection.execute(
+                "INSERT INTO idp_config_entries "
+                "(config_namespace, entry_key, entry_value) "
+                "VALUES (?, ?, ?) ON CONFLICT(config_namespace, entry_key) "
+                "DO UPDATE SET entry_value = excluded.entry_value",
+                (namespace, entry_key, entry_value),
+            )
+            self._connection.commit()
 
     def get(self, namespace: str, entry_key: str) -> str | None:
         """Return one config value, if present."""
-        row = self._connection.execute(
-            "SELECT entry_value FROM idp_config_entries "
-            "WHERE config_namespace = ? AND entry_key = ?",
-            (namespace, entry_key),
-        ).fetchone()
+        with self._connection_lock:
+            row = self._connection.execute(
+                "SELECT entry_value FROM idp_config_entries "
+                "WHERE config_namespace = ? AND entry_key = ?",
+                (namespace, entry_key),
+            ).fetchone()
         return row[0] if row else None
 
     def get_all(self, namespace: str) -> dict[str, str]:
         """Return every config value in one namespace."""
-        rows = self._connection.execute(
-            "SELECT entry_key, entry_value FROM idp_config_entries "
-            "WHERE config_namespace = ?",
-            (namespace,),
-        ).fetchall()
+        with self._connection_lock:
+            rows = self._connection.execute(
+                "SELECT entry_key, entry_value FROM idp_config_entries "
+                "WHERE config_namespace = ?",
+                (namespace,),
+            ).fetchall()
         return {entry_key: entry_value for entry_key, entry_value in rows}
 
     def close(self) -> None:
-        """Close the SQLite connection."""
-        self._connection.close()
+        """Close the SQLite connection after all in-flight operations finish."""
+        with self._connection_lock:
+            self._connection.close()

services/account_unification/tests/test_audit.py

@@ -1,6 +1,7 @@
 """Merge operations are fully audit-logged (in-memory and SQLite sinks)."""
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import closing
 
 from app.audit import AuditLogger, AuditSink, InMemoryAuditSink, SqliteAuditSink
@@ -82,3 +83,23 @@ def test_sqlite_audit_sink_persists(tmp_path):
         with closing(SqliteAuditSink(str(db))) as reopened:
             events = reopened.events_for(result.audit_id)
             assert any(e.event_type == "merge_completed" for e in events)
+
+
+def test_sqlite_audit_sink_supports_fastapi_worker_threads(tmp_path):
+    """A sink created during startup remains usable from request workers."""
+    db = tmp_path / "threaded-audit.db"
+    with closing(SqliteAuditSink(str(db))) as sink:
+        audit = AuditLogger(sink)
+        audit_id = audit.new_correlation_id()
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            executor.submit(
+                audit.emit,
+                audit_id=audit_id,
+                event_type="threaded_test_event",
+                actor="admin@cwl",
+                payload={"result": "ok"},
+            ).result(timeout=5)
+            events = executor.submit(sink.events_for, audit_id).result(timeout=5)
+
+        assert [event.event_type for event in events] == ["threaded_test_event"]

services/account_unification/tests/test_config.py

@@ -1,6 +1,7 @@
 """Config comes only from the KV store; bootstrap points at it."""
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import closing
 
 import pytest
@@ -79,6 +80,26 @@ def test_bootstrap_points_at_sqlite_store(tmp_path):
         assert config.keycloak_realm == "cwl"
 
 
+def test_sqlite_kv_store_supports_fastapi_worker_threads(tmp_path):
+    """A store created during startup remains usable from request workers."""
+    db = tmp_path / "threaded-store.db"
+    with closing(SqliteKvStore(str(db))) as store:
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            executor.submit(
+                store.put,
+                "account_unification",
+                "operator_api_token",
+                "opaque-test-value",
+            ).result(timeout=5)
+            stored = executor.submit(
+                store.get,
+                "account_unification",
+                "operator_api_token",
+            ).result(timeout=5)
+
+        assert stored == "opaque-test-value"
+
+
 def test_unsupported_standalone_backend_fails_loudly():
     descriptor = BootstrapDescriptor(
         backend="postgres",