diff --git a/openadapt_capture/db/__init__.py b/openadapt_capture/db/__init__.py index f5b10bb..ba157a3 100644 --- a/openadapt_capture/db/__init__.py +++ b/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 diff --git a/openadapt_capture/db/crud.py b/openadapt_capture/db/crud.py index 6de6b5e..a6d7fd9 100644 --- a/openadapt_capture/db/crud.py +++ b/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) diff --git a/tests/test_db_lock_retry.py b/tests/test_db_lock_retry.py new file mode 100644 index 0000000..cf7a9c9 --- /dev/null +++ b/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()