From fdcd0d31ab5f8522e08750ae9e63836ae306f982 Mon Sep 17 00:00:00 2001 From: timmy Date: Mon, 10 Aug 2026 22:54:08 +0000 Subject: [PATCH] perf: reuse unchanged live snapshot values (Closes #519) --- src/live_snapshot_store.py | 37 +++++++++++++++++++++++++++ src/main.py | 38 ++++++++++++++++++++++++++-- tests/test_live_snapshot.py | 42 +++++++++++++++++++++++++++++++ tests/test_live_snapshot_store.py | 21 ++++++++++++++++ 4 files changed, 136 insertions(+), 2 deletions(-) diff --git a/src/live_snapshot_store.py b/src/live_snapshot_store.py index 5bf8807..0fe155f 100644 --- a/src/live_snapshot_store.py +++ b/src/live_snapshot_store.py @@ -31,6 +31,19 @@ class LiveSnapshotState: lease_expires_at: float | None +@dataclass(frozen=True) +class LiveSnapshotMetadata: + """Shared snapshot coordination fields without the potentially large value.""" + + created_at: dict[str, float | None] + failure_count: dict[str, int] + retry_at: dict[str, float | None] + revisions: dict[str, int] + generation: str + refreshing_sections: set[str] + lease_expires_at: float | None + + class LiveSnapshotStore: """A private SQLite snapshot with an expiring, cross-process refresh lease.""" @@ -145,6 +158,30 @@ class LiveSnapshotStore: lease_expires_at=lease["expires_at"] if lease else None, ) + def load_metadata(self) -> LiveSnapshotMetadata: + """Load freshness and coordination state without reading the snapshot value.""" + now = self.clock() + with self._connect() as connection: + row = connection.execute( + """SELECT created_at_json, failure_count_json, retry_at_json, + revisions_json, generation FROM live_snapshot WHERE singleton = 1""" + ).fetchone() + lease = connection.execute( + "SELECT sections_json, expires_at FROM live_refresh_lease " + "WHERE singleton = 1 AND expires_at > ?", + (now,), + ).fetchone() + assert row is not None + return LiveSnapshotMetadata( + created_at=json.loads(row["created_at_json"]), + failure_count=json.loads(row["failure_count_json"]), + retry_at=json.loads(row["retry_at_json"]), + revisions=json.loads(row["revisions_json"]), + generation=row["generation"], + refreshing_sections=set(json.loads(lease["sections_json"])) if lease else set(), + lease_expires_at=lease["expires_at"] if lease else None, + ) + def publish_refresh( self, owner: str, diff --git a/src/main.py b/src/main.py index 22da575..ee8fc1b 100644 --- a/src/main.py +++ b/src/main.py @@ -39,7 +39,12 @@ from src.gitea_proxy import ( ) from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy from src.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, client_source -from src.live_snapshot_store import LiveSnapshotState, LiveSnapshotStore, RefreshLeaseLost +from src.live_snapshot_store import ( + LiveSnapshotMetadata, + LiveSnapshotState, + LiveSnapshotStore, + RefreshLeaseLost, +) from src.models import Issue, Milestone, PullRequest, Repo, User from src.passkey_store import PasskeyStore from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit @@ -2371,6 +2376,35 @@ def _apply_shared_live_state(state: LiveSnapshotState) -> None: _live_snapshot_created_at = max(successful) if successful else None +def _apply_shared_live_metadata(state: LiveSnapshotMetadata) -> None: + """Update coordination state while retaining the matching local snapshot value.""" + global _live_snapshot_created_at + global _live_section_created_at, _live_section_failure_count, _live_section_retry_at + global _live_section_revisions, _live_revision_generation + global _live_snapshot_refreshing_sections + _live_section_created_at = state.created_at + _live_section_failure_count = state.failure_count + _live_section_retry_at = state.retry_at + _live_section_revisions = state.revisions + _live_revision_generation = state.generation + _live_snapshot_refreshing_sections = state.refreshing_sections + successful = [value for value in state.created_at.values() if value is not None] + _live_snapshot_created_at = max(successful) if successful else None + + +async def _sync_shared_live_state() -> None: + """Load the large shared value only when this worker's copy is out of date.""" + metadata = await asyncio.to_thread(_live_snapshot_store.load_metadata) + if ( + _live_snapshot_value is None + or metadata.generation != _live_revision_generation + or metadata.revisions != _live_section_revisions + ): + _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load)) + else: + _apply_shared_live_metadata(metadata) + + def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict: if previous is None: return refreshed @@ -2603,7 +2637,7 @@ async def _live_snapshot_response( "events": events_revision, "notifications": notifications_revision, } - _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load)) + await _sync_shared_live_state() now = _live_snapshot_clock() due_sections = _due_live_sections(now) if _live_snapshot_value is not None and not due_sections: diff --git a/tests/test_live_snapshot.py b/tests/test_live_snapshot.py index 12e1fdf..baf2b15 100644 --- a/tests/test_live_snapshot.py +++ b/tests/test_live_snapshot.py @@ -137,6 +137,48 @@ async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch assert "freshness" in second +@pytest.mark.anyio +async def test_warm_unchanged_polls_only_load_shared_metadata(monkeypatch): + async def user(): + return {"id": 1, "login": "timmy"} + + async def empty(): + return [] + + monkeypatch.setattr(main, "current_user", user) + monkeypatch.setattr(main, "repos", empty) + monkeypatch.setattr(main, "issues", empty) + monkeypatch.setattr(main, "pull_requests", empty) + monkeypatch.setattr(main, "activity_events", lambda _user: empty()) + monkeypatch.setattr(main, "notifications", empty) + first = payload(await main.live_snapshot()) + full_load = main._live_snapshot_store.load + metadata_load = main._live_snapshot_store.load_metadata + calls = {"full": 0, "metadata": 0} + + def counted_full_load(): + calls["full"] += 1 + return full_load() + + def counted_metadata_load(): + calls["metadata"] += 1 + return metadata_load() + + monkeypatch.setattr(main._live_snapshot_store, "load", counted_full_load) + monkeypatch.setattr(main._live_snapshot_store, "load_metadata", counted_metadata_load) + + for _ in range(10): + response = await main.live_snapshot( + context_revision=first["revisions"]["context"], + events_revision=first["revisions"]["events"], + notifications_revision=first["revisions"]["notifications"], + ) + assert response.status_code == 200 + assert "context" not in payload(response) + + assert calls == {"full": 0, "metadata": 10} + + @pytest.mark.anyio async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path): async def user(): diff --git a/tests/test_live_snapshot_store.py b/tests/test_live_snapshot_store.py index 0b62586..0796306 100644 --- a/tests/test_live_snapshot_store.py +++ b/tests/test_live_snapshot_store.py @@ -1,5 +1,6 @@ import threading import os +import sqlite3 import pytest @@ -7,6 +8,26 @@ from src import live_snapshot_store from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost +def test_metadata_load_does_not_retrieve_or_decode_snapshot_value(tmp_path): + store = LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: 100.0) + with sqlite3.connect(store.path) as connection: + connection.execute( + "UPDATE live_snapshot SET value_json = ? WHERE singleton = 1", + ("not-json",), + ) + + metadata = store.load_metadata() + + assert metadata.created_at == { + section: None for section in ("context", "events", "notifications") + } + assert metadata.revisions == { + section: 0 for section in ("context", "events", "notifications") + } + with pytest.raises(ValueError): + store.load() + + def test_independent_stores_racing_for_refresh_have_one_lease_winner(tmp_path): path = tmp_path / "live.sqlite3" first = LiveSnapshotStore(path, clock=lambda: 100.0)