Avoid decoding unchanged live snapshots on every poll #520
|
|
@ -31,6 +31,19 @@ class LiveSnapshotState:
|
||||||
lease_expires_at: float | None
|
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:
|
class LiveSnapshotStore:
|
||||||
"""A private SQLite snapshot with an expiring, cross-process refresh lease."""
|
"""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,
|
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(
|
def publish_refresh(
|
||||||
self,
|
self,
|
||||||
owner: str,
|
owner: str,
|
||||||
|
|
|
||||||
38
src/main.py
38
src/main.py
|
|
@ -39,7 +39,12 @@ from src.gitea_proxy import (
|
||||||
)
|
)
|
||||||
from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy
|
from src.idempotency import IdempotencyLedger, IdempotencyLedgerBusy
|
||||||
from src.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, client_source
|
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.models import Issue, Milestone, PullRequest, Repo, User
|
||||||
from src.passkey_store import PasskeyStore
|
from src.passkey_store import PasskeyStore
|
||||||
from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
|
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
|
_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:
|
def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
|
||||||
if previous is None:
|
if previous is None:
|
||||||
return refreshed
|
return refreshed
|
||||||
|
|
@ -2603,7 +2637,7 @@ async def _live_snapshot_response(
|
||||||
"events": events_revision,
|
"events": events_revision,
|
||||||
"notifications": notifications_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()
|
now = _live_snapshot_clock()
|
||||||
due_sections = _due_live_sections(now)
|
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:
|
||||||
|
|
|
||||||
|
|
@ -137,6 +137,48 @@ async def test_live_snapshot_omits_section_bodies_at_known_revisions(monkeypatch
|
||||||
assert "freshness" in second
|
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
|
@pytest.mark.anyio
|
||||||
async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path):
|
async def test_live_snapshot_token_from_another_worker_never_suppresses_content(monkeypatch, tmp_path):
|
||||||
async def user():
|
async def user():
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,6 @@
|
||||||
import threading
|
import threading
|
||||||
import os
|
import os
|
||||||
|
import sqlite3
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
@ -7,6 +8,26 @@ from src import live_snapshot_store
|
||||||
from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost
|
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):
|
def test_independent_stores_racing_for_refresh_have_one_lease_winner(tmp_path):
|
||||||
path = tmp_path / "live.sqlite3"
|
path = tmp_path / "live.sqlite3"
|
||||||
first = LiveSnapshotStore(path, clock=lambda: 100.0)
|
first = LiveSnapshotStore(path, clock=lambda: 100.0)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user