From ed9c78b32cb588e94bb03f80a0fc08ad0d7e1989 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 13:03:33 -0400 Subject: [PATCH 1/4] fix(recorder): survive SQLite writer contention in the video writer The video writer process is the last writer a recording starts, and the first thing it does is persist the video start time to the per-capture database. By then the screen, action, performance and memory writers are already committing to that same file, and SQLite gives them one write lock between them. Every other writer waits and retries for that lock. This one statement did not: it took the connection's bounded wait, and an expiry raised straight out of the writer's startup callback and killed the process. Recording then failed with "Recording tasks exited before readiness", the reason stayed in the dead child, and the parent reported an exit code. Three changes: Generalize the existing insert retry into one write-transaction helper and route the video start time update through it, so this path recovers from contention like the inserts do and still fails loud when the lock does not clear. Write it as a single UPDATE and report a missing row by the affected row count, rather than reading the row into a session the retry must unwind. Each attempt logs how long it waited, because that wait is the only measure of how close a capture came to losing the race. Carry a child's traceback back to its parent. WrapStdout now writes the formatted traceback to a synchronous queue before the child exits, and both the readiness wait and the child-failure error quote it. A child that dies during startup is now readable from the parent's log alone. Reap what a recording starts. record() force-stops surviving children and releases its queues in a finally, and Recorder does the same after joining, in case record() raised before its own teardown. A surviving child keeps the inherited standard output open and stops the parent's interpreter exiting at all, because multiprocessing joins live children at exit with no timeout; a queue feeder thread whose reader is gone blocks the same way. A failed recording must report the failure, not hang the program that ran it. Co-Authored-By: Claude Opus 5 --- openadapt_capture/db/crud.py | 95 +- openadapt_capture/recorder.py | 952 ++++++++++-------- openadapt_capture/utils.py | 49 +- tests/test_child_process_failure_reporting.py | 132 +++ tests/test_db_lock_retry.py | 98 +- 5 files changed, 896 insertions(+), 430 deletions(-) create mode 100644 tests/test_child_process_failure_reporting.py diff --git a/openadapt_capture/db/crud.py b/openadapt_capture/db/crud.py index a6d7fd9..5d1ec19 100644 --- a/openadapt_capture/db/crud.py +++ b/openadapt_capture/db/crud.py @@ -6,8 +6,8 @@ import json import sqlite3 -from time import sleep -from typing import Any, TypeVar +from time import monotonic, sleep +from typing import Any, Callable, TypeVar import sqlalchemy as sa from loguru import logger @@ -26,6 +26,7 @@ # Type variable for generic model queries BaseModelType = TypeVar("BaseModelType") +T = TypeVar("T") BATCH_SIZE = 1 @@ -69,36 +70,68 @@ def _is_sqlite_lock_error(error: sa.exc.OperationalError) -> bool: ) -def _execute_insert_with_lock_retry( +def _write_with_lock_retry( session: SaSession, - table: sa.Table, - to_insert: list[dict[str, Any]], -) -> sa.engine.Result: - """Commit one insert, with bounded recovery from SQLite writer contention.""" + write: Callable[[], T], + statement_label: str, +) -> T: + """Run one write transaction, recovering from bounded SQLite contention. + + ``write`` must perform every statement of the transaction and must be safe + to run again from the start: a rollback discards its partial work before + each 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. + """ for attempt in range(len(SQLITE_LOCK_RETRY_DELAYS_SECONDS) + 1): + started_at = monotonic() try: - result = session.execute(sa.insert(table), to_insert) + result = write() session.commit() return result except sa.exc.OperationalError as exc: if not _is_sqlite_lock_error(exc): raise + waited = monotonic() - started_at # 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): + 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" + ) raise delay = SQLITE_LOCK_RETRY_DELAYS_SECONDS[attempt] logger.warning( - "SQLite writer lock during insert; retrying in " + 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)})" ) sleep(delay) - raise AssertionError("unreachable SQLite insert retry state") + raise AssertionError("unreachable SQLite write retry state") + + +def _execute_insert_with_lock_retry( + session: SaSession, + table: sa.Table, + to_insert: list[dict[str, Any]], +) -> sa.engine.Result: + """Commit one insert, with bounded recovery from SQLite writer contention.""" + return _write_with_lock_retry( + session, + lambda: session.execute(sa.insert(table), to_insert), + "insert", + ) def _insert( @@ -334,22 +367,36 @@ def update_video_start_time( recording (Recording): The recording object to update. video_start_time (float): The new video start time to set. """ - # Find the recording by its timestamp - recording = session.query(Recording).filter(Recording.id == recording.id).first() - - if not recording: - logger.error(f"No recording found with id {recording.id}.") - return + recording_id = recording.id + + # This is the first thing the video writer process does, and it runs while + # the screen, action, performance and memory writer processes are already + # committing to the same database file. It therefore has to wait for the + # one SQLite write lock like every other writer, and it fails the whole + # recording if that wait is not bounded and retried. + # + # Read nothing first: a read that finds no row is answered below by the + # affected row count, and a session that has already read holds a + # transaction the retry would have to unwind. + session.rollback() + + def _write() -> int: + result = session.execute( + sa.update(Recording) + .where(Recording.id == recording_id) + .values(video_start_time=video_start_time) + ) + return result.rowcount - # Update the video start time - recording.video_start_time = video_start_time + updated_rows = _write_with_lock_retry( + session, + _write, + "video start time update", + ) - # the function is called from a different process which uses a different - # session from the one used to create the recording object, so we need to - # add the recording object to the session - session.add(recording) - # Commit the changes to the database - session.commit() + if not updated_rows: + logger.error(f"No recording found with id {recording_id}.") + return logger.info( f"Updated video start time for recording {recording.timestamp} to" diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 4f12bb3..e515b98 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -524,6 +524,8 @@ def __bool__(self): STARTUP_WAIT_POLL_SECONDS = 0.1 STARTUP_READY_TIMEOUT_SECONDS = 30.0 PRE_READY_TASK_JOIN_TIMEOUT_SECONDS = 2.0 +PROCESS_REAP_TIMEOUT_SECONDS = 2.0 +QUEUE_FEEDER_JOIN_TIMEOUT_SECONDS = 5.0 TERMINAL_FRAME_SEAL_TIMEOUT_SECONDS = 10.0 stop_sequence_detected = False @@ -544,11 +546,100 @@ def _run_task_fail_loud( terminate_processing.set() +def _drain_process_errors( + process_errors: Any | None, + collected: dict[str, str], +) -> dict[str, str]: + """Collect the tracebacks that child processes reported before they died. + + ``WrapStdout`` writes each traceback synchronously, so a process the parent + has already seen exit has finished reporting. Reading is therefore never a + race, and the first report from a task is the one that killed it. + """ + if process_errors is None: + return collected + while True: + try: + if process_errors.empty(): + break + task_name, detail = process_errors.get() + except (EOFError, OSError, ValueError): + break + collected.setdefault(task_name or "unnamed child process", detail) + return collected + + +def _describe_child_failure(task_name: str, detail: str | None) -> str: + """Render one child process failure with its own traceback, when reported.""" + if not detail: + return ( + f"{task_name} reported no traceback; it was stopped rather than " + "raised, or it died before it could report" + ) + return f"{task_name} raised:\n{detail.rstrip()}" + + +def _force_reap_processes(task_by_name: dict[str, Any]) -> list[str]: + """Stop every child process this recording started, and wait for each. + + A child that outlives the recording keeps the parent's inherited standard + output open and keeps the parent's own interpreter from exiting, because + multiprocessing joins live children at exit with no timeout. A recording + that has already failed must not also hang the program that ran it. + + Returns the names of any processes still alive after being killed. + """ + survivors: list[str] = [] + for task_name, task in task_by_name.items(): + if not isinstance(task, multiprocessing.process.BaseProcess): + continue + if task.exitcode is not None or not task.is_alive(): + continue + logger.warning(f"reaping {task_name!r}, which outlived the recording") + task.terminate() + task.join(timeout=PROCESS_REAP_TIMEOUT_SECONDS) + if task.is_alive(): + task.kill() + task.join(timeout=PROCESS_REAP_TIMEOUT_SECONDS) + if task.is_alive(): + survivors.append(task_name) + if survivors: + logger.error(f"child processes survived being killed: {sorted(survivors)}") + return survivors + + +def _release_queues(queues: list[Any]) -> None: + """Release the recording's cross-process queues so the parent can exit. + + Every queue keeps a background thread that feeds bytes into a pipe. When + the process at the far end dies, that pipe fills, the feeder thread blocks + on it forever, and the interpreter waits for that thread at exit. Closing + each queue and giving its feeder a bounded wait keeps the parent's exit + bounded too; a feeder that does not finish is abandoned rather than waited + on, since the recording is over and its buffered bytes have no reader. + """ + for pending in queues: + try: + pending.close() + except (OSError, ValueError): + continue + joiner = threading.Thread(target=pending.join_thread, daemon=True) + joiner.start() + joiner.join(timeout=QUEUE_FEEDER_JOIN_TIMEOUT_SECONDS) + if joiner.is_alive(): + logger.warning("abandoning a queue feeder that did not drain in time") + try: + pending.cancel_join_thread() + except (OSError, ValueError): + pass + + def _wait_for_tasks_started( task_by_name: dict[str, Any], task_started_events: dict[str, Any], terminate_processing: Any, task_errors: queue.Queue | None = None, + process_errors: Any | None = None, *, timeout: float = STARTUP_READY_TIMEOUT_SECONDS, ) -> bool: @@ -591,6 +682,9 @@ def _wait_for_tasks_started( ] if stopped_before_ready: logger.error(f"Recording tasks exited before readiness: {stopped_before_ready}") + reported = _drain_process_errors(process_errors, {}) + for task_name in stopped_before_ready: + logger.error(_describe_child_failure(task_name, reported.get(task_name))) terminate_processing.set() return False @@ -639,18 +733,30 @@ def _join_tasks( return lingering -def _raise_for_failed_processes(task_by_name: dict[str, Any]) -> None: - """Surface required child-process failures through the recording boundary.""" +def _raise_for_failed_processes( + task_by_name: dict[str, Any], + process_errors: Any | None = None, +) -> None: + """Surface required child-process failures through the recording boundary. + + An exit code alone says a child died, not why. Each child's own traceback + is quoted here so the reason crosses the recording boundary with the error, + rather than being left in whatever the child's stderr reached. + """ failures = { name: task.exitcode for name, task in task_by_name.items() if isinstance(task, multiprocessing.process.BaseProcess) and task.exitcode not in (None, 0) } if failures: + reported = _drain_process_errors(process_errors, {}) detail = ", ".join( f"{name} (exit code {exitcode})" for name, exitcode in sorted(failures.items()) ) - raise RuntimeError(f"Recording child process failed: {detail}") + error = RuntimeError(f"Recording child process failed: {detail}") + for name in sorted(failures): + add_exception_note(error, _describe_child_failure(name, reported.get(name))) + raise error def collect_stats(performance_snapshots: list[tracemalloc.Snapshot]) -> None: @@ -2453,6 +2559,7 @@ def record( window_owner: str | None = None, window_title: str | None = None, structural_observer: StructuralObserver | None = None, + child_registry: dict[str, Any] | None = None, ) -> int | None: """Record native screenshots, action events, and window events. @@ -2471,6 +2578,9 @@ def record( structural_observer: Optional injected accessibility observer. When omitted, the platform factory follows ``RECORD_STRUCTURAL_OBSERVATIONS``. + child_registry: Optional dict the caller owns, filled with every task + this recording starts. It lets the caller reap surviving child + processes even when this function raises. """ if config.RECORD_BROWSER_EVENTS: # Fail before encoder checks, display access, database creation, or any @@ -2611,475 +2721,508 @@ def record( perf_q = sq.SynchronizedQueue() if terminate_processing is None: terminate_processing = multiprocessing.Event() - task_by_name = {} + writer_queues = [ + screen_write_q, + action_write_q, + window_write_q, + browser_write_q, + video_write_q, + perf_q, + ] + # The caller keeps this dict so it can reap anything this recording leaves + # behind, even when record() itself raises. + task_by_name = {} if child_registry is None else child_registry task_started_events = {} task_errors: queue.Queue = queue.Queue() + # Writes are synchronous, so a child's traceback reaches this queue before + # the child exits, and reading it is never a race against the child. + process_errors = multiprocessing.SimpleQueue() _screen_timing = _ScreenTimingStats() # running stats, no unbounded list - # In window-scoped mode the screen reader emits the target window's - # bounds timeline itself; the active-window poller would record a - # DIFFERENT window (whichever is focused), so it stays off. - if config.RECORD_WINDOW_DATA and window_scope is None: - window_event_reader = threading.Thread( + # Nothing this recording starts may outlive it. A surviving child keeps + # the standard output it inherited open, and multiprocessing joins live + # children at interpreter exit with no timeout, so one leaked writer + # hangs the program that ran the recording instead of letting it report + # the failure. The same applies to each queue's feeder thread once the + # process at the far end is gone. + try: + + # In window-scoped mode the screen reader emits the target window's + # bounds timeline itself; the active-window poller would record a + # DIFFERENT window (whichever is focused), so it stays off. + if config.RECORD_WINDOW_DATA and window_scope is None: + window_event_reader = threading.Thread( + target=_run_task_fail_loud, + daemon=True, + args=( + "window_event_reader", + read_window_events, + ( + event_q, + terminate_processing, + recording, + task_started_events.setdefault("window_event_reader", threading.Event()), + ), + terminate_processing, + task_errors, + ), + ) + window_event_reader.start() + task_by_name["window_event_reader"] = window_event_reader + + screen_event_reader = threading.Thread( target=_run_task_fail_loud, daemon=True, args=( - "window_event_reader", - read_window_events, + "screen_event_reader", + read_screen_events, ( event_q, terminate_processing, recording, - task_started_events.setdefault("window_event_reader", threading.Event()), + task_started_events.setdefault("screen_event_reader", threading.Event()), + _screen_timing, + window_scope, + desktop_scope, + input_finished, + input_frame_boundary, + terminal_frame_finished, + terminal_frame_cancelled, ), terminate_processing, task_errors, ), ) - window_event_reader.start() - task_by_name["window_event_reader"] = window_event_reader + screen_event_reader.start() + task_by_name["screen_event_reader"] = screen_event_reader - screen_event_reader = threading.Thread( - target=_run_task_fail_loud, - daemon=True, - args=( - "screen_event_reader", - read_screen_events, - ( - event_q, + input_reader_args = ( + event_q, + terminate_processing, + recording, + task_started_events.setdefault("input_event_reader", threading.Event()), + window_scope or desktop_scope, + structural_observer, + input_finished, + input_frame_boundary, + terminal_frame_finished, + terminal_frame_cancelled, + ) + input_event_reader = threading.Thread( + target=_run_task_fail_loud, + daemon=True, + args=( + "input_event_reader", + read_input_events, + input_reader_args, terminate_processing, - recording, - task_started_events.setdefault("screen_event_reader", threading.Event()), - _screen_timing, - window_scope, - desktop_scope, - input_finished, - input_frame_boundary, - terminal_frame_finished, - terminal_frame_cancelled, + task_errors, ), - terminate_processing, - task_errors, - ), - ) - screen_event_reader.start() - task_by_name["screen_event_reader"] = screen_event_reader - - input_reader_args = ( - event_q, - terminate_processing, - recording, - task_started_events.setdefault("input_event_reader", threading.Event()), - window_scope or desktop_scope, - structural_observer, - input_finished, - input_frame_boundary, - terminal_frame_finished, - terminal_frame_cancelled, - ) - input_event_reader = threading.Thread( - target=_run_task_fail_loud, - daemon=True, - args=( - "input_event_reader", - read_input_events, - input_reader_args, - terminate_processing, - task_errors, - ), - ) - input_event_reader.start() - task_by_name["input_event_reader"] = input_event_reader - - if num_action_events is None: - num_action_events = multiprocessing.Value("i", 0) - if num_screen_events is None: - num_screen_events = multiprocessing.Value("i", 0) - if num_window_events is None: - num_window_events = multiprocessing.Value("i", 0) - if num_browser_events is None: - num_browser_events = multiprocessing.Value("i", 0) - if num_video_events is None: - num_video_events = multiprocessing.Value("i", 0) - - event_processor_args = ( - event_q, - screen_write_q, - action_write_q, - window_write_q, - browser_write_q, - video_write_q, - perf_q, - recording, - terminate_processing, - task_started_events.setdefault("event_processor", threading.Event()), - num_screen_events, - num_action_events, - num_window_events, - num_browser_events, - num_video_events, - producers_finished, - processing_aborted, - ) - event_processor = threading.Thread( - target=_run_task_fail_loud, - daemon=True, - args=( - "event_processor", - process_events, - event_processor_args, - terminate_processing, - task_errors, - ), - ) - event_processor.start() - task_by_name["event_processor"] = event_processor - - screen_event_writer = multiprocessing.Process( - target=utils.WrapStdout( - partial(write_events, ready_after_first_event=True) - ), - args=( - "screen", - partial(write_screen_event, record_images=bool(config.RECORD_IMAGES)), + ) + input_event_reader.start() + task_by_name["input_event_reader"] = input_event_reader + + if num_action_events is None: + num_action_events = multiprocessing.Value("i", 0) + if num_screen_events is None: + num_screen_events = multiprocessing.Value("i", 0) + if num_window_events is None: + num_window_events = multiprocessing.Value("i", 0) + if num_browser_events is None: + num_browser_events = multiprocessing.Value("i", 0) + if num_video_events is None: + num_video_events = multiprocessing.Value("i", 0) + + event_processor_args = ( + event_q, screen_write_q, - num_screen_events, - perf_q, - recording, - db_path, - terminate_writers, - task_started_events.setdefault("screen_event_writer", multiprocessing.Event()), - ), - ) - screen_event_writer.start() - task_by_name["screen_event_writer"] = screen_event_writer - - action_event_writer = multiprocessing.Process( - target=utils.WrapStdout(write_events), - args=( - "action", - write_action_event, action_write_q, - num_action_events, + window_write_q, + browser_write_q, + video_write_q, perf_q, recording, - db_path, - terminate_writers, - task_started_events.setdefault("action_event_writer", multiprocessing.Event()), - ), - ) - action_event_writer.start() - task_by_name["action_event_writer"] = action_event_writer - - if config.RECORD_WINDOW_DATA or window_scope is not None: - window_event_writer = multiprocessing.Process( - target=utils.WrapStdout( - partial(write_events, ready_after_first_event=True) - ), + terminate_processing, + task_started_events.setdefault("event_processor", threading.Event()), + num_screen_events, + num_action_events, + num_window_events, + num_browser_events, + num_video_events, + producers_finished, + processing_aborted, + ) + event_processor = threading.Thread( + target=_run_task_fail_loud, + daemon=True, args=( - "window", - write_window_event, - window_write_q, - num_window_events, - perf_q, - recording, - db_path, - terminate_writers, - task_started_events.setdefault("window_event_writer", multiprocessing.Event()), + "event_processor", + process_events, + event_processor_args, + terminate_processing, + task_errors, ), ) - window_event_writer.start() - task_by_name["window_event_writer"] = window_event_writer + event_processor.start() + task_by_name["event_processor"] = event_processor - if config.RECORD_VIDEO: - video_writer = multiprocessing.Process( + screen_event_writer = multiprocessing.Process( target=utils.WrapStdout( - partial(write_events, ready_after_first_event=True) + partial(write_events, ready_after_first_event=True), + "screen_event_writer", + process_errors, ), args=( - "screen/video", - write_video_event, - video_write_q, - num_video_events, + "screen", + partial(write_screen_event, record_images=bool(config.RECORD_IMAGES)), + screen_write_q, + num_screen_events, perf_q, recording, db_path, terminate_writers, - task_started_events.setdefault("video_writer", multiprocessing.Event()), - partial( - video_pre_callback, - video_dir=capture_dir, - # Window-scoped frames are the window's pixels, not the - # monitor's: size the stream from the initial frame. - frame_size=( - initial_window_frame.size if initial_window_frame is not None else None - ), - provision=video_provision, - timeout_seconds=config.VIDEO_FFMPEG_TIMEOUT_SECONDS, - ), - video_post_callback, + task_started_events.setdefault("screen_event_writer", multiprocessing.Event()), ), ) - video_writer.start() - task_by_name["video_writer"] = video_writer + screen_event_writer.start() + task_by_name["screen_event_writer"] = screen_event_writer - if config.RECORD_AUDIO: - audio_recorder = multiprocessing.Process( - target=utils.WrapStdout(record_audio), + action_event_writer = multiprocessing.Process( + target=utils.WrapStdout(write_events, "action_event_writer", process_errors), args=( + "action", + write_action_event, + action_write_q, + num_action_events, + perf_q, recording, db_path, - terminate_processing, - task_started_events.setdefault("audio_event_writer", multiprocessing.Event()), + terminate_writers, + task_started_events.setdefault("action_event_writer", multiprocessing.Event()), ), ) - audio_recorder.start() - task_by_name["audio_recorder"] = audio_recorder + action_event_writer.start() + task_by_name["action_event_writer"] = action_event_writer + + if config.RECORD_WINDOW_DATA or window_scope is not None: + window_event_writer = multiprocessing.Process( + target=utils.WrapStdout( + partial(write_events, ready_after_first_event=True), + "window_event_writer", + process_errors, + ), + args=( + "window", + write_window_event, + window_write_q, + num_window_events, + perf_q, + recording, + db_path, + terminate_writers, + task_started_events.setdefault("window_event_writer", multiprocessing.Event()), + ), + ) + window_event_writer.start() + task_by_name["window_event_writer"] = window_event_writer + + if config.RECORD_VIDEO: + video_writer = multiprocessing.Process( + target=utils.WrapStdout( + partial(write_events, ready_after_first_event=True), + "video_writer", + process_errors, + ), + args=( + "screen/video", + write_video_event, + video_write_q, + num_video_events, + perf_q, + recording, + db_path, + terminate_writers, + task_started_events.setdefault("video_writer", multiprocessing.Event()), + partial( + video_pre_callback, + video_dir=capture_dir, + # Window-scoped frames are the window's pixels, not the + # monitor's: size the stream from the initial frame. + frame_size=( + initial_window_frame.size if initial_window_frame is not None else None + ), + provision=video_provision, + timeout_seconds=config.VIDEO_FFMPEG_TIMEOUT_SECONDS, + ), + video_post_callback, + ), + ) + video_writer.start() + task_by_name["video_writer"] = video_writer - terminate_perf_event = multiprocessing.Event() - perf_stats_writer = multiprocessing.Process( - target=utils.WrapStdout(performance_stats_writer), - args=( - perf_q, - recording, - db_path, - terminate_perf_event, - task_started_events.setdefault("perf_stats_writer", multiprocessing.Event()), - ), - ) - perf_stats_writer.start() - task_by_name["perf_stats_writer"] = perf_stats_writer + if config.RECORD_AUDIO: + audio_recorder = multiprocessing.Process( + target=utils.WrapStdout(record_audio, "audio_recorder", process_errors), + args=( + recording, + db_path, + terminate_processing, + task_started_events.setdefault("audio_event_writer", multiprocessing.Event()), + ), + ) + audio_recorder.start() + task_by_name["audio_recorder"] = audio_recorder - if config.PLOT_PERFORMANCE: - record_pid = os.getpid() - mem_writer = multiprocessing.Process( - target=utils.WrapStdout(memory_writer), + terminate_perf_event = multiprocessing.Event() + perf_stats_writer = multiprocessing.Process( + target=utils.WrapStdout( + performance_stats_writer, "perf_stats_writer", process_errors + ), args=( + perf_q, recording, db_path, terminate_perf_event, - record_pid, - task_started_events.setdefault("mem_writer", multiprocessing.Event()), + task_started_events.setdefault("perf_stats_writer", multiprocessing.Event()), ), ) - mem_writer.start() - task_by_name["mem_writer"] = mem_writer - - if log_memory: - performance_snapshots = [] - _tracker = tracker.SummaryTracker() - tracemalloc.start() - collect_stats(performance_snapshots) - - # TODO: discard events until everything is ready - - global stop_sequence_detected - stop_sequence_detected = False - startup_ready = _wait_for_tasks_started( - task_by_name, - task_started_events, - terminate_processing, - task_errors, - ) - if startup_ready: - for _ in range(5): - logger.info("*" * 40) - logger.info("All readers and writers have started. Waiting for input events...") + perf_stats_writer.start() + task_by_name["perf_stats_writer"] = perf_stats_writer + + if config.PLOT_PERFORMANCE: + record_pid = os.getpid() + mem_writer = multiprocessing.Process( + target=utils.WrapStdout(memory_writer, "mem_writer", process_errors), + args=( + recording, + db_path, + terminate_perf_event, + record_pid, + task_started_events.setdefault("mem_writer", multiprocessing.Event()), + ), + ) + mem_writer.start() + task_by_name["mem_writer"] = mem_writer - if status_pipe: - status_pipe.send({"type": "record.started"}) + if log_memory: + performance_snapshots = [] + _tracker = tracker.SummaryTracker() + tracemalloc.start() + collect_stats(performance_snapshots) - try: - while not (stop_sequence_detected or terminate_processing.is_set()): - terminate_processing.wait(1) - except KeyboardInterrupt: - terminate_processing.set() - else: - logger.info("Tearing down recording after incomplete startup") - terminal_frame_cancelled.set() - input_frame_boundary.fail( - WindowCaptureError( - "recording startup ended before the native frame boundary was ready" - ) + # TODO: discard events until everything is ready + + global stop_sequence_detected + stop_sequence_detected = False + startup_ready = _wait_for_tasks_started( + task_by_name, + task_started_events, + terminate_processing, + task_errors, + process_errors, ) - terminate_processing.set() + if startup_ready: + for _ in range(5): + logger.info("*" * 40) + logger.info("All readers and writers have started. Waiting for input events...") - if status_pipe: - status_pipe.send({"type": "record.stopping"}) + if status_pipe: + status_pipe.send({"type": "record.started"}) - if log_memory: - collect_stats(performance_snapshots) - log_memory_usage(_tracker, performance_snapshots) + try: + while not (stop_sequence_detected or terminate_processing.is_set()): + terminate_processing.wait(1) + except KeyboardInterrupt: + terminate_processing.set() + else: + logger.info("Tearing down recording after incomplete startup") + terminal_frame_cancelled.set() + input_frame_boundary.fail( + WindowCaptureError( + "recording startup ended before the native frame boundary was ready" + ) + ) + terminate_processing.set() + + if status_pipe: + status_pipe.send({"type": "record.stopping"}) + + if log_memory: + collect_stats(performance_snapshots) + log_memory_usage(_tracker, performance_snapshots) + + pre_ready_timeout = None if startup_ready else PRE_READY_TASK_JOIN_TIMEOUT_SECONDS + lingering_tasks = _join_tasks( + task_by_name, + [ + "window_event_reader", + "input_event_reader", + "screen_event_reader", + "audio_recorder", + ], + timeout=pre_ready_timeout, + ) - pre_ready_timeout = None if startup_ready else PRE_READY_TASK_JOIN_TIMEOUT_SECONDS - lingering_tasks = _join_tasks( - task_by_name, - [ + journal_producers = { "window_event_reader", "input_event_reader", "screen_event_reader", - "audio_recorder", - ], - timeout=pre_ready_timeout, - ) - - journal_producers = { - "window_event_reader", - "input_event_reader", - "screen_event_reader", - } - lingering_producers = sorted(journal_producers.intersection(lingering_tasks)) - producer_shutdown_error = None - if lingering_producers: - producer_shutdown_error = RuntimeError( - "recording startup failed with live journal producers: " - + ", ".join(lingering_producers) + } + lingering_producers = sorted(journal_producers.intersection(lingering_tasks)) + producer_shutdown_error = None + if lingering_producers: + producer_shutdown_error = RuntimeError( + "recording startup failed with live journal producers: " + + ", ".join(lingering_producers) + ) + processing_aborted.set() + else: + # The processor can now drain every completed reservation. No producer + # can append a later event after it observes an empty journal. + producers_finished.set() + _join_tasks( + task_by_name, + ["event_processor"], + timeout=pre_ready_timeout, ) - processing_aborted.set() - else: - # The processor can now drain every completed reservation. No producer - # can append a later event after it observes an empty journal. - producers_finished.set() - _join_tasks( - task_by_name, - ["event_processor"], - timeout=pre_ready_timeout, - ) - # No writer can stop while the event processor can still enqueue work. - # Signal writer completion only after all producers have exited. - terminate_writers.set() - _join_tasks( - task_by_name, - [ - "screen_event_writer", - "action_event_writer", - "window_event_writer", - "video_writer", - ], - timeout=pre_ready_timeout, - ) + # No writer can stop while the event processor can still enqueue work. + # Signal writer completion only after all producers have exited. + terminate_writers.set() + _join_tasks( + task_by_name, + [ + "screen_event_writer", + "action_event_writer", + "window_event_writer", + "video_writer", + ], + timeout=pre_ready_timeout, + ) - terminate_perf_event.set() - _join_tasks( - task_by_name, - [ - "perf_stats_writer", - "mem_writer", - ], - timeout=pre_ready_timeout, - ) + terminate_perf_event.set() + _join_tasks( + task_by_name, + [ + "perf_stats_writer", + "mem_writer", + ], + timeout=pre_ready_timeout, + ) - if not task_errors.empty(): - task_name, task_error = task_errors.get_nowait() - add_exception_note(task_error, f"recording task {task_name!r} failed") + if not task_errors.empty(): + task_name, task_error = task_errors.get_nowait() + add_exception_note(task_error, f"recording task {task_name!r} failed") + if producer_shutdown_error is not None: + add_exception_note(task_error, str(producer_shutdown_error)) + raise task_error if producer_shutdown_error is not None: - add_exception_note(task_error, str(producer_shutdown_error)) - raise task_error - if producer_shutdown_error is not None: - raise producer_shutdown_error - _raise_for_failed_processes(task_by_name) - if window_scope is not None: - window_scope.assert_current() - elif desktop_scope is not None: - # Close the interval between the last captured frame and operator stop. - # A topology change in that interval still invalidates the session. - desktop_scope.assert_current(force=True) + raise producer_shutdown_error + _raise_for_failed_processes(task_by_name, process_errors) + if window_scope is not None: + window_scope.assert_current() + elif desktop_scope is not None: + # Close the interval between the last captured frame and operator stop. + # A topology change in that interval still invalidates the session. + desktop_scope.assert_current(force=True) + + if config.PLOT_PERFORMANCE and startup_ready: + from openadapt_capture import plotting + + session = get_session_for_path(db_path) + plotting.plot_performance( + session, + recording, + save_dir=capture_dir, + ) - if config.PLOT_PERFORMANCE and startup_ready: - from openadapt_capture import plotting + logger.info(f"Saved {recording_timestamp=}") session = get_session_for_path(db_path) - plotting.plot_performance( - session, - recording, - save_dir=capture_dir, - ) - - logger.info(f"Saved {recording_timestamp=}") - - session = get_session_for_path(db_path) - crud.post_process_events(session, recording) - - # --- Profiling summary --- - _profile_duration = time.perf_counter() - _profile_start - _profile_data = { - "duration_seconds": round(_profile_duration, 2), - "main_thread": _profile_is_main_thread, - "platform": sys.platform, - "python_version": sys.version, - "threads_started": list(task_by_name.keys()), - "thread_count": threading.active_count(), - "event_counts": { - "action": num_action_events.value, - "screen": num_screen_events.value, - "window": num_window_events.value, - "browser": num_browser_events.value, - "video": num_video_events.value, - }, - "screen_timing": {}, - "config": { - "RECORD_VIDEO": config.RECORD_VIDEO, - "RECORD_AUDIO": config.RECORD_AUDIO, - "RECORD_IMAGES": config.RECORD_IMAGES, - "RECORD_WINDOW_DATA": config.RECORD_WINDOW_DATA, - "RECORD_WINDOW_OWNER": window_owner or config.RECORD_WINDOW_OWNER, - "RECORD_WINDOW_TITLE": window_title or config.RECORD_WINDOW_TITLE, - "RECORD_BROWSER_EVENTS": config.RECORD_BROWSER_EVENTS, - "RECORD_FULL_VIDEO": config.RECORD_FULL_VIDEO, - "PLOT_PERFORMANCE": config.PLOT_PERFORMANCE, - "SCREEN_CAPTURE_FPS": config.SCREEN_CAPTURE_FPS, - }, - "capture_dir": capture_dir, - } - # Compute screen timing stats - if _screen_timing: - _profile_data["screen_timing"] = _screen_timing.to_dict() - - _profile_path = os.path.join(capture_dir, "profiling.json") - try: - import json as _json - - with open(_profile_path, "w") as _f: - _json.dump(_profile_data, _f, indent=2) - logger.info(f"Profiling saved to {_profile_path}") - - # Print compact summary - print("\n=== Recording Profile ===") - print(f"Duration: {_profile_duration:.1f}s") - print(f"Main thread: {_profile_is_main_thread}") - print(f"Threads started: {len(task_by_name)}") - for k, v in _profile_data["event_counts"].items(): - rate = v / _profile_duration if _profile_duration > 0 else 0 - print(f" {k}: {v} events ({rate:.1f}/s)") + crud.post_process_events(session, recording) + + # --- Profiling summary --- + _profile_duration = time.perf_counter() - _profile_start + _profile_data = { + "duration_seconds": round(_profile_duration, 2), + "main_thread": _profile_is_main_thread, + "platform": sys.platform, + "python_version": sys.version, + "threads_started": list(task_by_name.keys()), + "thread_count": threading.active_count(), + "event_counts": { + "action": num_action_events.value, + "screen": num_screen_events.value, + "window": num_window_events.value, + "browser": num_browser_events.value, + "video": num_video_events.value, + }, + "screen_timing": {}, + "config": { + "RECORD_VIDEO": config.RECORD_VIDEO, + "RECORD_AUDIO": config.RECORD_AUDIO, + "RECORD_IMAGES": config.RECORD_IMAGES, + "RECORD_WINDOW_DATA": config.RECORD_WINDOW_DATA, + "RECORD_WINDOW_OWNER": window_owner or config.RECORD_WINDOW_OWNER, + "RECORD_WINDOW_TITLE": window_title or config.RECORD_WINDOW_TITLE, + "RECORD_BROWSER_EVENTS": config.RECORD_BROWSER_EVENTS, + "RECORD_FULL_VIDEO": config.RECORD_FULL_VIDEO, + "PLOT_PERFORMANCE": config.PLOT_PERFORMANCE, + "SCREEN_CAPTURE_FPS": config.SCREEN_CAPTURE_FPS, + }, + "capture_dir": capture_dir, + } + # Compute screen timing stats if _screen_timing: - st = _profile_data["screen_timing"] + _profile_data["screen_timing"] = _screen_timing.to_dict() + + _profile_path = os.path.join(capture_dir, "profiling.json") + try: + import json as _json + + with open(_profile_path, "w") as _f: + _json.dump(_profile_data, _f, indent=2) + logger.info(f"Profiling saved to {_profile_path}") + + # Print compact summary + print("\n=== Recording Profile ===") + print(f"Duration: {_profile_duration:.1f}s") + print(f"Main thread: {_profile_is_main_thread}") + print(f"Threads started: {len(task_by_name)}") + for k, v in _profile_data["event_counts"].items(): + rate = v / _profile_duration if _profile_duration > 0 else 0 + print(f" {k}: {v} events ({rate:.1f}/s)") + if _screen_timing: + st = _profile_data["screen_timing"] + print( + f" screenshot: avg={st['screenshot_avg_ms']}ms " + f"max={st['screenshot_max_ms']}ms " + f"min={st['screenshot_min_ms']}ms" + ) print( - f" screenshot: avg={st['screenshot_avg_ms']}ms " - f"max={st['screenshot_max_ms']}ms " - f"min={st['screenshot_min_ms']}ms" + f"Config: WINDOW_DATA={config.RECORD_WINDOW_DATA} " + f"VIDEO={config.RECORD_VIDEO} " + f"PLOT_PERF={config.PLOT_PERFORMANCE} " + f"FPS={config.SCREEN_CAPTURE_FPS}" ) - print( - f"Config: WINDOW_DATA={config.RECORD_WINDOW_DATA} " - f"VIDEO={config.RECORD_VIDEO} " - f"PLOT_PERF={config.PLOT_PERFORMANCE} " - f"FPS={config.SCREEN_CAPTURE_FPS}" - ) - print("=========================\n") + print("=========================\n") - # Auto-send profiling via wormhole if requested - if send_profile: - _send_profiling_via_wormhole(_profile_path) - except Exception as exc: - logger.warning(f"Profiling save/send failed: {exc}") + # Auto-send profiling via wormhole if requested + if send_profile: + _send_profiling_via_wormhole(_profile_path) + except Exception as exc: + logger.warning(f"Profiling save/send failed: {exc}") - if terminate_recording is not None: - terminate_recording.set() + if terminate_recording is not None: + terminate_recording.set() - # TODO: consolidate terminate_recording and status_pipe - if status_pipe: - status_pipe.send({"type": "record.stopped"}) - return event_q.last_source_ordinal if window_scope is not None else None + # TODO: consolidate terminate_recording and status_pipe + if status_pipe: + status_pipe.send({"type": "record.stopped"}) + return event_q.last_source_ordinal if window_scope is not None else None + finally: + _force_reap_processes(task_by_name) + _release_queues(writer_queues) class Recorder: @@ -3199,6 +3342,9 @@ def __init__( self._status_thread: threading.Thread | None = None self._capture = None # lazy CaptureSession self._last_source_ordinal: int | None = None + # Every task record() starts, so teardown can reap a survivor even when + # record() raised before its own teardown ran. + self._child_registry: dict[str, Any] = {} self._worker_error: BaseException | None = None self._worker_error_lock = threading.Lock() self._structural_observer = structural_observer @@ -3556,6 +3702,7 @@ def _run_record(self) -> None: num_video_events=self._num_video_events, send_profile=self._send_profile, structural_observer=self._structural_observer, + child_registry=self._child_registry, ) if last_source_ordinal is not None: self._last_source_ordinal = last_source_ordinal @@ -3651,6 +3798,12 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None: self._terminate_processing.set() if self._record_thread is not None: self._record_thread.join() + # record() reaps its own children, but it can raise before it gets + # there. Leaving one alive keeps this process from exiting at all, + # because multiprocessing joins live children at exit without a + # timeout, so a failed recording would hang its caller instead of + # reporting the failure. + _force_reap_processes(self._child_registry) self._stopped_event.set() # ensure status thread exits if self._status_thread is not None: self._status_thread.join(timeout=5) @@ -3674,6 +3827,7 @@ def stop(self) -> None: self._terminate_processing.set() if self._record_thread is not None: self._record_thread.join() + _force_reap_processes(self._child_registry) try: self.check_health() finally: diff --git a/openadapt_capture/utils.py b/openadapt_capture/utils.py index 05fbc26..1bc6581 100644 --- a/openadapt_capture/utils.py +++ b/openadapt_capture/utils.py @@ -7,6 +7,7 @@ import sys import threading import time +import traceback from functools import wraps from typing import Any, Callable @@ -158,17 +159,53 @@ def get_double_click_distance_pixels() -> int: class WrapStdout: """Wrapper for multiprocessing process targets. - Ensures that stdout/stderr are properly redirected in child processes. - Copied from legacy OpenAdapt utils.py. + Ensures that stdout/stderr are properly redirected in child processes, and + reports why a child died to the process that started it. + + A child that raises leaves the parent nothing but an exit code. Its + traceback goes to the child's own stderr, which a test runner, a service + manager, or a spawn-based launcher may discard, reorder, or interleave past + recognition. Sending the formatted traceback back over ``error_sink`` puts + the cause in the parent's hands, so the parent can name it in the error it + raises instead of quoting a number. """ - def __init__(self, fn: Callable) -> None: - """Initialize with the function to wrap.""" + def __init__( + self, + fn: Callable, + task_name: str | None = None, + error_sink: Any | None = None, + ) -> None: + """Initialize with the function to wrap. + + Args: + fn: The process target to run. + task_name: Name the parent knows this child by. + error_sink: A ``multiprocessing.SimpleQueue`` the parent reads. It + is written synchronously, before the child exits, so a + traceback is never lost to a background flush. + """ self.fn = fn + self.task_name = task_name + self.error_sink = error_sink def __call__(self, *args: Any, **kwargs: Any) -> Any: - """Call the wrapped function.""" - return self.fn(*args, **kwargs) + """Call the wrapped function, reporting any failure to the parent.""" + try: + return self.fn(*args, **kwargs) + except BaseException: + self._report(traceback.format_exc()) + raise + + def _report(self, detail: str) -> None: + if self.error_sink is None: + return + try: + self.error_sink.put((self.task_name, detail)) + except BaseException: + # The sink is a diagnostic. Losing it must never replace the real + # failure with a reporting failure. + pass def trace(logger: Any) -> Callable: diff --git a/tests/test_child_process_failure_reporting.py b/tests/test_child_process_failure_reporting.py new file mode 100644 index 0000000..6ecbc16 --- /dev/null +++ b/tests/test_child_process_failure_reporting.py @@ -0,0 +1,132 @@ +"""Contracts for what a dying recorder child process tells its parent. + +A recording runs its writers as separate processes. When one of them raises +during startup, the parent used to learn only an exit code, and the reason lived +in whatever the child's stderr happened to reach. That cost a full diagnosis +cycle for a failure that was fully explained in the child. Worse, a child left +running after such a failure keeps the standard output it inherited open and +stops the parent's own interpreter from exiting, so the failure never gets +reported at all. + +These tests pin three things: + +- a child's traceback reaches the parent and lands in the error the parent + raises, +- a child that ignores a stop request is killed rather than waited on forever, +- a queue whose reader is gone does not hold the parent open. + +They need no display, listeners, or injected input. +""" + +from __future__ import annotations + +import multiprocessing +import time + +import pytest + +from openadapt_capture import utils +from openadapt_capture.extensions import synchronized_queue as sq +from openadapt_capture.recorder import ( + _describe_child_failure, + _drain_process_errors, + _force_reap_processes, + _raise_for_failed_processes, + _release_queues, +) + +JOIN_TIMEOUT = 30.0 + + +class _DistinctiveStartupError(RuntimeError): + """Named so the assertions cannot pass on some other failure.""" + + +def _raise_distinctively() -> None: + raise _DistinctiveStartupError("the video encoder never opened") + + +def _ignore_the_stop_request() -> None: + while True: + time.sleep(0.05) + + +def _fill_a_queue_and_leave(write_q: sq.SynchronizedQueue) -> None: + write_q.put(b"x" * 4096) + + +def test_a_child_traceback_reaches_the_parent(): + """The parent must be able to read why its child died, not just that it did.""" + error_sink = multiprocessing.SimpleQueue() + child = multiprocessing.Process( + target=utils.WrapStdout(_raise_distinctively, "video_writer", error_sink), + ) + child.start() + child.join(timeout=JOIN_TIMEOUT) + + assert child.exitcode == 1 + reported = _drain_process_errors(error_sink, {}) + assert "video_writer" in reported + assert "_DistinctiveStartupError" in reported["video_writer"] + assert "the video encoder never opened" in reported["video_writer"] + + +def test_the_raised_recording_error_quotes_the_child_traceback(): + """The traceback must cross the recording boundary, not stop at the child.""" + error_sink = multiprocessing.SimpleQueue() + child = multiprocessing.Process( + target=utils.WrapStdout(_raise_distinctively, "video_writer", error_sink), + ) + child.start() + child.join(timeout=JOIN_TIMEOUT) + + with pytest.raises(RuntimeError) as raised: + _raise_for_failed_processes({"video_writer": child}, error_sink) + + notes = "\n".join(getattr(raised.value, "__notes__", [])) + assert "video_writer (exit code 1)" in str(raised.value) + assert "_DistinctiveStartupError" in notes + assert "the video encoder never opened" in notes + + +def test_a_child_stopped_by_a_signal_is_reported_as_unexplained(): + """Say a child was stopped, rather than imply it reported nothing useful.""" + described = _describe_child_failure("mem_writer", None) + assert "mem_writer" in described + assert "stopped rather than raised" in described + + +def test_a_child_that_ignores_the_stop_request_is_killed(): + """A survivor would hold the parent's output open and block its exit.""" + child = multiprocessing.Process(target=_ignore_the_stop_request, daemon=True) + child.start() + try: + assert _force_reap_processes({"perf_stats_writer": child}) == [] + assert not child.is_alive() + finally: + if child.is_alive(): + child.kill() + child.join(timeout=JOIN_TIMEOUT) + + +def test_reaping_leaves_a_finished_child_alone(): + """Normal teardown already joined its writers; reaping must be a no-op.""" + child = multiprocessing.Process(target=time.sleep, args=(0,)) + child.start() + child.join(timeout=JOIN_TIMEOUT) + + assert _force_reap_processes({"action_event_writer": child}) == [] + assert child.exitcode == 0 + + +def test_a_queue_whose_reader_died_does_not_hold_the_parent_open(): + """The feeder thread must be released once nothing will ever read it.""" + write_q = sq.SynchronizedQueue() + reader = multiprocessing.Process(target=_fill_a_queue_and_leave, args=(write_q,)) + reader.start() + reader.join(timeout=JOIN_TIMEOUT) + write_q.put(b"y" * 4096) + + started_at = time.monotonic() + _release_queues([write_q]) + assert time.monotonic() - started_at < JOIN_TIMEOUT diff --git a/tests/test_db_lock_retry.py b/tests/test_db_lock_retry.py index cf7a9c9..f3b2783 100644 --- a/tests/test_db_lock_retry.py +++ b/tests/test_db_lock_retry.py @@ -9,7 +9,7 @@ from openadapt_capture import db from openadapt_capture.db import crud -from openadapt_capture.db.models import MemoryStat +from openadapt_capture.db.models import MemoryStat, Recording def _operational_error(message): @@ -105,3 +105,99 @@ def test_persistent_sqlite_writer_lock_still_fails(tmp_path, monkeypatch): locking_connection.close() session.close() engine.dispose() + + +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) + db_path = tmp_path / "recording.db" + engine, Session = db.create_db(str(db_path)) + setup_session = Session() + recording = crud.insert_recording( + setup_session, + { + "timestamp": 2.0, + "monitor_width": 100, + "monitor_height": 100, + "platform": "test", + "task_description": "Video start time under contention", + }, + ) + setup_session.close() + + video_writer_session = Session() + competitor = sqlite3.connect(db_path, timeout=0.01) + competitor.execute("BEGIN IMMEDIATE") + competitor.execute( + "UPDATE recording SET task_description = task_description WHERE id = ?", + (recording.id,), + ) + return engine, video_writer_session, competitor, recording + + +def test_video_start_time_recovers_from_a_transient_writer_lock(tmp_path, monkeypatch): + """The video writer must survive another writer holding the write lock. + + Without this recovery the recorder's ``video_writer`` process died of an + unhandled OperationalError inside its startup callback, before it could + announce readiness, and the whole recording failed. + """ + engine, session, competitor, recording = _recording_under_a_competing_writer( + tmp_path, monkeypatch + ) + retry_delays = [] + + def release_lock(delay): + retry_delays.append(delay) + competitor.commit() + + 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]] + stored = session.execute( + sa.select(Recording.video_start_time).where(Recording.id == recording.id) + ).scalar_one() + assert stored == pytest.approx(1234.5) + finally: + competitor.close() + session.close() + engine.dispose() + + +def test_video_start_time_fails_loud_under_a_held_lock(tmp_path, monkeypatch): + """A lock that never clears must surface, never be silently skipped.""" + engine, session, competitor, recording = _recording_under_a_competing_writer( + tmp_path, monkeypatch + ) + retry_delays = [] + monkeypatch.setattr(crud, "sleep", retry_delays.append) + try: + 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) + competitor.rollback() + stored = session.execute( + sa.select(Recording.video_start_time).where(Recording.id == recording.id) + ).scalar_one() + assert stored is None + finally: + competitor.close() + session.close() + engine.dispose() + + +def test_video_start_time_reports_a_missing_recording(tmp_path, monkeypatch): + """A recording row that is not there must be reported, not written blind.""" + monkeypatch.setattr(db, "SQLITE_BUSY_TIMEOUT_SECONDS", 0.01) + db_path = tmp_path / "recording.db" + engine, Session = db.create_db(str(db_path)) + session = Session() + absent = Recording(id=404, timestamp=3.0) + try: + crud.update_video_start_time(session, absent, 1234.5) + assert session.execute(sa.select(sa.func.count(Recording.id))).scalar_one() == 0 + finally: + session.close() + engine.dispose() From 3396bf21aa605cb6269d707cc2feb2fffb8de2a7 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 13:12:48 -0400 Subject: [PATCH 2/4] test(db): drive the real video writer through a held write lock The Windows qualification lane reproduces this failure only under load: it went green on both the fixed and the unfixed code in back-to-back dispatches on a quiet runner. Pin the contract where it is deterministic instead. This runs the production write_events body with the production video_pre_callback against a database whose write lock another connection holds, and asserts the writer announces readiness. Against the unfixed code it fails with "the video writer never announced readiness", which is the symptom the recorder reported in runs 33186627127 and 33189604580. Encoding a frame would need a real FFmpeg process, so the test supplies a preflighted provision and stops at the startup callback, which is where the writer died. Co-Authored-By: Claude Opus 5 --- tests/test_db_lock_retry.py | 94 ++++++++++++++++++++++++++++++++++++- 1 file changed, 92 insertions(+), 2 deletions(-) diff --git a/tests/test_db_lock_retry.py b/tests/test_db_lock_retry.py index f3b2783..0e41e90 100644 --- a/tests/test_db_lock_retry.py +++ b/tests/test_db_lock_retry.py @@ -2,14 +2,26 @@ from __future__ import annotations +import multiprocessing +import signal import sqlite3 +import threading +from functools import partial +from types import SimpleNamespace import pytest import sqlalchemy as sa -from openadapt_capture import db +from openadapt_capture import db, 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 ( + video_post_callback, + video_pre_callback, + write_events, + write_video_event, +) def _operational_error(message): @@ -126,7 +138,8 @@ def _recording_under_a_competing_writer(tmp_path, monkeypatch): setup_session.close() video_writer_session = Session() - competitor = sqlite3.connect(db_path, timeout=0.01) + # check_same_thread: one test releases this lock from the writer thread. + competitor = sqlite3.connect(db_path, timeout=0.01, check_same_thread=False) competitor.execute("BEGIN IMMEDIATE") competitor.execute( "UPDATE recording SET task_description = task_description WHERE id = ?", @@ -201,3 +214,80 @@ def test_video_start_time_reports_a_missing_recording(tmp_path, monkeypatch): finally: session.close() engine.dispose() + + +def test_the_real_video_writer_starts_through_a_writer_lock(tmp_path, monkeypatch): + """The recorder's own video writer body must survive contention at startup. + + This drives the production ``write_events`` target with the production + ``video_pre_callback``, against a database whose write lock another + connection holds. It isolates the startup callback: encoding a frame would + need a real FFmpeg process, and the callback is where the writer died. + + Run it in a thread rather than a process so the short busy timeout and the + lock release apply. The process boundary itself is covered in + tests/test_child_process_failure_reporting.py. + """ + monkeypatch.setattr(signal, "signal", lambda *_args, **_kwargs: None) + engine, session, competitor, recording = _recording_under_a_competing_writer( + tmp_path, monkeypatch + ) + session.close() + + released = [] + + def release_lock(delay): + released.append(delay) + competitor.commit() + + monkeypatch.setattr(crud, "sleep", release_lock) + + terminate = multiprocessing.Event() + started = multiprocessing.Event() + perf_queue = sq.SynchronizedQueue() + writer = threading.Thread( + target=write_events, + args=( + "screen/video", + write_video_event, + sq.SynchronizedQueue(), + multiprocessing.Value("i", 0), + perf_queue, + SimpleNamespace(id=recording.id, timestamp=recording.timestamp), + str(tmp_path / "recording.db"), + terminate, + started, + 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, + ), + ) + writer.start() + try: + assert started.wait(timeout=30.0), "the video writer never announced readiness" + finally: + terminate.set() + writer.join(timeout=30.0) + try: + assert not writer.is_alive(), "the video writer hung" + assert released == [crud.SQLITE_LOCK_RETRY_DELAYS_SECONDS[0]] + 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() + assert stored is not None + finally: + while not perf_queue.empty(): + perf_queue.get() + competitor.close() + engine.dispose() From 9d36dcdc5c1fff48500254ae74c03004d7dbc8fb Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 13:12:48 -0400 Subject: [PATCH 3/4] ci: return the hosted Windows live recorder lane to the release gate The lane moved to live-qualification.yml on 2026-08-28 because the video_writer child died during startup and blocked every release while it was open. That defect is fixed, so hosted Windows rejoins the required matrix beside hosted macOS and check_release_ci.py requires its job again. Keep the twelve-minute bound on the trial step. It is cheap, and a step that stops making progress should report in twelve minutes rather than hold a runner for the job's full thirty-five. Co-Authored-By: Claude Opus 5 --- .github/workflows/live-qualification.yml | 162 +----------------- .../workflows/production-qualification.yml | 24 +-- docs/DESIGN.md | 16 +- scripts/check_release_ci.py | 1 + 4 files changed, 16 insertions(+), 187 deletions(-) diff --git a/.github/workflows/live-qualification.yml b/.github/workflows/live-qualification.yml index 5025c58..f899d37 100644 --- a/.github/workflows/live-qualification.yml +++ b/.github/workflows/live-qualification.yml @@ -67,6 +67,7 @@ jobs: build-candidate: name: Build candidate distributions + if: vars.CAPTURE_SELF_HOSTED_QUALIFIED_RUNNERS == '1' runs-on: ubuntu-latest timeout-minutes: 15 steps: @@ -427,164 +428,3 @@ jobs: path: evidence/ if-no-files-found: error retention-days: 30 - - # Hosted Windows runs the same four live recorder tests as the required - # macOS lane, but it does not gate a release. On 2026-08-28 it failed the - # same way in both qualification dispatches of ecdd9c02 (runs 33186627127 - # and 33189604580): the video_writer child exits 1 during startup, the - # recorder reports "Recording tasks exited before readiness", and an - # orphaned child then holds the step open until the job timeout. The same - # four tests pass in test.yml's windows-latest leg on the same commit, so - # this reproduces only from an isolated wheel install. It is a real defect - # and it stays visible here until it is fixed, rather than blocking every - # release while it is open. - hosted-live-recorder-windows: - name: Hosted live recorder qualification (${{ matrix.os }}) - needs: build-candidate - strategy: - fail-fast: false - matrix: - os: [windows-latest] - runs-on: ${{ matrix.os }} - timeout-minutes: 35 - steps: - - name: Checkout the exact candidate source - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - - - name: Set up Python - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 - with: - python-version: "3.12" - - - name: Install exact uv - uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2 - with: - version: "0.11.29" - - - name: Download the exact candidate - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1 - with: - name: capture-candidate-${{ github.sha }} - path: dist - - - name: Install the exact wheel into an isolated environment - shell: bash - run: | - set -euo pipefail - qualification_root="${HOME}/capture-qualification-${GITHUB_RUN_ID}" - if [ "${RUNNER_OS}" = "Windows" ]; then - qualification_python="${qualification_root}/Scripts/python.exe" - workspace="$(cygpath -u "${GITHUB_WORKSPACE}")" - else - qualification_python="${qualification_root}/bin/python" - workspace="${GITHUB_WORKSPACE}" - fi - candidate_wheels=(dist/*.whl) - if [ "${#candidate_wheels[@]}" -ne 1 ]; then - echo "Expected exactly one candidate wheel." - exit 1 - fi - uv venv --clear --python 3.12 "${qualification_root}" - uv pip install --python "${qualification_python}" \ - "${candidate_wheels[0]}" \ - pytest==9.1.1 pytest-timeout==2.4.0 pynput==1.8.2 - echo "QUALIFICATION_PYTHON=${qualification_python}" >> "${GITHUB_ENV}" - echo "QUALIFICATION_WORKSPACE=${workspace}" >> "${GITHUB_ENV}" - - - name: Install the reviewed external video tools - shell: bash - run: | - set -euo pipefail - if [ "${RUNNER_OS}" = "Windows" ]; then - choco install ffmpeg -y --no-progress - else - brew install ffmpeg - fi - - - name: Require the reviewed external video tools - shell: bash - run: | - set -euo pipefail - mkdir -p evidence - command -v ffmpeg - command -v ffprobe - ffmpeg -version > evidence/ffmpeg-version.txt - ffprobe -version > evidence/ffprobe-version.txt - - # A GitHub-hosted runner exposes exactly one virtual display. The - # multiple-monitor contract is proven only by the self-hosted lanes in - # live-qualification.yml. This step still records the exact topology the - # trials ran against, and still fails on a topology that changes during - # qualification. - - name: Record a stable single-monitor desktop - shell: bash - run: | - set -euo pipefail - "${QUALIFICATION_PYTHON}" scripts/check_display_topology.py \ - --minimum-monitors 1 \ - --output evidence/display-topology.json - - # These four live tests drive the real recorder against the real display - # of the hosted runner: they start it, wait for a real first frame, prove - # the capture database is created, prove memory stays bounded, and prove - # a clean shutdown. They are named one by one, and the trial evidence is - # rejected below unless exactly four of them ran, so a rename or a - # deletion fails the gate instead of silently shrinking it. - # - # The three input-injection tests of this file are deliberately NOT here. - # Injected input does not reach the native low-level hooks in a hosted - # runner session, measured on all three hosted operating systems on - # 2026-08-28. live-qualification.yml runs them on a qualified host. - # A failing trial can leave a recorder child process holding this step's - # output pipe open, and the step then sits until the job timeout rather - # than reporting the failure. Three trials take under three minutes on a - # healthy runner, so bound the step itself. - - name: Run three counted live recorder trials - timeout-minutes: 12 - shell: bash - env: - OPENADAPT_CAPTURE_PRODUCTION_QUALIFICATION: "1" - run: | - set -euo pipefail - cd "${HOME}" - for trial in 1 2 3; do - echo "::group::${RUNNER_OS} trial ${trial}" - "${QUALIFICATION_PYTHON}" -m pytest \ - "${QUALIFICATION_WORKSPACE}/tests/test_performance.py" \ - -m slow -v --timeout=300 --import-mode=importlib \ - -k "test_initial_frame_ready_without_input or test_shutdown_time or test_memory_bounded or test_db_file_created" \ - "--junitxml=${QUALIFICATION_WORKSPACE}/evidence/trial-${trial}-${{ matrix.os }}.xml" - echo "::endgroup::" - done - - - name: Reject skipped, incomplete, or shrunken qualification trials - shell: bash - run: | - set -euo pipefail - for trial in 1 2 3; do - python scripts/check_junit_no_skips.py \ - --expected-tests 4 \ - "evidence/trial-${trial}-${{ matrix.os }}.xml" - done - - - name: Aggregate the counted trial evidence - shell: bash - run: | - set -euo pipefail - python scripts/aggregate_qualification_trials.py \ - --os "${{ matrix.os }}" \ - --candidate-sha "${GITHUB_SHA}" \ - --expected-trials 3 \ - "evidence/trial-1-${{ matrix.os }}.xml" \ - "evidence/trial-2-${{ matrix.os }}.xml" \ - "evidence/trial-3-${{ matrix.os }}.xml" \ - --output "evidence/qualification-summary-${{ matrix.os }}.json" - - - name: Upload hosted live recorder evidence - if: always() - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 - with: - name: capture-hosted-live-${{ matrix.os }}-${{ github.sha }} - path: evidence/ - if-no-files-found: error - retention-days: 30 diff --git a/.github/workflows/production-qualification.yml b/.github/workflows/production-qualification.yml index 99391cc..7582cfa 100644 --- a/.github/workflows/production-qualification.yml +++ b/.github/workflows/production-qualification.yml @@ -32,16 +32,10 @@ # boundary; # - that exact wheel installs, imports, exposes its CLI, and uninstalls on # clean Linux, macOS, and Windows machines; -# - the real recorder starts against a real display on hosted macOS, -# produces a real first frame, creates its capture database, holds -# memory bounded, and shuts down cleanly, in three counted trials. -# -# The same lane on hosted Windows is in live-qualification.yml, not here. -# It reproduces a video_writer startup failure that test.yml's -# windows-latest leg does not hit, and test.yml is already required on the -# same exact commit, so Windows live recorder coverage stays in the required -# evidence while that defect is open. The comment above that lane has the -# run ids and the failure signature. +# - the real recorder starts against a real display on hosted macOS and +# Windows, produces a real first frame, creates its capture database, +# holds memory bounded, and shuts down cleanly, three counted trials per +# operating system. # # scripts/check_release_ci.py additionally requires a successful test.yml run # on the same exact commit, which is where the complete headless suite and the @@ -176,7 +170,7 @@ jobs: strategy: fail-fast: false matrix: - os: [macos-latest] + os: [macos-latest, windows-latest] runs-on: ${{ matrix.os }} timeout-minutes: 35 steps: @@ -267,10 +261,10 @@ jobs: # Injected input does not reach the native low-level hooks in a hosted # runner session, measured on all three hosted operating systems on # 2026-08-28. live-qualification.yml runs them on a qualified host. - # A failing trial can leave a recorder child process holding this step's - # output pipe open, and the step then sits until the job timeout rather - # than reporting the failure. Three trials take under three minutes on a - # healthy runner, so bound the step itself. + # Three trials take under three minutes on a healthy runner, so the step + # carries its own bound well under the job's. A step that stops making + # progress reports a failure in twelve minutes instead of holding the + # runner for the job's full thirty-five. - name: Run three counted live recorder trials timeout-minutes: 12 shell: bash diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 7917f80..f5bdc9f 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -235,9 +235,10 @@ commit. Every job runs on a GitHub-hosted runner. It must: - build and validate one wheel and sdist; - install and uninstall that exact wheel in a clean environment on Linux, macOS, and Windows; -- start the real recorder against the real display of a hosted macOS runner, - prove a real first frame, prove the capture database, prove bounded memory, - and prove a clean shutdown, in three counted trials with no skip; +- start the real recorder against the real display of a hosted macOS and a + hosted Windows runner, prove a real first frame, prove the capture database, + prove bounded memory, and prove a clean shutdown, in three counted trials per + operating system with no skip; - record the exact display topology each trial ran against; and - retain machine-readable test and topology evidence. @@ -253,14 +254,7 @@ commit. Missing, stale, skipped, partial, or failed evidence blocks publication. `live-qualification.yml` runs the interactive lanes that need a physical desktop. It runs weekly and on demand, and it is not a release gate. -It also carries the same live recorder lane on hosted Windows. That lane -reproduces a `video_writer` startup failure that `test.yml` does not hit on -the same commit, so it does not gate a release while that defect is open. -`test.yml` runs the same four tests on `windows-latest` and is required on -the exact commit, so Windows live recorder coverage stays in the required -evidence. - -The self-hosted lanes prove, and the release gate therefore does not prove: +Those lanes prove, and the release gate therefore does not prove: - that global input injected through the operating system reaches the native listeners and is written into the capture; diff --git a/scripts/check_release_ci.py b/scripts/check_release_ci.py index e976e49..068d054 100644 --- a/scripts/check_release_ci.py +++ b/scripts/check_release_ci.py @@ -26,6 +26,7 @@ "Clean candidate wheel (macos-latest)", "Clean candidate wheel (windows-latest)", "Hosted live recorder qualification (macos-latest)", + "Hosted live recorder qualification (windows-latest)", } ) ACTIVE_STATES = frozenset({"queued", "in_progress", "waiting", "pending", "requested"}) From b9d8f4104891d1b6743addba6af196fad37ab540 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 13:17:34 -0400 Subject: [PATCH 4/4] fix(recorder): report a child traceback where Python 3.10 can see it BaseException.add_note arrived in Python 3.11 and this package supports 3.10, where a note is attached to nothing and printed nowhere. The child traceback belonged in the error message all along, since that is what a log, a test report, and a stack trace all show. Caught by the 3.10 leg of tests/test_child_process_failure_reporting.py. Co-Authored-By: Claude Opus 5 --- openadapt_capture/recorder.py | 16 ++++++++++------ tests/test_child_process_failure_reporting.py | 10 ++++++---- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index e515b98..03dee1a 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -740,8 +740,11 @@ def _raise_for_failed_processes( """Surface required child-process failures through the recording boundary. An exit code alone says a child died, not why. Each child's own traceback - is quoted here so the reason crosses the recording boundary with the error, - rather than being left in whatever the child's stderr reached. + goes into the error message, so the reason crosses the recording boundary + with the error rather than being left in whatever the child's stderr + reached. It is the message and not an exception note because + ``BaseException.add_note`` arrived in Python 3.11 and this package supports + 3.10, where a note is discarded without a word. """ failures = { name: task.exitcode @@ -753,10 +756,11 @@ def _raise_for_failed_processes( detail = ", ".join( f"{name} (exit code {exitcode})" for name, exitcode in sorted(failures.items()) ) - error = RuntimeError(f"Recording child process failed: {detail}") - for name in sorted(failures): - add_exception_note(error, _describe_child_failure(name, reported.get(name))) - raise error + message = "\n".join( + [f"Recording child process failed: {detail}"] + + [_describe_child_failure(name, reported.get(name)) for name in sorted(failures)] + ) + raise RuntimeError(message) def collect_stats(performance_snapshots: list[tracemalloc.Snapshot]) -> None: diff --git a/tests/test_child_process_failure_reporting.py b/tests/test_child_process_failure_reporting.py index 6ecbc16..bb1018d 100644 --- a/tests/test_child_process_failure_reporting.py +++ b/tests/test_child_process_failure_reporting.py @@ -83,10 +83,12 @@ def test_the_raised_recording_error_quotes_the_child_traceback(): with pytest.raises(RuntimeError) as raised: _raise_for_failed_processes({"video_writer": child}, error_sink) - notes = "\n".join(getattr(raised.value, "__notes__", [])) - assert "video_writer (exit code 1)" in str(raised.value) - assert "_DistinctiveStartupError" in notes - assert "the video encoder never opened" in notes + # The message, not an exception note: notes need Python 3.11, and this + # package supports 3.10, where a note is discarded without a word. + reported = str(raised.value) + assert "video_writer (exit code 1)" in reported + assert "_DistinctiveStartupError" in reported + assert "the video encoder never opened" in reported def test_a_child_stopped_by_a_signal_is_reported_as_unexplained():