From b38b9fdb39ce5a0545629a3c1559803b1c388959 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 17:29:07 -0500 Subject: [PATCH] perf(performance-monitor): move blocking sqlite3 I/O off the event loop `PerformanceMonitor` used the fully synchronous `sqlite3` driver directly inside six `async def` methods. Each call ran connect + statement + commit (which fsyncs) + close on the event loop, so nothing else on the loop could be scheduled for the duration of the disk write. This is on a live request path: `api/v1/router.py:69` imports the monitor and awaits `record_metric()` at `:1165` and `:1183`, and `record_metric` calls `_store_metric` unconditionally. Each method's database work is now a nested synchronous function dispatched via `await asyncio.to_thread(...)`. Statement text, transaction boundaries, return values and error handling are unchanged; only the thread the work runs on changes. Connections are created and closed inside each call, so there is no shared handle a caller could tear down mid-flight. Sites moved off-loop: _store_metric, _store_alert, _basic_cleanup, get_current_performance_summary, _get_recent_metrics_summary, _store_benchmark_result `_init_database` is left alone: it is a synchronous method called from `__init__`, so it never runs on the event loop. Tests: 113 pre-existing pass with zero edits, plus 6 new in `TestSqliteDoesNotBlockEventLoop`. Non-vacuity was proven by behavioural mutation: replacing all six `to_thread` dispatches with direct calls fails exactly the 3 heartbeat tests and no others. The other 3 new tests are preserved-behaviour guards -- they pass under both old and new code and exist to pin behaviour the change must not regress, not to prove the change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/performance_monitor.py | 31 +++- tests/unit/test_performance_monitor.py | 142 ++++++++++++++++++ 2 files changed, 166 insertions(+), 7 deletions(-) diff --git a/src/youtube_extension/backend/services/performance_monitor.py b/src/youtube_extension/backend/services/performance_monitor.py index 26f6f1e64..1c89fccb0 100644 --- a/src/youtube_extension/backend/services/performance_monitor.py +++ b/src/youtube_extension/backend/services/performance_monitor.py @@ -270,7 +270,8 @@ async def record_metric(self, async def _store_metric(self, metric: PerformanceMetric): """Store metric in database""" - try: + + def _write() -> None: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() @@ -290,6 +291,8 @@ async def _store_metric(self, metric: PerformanceMetric): conn.commit() conn.close() + try: + await asyncio.to_thread(_write) except Exception as e: logger.error(f"Failed to store metric in database: {e}") @@ -353,7 +356,8 @@ async def _trigger_alert(self, alert: PerformanceAlert): async def _store_alert(self, alert: PerformanceAlert): """Store alert in database""" - try: + + def _write() -> None: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() @@ -375,6 +379,8 @@ async def _store_alert(self, alert: PerformanceAlert): conn.commit() conn.close() + try: + await asyncio.to_thread(_write) except Exception as e: logger.error(f"Failed to store alert in database: {e}") @@ -498,7 +504,8 @@ async def _cleanup_old_metrics(self): async def _basic_cleanup(self): """Basic cleanup fallback when service is not available""" - try: + + def _purge() -> None: cutoff_date = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat() conn = sqlite3.connect(self.db_path) @@ -518,8 +525,9 @@ async def _basic_cleanup(self): conn.commit() conn.close() + try: + await asyncio.to_thread(_purge) logger.info("Basic cleanup completed successfully") - except Exception as e: logger.error(f"Error in basic cleanup: {e}") @@ -562,7 +570,8 @@ async def trigger_manual_cleanup(self) -> dict[str, Any]: async def get_current_performance_summary(self) -> dict[str, dict[str, float]]: """Get current performance summary (last hour averages)""" - try: + + def _query() -> dict[str, dict[str, float]]: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() @@ -582,6 +591,8 @@ async def get_current_performance_summary(self) -> dict[str, dict[str, float]]: conn.close() return dict(results) + try: + return await asyncio.to_thread(_query) except Exception as e: logger.error(f"Error getting performance summary: {e}") return {} @@ -658,7 +669,8 @@ async def _get_target_progress(self) -> dict[str, Any]: async def _get_recent_metrics_summary(self) -> dict[str, Any]: """Get recent metrics summary""" - try: + + def _query() -> dict[str, Any]: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() @@ -683,6 +695,8 @@ async def _get_recent_metrics_summary(self) -> dict[str, Any]: conn.close() return metrics_summary + try: + return await asyncio.to_thread(_query) except Exception as e: logger.error(f"Error getting recent metrics summary: {e}") return {} @@ -798,7 +812,8 @@ async def run_performance_benchmark(self, async def _store_benchmark_result(self, result: dict[str, Any]): """Store benchmark result in database""" - try: + + def _write() -> None: conn = sqlite3.connect(self.db_path) cursor = conn.cursor() @@ -824,6 +839,8 @@ async def _store_benchmark_result(self, result: dict[str, Any]): conn.commit() conn.close() + try: + await asyncio.to_thread(_write) except Exception as e: logger.error(f"Failed to store benchmark result: {e}") diff --git a/tests/unit/test_performance_monitor.py b/tests/unit/test_performance_monitor.py index 682c578f8..b5fcf603c 100644 --- a/tests/unit/test_performance_monitor.py +++ b/tests/unit/test_performance_monitor.py @@ -17,14 +17,20 @@ sys.modules['psutil'] = _real_psutil sys.modules.pop('youtube_extension.backend.services.performance_monitor', None) +import asyncio +import contextlib +import sqlite3 +import time from collections import deque from datetime import datetime, timezone from pathlib import Path +from unittest.mock import patch import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src")) +from youtube_extension.backend.services import performance_monitor as perf_mod from youtube_extension.backend.services.performance_monitor import ( PerformanceAlert, PerformanceMetric, @@ -1002,3 +1008,139 @@ def my_sync_func(): result = my_sync_func() assert result == "hello" + + +# =========================================================================== +# SQLite I/O must not block the event loop +# =========================================================================== + + +class TestSqliteDoesNotBlockEventLoop: + """`sqlite3` is fully synchronous: connect + INSERT + commit (fsync) + close. + + `PerformanceMonitor.record_metric` runs on live request paths + (`api/v1/router.py:1165` and `:1183`), so every one of those calls used to + stall the event loop for the duration of the write. These tests measure + whether an independent coroutine keeps getting scheduled while the database + work is in flight. + """ + + _DB_SECONDS = 0.15 + _HEARTBEAT_INTERVAL = 0.01 + + @pytest.fixture + def monitor(self, tmp_path): + return PerformanceMonitor(db_path=str(tmp_path / "perf.db")) + + def _slow_connect(self, real_connect): + """Wrap sqlite3.connect so the *synchronous* work takes a visible time.""" + + def _connect(*args, **kwargs): + time.sleep(self._DB_SECONDS) + return real_connect(*args, **kwargs) + + return _connect + + async def _count_heartbeats_during(self, coro): + ticks = 0 + stop = False + + async def heartbeat(): + nonlocal ticks + while not stop: + await asyncio.sleep(self._HEARTBEAT_INTERVAL) + ticks += 1 + + beat = asyncio.create_task(heartbeat()) + await asyncio.sleep(0) # let the heartbeat reach its first await first + try: + result = await coro + finally: + stop = True + beat.cancel() + with contextlib.suppress(asyncio.CancelledError): + await beat + return result, ticks + + async def test_store_metric_does_not_block_event_loop(self, monitor): + metric = PerformanceMetric( + component="api", + metric_name="api_response_time", + value=12.5, + timestamp=datetime.now(timezone.utc), + unit="ms", + tags={}, + ) + real = perf_mod.sqlite3.connect + with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)): + _, ticks = await self._count_heartbeats_during(monitor._store_metric(metric)) + assert ticks > 0, "event loop was blocked for the whole sqlite write" + + async def test_record_metric_does_not_block_event_loop(self, monitor): + """The public request-path entry point, not just the private writer.""" + real = perf_mod.sqlite3.connect + with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)): + _, ticks = await self._count_heartbeats_during( + monitor.record_metric("api", "api_response_time", 12.5) + ) + assert ticks > 0, "record_metric blocked the event loop" + + async def test_read_path_does_not_block_event_loop(self, monitor): + real = perf_mod.sqlite3.connect + with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)): + _, ticks = await self._count_heartbeats_during( + monitor.get_current_performance_summary() + ) + assert ticks > 0, "the read path blocked the event loop" + + # --- preserved-behaviour guards (pass under old and new code alike) ----- + + async def test_store_metric_still_writes_the_row(self, monitor): + metric = PerformanceMetric( + component="api", + metric_name="api_response_time", + value=42.0, + timestamp=datetime.now(timezone.utc), + unit="ms", + tags={"route": "/x"}, + ) + await monitor._store_metric(metric) + + conn = sqlite3.connect(monitor.db_path) + rows = conn.execute( + "SELECT component, metric_name, value, unit FROM performance_metrics" + ).fetchall() + conn.close() + assert rows == [("api", "api_response_time", 42.0, "ms")] + + async def test_summary_reflects_stored_metrics(self, monitor): + for value in (10.0, 20.0): + await monitor._store_metric( + PerformanceMetric( + component="api", + metric_name="api_response_time", + value=value, + timestamp=datetime.now(timezone.utc), + unit="ms", + tags={}, + ) + ) + summary = await monitor.get_current_performance_summary() + assert summary["api"]["api_response_time"] == 15.0 + + async def test_store_metric_swallows_database_errors(self, monitor): + """Error handling must still absorb failures rather than propagate.""" + metric = PerformanceMetric( + component="api", + metric_name="api_response_time", + value=1.0, + timestamp=datetime.now(timezone.utc), + unit="ms", + tags={}, + ) + + def _boom(*_args, **_kwargs): + raise sqlite3.OperationalError("disk I/O error") + + with patch.object(perf_mod.sqlite3, "connect", _boom): + await monitor._store_metric(metric) # must not raise