Merge pull request 'Avoid decoding unchanged live snapshots on every poll' (#520) from timmy/519-metadata-only-live-polls into main

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,

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:

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():

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)