perf: share live refreshes across workers (#433)

README.md

@@ -221,12 +221,19 @@ below the route deadlines. Streaming diff reads share the same read capacity,
 while POST, PATCH, PUT, and DELETE requests are never coalesced.
 
 Live snapshots expose independent `context`, `events`, and `notifications`
-revision tokens. Each bounded opaque token includes a per-process generation, so a token
-from another worker or from before a restart cannot suppress different content. The browser
-sends its known tokens on later polls, so `/api/v1/live` can omit unchanged section bodies
-while still returning current freshness and retry metadata. The client retains omitted data
-and only rebuilds or persists the sections that changed. Malformed or oversized tokens are
-rejected before any upstream work.
+revision tokens. Snapshot content, freshness/backoff metadata, and revisions are published
+atomically through a private SQLite store shared by all application workers. An expiring
+refresh lease ensures only one worker loads currently due sections; other workers serve the
+same stale snapshot while that refresh runs, and can recover an abandoned lease after expiry.
+Set `STACKCHAIN_LIVE_SNAPSHOT_DB` to override the default
+`STACKCHAIN_STATE_DIR/live-snapshot.sqlite3`; keep the containing directory on private,
+worker-shared writable storage. The database and directory are restricted to the service
+account and never contain the Gitea token. Each bounded opaque revision token includes the
+store generation, so a token from a different deployment or before replacement of the store
+cannot suppress different content. The browser sends its known tokens on later polls, so
+`/api/v1/live` can omit unchanged section bodies while still returning current freshness and
+retry metadata. The client retains omitted data and only rebuilds or persists the sections
+that changed. Malformed or oversized tokens are rejected before any upstream work.
 
 ## Offline mobile shell
 

src/live_snapshot_store.py

@@ -0,0 +1,287 @@
+"""Process-shared live snapshot state and refresh coordination."""
+
+from __future__ import annotations
+
+import json
+import os
+import secrets
+import sqlite3
+import stat
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Iterable
+
+SECTIONS = ("context", "events", "notifications")
+
+
+class RefreshLeaseLost(RuntimeError):
+    """The refresh is no longer authorized to publish shared state."""
+
+
+@dataclass(frozen=True)
+class LiveSnapshotState:
+    value: dict | None
+    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."""
+
+    def __init__(
+        self,
+        path: str | os.PathLike[str],
+        *,
+        clock: Callable[[], float] | None = None,
+    ):
+        self.path = Path(path)
+        self.clock = clock or time.time
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path, timeout=1.0, isolation_level=None)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA busy_timeout = 1000")
+        return connection
+
+    def _initialize(self) -> None:
+        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        try:
+            os.chmod(self.path.parent, stat.S_IRWXU)
+        except OSError:
+            pass
+        old_umask = os.umask(0o077)
+        try:
+            with self._connect() as connection:
+                connection.executescript(
+                    """
+                    CREATE TABLE IF NOT EXISTS live_snapshot (
+                        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                        value_json TEXT,
+                        created_at_json TEXT NOT NULL,
+                        failure_count_json TEXT NOT NULL,
+                        retry_at_json TEXT NOT NULL,
+                        revisions_json TEXT NOT NULL,
+                        generation TEXT NOT NULL
+                    );
+                    CREATE TABLE IF NOT EXISTS live_refresh_lease (
+                        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                        owner TEXT NOT NULL,
+                        sections_json TEXT NOT NULL,
+                        expires_at REAL NOT NULL
+                    );
+                    CREATE TABLE IF NOT EXISTS live_read_notification (
+                        notification_id INTEGER PRIMARY KEY
+                    );
+                    """
+                )
+                connection.execute(
+                    """INSERT OR IGNORE INTO live_snapshot VALUES
+                    (1, NULL, ?, ?, ?, ?, ?)""",
+                    (
+                        json.dumps({section: None for section in SECTIONS}),
+                        json.dumps({section: 0 for section in SECTIONS}),
+                        json.dumps({section: None for section in SECTIONS}),
+                        json.dumps({section: 0 for section in SECTIONS}),
+                        secrets.token_hex(8),
+                    ),
+                )
+        finally:
+            os.umask(old_umask)
+        os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR)
+
+    def try_acquire_refresh(
+        self, sections: Iterable[str], *, lease_seconds: float
+    ) -> str | None:
+        requested = sorted(set(sections))
+        if not requested or any(section not in SECTIONS for section in requested):
+            raise ValueError("refresh lease requires known sections")
+        if lease_seconds <= 0:
+            raise ValueError("refresh lease duration must be positive")
+        now = self.clock()
+        owner = secrets.token_hex(16)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            active = connection.execute(
+                "SELECT expires_at FROM live_refresh_lease WHERE singleton = 1"
+            ).fetchone()
+            if active is not None and active["expires_at"] > now:
+                connection.rollback()
+                return None
+            connection.execute("DELETE FROM live_refresh_lease WHERE singleton = 1")
+            connection.execute(
+                "INSERT INTO live_refresh_lease VALUES (1, ?, ?, ?)",
+                (owner, json.dumps(requested, separators=(",", ":")), now + lease_seconds),
+            )
+            connection.commit()
+        return owner
+
+    def load(self) -> LiveSnapshotState:
+        now = self.clock()
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT * 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 LiveSnapshotState(
+            value=json.loads(row["value_json"]) if row["value_json"] is not None else None,
+            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,
+        *,
+        value: dict,
+        created_at: dict[str, float | None],
+        failure_count: dict[str, int],
+        retry_at: dict[str, float | None],
+        changed_sections: Iterable[str],
+    ) -> LiveSnapshotState:
+        changed = set(changed_sections)
+        if changed - set(SECTIONS):
+            raise ValueError("publication contains unknown sections")
+        if any(set(mapping) != set(SECTIONS) for mapping in (created_at, failure_count, retry_at)):
+            raise ValueError("publication metadata must contain every section")
+        now = self.clock()
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            lease = connection.execute(
+                "SELECT owner, expires_at FROM live_refresh_lease WHERE singleton = 1"
+            ).fetchone()
+            if lease is None or lease["owner"] != owner or lease["expires_at"] <= now:
+                connection.rollback()
+                raise RefreshLeaseLost("live refresh lease expired or changed owner")
+            row = connection.execute(
+                "SELECT revisions_json FROM live_snapshot WHERE singleton = 1"
+            ).fetchone()
+            revisions = json.loads(row["revisions_json"])
+            notifications = value.get("notifications")
+            if isinstance(notifications, list):
+                returned_ids = {
+                    item.get("id") for item in notifications if isinstance(item, dict)
+                }
+                read_ids = {
+                    item["notification_id"]
+                    for item in connection.execute(
+                        "SELECT notification_id FROM live_read_notification"
+                    )
+                }
+                value = dict(value)
+                value["notifications"] = [
+                    item for item in notifications
+                    if not isinstance(item, dict) or item.get("id") not in read_ids
+                ]
+                if returned_ids:
+                    placeholders = ",".join("?" for _ in returned_ids)
+                    connection.execute(
+                        f"DELETE FROM live_read_notification WHERE notification_id NOT IN ({placeholders})",
+                        tuple(returned_ids),
+                    )
+                else:
+                    connection.execute("DELETE FROM live_read_notification")
+            for section in changed:
+                revisions[section] += 1
+            connection.execute(
+                """UPDATE live_snapshot SET value_json = ?, created_at_json = ?,
+                failure_count_json = ?, retry_at_json = ?, revisions_json = ?
+                WHERE singleton = 1""",
+                (
+                    json.dumps(value, separators=(",", ":")),
+                    json.dumps(created_at, separators=(",", ":")),
+                    json.dumps(failure_count, separators=(",", ":")),
+                    json.dumps(retry_at, separators=(",", ":")),
+                    json.dumps(revisions, separators=(",", ":")),
+                ),
+            )
+            connection.execute(
+                "DELETE FROM live_refresh_lease WHERE singleton = 1 AND owner = ?", (owner,)
+            )
+            connection.commit()
+        return self.load()
+
+    def remove_notification(self, notification_id: int) -> LiveSnapshotState:
+        """Filter a confirmed read from every worker and subsequent stale refresh."""
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute(
+                "INSERT OR IGNORE INTO live_read_notification VALUES (?)",
+                (notification_id,),
+            )
+            row = connection.execute(
+                "SELECT value_json, revisions_json FROM live_snapshot WHERE singleton = 1"
+            ).fetchone()
+            value = json.loads(row["value_json"]) if row["value_json"] else None
+            revisions = json.loads(row["revisions_json"])
+            if value is not None and isinstance(value.get("notifications"), list):
+                previous = value["notifications"]
+                retained = [
+                    item for item in previous
+                    if not isinstance(item, dict) or item.get("id") != notification_id
+                ]
+                if retained != previous:
+                    value["notifications"] = retained
+                    revisions["notifications"] += 1
+                    connection.execute(
+                        "UPDATE live_snapshot SET value_json = ?, revisions_json = ? WHERE singleton = 1",
+                        (
+                            json.dumps(value, separators=(",", ":")),
+                            json.dumps(revisions, separators=(",", ":")),
+                        ),
+                    )
+            connection.commit()
+        return self.load()
+
+    def release_refresh(self, owner: str) -> None:
+        """Relinquish a lease that became unnecessary after a coherent recheck."""
+        with self._connect() as connection:
+            connection.execute(
+                "DELETE FROM live_refresh_lease WHERE singleton = 1 AND owner = ?",
+                (owner,),
+            )
+
+    def fail_refresh(
+        self,
+        owner: str,
+        *,
+        failure_count: dict[str, int],
+        retry_at: dict[str, float | None],
+    ) -> LiveSnapshotState:
+        """Publish backoff metadata and relinquish a failed refresh atomically."""
+        now = self.clock()
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            lease = connection.execute(
+                "SELECT owner, expires_at FROM live_refresh_lease WHERE singleton = 1"
+            ).fetchone()
+            if lease is None or lease["owner"] != owner or lease["expires_at"] <= now:
+                connection.rollback()
+                raise RefreshLeaseLost("live refresh lease expired or changed owner")
+            connection.execute(
+                """UPDATE live_snapshot SET failure_count_json = ?, retry_at_json = ?
+                WHERE singleton = 1""",
+                (
+                    json.dumps(failure_count, separators=(",", ":")),
+                    json.dumps(retry_at, separators=(",", ":")),
+                ),
+            )
+            connection.execute("DELETE FROM live_refresh_lease WHERE singleton = 1")
+            connection.commit()
+        return self.load()

src/main.py

@@ -34,6 +34,7 @@ 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.models import Issue, Milestone, PullRequest, Repo, User
 from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
 from src.suggestion_engine import compute
@@ -117,6 +118,11 @@ _authored_action_operations: dict[
     str, tuple[tuple[Any, ...], asyncio.Task, float]
 ] = {}
 _state_dir = Path(os.getenv("STACKCHAIN_STATE_DIR", ".stackchain-state"))
+_live_snapshot_clock = time.time
+_live_snapshot_store = LiveSnapshotStore(
+    os.getenv("STACKCHAIN_LIVE_SNAPSHOT_DB", str(_state_dir / "live-snapshot.sqlite3")),
+    clock=lambda: _live_snapshot_clock(),
+)
 _idempotency_ledger = IdempotencyLedger(
     os.getenv("STACKCHAIN_IDEMPOTENCY_DB", str(_state_dir / "idempotency.sqlite3")),
     ttl_seconds=AUTHORED_ACTION_IDEMPOTENCY_TTL_SECONDS,
@@ -1530,7 +1536,7 @@ def _record_live_section_failure(section: str) -> None:
         LIVE_SNAPSHOT_RETRY_BASE_SECONDS
         * (2 ** (_live_section_failure_count[section] - 1)),
     )
-    _live_section_retry_at[section] = time.monotonic() + delay
+    _live_section_retry_at[section] = _live_snapshot_clock() + delay
 
 
 def _due_live_sections(now: float) -> set[str]:
@@ -1545,6 +1551,23 @@ def _due_live_sections(now: float) -> set[str]:
     return due
 
 
+def _apply_shared_live_state(state: LiveSnapshotState) -> None:
+    """Replace this worker's fast local view with one coherent SQLite read."""
+    global _live_snapshot_value, _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_snapshot_value = state.value
+    _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
+
+
 def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
     if previous is None:
         return refreshed
@@ -1567,15 +1590,32 @@ def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
     return merged
 
 
-async def _refresh_live_snapshot(sections: set[str]) -> dict:
+async def _refresh_live_snapshot(sections: set[str], lease_owner: str) -> dict:
     global _live_snapshot_value, _live_snapshot_created_at
     try:
         refreshed = await _build_live_snapshot_before_deadline(sections)
+    except asyncio.CancelledError:
+        try:
+            await asyncio.shield(
+                asyncio.to_thread(_live_snapshot_store.release_refresh, lease_owner)
+            )
+        finally:
+            raise
     except Exception:
         for section in sections:
             _record_live_section_failure(section)
+        try:
+            state = await asyncio.to_thread(
+                _live_snapshot_store.fail_refresh,
+                lease_owner,
+                failure_count=_live_section_failure_count,
+                retry_at=_live_section_retry_at,
+            )
+            _apply_shared_live_state(state)
+        except RefreshLeaseLost:
+            _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load))
         raise
-    now = time.monotonic()
+    now = _live_snapshot_clock()
     refreshed_states = refreshed.get("sections") or {}
     for section in sections:
         if refreshed_states.get(section) == "fresh":
@@ -1587,6 +1627,7 @@ async def _refresh_live_snapshot(sections: set[str]) -> dict:
     previous = _live_snapshot_value
     result = _merge_live_snapshot(previous, refreshed)
     result = _without_read_notifications(result)
+    changed_sections = set()
     for section in sections:
         if section not in refreshed_states:
             continue
@@ -1596,11 +1637,24 @@ async def _refresh_live_snapshot(sections: set[str]) -> dict:
                 "notification_pagination"
             ) != result.get("notification_pagination")
         if changed:
-            _live_section_revisions[section] += 1
-    _live_snapshot_value = result
-    successful_times = [value for value in _live_section_created_at.values() if value is not None]
-    _live_snapshot_created_at = max(successful_times) if successful_times else None
-    return result
+            changed_sections.add(section)
+    try:
+        state = await asyncio.to_thread(
+            _live_snapshot_store.publish_refresh,
+            lease_owner,
+            value=result,
+            created_at=_live_section_created_at,
+            failure_count=_live_section_failure_count,
+            retry_at=_live_section_retry_at,
+            changed_sections=changed_sections,
+        )
+    except RefreshLeaseLost:
+        state = await asyncio.to_thread(_live_snapshot_store.load)
+        if state.value is None:
+            raise
+    _apply_shared_live_state(state)
+    assert state.value is not None
+    return state.value
 
 
 def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
@@ -1609,12 +1663,22 @@ def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
     task.exception()
 
 
-def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
+def _start_live_snapshot_refresh(
+    sections: set[str], lease_owner: str | None = None
+) -> asyncio.Task:
     global _live_snapshot_task
     global _live_snapshot_refreshing_sections
     if _live_snapshot_task is None or _live_snapshot_task.done():
+        if lease_owner is None:
+            lease_owner = _live_snapshot_store.try_acquire_refresh(
+                sections, lease_seconds=CONTEXT_TIMEOUT_SECONDS + 1.0
+            )
+            if lease_owner is None:
+                raise RuntimeError("live snapshot refresh lease is already held")
         _live_snapshot_refreshing_sections = set(sections)
-        _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot(sections))
+        _live_snapshot_task = asyncio.create_task(
+            _refresh_live_snapshot(sections, lease_owner)
+        )
         _live_snapshot_task.add_done_callback(_consume_live_snapshot_failure)
     return _live_snapshot_task
 
@@ -1627,7 +1691,7 @@ def _live_snapshot_payload(
     known_revisions: dict[str, str | None] | None = None,
 ) -> dict:
     payload = dict(value)
-    now = time.monotonic()
+    now = _live_snapshot_clock()
     section_freshness = {}
     for section in LIVE_SNAPSHOT_SECTIONS:
         created_at = _live_section_created_at[section]
@@ -1674,6 +1738,13 @@ def _live_snapshot_payload(
 def _remove_notification_from_live_snapshot(thread_id: int) -> None:
     global _live_snapshot_value, _read_notification_ids
     _read_notification_ids = _read_notification_ids | {thread_id}
+    try:
+        shared = _live_snapshot_store.remove_notification(thread_id)
+        if shared.value is not None or _live_snapshot_value is None:
+            _apply_shared_live_state(shared)
+    except (OSError, sqlite3.Error):
+        # The upstream mutation already succeeded; keep process-local filtering.
+        pass
     if _live_snapshot_value is None:
         return
     retained_notifications = _live_snapshot_value.get("notifications")
@@ -1711,8 +1782,7 @@ def _without_read_notifications(snapshot: dict) -> dict:
     return updated
 
 
-@app.get("/api/v1/live")
-async def live_snapshot(
+async def _live_snapshot_response(
     context_revision: str | None = Query(
         default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
     ),
@@ -1730,12 +1800,10 @@ async def live_snapshot(
         "events": events_revision,
         "notifications": notifications_revision,
     }
-    now = time.monotonic()
+    _apply_shared_live_state(await asyncio.to_thread(_live_snapshot_store.load))
+    now = _live_snapshot_clock()
     due_sections = _due_live_sections(now)
-    if (
-        _live_snapshot_value is not None
-        and not due_sections
-    ):
+    if _live_snapshot_value is not None and not due_sections:
         return JSONResponse(
             _live_snapshot_payload(
                 _live_snapshot_value,
@@ -1747,7 +1815,77 @@ async def live_snapshot(
                 known_revisions=known_revisions,
             )
         )
-    task = _start_live_snapshot_refresh(due_sections)
+    if not due_sections:
+        retries = [
+            retry_at - now
+            for retry_at in _live_section_retry_at.values()
+            if retry_at is not None and retry_at > now
+        ]
+        return JSONResponse(
+            {"error": "Gitea live snapshot is temporarily unavailable"},
+            status_code=503,
+            headers={"Retry-After": str(max(1, math.ceil(min(retries, default=1.0))))},
+        )
+
+    task = _live_snapshot_task
+    if task is None or task.done():
+        lease_owner = await asyncio.to_thread(
+            _live_snapshot_store.try_acquire_refresh,
+            due_sections,
+            lease_seconds=CONTEXT_TIMEOUT_SECONDS + 1.0,
+        )
+        if lease_owner is not None:
+            latest = await asyncio.to_thread(_live_snapshot_store.load)
+            _apply_shared_live_state(latest)
+            due_sections = _due_live_sections(_live_snapshot_clock())
+            if not due_sections:
+                await asyncio.to_thread(
+                    _live_snapshot_store.release_refresh, lease_owner
+                )
+                return JSONResponse(
+                    _live_snapshot_payload(
+                        _live_snapshot_value,
+                        stale=any(
+                            state != "fresh"
+                            for state in (_live_snapshot_value or {}).get("sections", {}).values()
+                        ),
+                        revalidating=False,
+                        known_revisions=known_revisions,
+                    )
+                )
+            task = _start_live_snapshot_refresh(due_sections, lease_owner)
+        else:
+            shared = await asyncio.to_thread(_live_snapshot_store.load)
+            _apply_shared_live_state(shared)
+            if _live_snapshot_value is not None:
+                return JSONResponse(
+                    _live_snapshot_payload(
+                        _live_snapshot_value,
+                        stale=bool(_due_live_sections(_live_snapshot_clock())),
+                        revalidating=bool(shared.refreshing_sections),
+                        known_revisions=known_revisions,
+                    )
+                )
+            deadline = time.perf_counter() + CONTEXT_TIMEOUT_SECONDS
+            while time.perf_counter() < deadline:
+                await asyncio.sleep(0.01)
+                shared = await asyncio.to_thread(_live_snapshot_store.load)
+                if shared.value is not None:
+                    _apply_shared_live_state(shared)
+                    return JSONResponse(
+                        _live_snapshot_payload(
+                            shared.value,
+                            stale=bool(_due_live_sections(_live_snapshot_clock())),
+                            revalidating=bool(shared.refreshing_sections),
+                            known_revisions=known_revisions,
+                        )
+                    )
+            return JSONResponse(
+                {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
+                status_code=503,
+                headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
+            )
+    assert task is not None
     if _live_snapshot_value is not None:
         return JSONResponse(
             _live_snapshot_payload(
@@ -1780,6 +1918,32 @@ async def live_snapshot(
         )
 
 
+@app.get("/api/v1/live")
+async def live_snapshot(
+    context_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
+    events_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
+    notifications_revision: str | None = Query(
+        default=None, max_length=64, pattern=r"^[0-9a-f]{16}\.[0-9]{1,20}$"
+    ),
+) -> JSONResponse:
+    try:
+        return await _live_snapshot_response(
+            context_revision=context_revision,
+            events_revision=events_revision,
+            notifications_revision=notifications_revision,
+        )
+    except (OSError, sqlite3.Error):
+        return JSONResponse(
+            {"error": "Gitea live snapshot state is temporarily unavailable"},
+            status_code=503,
+            headers={"Retry-After": "1"},
+        )
+
+
 
 @app.get("/api/v1/events")
 async def event_stream():

tests/test_live_snapshot.py

@@ -1,14 +1,21 @@
 import asyncio
 import json
+import sqlite3
 
 import httpx
 import pytest
 
 from src import main
+from src.live_snapshot_store import LiveSnapshotStore
 
 
 @pytest.fixture(autouse=True)
-def reset_live_snapshot_task():
+def reset_live_snapshot_task(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        main,
+        "_live_snapshot_store",
+        LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()),
+    )
     main._live_snapshot_task = None
     main._live_snapshot_value = None
     main._live_snapshot_created_at = None
@@ -131,7 +138,7 @@ async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch
 
 
 @pytest.mark.anyio
-async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch):
+async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path):
     async def user():
         return {"id": 1, "login": "timmy"}
 
@@ -144,8 +151,6 @@ async def test_live_snapshot_token_from_another_worker_never_suppresses_content(
     monkeypatch.setattr(main, "pull_requests", empty)
     monkeypatch.setattr(main, "activity_events", lambda _user: empty())
     monkeypatch.setattr(main, "notifications", empty)
-    monkeypatch.setattr(main, "_live_revision_generation", "worker-a")
-
     worker_a = payload(await main.live_snapshot())
     worker_a_token = worker_a["revisions"]["context"]
 
@@ -157,12 +162,15 @@ async def test_live_snapshot_token_from_another_worker_never_suppresses_content(
     main._live_section_revisions = {
         section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
     }
-    monkeypatch.setattr(main, "_live_revision_generation", "worker-b")
+    main._live_snapshot_store = LiveSnapshotStore(
+        tmp_path / "other-worker.sqlite3", clock=lambda: main._live_snapshot_clock()
+    )
 
     worker_b = payload(await main.live_snapshot(context_revision=worker_a_token))
 
-    assert worker_a_token == "worker-a.1"
-    assert worker_b["revisions"]["context"] == "worker-b.1"
+    assert worker_a_token.endswith(".1")
+    assert worker_b["revisions"]["context"].endswith(".1")
+    assert worker_b["revisions"]["context"] != worker_a_token
     assert worker_b["context"]["user"]["login"] == "timmy"
 
 
@@ -302,7 +310,10 @@ async def test_live_snapshot_reuses_completed_snapshot_inside_freshness_window(m
     first = asyncio.create_task(main.live_snapshot())
     await asyncio.sleep(0)
     second = asyncio.create_task(main.live_snapshot())
-    await asyncio.sleep(0)
+    for _ in range(100):
+        if user_calls:
+            break
+        await asyncio.to_thread(lambda: None)
     assert user_calls == 1
 
     release.set()
@@ -332,7 +343,7 @@ async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(
             "sections": {},
         }
 
-    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
     monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
 
     first = payload(await main.live_snapshot())
@@ -373,13 +384,14 @@ async def test_failed_revalidation_enters_cooldown_and_keeps_last_snapshot(monke
             },
         }
 
-    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
     monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
 
     await main.live_snapshot()
     now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
     stale = payload(await main.live_snapshot())
-    await asyncio.sleep(0)
+    assert main._live_snapshot_task is not None
+    await asyncio.gather(main._live_snapshot_task, return_exceptions=True)
     degraded = payload(await main.live_snapshot())
 
     assert stale["freshness"]["revalidating"] is True
@@ -421,13 +433,14 @@ async def test_partial_refresh_updates_fresh_sections_and_retains_failed_section
             },
         }
 
-    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
     monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
 
     await main.live_snapshot()
     now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
     await main.live_snapshot()
-    await asyncio.sleep(0)
+    assert main._live_snapshot_task is not None
+    await main._live_snapshot_task
     result = payload(await main.live_snapshot())
 
     assert result["context"] == {"generation": 2}
@@ -470,7 +483,7 @@ async def test_notification_cooldown_does_not_stop_due_work_and_activity_refresh
             raise ConnectionError("notifications unavailable")
         return []
 
-    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
     monkeypatch.setattr(main, "current_user", user)
     monkeypatch.setattr(main, "repos", work)
     monkeypatch.setattr(main, "issues", empty_work)
@@ -537,20 +550,19 @@ async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state
             "sections": {"context": "fresh", "events": "fresh", "notifications": "fresh"},
         }
 
-    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
     monkeypatch.setattr(main, "_build_live_snapshot", snapshot)
 
     await main.live_snapshot()
     now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
     await main.live_snapshot()
-    await asyncio.sleep(0)
+    assert main._live_snapshot_task is not None
+    await main._live_snapshot_task
     now += main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS
 
     first = asyncio.create_task(main.live_snapshot())
     second = asyncio.create_task(main.live_snapshot())
-    await asyncio.sleep(0)
-    await asyncio.sleep(0)
-    assert retry_started.is_set()
+    await asyncio.wait_for(retry_started.wait(), timeout=1.0)
     assert builds == 3
     assert payload(await first)["freshness"]["revalidating"] is True
     assert payload(await second)["freshness"]["revalidating"] is True
@@ -640,3 +652,123 @@ async def test_shared_snapshot_deadline_cancels_upstream_work_for_all_waiters(mo
             task.cancel()
             with pytest.raises(asyncio.CancelledError):
                 await task
+
+
+@pytest.mark.anyio
+async def test_cold_refresh_cooldown_remains_a_retryable_503(monkeypatch):
+    now = 100.0
+
+    async def unavailable(_sections=None):
+        raise ConnectionError("private upstream failure")
+
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: now)
+    monkeypatch.setattr(main, "_build_live_snapshot", unavailable)
+
+    first = await main.live_snapshot()
+    second = await main.live_snapshot()
+
+    assert first.status_code == 503
+    assert second.status_code == 503
+    assert payload(second) == {
+        "error": "Gitea live snapshot is temporarily unavailable"
+    }
+
+
+@pytest.mark.anyio
+async def test_cancelled_refresh_releases_shared_lease(monkeypatch):
+    started = asyncio.Event()
+
+    async def blocked(_sections=None):
+        started.set()
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(main, "_build_live_snapshot", blocked)
+    request = asyncio.create_task(main.live_snapshot())
+    await started.wait()
+    assert main._live_snapshot_task is not None
+
+    main._live_snapshot_task.cancel()
+    await asyncio.gather(request, main._live_snapshot_task, return_exceptions=True)
+
+    replacement = main._live_snapshot_store.try_acquire_refresh(
+        {"context"}, lease_seconds=1
+    )
+    assert replacement is not None
+
+
+@pytest.mark.anyio
+async def test_shared_store_outage_returns_retryable_503(monkeypatch):
+    def unavailable():
+        raise sqlite3.OperationalError("private database detail")
+
+    monkeypatch.setattr(main._live_snapshot_store, "load", unavailable)
+
+    response = await main.live_snapshot()
+
+    assert response.status_code == 503
+    assert response.headers["retry-after"] == "1"
+    assert payload(response) == {
+        "error": "Gitea live snapshot state is temporarily unavailable"
+    }
+
+
+@pytest.mark.anyio
+async def test_independent_workers_reuse_shared_live_snapshot(monkeypatch, tmp_path):
+    calls = 0
+
+    async def user():
+        nonlocal calls
+        calls += 1
+        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)
+    path = tmp_path / "worker-shared.sqlite3"
+    monkeypatch.setattr(main, "_live_snapshot_store", LiveSnapshotStore(path, clock=lambda: 100.0))
+
+    first = payload(await main.live_snapshot())
+    main._live_snapshot_value = None
+    main._live_snapshot_created_at = None
+    main._live_section_created_at = {section: None for section in main.LIVE_SNAPSHOT_SECTIONS}
+    main._live_snapshot_store = LiveSnapshotStore(path, clock=lambda: 100.0)
+    second = payload(await main.live_snapshot())
+
+    assert calls == 1
+    assert second["context"] == first["context"]
+    assert second["revisions"] == first["revisions"]
+
+
+@pytest.mark.anyio
+async def test_live_snapshot_persists_reboot_stable_wall_timestamps(monkeypatch, tmp_path):
+    epoch = 1_700_000_000.0
+
+    async def user():
+        return {"id": 1, "login": "timmy"}
+
+    async def empty():
+        return []
+
+    monkeypatch.setattr(main, "_live_snapshot_clock", lambda: epoch, raising=False)
+    monkeypatch.setattr(
+        main,
+        "_live_snapshot_store",
+        LiveSnapshotStore(tmp_path / "wall.sqlite3", clock=lambda: main._live_snapshot_clock()),
+    )
+    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)
+
+    response = await main.live_snapshot()
+
+    assert response.status_code == 200
+    assert set(main._live_snapshot_store.load().created_at.values()) == {epoch}

tests/test_live_snapshot_store.py

@@ -0,0 +1,134 @@
+import threading
+import os
+
+import pytest
+
+from src import live_snapshot_store
+from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost
+
+
+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)
+    second = LiveSnapshotStore(path, clock=lambda: 100.0)
+    barrier = threading.Barrier(2)
+    results = []
+
+    def acquire(store):
+        barrier.wait()
+        results.append(store.try_acquire_refresh({"context", "events"}, lease_seconds=5))
+
+    threads = [threading.Thread(target=acquire, args=(store,)) for store in (first, second)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert sum(result is not None for result in results) == 1
+
+
+def test_snapshot_and_shared_revisions_publish_atomically(tmp_path):
+    path = tmp_path / "live.sqlite3"
+    writer = LiveSnapshotStore(path, clock=lambda: 100.0)
+    reader = LiveSnapshotStore(path, clock=lambda: 100.0)
+    lease = writer.try_acquire_refresh(set(("context", "events", "notifications")), lease_seconds=5)
+    assert lease is not None
+
+    state = writer.publish_refresh(
+        lease,
+        value={
+            "context": {"generation": 1},
+            "events": [],
+            "notifications": [],
+            "sections": {section: "fresh" for section in ("context", "events", "notifications")},
+        },
+        created_at={section: 100.0 for section in ("context", "events", "notifications")},
+        failure_count={section: 0 for section in ("context", "events", "notifications")},
+        retry_at={section: None for section in ("context", "events", "notifications")},
+        changed_sections=set(("context", "events", "notifications")),
+    )
+
+    observed = reader.load()
+    assert observed.value == state.value
+    assert observed.revisions == {"context": 1, "events": 1, "notifications": 1}
+    assert observed.refreshing_sections == set()
+    with pytest.raises(RefreshLeaseLost):
+        writer.publish_refresh(
+            lease,
+            value={"context": {"generation": 2}},
+            created_at=observed.created_at,
+            failure_count=observed.failure_count,
+            retry_at=observed.retry_at,
+            changed_sections={"context"},
+        )
+
+
+def test_expired_refresh_lease_can_be_recovered(tmp_path):
+    now = 100.0
+    path = tmp_path / "live.sqlite3"
+    abandoned = LiveSnapshotStore(path, clock=lambda: now)
+    recovery = LiveSnapshotStore(path, clock=lambda: now)
+
+    first = abandoned.try_acquire_refresh({"context"}, lease_seconds=5)
+    assert first is not None
+    assert recovery.try_acquire_refresh({"context"}, lease_seconds=5) is None
+    now = 106.0
+    second = recovery.try_acquire_refresh({"context"}, lease_seconds=5)
+
+    assert second is not None
+    assert second != first
+
+
+def test_default_store_clock_uses_reboot_stable_wall_time(tmp_path, monkeypatch):
+    epoch = 1_700_000_000.0
+    monkeypatch.setattr(live_snapshot_store.time, "time", lambda: epoch)
+    store = LiveSnapshotStore(tmp_path / "live.sqlite3")
+
+    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)
+
+    assert lease is not None
+    assert store.load().lease_expires_at == epoch + 5
+
+
+def test_store_is_private_and_does_not_persist_upstream_token(tmp_path, monkeypatch):
+    token = "gitea-super-secret-token-material"
+    monkeypatch.setenv("GITEA_TOKEN", token)
+    path = tmp_path / "private-state" / "live.sqlite3"
+
+    LiveSnapshotStore(path)
+
+    assert os.stat(path).st_mode & 0o777 == 0o600
+    assert os.stat(path.parent).st_mode & 0o777 == 0o700
+    assert token.encode() not in path.read_bytes()
+
+
+def test_read_notification_filter_is_shared_with_future_publications(tmp_path):
+    path = tmp_path / "live.sqlite3"
+    first = LiveSnapshotStore(path, clock=lambda: 100.0)
+    second = LiveSnapshotStore(path, clock=lambda: 100.0)
+    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
+    assert lease is not None
+    metadata = {section: None for section in ("context", "events", "notifications")}
+    first.publish_refresh(
+        lease,
+        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
+        created_at=metadata,
+        failure_count={section: 0 for section in metadata},
+        retry_at=metadata,
+        changed_sections={"notifications"},
+    )
+
+    removed = second.remove_notification(7)
+    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
+    assert lease is not None
+    republished = first.publish_refresh(
+        lease,
+        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
+        created_at=metadata,
+        failure_count={section: 0 for section in metadata},
+        retry_at=metadata,
+        changed_sections={"notifications"},
+    )
+
+    assert removed.value["notifications"] == [{"id": 8}]
+    assert republished.value["notifications"] == [{"id": 8}]

tests/test_notification_read.py

@@ -5,6 +5,17 @@ import httpx
 import pytest
 
 from src import gitea_proxy, main
+from src.live_snapshot_store import LiveSnapshotStore
+
+
+@pytest.fixture(autouse=True)
+def isolated_live_snapshot_store(tmp_path, monkeypatch):
+    monkeypatch.setattr(
+        main,
+        "_live_snapshot_store",
+        LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: main._live_snapshot_clock()),
+    )
+    monkeypatch.setattr(main, "_read_notification_ids", set())
 
 
 @pytest.mark.anyio