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/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..03dee1a 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,34 @@ 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 + 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 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}") + 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: @@ -2453,6 +2563,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 +2582,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 +2725,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 +3346,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 +3706,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 +3802,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 +3831,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/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"}) diff --git a/tests/test_child_process_failure_reporting.py b/tests/test_child_process_failure_reporting.py new file mode 100644 index 0000000..bb1018d --- /dev/null +++ b/tests/test_child_process_failure_reporting.py @@ -0,0 +1,134 @@ +"""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) + + # 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(): + """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..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 +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): @@ -105,3 +117,177 @@ 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() + # 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 = ?", + (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() + + +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()