fix(db): bound the SQLite write-lock wait by time, not by attempts

CHANGELOG.md

@@ -51,6 +51,12 @@ _This release is published under the MIT License._
 - **recorder**: Survive SQLite writer contention in the video writer
   ([#116](https://github.com/OpenAdaptAI/openadapt-capture/pull/116),
   [`1695be7`](https://github.com/OpenAdaptAI/openadapt-capture/commit/1695be789988c36370aba752685b8ccf61330120))
+- **db**: Bound the SQLite write-lock wait by time rather than by a count of
+  attempts, and give a live capture a write-ahead log so the lock is free far
+  more often. A recorder writer used to starve for about twenty-two seconds and
+  then exhaust its three retries, which killed the writer process or cost it
+  the startup readiness deadline
+  ([#122](https://github.com/OpenAdaptAI/openadapt-capture/pull/122))
 - **release**: Let the changelog document the pending release candidate
   ([#110](https://github.com/OpenAdaptAI/openadapt-capture/pull/110),
   [`854f015`](https://github.com/OpenAdaptAI/openadapt-capture/commit/854f015fd994b0aa1992948791c18c2e6c571029))

openadapt_capture/db/__init__.py

@@ -7,6 +7,7 @@
 from pathlib import Path
 
 import sqlalchemy as sa
+from loguru import logger
 from sqlalchemy import create_engine, inspect, text
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
@@ -20,9 +21,32 @@
     "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
+# How long ONE statement waits for the write lock before it reports
+# "database is locked".
+#
+# This is deliberately short. It is not the whole wait: crud spends a declared
+# budget on many short attempts rather than a few long ones, and this value is
+# the cost of one of those attempts. A long value spends the whole budget on a
+# handful of samples of a lock that is busy most of the time, which is how a
+# writer used to reach its last retry ~22 seconds after its first.
+SQLITE_BUSY_TIMEOUT_SECONDS = 0.5
+
+# Journal mode for a live per-capture database.
+#
+# Every recorder writer process commits to the one capture file, and under the
+# default rollback journal each commit creates, syncs and deletes a journal
+# file beside it. On Windows that file churn is scanned by the filesystem
+# filter driver, one screenshot row costs about half a second to commit, and a
+# writer draining a backlog holds the single write lock at essentially full
+# duty cycle. Other writers then starve for tens of seconds.
+#
+# A write-ahead log appends instead, so a commit is cheap, readers never block
+# a writer, and the lock is free far more often. The mode is recorded in the
+# database header, so every later connection to the same file inherits it
+# without setting it. A capture is returned to the rollback journal when it is
+# finalized (see finalize_capture_database), so a sealed capture keeps the
+# on-disk shape it has always had and carries no sidecar files.
+SQLITE_CAPTURE_JOURNAL_MODE = "WAL"
 
 
 class BaseModel:
@@ -75,6 +99,25 @@ def get_engine(db_url: str, echo: bool = False) -> sa.engine:
         },
         echo=echo,
     )
+
+    @sa.event.listens_for(engine, "connect")
+    def _match_synchronous_to_the_journal_mode(dbapi_connection, _record) -> None:
+        """Relax the sync only for a database that already keeps a write log.
+
+        A write-ahead log makes NORMAL safe against a process crash: only a
+        power loss can cost the last commits, and a capture interrupted by a
+        power loss is incomplete anyway. A rollback-journal database keeps the
+        default FULL, so an existing capture's durability is unchanged.
+        """
+        cursor = dbapi_connection.cursor()
+        try:
+            cursor.execute("PRAGMA journal_mode")
+            row = cursor.fetchone()
+            if row and str(row[0]).lower() == "wal":
+                cursor.execute("PRAGMA synchronous=NORMAL")
+        finally:
+            cursor.close()
+
     return engine
 
 
@@ -123,14 +166,18 @@ def migrate_missing_columns(engine: sa.engine) -> None:
                 )
 
 
-def create_db(db_path: str, echo: bool = False) -> tuple:
+def create_db(db_path: str, echo: bool = False, journal_mode: str | None = None) -> tuple:
     """Create a new database at the given path, returning (engine, Session).
 
     Creates all tables defined in the models.
 
     Args:
         db_path: Path to the SQLite database file.
         echo: Whether to echo SQL statements.
+        journal_mode: Journal mode to record in the new database header. Pass
+            ``SQLITE_CAPTURE_JOURNAL_MODE`` for a database several writer
+            processes will share; leave it unset for a database with one
+            writer, whose exact bytes are then unchanged by this argument.
 
     Returns:
         tuple of (engine, Session class).
@@ -141,6 +188,23 @@ def create_db(db_path: str, echo: bool = False) -> tuple:
     # Import models to ensure they are registered with Base
     from openadapt_capture.db import models  # noqa: F401
 
+    if journal_mode is not None:
+        with engine.connect() as connection:
+            selected = connection.exec_driver_sql(
+                f"PRAGMA journal_mode={journal_mode}"
+            ).scalar()
+        # A journal mode is a request, not a guarantee: a database on a
+        # filesystem without shared memory keeps the rollback journal. The
+        # capture still works there, only with the contention this mode is here
+        # to remove, so say so once rather than failing a recording over it.
+        if str(selected).lower() != journal_mode.lower():
+            logger.warning(
+                f"{db_path} kept journal mode {selected!r} rather than "
+                f"{journal_mode!r}; expect slower commits and more "
+                "writer-lock contention"
+            )
+        engine.dispose()
+
     Base.metadata.create_all(engine)
     # Reconcile schemas of pre-existing DBs that predate newer columns.
     migrate_missing_columns(engine)
@@ -178,6 +242,61 @@ def get_session_for_path(db_path: str, echo: bool = False):
         raise
 
 
+def close_capture_session(session) -> None:
+    """Close a session and release the connection its engine pooled.
+
+    ``get_session_for_path`` builds an engine per call, so closing the session
+    alone returns its connection to that engine's pool and leaves the file
+    open. An open connection stops a capture being finalized, and on Windows it
+    also makes the capture directory undeletable.
+
+    Args:
+        session: A session from ``get_session_for_path``.
+    """
+    bind = session.get_bind()
+    session.close()
+    bind.dispose()
+
+
+def finalize_capture_database(db_path: str) -> None:
+    """Fold the write log back into the capture file and drop the sidecars.
+
+    A live capture uses a write-ahead log, which keeps ``recording.db-wal`` and
+    ``recording.db-shm`` beside the database. A finalized capture must not: the
+    seal inventories every regular file under the capture directory, and the
+    shared-memory file is created and removed by whoever opens the database
+    next, so a sealed capture that listed one would fail its own validation.
+
+    Call this once, after every writer process has exited and before the
+    capture is verified and sealed. It fails loud rather than sealing a capture
+    whose sidecars are still there.
+
+    Args:
+        db_path: Path to the per-capture SQLite database file.
+
+    Raises:
+        RuntimeError: The write log did not fold back into the database.
+    """
+    database = sqlite3.connect(db_path, timeout=SQLITE_BUSY_TIMEOUT_SECONDS)
+    try:
+        database.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+        database.execute("PRAGMA journal_mode=DELETE")
+        database.commit()
+    finally:
+        database.close()
+
+    surviving = [
+        suffix
+        for suffix in ("-wal", "-shm")
+        if Path(f"{db_path}{suffix}").exists()
+    ]
+    if surviving:
+        raise RuntimeError(
+            "The finalized Capture database still has a write log: "
+            f"{', '.join(surviving)}. A writer is still holding it open."
+        )
+
+
 def get_immutable_session_for_path(db_path: str, echo: bool = False):
     """Open an already-verified SQLite snapshot without schema migration."""
 

openadapt_capture/db/crud.py

@@ -30,11 +30,39 @@
 
 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)
+# The whole share of a startup-readiness deadline that ONE write transaction
+# may spend waiting for the single SQLite write lock.
+#
+# Express the wait as time, never as a number of attempts. An attempt count
+# gives no bound at all, because the cost of an attempt is whatever SQLite's
+# busy handler decides: measured on a hosted Windows runner, one attempt with
+# a five-second busy timeout took about seven seconds, so three attempts spent
+# about twenty-two seconds and then gave up. Twenty-two seconds is both too
+# long to fit a thirty-second readiness deadline and, at three samples of a
+# busy lock, far too few chances to win the race.
+#
+# The same budget spent on short attempts samples the lock some tens of times
+# instead of three. recorder.py checks this budget against its own readiness
+# deadline when it is imported.
+SQLITE_WRITE_LOCK_BUDGET_SECONDS = 20.0
+
+# The most one attempt is allowed to cost.
+#
+# Every statement of the transaction may wait the connection's busy timeout,
+# and SQLite's busy handler overshoots that timeout under contention. The
+# helper below refuses to BEGIN an attempt unless this much of the budget
+# remains, which is what makes the budget an upper bound on the total wait
+# rather than an estimate of it.
+#
+# tests/test_db_lock_retry.py measures a real attempt against a real held lock
+# and fails if it costs more than this.
+SQLITE_LOCK_ATTEMPT_CEILING_SECONDS = 2.0
+
+# Back off between attempts so the writers do not resample the lock in step,
+# and so a writer that has just lost gives the winner room to commit. The delay
+# grows to a cap; the budget, not the delay, decides when to stop.
+SQLITE_LOCK_FIRST_RETRY_SECONDS = 0.05
+SQLITE_LOCK_MAX_RETRY_SECONDS = 0.5
 _SQLITE_LOCK_PRIMARY_CODES = {
     getattr(sqlite3, "SQLITE_BUSY", 5),
     getattr(sqlite3, "SQLITE_LOCKED", 6),
@@ -84,10 +112,19 @@ def _write_with_lock_retry(
     Every recorder writer process writes the one per-capture database file, so
     each of them competes for the single SQLite write lock. A connection whose
     bounded wait expires reports "database is locked", which is contention, not
-    corruption. Each log line carries how long that attempt waited, because the
-    wait is the only measure of how close a capture is to losing this race.
+    corruption.
+
+    The retry runs against a clock, not a counter. It re-enters the race for as
+    long as ``SQLITE_WRITE_LOCK_BUDGET_SECONDS`` allows, and it stops as soon
+    as too little of that budget remains to finish another attempt. The total
+    wait is therefore never more than the budget, whatever one attempt costs on
+    the machine underneath.
     """
-    for attempt in range(len(SQLITE_LOCK_RETRY_DELAYS_SECONDS) + 1):
+    deadline = monotonic() + SQLITE_WRITE_LOCK_BUDGET_SECONDS
+    backoff = SQLITE_LOCK_FIRST_RETRY_SECONDS
+    attempt = 0
+    while True:
+        attempt += 1
         started_at = monotonic()
         try:
             result = write()
@@ -101,24 +138,27 @@ def _write_with_lock_retry(
             # 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):
+
+            # Begin another attempt only if the budget can pay for one. This
+            # test, not the count of attempts, is what bounds the total wait.
+            remaining = deadline - monotonic()
+            if remaining <= SQLITE_LOCK_ATTEMPT_CEILING_SECONDS:
                 logger.error(
-                    f"SQLite writer lock during {statement_label} did not clear: "
-                    f"attempt {attempt + 1} waited {waited:.2f}s and every retry "
-                    "is spent"
+                    f"SQLite writer lock during {statement_label} did not clear "
+                    f"within {SQLITE_WRITE_LOCK_BUDGET_SECONDS:.1f}s: attempt "
+                    f"{attempt} waited {waited:.2f}s and the budget is spent"
                 )
                 raise
 
-            delay = SQLITE_LOCK_RETRY_DELAYS_SECONDS[attempt]
+            # Never sleep away the room the next attempt needs.
+            delay = min(backoff, remaining - SQLITE_LOCK_ATTEMPT_CEILING_SECONDS)
             logger.warning(
                 f"SQLite writer lock during {statement_label} after waiting "
-                f"{waited:.2f}s; retrying in "
-                f"{delay:.2f}s ({attempt + 1}/"
-                f"{len(SQLITE_LOCK_RETRY_DELAYS_SECONDS)})"
+                f"{waited:.2f}s; retrying in {delay:.2f}s "
+                f"(attempt {attempt}, {remaining:.1f}s of budget left)"
             )
             sleep(delay)
-
-    raise AssertionError("unreachable SQLite write retry state")
+            backoff = min(backoff * 2, SQLITE_LOCK_MAX_RETRY_SECONDS)
 
 
 def _execute_insert_with_lock_retry(

openadapt_capture/recorder.py

@@ -46,7 +46,14 @@
 
 from openadapt_capture import platform, utils, video, window
 from openadapt_capture.config import config
-from openadapt_capture.db import create_db, crud, get_session_for_path
+from openadapt_capture.db import (
+    SQLITE_CAPTURE_JOURNAL_MODE,
+    close_capture_session,
+    create_db,
+    crud,
+    finalize_capture_database,
+    get_session_for_path,
+)
 from openadapt_capture.db.models import ActionEvent, Recording
 from openadapt_capture.desktop_capture import DesktopCaptureScope
 from openadapt_capture.extensions import synchronized_queue as sq
@@ -523,6 +530,19 @@ def __bool__(self):
 NUM_MEMORY_STATS_TO_LOG = 3
 STARTUP_WAIT_POLL_SECONDS = 0.1
 STARTUP_READY_TIMEOUT_SECONDS = 30.0
+
+# A writer announces readiness only after its first database write returns, so
+# the database's wait for the write lock is spent inside the deadline above.
+# Check the two against each other here rather than trusting a comment beside
+# either one: whichever a later change moves, the package refuses to import
+# with a budget that cannot fit.
+if crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS >= STARTUP_READY_TIMEOUT_SECONDS:
+    raise RuntimeError(
+        "The SQLite write-lock budget "
+        f"({crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS:.1f}s) must leave a writer "
+        "time to announce readiness within "
+        f"{STARTUP_READY_TIMEOUT_SECONDS:.1f}s."
+    )
 PRE_READY_TASK_JOIN_TIMEOUT_SECONDS = 2.0
 PROCESS_REAP_TIMEOUT_SECONDS = 2.0
 QUEUE_FEEDER_JOIN_TIMEOUT_SECONDS = 5.0
@@ -2181,7 +2201,8 @@ def create_recording(
         capture_config["capture_desktop"] = desktop_capture_info
     if capture_config:
         recording_data["config"] = capture_config
-    engine, Session = create_db(db_path)
+    # Several writer processes share this file: give it a write log.
+    engine, Session = create_db(db_path, journal_mode=SQLITE_CAPTURE_JOURNAL_MODE)
     session = Session()
     recording = crud.insert_recording(session, recording_data)
     logger.info(f"{recording=}")
@@ -3134,16 +3155,23 @@ def record(
             from openadapt_capture import plotting
 
             session = get_session_for_path(db_path)
-            plotting.plot_performance(
-                session,
-                recording,
-                save_dir=capture_dir,
-            )
+            try:
+                plotting.plot_performance(
+                    session,
+                    recording,
+                    save_dir=capture_dir,
+                )
+            finally:
+                close_capture_session(session)
 
         logger.info(f"Saved {recording_timestamp=}")
 
         session = get_session_for_path(db_path)
-        crud.post_process_events(session, recording)
+        try:
+            crud.post_process_events(session, recording)
+        finally:
+            # Release the file before the capture is finalized and sealed.
+            close_capture_session(session)
 
         # --- Profiling summary ---
         _profile_duration = time.perf_counter() - _profile_start
@@ -3712,6 +3740,11 @@ def _run_record(self) -> None:
                 self._last_source_ordinal = last_source_ordinal
             self.check_health()
             if self._ready_event.is_set():
+                # Every writer has exited by here, so fold the write log back
+                # into the database before anything reads or inventories it.
+                finalize_capture_database(
+                    os.path.join(self.capture_dir, "recording.db")
+                )
                 self._verify_completed_capture()
                 finalized_at = self._stage_completed_control_state()
                 self._seal_completed_capture()