fix(db): retry transient SQLite writer locks

openadapt_capture/db/__init__.py

@@ -20,6 +20,10 @@
     "pk": "pk_%(table_name)s",
 }
 
+# Keep SQLite's existing bounded wait explicit so lock-recovery tests can use a
+# short timeout without making production captures fail faster.
+SQLITE_BUSY_TIMEOUT_SECONDS = 5.0
+
 
 class BaseModel:
     """The base model for database tables."""
@@ -65,7 +69,10 @@ def get_engine(db_url: str, echo: bool = False) -> sa.engine:
     """
     engine = create_engine(
         db_url,
-        connect_args={"check_same_thread": False},
+        connect_args={
+            "check_same_thread": False,
+            "timeout": SQLITE_BUSY_TIMEOUT_SECONDS,
+        },
         echo=echo,
     )
     return engine

openadapt_capture/db/crud.py

@@ -5,6 +5,8 @@
 """
 
 import json
+import sqlite3
+from time import sleep
 from typing import Any, TypeVar
 
 import sqlalchemy as sa
@@ -27,6 +29,16 @@
 
 BATCH_SIZE = 1
 
+# A SQLite connection already waits up to five seconds for a writer lock. Two
+# more bounded attempts cover a short competing transaction without hiding a
+# lock that persists. The worst-case database wait remains below the recorder's
+# 30-second shutdown contract.
+SQLITE_LOCK_RETRY_DELAYS_SECONDS = (0.05, 0.2)
+_SQLITE_LOCK_PRIMARY_CODES = {
+    getattr(sqlite3, "SQLITE_BUSY", 5),
+    getattr(sqlite3, "SQLITE_LOCKED", 6),
+}
+
 action_events = []
 screenshots = []
 window_events = []
@@ -35,6 +47,60 @@
 memory_stats = []
 
 
+def _is_sqlite_lock_error(error: sa.exc.OperationalError) -> bool:
+    """Return whether an OperationalError is SQLite lock contention."""
+    original = error.orig
+    if not isinstance(original, sqlite3.OperationalError):
+        return False
+
+    error_code = getattr(original, "sqlite_errorcode", None)
+    if isinstance(error_code, int):
+        primary_code = error_code & 0xFF
+        return primary_code in _SQLITE_LOCK_PRIMARY_CODES
+
+    message = str(original).lower()
+    return any(
+        lock_message in message
+        for lock_message in (
+            "database is locked",
+            "database table is locked",
+            "database schema is locked",
+        )
+    )
+
+
+def _execute_insert_with_lock_retry(
+    session: SaSession,
+    table: sa.Table,
+    to_insert: list[dict[str, Any]],
+) -> sa.engine.Result:
+    """Commit one insert, with bounded recovery from SQLite writer contention."""
+    for attempt in range(len(SQLITE_LOCK_RETRY_DELAYS_SECONDS) + 1):
+        try:
+            result = session.execute(sa.insert(table), to_insert)
+            session.commit()
+            return result
+        except sa.exc.OperationalError as exc:
+            if not _is_sqlite_lock_error(exc):
+                raise
+
+            # A failed execute or commit can leave the Session transaction
+            # unusable. Roll it back before either retrying or failing loud.
+            session.rollback()
+            if attempt == len(SQLITE_LOCK_RETRY_DELAYS_SECONDS):
+                raise
+
+            delay = SQLITE_LOCK_RETRY_DELAYS_SECONDS[attempt]
+            logger.warning(
+                "SQLite writer lock during insert; retrying in "
+                f"{delay:.2f}s ({attempt + 1}/"
+                f"{len(SQLITE_LOCK_RETRY_DELAYS_SECONDS)})"
+            )
+            sleep(delay)
+
+    raise AssertionError("unreachable SQLite insert retry state")
+
+
 def _insert(
     session: SaSession,
     event_data: dict[str, Any],
@@ -69,8 +135,7 @@ def _insert(
 
     if buffer is None or len(buffer) >= BATCH_SIZE:
         to_insert = buffer or [db_obj]
-        result = session.execute(sa.insert(table), to_insert)
-        session.commit()
+        result = _execute_insert_with_lock_retry(session, table, to_insert)
         if buffer:
             buffer.clear()
         # Note: this does not contain the inserted row(s)

tests/test_db_lock_retry.py

@@ -0,0 +1,107 @@
+"""Bounded SQLite writer-lock recovery contracts."""
+
+from __future__ import annotations
+
+import sqlite3
+
+import pytest
+import sqlalchemy as sa
+
+from openadapt_capture import db
+from openadapt_capture.db import crud
+from openadapt_capture.db.models import MemoryStat
+
+
+def _operational_error(message):
+    return sa.exc.OperationalError(
+        "INSERT INTO memory_stat (...) VALUES (...)",
+        {},
+        sqlite3.OperationalError(message),
+    )
+
+
+def _locked_memory_stat_database(tmp_path, monkeypatch):
+    monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.01)
+    db_path = tmp_path / "recording.db"
+    engine, Session = db.create_db(str(db_path))
+    setup_session = Session()
+    recording = crud.insert_recording(
+        setup_session,
+        {
+            "timestamp": 1.0,
+            "monitor_width": 100,
+            "monitor_height": 100,
+            "platform": "test",
+            "task_description": "SQLite lock retry",
+        },
+    )
+    recording_id = recording.id
+    setup_session.close()
+
+    writer_session = Session()
+    writer_session.execute(sa.text("SELECT 1"))
+    locking_connection = sqlite3.connect(db_path)
+    locking_connection.execute("BEGIN EXCLUSIVE")
+    locking_connection.execute(
+        "UPDATE recording SET task_description = task_description WHERE id = ?",
+        (recording_id,),
+    )
+    event_data = {
+        "recording_timestamp": 1.0,
+        "recording_id": recording_id,
+        "memory_usage_bytes": 1,
+        "timestamp": 1,
+    }
+    return engine, writer_session, locking_connection, event_data
+
+
+@pytest.mark.parametrize(
+    "message",
+    (
+        "database is locked",
+        "database table is locked: memory_stat",
+        "database schema is locked: main",
+    ),
+)
+def test_python_310_sqlite_lock_messages_are_retryable(message):
+    assert crud._is_sqlite_lock_error(_operational_error(message))
+
+
+def test_transient_sqlite_writer_lock_recovers(tmp_path, monkeypatch):
+    engine, session, locking_connection, event_data = _locked_memory_stat_database(
+        tmp_path, monkeypatch
+    )
+    retry_delays = []
+
+    def release_lock(delay):
+        retry_delays.append(delay)
+        locking_connection.commit()
+
+    monkeypatch.setattr(crud, "sleep", release_lock)
+    try:
+        crud._insert(session, event_data, MemoryStat)
+        assert retry_delays == [crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS[0]]
+        assert session.query(MemoryStat).count() == 1
+    finally:
+        locking_connection.close()
+        session.close()
+        engine.dispose()
+
+
+def test_persistent_sqlite_writer_lock_still_fails(tmp_path, monkeypatch):
+    engine, session, locking_connection, event_data = _locked_memory_stat_database(
+        tmp_path, monkeypatch
+    )
+    retry_delays = []
+    monkeypatch.setattr(crud, "sleep", retry_delays.append)
+    try:
+        with pytest.raises(sa.exc.OperationalError, match="database is locked"):
+            crud._insert(session, event_data, MemoryStat)
+
+        assert retry_delays == list(crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS)
+        locking_connection.rollback()
+        assert session.query(MemoryStat).count() == 0
+    finally:
+        locking_connection.close()
+        session.close()
+        engine.dispose()