diff --git a/README.md b/README.md index 2c3225b..7dda6ec 100644 --- a/README.md +++ b/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 diff --git a/src/live_snapshot_store.py b/src/live_snapshot_store.py new file mode 100644 index 0000000..1c16d76 --- /dev/null +++ b/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() diff --git a/src/main.py b/src/main.py index eda2c7b..6e9c568 100644 --- a/src/main.py +++ b/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(): diff --git a/tests/test_live_snapshot.py b/tests/test_live_snapshot.py index f46e2f9..12e1fdf 100644 --- a/tests/test_live_snapshot.py +++ b/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} diff --git a/tests/test_live_snapshot_store.py b/tests/test_live_snapshot_store.py new file mode 100644 index 0000000..1ca6f2c --- /dev/null +++ b/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}] diff --git a/tests/test_notification_read.py b/tests/test_notification_read.py index ee924cd..377d1eb 100644 --- a/tests/test_notification_read.py +++ b/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