diff --git a/services/account_unification/app/audit.py b/services/account_unification/app/audit.py index dc967f6..c89583b 100644 --- a/services/account_unification/app/audit.py +++ b/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: diff --git a/services/account_unification/app/kv_store.py b/services/account_unification/app/kv_store.py index ad9bd7a..3dc791d 100644 --- a/services/account_unification/app/kv_store.py +++ b/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() diff --git a/services/account_unification/tests/test_audit.py b/services/account_unification/tests/test_audit.py index a12de6d..1a84fcc 100644 --- a/services/account_unification/tests/test_audit.py +++ b/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"] diff --git a/services/account_unification/tests/test_config.py b/services/account_unification/tests/test_config.py index 2620fd4..2047eda 100644 --- a/services/account_unification/tests/test_config.py +++ b/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",