From a955d5b801d4433866e1616877a85c6af0ce8055 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 14:48:20 -0400 Subject: [PATCH 1/2] fix(db): bound the SQLite write-lock wait by time, not by attempts Two hosted Windows lanes failed on 0d14af17 for one reason. A writer's write transaction starved for about twenty-two seconds and the retry policy ran out of attempts while it was still starving. In the qualification lane video_writer spent 22.8s on three attempts at its video start-time update, never announced readiness, and failed the thirty-second startup deadline. In the tests lane mem_writer spent 21.9s on three attempts while the screen writer drained its screenshot backlog, exhausted them, and exited 1. Readiness had already succeeded there: the contention is not confined to startup, it is whenever one writer holds the file. The policy was a count of attempts with an inherited per-attempt wait, and a count of attempts bounds nothing. The runners measured one attempt at about seven seconds against a nominal five-second busy timeout, so three attempts cost twenty-two seconds -- too long to fit the readiness deadline, and only three samples of a lock that was busy nearly all the time. Spend a declared time budget instead. _write_with_lock_retry now runs against a clock: it re-enters the race for as long as SQLITE_WRITE_LOCK_BUDGET_SECONDS allows and stops as soon as too little budget remains to finish another attempt, so the total wait is never more than the budget whatever one attempt costs underneath. The busy timeout drops to 0.5s, which turns the same budget into tens of chances at the lock rather than three. recorder.py refuses to import if the budget cannot fit inside its readiness deadline. Reduce the contention as well as bound it. A live capture now keeps a write-ahead log. Under the default rollback journal every commit creates, syncs and deletes a journal file beside the capture; on Windows that churn is scanned by the filesystem filter driver, one screenshot row costs about half a second, and a writer draining a backlog holds the single write lock at nearly full duty cycle. A write log appends instead, and readers stop blocking writers. Measured here on six concurrent writers, it cut median commit latency from 0.72ms to 0.13ms and the worst wait from 2.3s to 0.57s. The write log is a property of a live capture, not of every database this package creates: create_db takes the journal mode as an argument, and only the recorder passes it. A fixture built by scripts/generate_synthetic_captures.py has one writer and its bytes are unchanged. finalize_capture_database folds the log back into the file before a capture is verified and sealed. build_artifact_manifest 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 capture sealed with its sidecars present would fail its own validation later. It fails loud rather than sealing one. record() also now closes the two sessions it opened and never closed, which otherwise held the file open past the end of the recording. tests/test_db_lock_retry.py gains the shape that was missing. The earlier regression test drove one writer against a lock held by one other connection, which passes against the defect. The new test starts the real memory, performance-stats and video writer bodies together against one database whose lock is already held, which is what production does, and it fails against the replaced policy because three attempts is three chances. Two more tests measure rather than assert the bound: one times a real attempt against the declared attempt ceiling, the other times the whole helper against the declared budget and fails both when it overruns and when it gives up early. Co-Authored-By: Claude Opus 5 --- openadapt_capture/db/__init__.py | 127 ++++++++++- openadapt_capture/db/crud.py | 76 +++++-- openadapt_capture/recorder.py | 49 ++++- tests/test_control.py | 78 ++++++- tests/test_db_lock_retry.py | 365 ++++++++++++++++++++++++++++++- 5 files changed, 657 insertions(+), 38 deletions(-) diff --git a/openadapt_capture/db/__init__.py b/openadapt_capture/db/__init__.py index ba157a3..0a53950 100644 --- a/openadapt_capture/db/__init__.py +++ b/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,7 +166,7 @@ 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. @@ -131,6 +174,10 @@ def create_db(db_path: str, echo: bool = False) -> tuple: 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.""" diff --git a/openadapt_capture/db/crud.py b/openadapt_capture/db/crud.py index 5d1ec19..df5a98e 100644 --- a/openadapt_capture/db/crud.py +++ b/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( diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 03dee1a..33a6f50 100644 --- a/openadapt_capture/recorder.py +++ b/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() diff --git a/tests/test_control.py b/tests/test_control.py index aa8ed84..00d926d 100644 --- a/tests/test_control.py +++ b/tests/test_control.py @@ -24,6 +24,7 @@ from openadapt_capture import control from openadapt_capture import recorder as recorder_module +from openadapt_capture.capture import CaptureSession from openadapt_capture.config import RecordingConfig, config_override from openadapt_capture.control import ( CaptureControlError, @@ -33,7 +34,12 @@ status_recording, stop_recording, ) -from openadapt_capture.db import create_db, crud +from openadapt_capture.db import ( + SQLITE_CAPTURE_JOURNAL_MODE, + create_db, + crud, + finalize_capture_database, +) from openadapt_capture.terminal import verify_capture_artifacts @@ -65,9 +71,12 @@ def _create_minimal_recording( capture_dir: Path, *, browser_messages: list[object] | None = None, + journal_mode: str | None = None, ) -> None: capture_dir.mkdir(parents=True, exist_ok=True) - engine, session_factory = create_db(str(capture_dir / "recording.db")) + engine, session_factory = create_db( + str(capture_dir / "recording.db"), journal_mode=journal_mode + ) session = session_factory() try: recording = crud.insert_recording( @@ -920,3 +929,68 @@ def test_browser_events_increment_the_persisted_count() -> None: assert counters[3].value == 1 assert queues[3].qsize() == 1 + + +def test_a_write_logged_capture_verifies_and_seals(tmp_path: Path) -> None: + """A capture recorded with a write log must still seal and revalidate. + + A live capture keeps a write-ahead log, which removes the writer-lock + contention that killed recorder children. That log lives in sidecar files + beside the database, and ``build_artifact_manifest`` inventories every + regular file under the capture directory. A capture that sealed while its + sidecars existed would fail its own validation later, because the + shared-memory file is created and removed by whoever opens the database + next. ``finalize_capture_database`` folds the log back in first. + + The live recorder lanes cover this on a real desktop. This covers it + without a display, listeners, or injected input. + """ + capture_dir = tmp_path / "capture" + _create_minimal_recording(capture_dir, journal_mode=SQLITE_CAPTURE_JOURNAL_MODE) + db_path = capture_dir / "recording.db" + + output = io.BytesIO() + Image.new("RGB", (2, 2), "black").save(output, format="PNG") + png = output.getvalue() + engine, session_factory = create_db(str(db_path)) + session = session_factory() + try: + recording = session.query(crud.Recording).one() + crud.insert_screenshot( + session, + recording, + recording.timestamp + 1, + { + "source_ordinal": 1, + "png_data": png, + "png_sha256": hashlib.sha256(png).hexdigest(), + }, + ) + assert (db_path.parent / f"{db_path.name}-wal").exists(), ( + "the capture kept no write log to fold back" + ) + finally: + session.close() + engine.dispose() + + finalize_capture_database(str(db_path)) + assert not (db_path.parent / f"{db_path.name}-wal").exists() + assert not (db_path.parent / f"{db_path.name}-shm").exists() + + recorder = recorder_module.Recorder( + str(capture_dir), + capture_video=False, + capture_images=True, + ) + recorder._num_screen_events.value = 1 + recorder._last_source_ordinal = 1 + recorder._verify_completed_capture() + recorder._stage_completed_control_state() + recorder._seal_completed_capture() + + # _seal_completed_capture validates the seal it just wrote, so reaching + # here already proves the sealed inventory matches the directory. Re-read + # it the way a later consumer does, and prove no sidecar came back. + CaptureSession.validate_sealed(str(capture_dir)) + assert not (db_path.parent / f"{db_path.name}-wal").exists() + assert not (db_path.parent / f"{db_path.name}-shm").exists() diff --git a/tests/test_db_lock_retry.py b/tests/test_db_lock_retry.py index 0e41e90..774291f 100644 --- a/tests/test_db_lock_retry.py +++ b/tests/test_db_lock_retry.py @@ -3,20 +3,25 @@ from __future__ import annotations import multiprocessing +import os import signal import sqlite3 import threading from functools import partial +from pathlib import Path +from time import monotonic, sleep from types import SimpleNamespace import pytest import sqlalchemy as sa -from openadapt_capture import db, video +from openadapt_capture import db, recorder, video from openadapt_capture.db import crud from openadapt_capture.db.models import MemoryStat, Recording from openadapt_capture.extensions import synchronized_queue as sq from openadapt_capture.recorder import ( + memory_writer, + performance_stats_writer, video_post_callback, video_pre_callback, write_events, @@ -34,6 +39,11 @@ def _operational_error(message): def _locked_memory_stat_database(tmp_path, monkeypatch): monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.01) + # Keep the budget and the attempt ceiling proportional to that timeout so a + # test of a lock that never clears finishes in a fraction of a second. The + # production values are measured by their own tests below. + monkeypatch.setattr(crud, "SQLITE_WRITE_LOCK_BUDGET_SECONDS", 1.0) + monkeypatch.setattr(crud, "SQLITE_LOCK_ATTEMPT_CEILING_SECONDS", 0.05) db_path = tmp_path / "recording.db" engine, Session = db.create_db(str(db_path)) setup_session = Session() @@ -92,7 +102,7 @@ def release_lock(delay): monkeypatch.setattr(crud, "sleep", release_lock) try: crud._insert(session, event_data, MemoryStat) - assert retry_delays == [crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS[0]] + assert retry_delays, "the insert did not retry the held lock" assert session.query(MemoryStat).count() == 1 finally: locking_connection.close() @@ -110,7 +120,7 @@ def test_persistent_sqlite_writer_lock_still_fails(tmp_path, monkeypatch): 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) + assert retry_delays, "the insert did not retry the held lock" locking_connection.rollback() assert session.query(MemoryStat).count() == 0 finally: @@ -122,6 +132,11 @@ def test_persistent_sqlite_writer_lock_still_fails(tmp_path, monkeypatch): def _recording_under_a_competing_writer(tmp_path, monkeypatch): """Create a capture database whose one write lock a second writer holds.""" monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.01) + # Keep the budget and the attempt ceiling proportional to that timeout so a + # test of a lock that never clears finishes in a fraction of a second. The + # production values are measured by their own tests below. + monkeypatch.setattr(crud, "SQLITE_WRITE_LOCK_BUDGET_SECONDS", 1.0) + monkeypatch.setattr(crud, "SQLITE_LOCK_ATTEMPT_CEILING_SECONDS", 0.05) db_path = tmp_path / "recording.db" engine, Session = db.create_db(str(db_path)) setup_session = Session() @@ -167,7 +182,7 @@ def release_lock(delay): monkeypatch.setattr(crud, "sleep", release_lock) try: crud.update_video_start_time(session, recording, 1234.5) - assert retry_delays == [crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS[0]] + assert retry_delays, "the update did not retry the held lock" stored = session.execute( sa.select(Recording.video_start_time).where(Recording.id == recording.id) ).scalar_one() @@ -189,7 +204,7 @@ def test_video_start_time_fails_loud_under_a_held_lock(tmp_path, monkeypatch): with pytest.raises(sa.exc.OperationalError, match="database is locked"): crud.update_video_start_time(session, recording, 1234.5) - assert retry_delays == list(crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS) + assert retry_delays, "the update did not retry the held lock" competitor.rollback() stored = session.execute( sa.select(Recording.video_start_time).where(Recording.id == recording.id) @@ -281,7 +296,7 @@ def release_lock(delay): writer.join(timeout=30.0) try: assert not writer.is_alive(), "the video writer hung" - assert released == [crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS[0]] + assert released, "the video writer did not retry the held lock" stored = db.get_session_for_path(str(tmp_path / "recording.db")).execute( sa.select(Recording.video_start_time).where(Recording.id == recording.id) ).scalar_one() @@ -291,3 +306,341 @@ def release_lock(delay): perf_queue.get() competitor.close() engine.dispose() + + +def _capture_database_with_a_recording(tmp_path, task_description): + """Create a per-capture database holding one recording row.""" + db_path = tmp_path / "recording.db" + engine, Session = db.create_db( + str(db_path), journal_mode=db.SQLITE_CAPTURE_JOURNAL_MODE + ) + setup_session = Session() + recording = crud.insert_recording( + setup_session, + { + "timestamp": 4.0, + "monitor_width": 100, + "monitor_height": 100, + "platform": "test", + "task_description": task_description, + }, + ) + detached = SimpleNamespace(id=recording.id, timestamp=recording.timestamp) + setup_session.close() + return engine, db_path, detached + + +def test_a_live_capture_database_keeps_a_write_log(tmp_path): + """The contention this module bounds is first of all reduced. + + Under the default rollback journal every commit creates, syncs and deletes + a journal file beside the capture. On Windows that churn costs about half a + second per screenshot row, and a writer draining a backlog then holds the + single write lock at nearly full duty cycle while the other writers starve. + """ + engine, db_path, _ = _capture_database_with_a_recording(tmp_path, "journal mode") + try: + with engine.connect() as connection: + mode = connection.exec_driver_sql("PRAGMA journal_mode").scalar() + sync = connection.exec_driver_sql("PRAGMA synchronous").scalar() + assert str(mode).lower() == "wal" + # 1 is NORMAL. A write log makes it safe against a process crash. + assert sync == 1 + finally: + engine.dispose() + + +def test_finalizing_a_capture_removes_the_write_log(tmp_path): + """A sealed capture carries no sidecar files. + + ``terminal.build_artifact_manifest`` inventories every regular file under + the capture directory, and the shared-memory file is created and removed by + whoever opens the database next. A sealed capture that listed one would + fail its own validation later. + """ + engine, db_path, recording = _capture_database_with_a_recording(tmp_path, "seal") + session = db.get_session_for_path(str(db_path)) + crud.insert_memory_stat(session, recording, 1, 1) + assert Path(f"{db_path}-wal").exists(), "the live capture kept no write log" + db.close_capture_session(session) + engine.dispose() + + db.finalize_capture_database(str(db_path)) + + assert not Path(f"{db_path}-wal").exists() + assert not Path(f"{db_path}-shm").exists() + # The finalized file must still read back through the read-only path the + # recorder verifies and seals it with. + read_only = sqlite3.connect(f"{db_path.resolve().as_uri()}?mode=ro", uri=True) + try: + assert read_only.execute("PRAGMA quick_check").fetchall() == [("ok",)] + assert read_only.execute("SELECT COUNT(*) FROM memory_stat").fetchone() == (1,) + finally: + read_only.close() + + +def _hold_the_write_lock(db_path, recording_id, release_from_another_thread=False): + """Take the one write lock and hold it.""" + competitor = sqlite3.connect( + db_path, timeout=0.01, check_same_thread=not release_from_another_thread + ) + competitor.execute("BEGIN IMMEDIATE") + competitor.execute( + "UPDATE recording SET task_description = task_description WHERE id = ?", + (recording_id,), + ) + return competitor + + +def _memory_stat_row(recording): + return { + "recording_id": recording.id, + "recording_timestamp": recording.timestamp, + "memory_usage_bytes": 1, + "timestamp": 1, + } + + +def test_one_locked_attempt_costs_less_than_the_declared_ceiling(tmp_path): + """Measure the number the budget's upper bound rests on. + + ``_write_with_lock_retry`` refuses to begin an attempt unless + ``SQLITE_LOCK_ATTEMPT_CEILING_SECONDS`` of budget remains. That is what + makes the budget an upper bound on the total wait rather than an estimate + of it, and it holds only while one real attempt against a real held lock + costs less than the ceiling. Measure it with the production busy timeout + instead of asserting it: this is the value that used to be inherited, and + a hosted Windows runner measured one attempt at about seven seconds. + """ + engine, db_path, recording = _capture_database_with_a_recording(tmp_path, "ceiling") + session = db.get_session_for_path(str(db_path)) + competitor = _hold_the_write_lock(db_path, recording.id) + try: + started = monotonic() + with pytest.raises(sa.exc.OperationalError): + session.execute(sa.insert(MemoryStat), [_memory_stat_row(recording)]) + session.commit() + attempt_cost = monotonic() - started + finally: + session.rollback() + competitor.rollback() + competitor.close() + session.close() + engine.dispose() + + assert attempt_cost <= crud.SQLITE_LOCK_ATTEMPT_CEILING_SECONDS, ( + f"one attempt against a held lock cost {attempt_cost:.2f}s, over the " + f"declared {crud.SQLITE_LOCK_ATTEMPT_CEILING_SECONDS:.2f}s ceiling" + ) + + +def test_the_total_wait_never_runs_past_the_declared_budget(tmp_path, monkeypatch): + """A lock that never clears must cost the budget and not a second more. + + The defect this replaces expressed the wait as a count of attempts, so the + total was whatever three attempts happened to cost: about twenty-two + seconds on a hosted Windows runner, against a thirty-second readiness + deadline. Measure the total against the declared budget instead. + """ + # Scale the busy timeout, the attempt ceiling and the budget together, so + # the loop behaves exactly as it does in production and the test still + # finishes in about two seconds. The production attempt cost is measured by + # test_one_locked_attempt_costs_less_than_the_declared_ceiling. + monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(crud, "SQLITE_LOCK_ATTEMPT_CEILING_SECONDS", 0.2) + monkeypatch.setattr(crud, "SQLITE_WRITE_LOCK_BUDGET_SECONDS", 2.0) + engine, db_path, recording = _capture_database_with_a_recording(tmp_path, "budget") + session = db.get_session_for_path(str(db_path)) + competitor = _hold_the_write_lock(db_path, recording.id) + try: + started = monotonic() + with pytest.raises(sa.exc.OperationalError, match="database is locked"): + crud._insert(session, _memory_stat_row(recording), MemoryStat) + elapsed = monotonic() - started + finally: + competitor.rollback() + competitor.close() + session.close() + engine.dispose() + + assert elapsed <= crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS, ( + f"a permanently held lock cost {elapsed:.2f}s, over the declared " + f"{crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS:.1f}s budget" + ) + # ... and it is spent, not abandoned. The replaced policy stopped after + # three attempts, which against this lock is under a tenth of the budget. + # That is the half of the contract a count of attempts cannot express. + assert elapsed >= ( + crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS + - 2 * crud.SQLITE_LOCK_ATTEMPT_CEILING_SECONDS + ), f"the helper gave up after {elapsed:.2f}s, long before its budget" + + +def test_the_write_lock_budget_fits_the_readiness_deadline(): + """The bound is only worth having if a writer can still announce readiness. + + ``recorder`` refuses to import when this does not hold, so this test states + the same contract where a reader of the database code can see it. + """ + assert ( + crud.SQLITE_WRITE_LOCK_BUDGET_SECONDS < recorder.STARTUP_READY_TIMEOUT_SECONDS + ) + + +def test_concurrent_writers_all_survive_a_busy_write_lock(tmp_path, monkeypatch): + """Start the real writers together against one already-busy database. + + This is the shape the recorder actually fails in, and the shape the earlier + regression test missed: production drives several writer bodies at the one + per-capture database at the same time, and the lock they compete for is + already held by whichever of them got there first. A test that drives a + single writer against a synthetic lock cannot see a writer starve. + + Two hosted Windows runs of 0d14af17 recorded what starving looks like. In + the qualification lane ``video_writer`` spent 22.8s on three attempts at + its start-time update, never announced readiness, and failed the 30s + deadline. In the tests lane ``mem_writer`` spent 21.9s on three attempts + while the screen writer drained its backlog, exhausted them and exited 1. + + The busy timeout, the attempt ceiling and the budget are scaled together + here so the whole test runs in a couple of seconds. What is NOT scaled is + the ratio that decides the outcome: the lock stays held for many multiples + of what one attempt costs, so a writer survives only by re-entering the + race, not by waiting longer in any single attempt. Against the replaced + policy every writer here dies, because three attempts is three chances. + """ + monkeypatch.setattr(signal, "signal", lambda *_args, **_kwargs: None) + monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.05) + monkeypatch.setattr(crud, "SQLITE_LOCK_ATTEMPT_CEILING_SECONDS", 0.2) + monkeypatch.setattr(crud, "SQLITE_WRITE_LOCK_BUDGET_SECONDS", 6.0) + hold_seconds = 1.5 + + engine, db_path, recording = _capture_database_with_a_recording( + tmp_path, "concurrent writers" + ) + competitor = _hold_the_write_lock( + db_path, recording.id, release_from_another_thread=True + ) + + terminate = multiprocessing.Event() + perf_queue = sq.SynchronizedQueue() + for index in range(4): + perf_queue.put(("screen", float(index), float(index) + 1.0)) + + started_events = { + name: multiprocessing.Event() + for name in ("mem_writer", "perf_stats_writer", "video_writer") + } + failures: dict[str, BaseException] = {} + + def _guard(name, target): + def _run(): + try: + target() + except BaseException as error: # noqa: BLE001 - reported below + failures[name] = error + + return _run + + writers = [ + threading.Thread( + target=_guard( + "mem_writer", + partial( + memory_writer, + recording, + str(db_path), + terminate, + os.getpid(), + started_events["mem_writer"], + ), + ) + ), + threading.Thread( + target=_guard( + "perf_stats_writer", + partial( + performance_stats_writer, + perf_queue, + recording, + str(db_path), + terminate, + started_events["perf_stats_writer"], + ), + ) + ), + threading.Thread( + target=_guard( + "video_writer", + partial( + write_events, + "screen/video", + write_video_event, + sq.SynchronizedQueue(), + multiprocessing.Value("i", 0), + sq.SynchronizedQueue(), + recording, + str(db_path), + terminate, + started_events["video_writer"], + partial( + video_pre_callback, + video_dir=str(tmp_path), + frame_size=(64, 48), + provision=video.FFmpegProvision( + executable="ffmpeg-is-never-run-by-this-test", + codec="mpeg4", + pixel_format="yuv420p", + muxer="mp4", + source="test", + ), + timeout_seconds=5.0, + ), + video_post_callback, + ), + ) + ), + ] + + releaser = threading.Timer(hold_seconds, competitor.commit) + releaser.start() + for writer in writers: + writer.start() + try: + # The video writer announces readiness from its startup callback, which + # is the write that starved on the qualification runner. + assert started_events["video_writer"].wait(timeout=20.0), ( + "the video writer never announced readiness through a busy lock" + ) + # The stats writers announce before their first write, so prove they + # survived it instead: this is exactly how mem_writer died. + reader = db.get_session_for_path(str(db_path)) + try: + deadline = monotonic() + 20.0 + while monotonic() < deadline: + reader.rollback() # start a fresh read of what is committed + committed = reader.execute( + sa.select(sa.func.count(MemoryStat.id)) + ).scalar_one() + if committed: + break + sleep(0.05) + else: # pragma: no cover - only reached on a regression + pytest.fail("the memory writer committed nothing through a busy lock") + finally: + db.close_capture_session(reader) + finally: + releaser.cancel() + terminate.set() + for writer in writers: + writer.join(timeout=30.0) + competitor.close() + engine.dispose() + while not perf_queue.empty(): + perf_queue.get() + + assert not failures, f"a writer died under contention: {failures}" + for writer in writers: + assert not writer.is_alive(), "a writer hung under contention" + for name, event in started_events.items(): + assert event.is_set(), f"{name} never announced readiness" From 87fb799d3399720196e4078d1c38df9316c893e2 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 14:49:03 -0400 Subject: [PATCH 2/2] docs(changelog): record the write-lock budget fix Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 94d9b42..0d30c6e 100644 --- a/CHANGELOG.md +++ b/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))