From 9519e160c1b410573e19ad03b77a1f97260c0b15 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 28 Aug 2026 16:29:57 -0400 Subject: [PATCH] Close capture sessions before database finalization --- CHANGELOG.md | 4 + openadapt_capture/recorder.py | 201 ++++++++++++++++++---------------- tests/test_db_lock_retry.py | 26 +++++ 3 files changed, 137 insertions(+), 94 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d30c6e..43b926d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,6 +57,10 @@ _This release is published under the MIT License._ then exhaust its three retries, which killed the writer process or cost it the startup readiness deadline ([#122](https://github.com/OpenAdaptAI/openadapt-capture/pull/122)) +- **db**: Close each recorder database session before finalization. The setup + session kept one idle write-log connection open until cyclic garbage + collection ran, so macOS could fail the final journal change with + `database is locked` after an otherwise complete capture. - **release**: Let the changelog document the pending release candidate ([#110](https://github.com/OpenAdaptAI/openadapt-capture/pull/110), [`854f015`](https://github.com/OpenAdaptAI/openadapt-capture/commit/854f015fd994b0aa1992948791c18c2e6c571029)) diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 9780bd4..6b5f0a8 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -1313,57 +1313,59 @@ def write_events( logger.info(f"{event_type=} starting") signal.signal(signal.SIGINT, signal.SIG_IGN) session = get_session_for_path(db_path) - - if pre_callback: - state = pre_callback(session, recording) - else: - state = None - - num_processed = 0 - progress = None - started = False - while not terminate_processing.is_set() or not write_q.empty(): - if terminate_processing.is_set() and progress is None: - # if processing is over, create a progress bar - total_events = num_events.value - progress = tqdm( - total=total_events, - desc=f"Writing {event_type} events...", - unit="event", - colour="green", - dynamic_ncols=True, - ) - # update the progress bar with the number of events that have already - # been processed - for _ in range(num_processed): - progress.update() - if not started and not ready_after_first_event: - started_event.set() - started = True - try: - event = write_q.get_nowait() - except queue.Empty: - continue - assert event.type == event_type, (event_type, event) - state = write_fn(session, recording, event, perf_q, **(state or {})) - num_processed += 1 - if not started: - started_event.set() - started = True - with num_events.get_lock(): - if progress is not None: - if progress.total < num_events.value: - # update the total number of events in the progress bar - progress.total = num_events.value - progress.refresh() - progress.update() - logger.debug(f"{event_type=} written") - - if post_callback: - post_callback(state) - - if progress is not None: - progress.close() + try: + if pre_callback: + state = pre_callback(session, recording) + else: + state = None + + num_processed = 0 + progress = None + started = False + while not terminate_processing.is_set() or not write_q.empty(): + if terminate_processing.is_set() and progress is None: + # if processing is over, create a progress bar + total_events = num_events.value + progress = tqdm( + total=total_events, + desc=f"Writing {event_type} events...", + unit="event", + colour="green", + dynamic_ncols=True, + ) + # update the progress bar with the number of events that have already + # been processed + for _ in range(num_processed): + progress.update() + if not started and not ready_after_first_event: + started_event.set() + started = True + try: + event = write_q.get_nowait() + except queue.Empty: + continue + assert event.type == event_type, (event_type, event) + state = write_fn(session, recording, event, perf_q, **(state or {})) + num_processed += 1 + if not started: + started_event.set() + started = True + with num_events.get_lock(): + if progress is not None: + if progress.total < num_events.value: + # update the total number of events in the progress bar + progress.total = num_events.value + progress.refresh() + progress.update() + logger.debug(f"{event_type=} written") + + if post_callback: + post_callback(state) + + if progress is not None: + progress.close() + finally: + close_capture_session(session) logger.info(f"{event_type=} done") @@ -2071,22 +2073,25 @@ def performance_stats_writer( signal.signal(signal.SIGINT, signal.SIG_IGN) started = False session = get_session_for_path(db_path) - while not terminate_processing.is_set() or not perf_q.empty(): - if not started: - started_event.set() - started = True - try: - event_type, start_time, end_time = perf_q.get_nowait() - except queue.Empty: - continue + try: + while not terminate_processing.is_set() or not perf_q.empty(): + if not started: + started_event.set() + started = True + try: + event_type, start_time, end_time = perf_q.get_nowait() + except queue.Empty: + continue - crud.insert_perf_stat( - session, - recording, - event_type, - start_time, - end_time, - ) + crud.insert_perf_stat( + session, + recording, + event_type, + start_time, + end_time, + ) + finally: + close_capture_session(session) logger.info("Performance stats writer done") @@ -2118,34 +2123,37 @@ def memory_writer( started = False session = get_session_for_path(db_path) - while not terminate_processing.is_set(): - if not started: - started_event.set() - started = True - memory_usage_bytes = 0 - - memory_info = process.memory_info() - rss = memory_info.rss # Resident Set Size: non-swapped physical memory - memory_usage_bytes += rss - - for child in process.children(recursive=True): - # after ctrl+c, children may terminate before the next line - try: - child_memory_info = child.memory_info() - except psutil.NoSuchProcess: - continue - child_rss = child_memory_info.rss - rss += child_rss + try: + while not terminate_processing.is_set(): + if not started: + started_event.set() + started = True + memory_usage_bytes = 0 + + memory_info = process.memory_info() + rss = memory_info.rss # Resident Set Size: non-swapped physical memory + memory_usage_bytes += rss + + for child in process.children(recursive=True): + # after ctrl+c, children may terminate before the next line + try: + child_memory_info = child.memory_info() + except psutil.NoSuchProcess: + continue + child_rss = child_memory_info.rss + rss += child_rss - timestamp = utils.get_timestamp() + timestamp = utils.get_timestamp() - crud.insert_memory_stat( - session, - recording, - rss, - timestamp, - ) - time.sleep(1) # sample once per second instead of tight loop + crud.insert_memory_stat( + session, + recording, + rss, + timestamp, + ) + time.sleep(1) # sample once per second instead of tight loop + finally: + close_capture_session(session) logger.info("Memory writer done") @@ -2210,8 +2218,13 @@ def create_recording( # Several writer processes share this file: give it a write log. engine, Session = create_db(db_path, journal_mode=SQLITE_CAPTURE_JOURNAL_MODE) session = Session() - recording = crud.insert_recording(session, recording_data) - logger.info(f"{recording=}") + try: + recording = crud.insert_recording(session, recording_data) + logger.info(f"{recording=}") + session.expunge(recording) + finally: + session.close() + engine.dispose() return recording, db_path diff --git a/tests/test_db_lock_retry.py b/tests/test_db_lock_retry.py index fcb461a..be753b9 100644 --- a/tests/test_db_lock_retry.py +++ b/tests/test_db_lock_retry.py @@ -379,6 +379,32 @@ def test_finalizing_a_capture_removes_the_write_log(tmp_path): read_only.close() +def test_create_recording_releases_its_setup_connection(tmp_path, monkeypatch): + """The live recorder must not hold its first WAL connection until GC. + + ``record`` returns from its writer pipeline before it changes the database + back to the rollback journal. An attached setup session can keep an idle + pooled connection open until cyclic garbage collection runs, which makes + that final journal-mode change fail intermittently on macOS. + """ + monkeypatch.setattr(recorder.utils, "get_monitor_dims", lambda: (100, 80)) + monkeypatch.setattr(recorder.platform, "get_display_pixel_ratio", lambda: 1.0) + monkeypatch.setattr(recorder.utils, "get_double_click_distance_pixels", lambda: 5.0) + monkeypatch.setattr(recorder.utils, "get_double_click_interval_seconds", lambda: 0.5) + capture_dir = tmp_path / "capture" + capture_dir.mkdir() + + recording, db_path = recorder.create_recording( + "setup connection lifecycle", + str(capture_dir), + ) + + assert sa.inspect(recording).detached + db.finalize_capture_database(db_path) + assert not Path(f"{db_path}-wal").exists() + assert not Path(f"{db_path}-shm").exists() + + def _hold_the_write_lock(db_path, recording_id, release_from_another_thread=False): """Take the one write lock and hold it.""" competitor = sqlite3.connect(