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)) 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"