perf: skip unchanged live snapshot sections (#285)
This commit is contained in:
1 parent
e28472b268
commit
fedddb246d
10 files changed
+188
-20
No files matched your search
+52
-5
@@ -98,6 +98,9 @@ _live_section_retry_at: dict[str, float | None] = {
|
||||
section: None for section in LIVE_SNAPSHOT_SECTIONS
|
||||
}
|
||||
_live_snapshot_refreshing_sections: set[str] = set()
|
||||
_live_section_revisions: dict[str, int] = {
|
||||
section: 0 for section in LIVE_SNAPSHOT_SECTIONS
|
||||
}
|
||||
_read_notification_ids: set[int] = set()
|
||||
_authored_action_operations: dict[
|
||||
str, tuple[tuple[Any, ...], asyncio.Task, float]
|
||||
@@ -1082,8 +1085,19 @@ async def _refresh_live_snapshot(sections: set[str]) -> dict:
|
||||
_live_section_retry_at[section] = None
|
||||
else:
|
||||
_record_live_section_failure(section)
|
||||
result = _merge_live_snapshot(_live_snapshot_value, refreshed)
|
||||
previous = _live_snapshot_value
|
||||
result = _merge_live_snapshot(previous, refreshed)
|
||||
result = _without_read_notifications(result)
|
||||
for section in sections:
|
||||
if section not in refreshed_states:
|
||||
continue
|
||||
changed = previous is None or previous.get(section) != result.get(section)
|
||||
if section == "notifications":
|
||||
changed = changed or previous is None or previous.get(
|
||||
"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
|
||||
@@ -1106,7 +1120,13 @@ def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
|
||||
return _live_snapshot_task
|
||||
|
||||
|
||||
def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> dict:
|
||||
def _live_snapshot_payload(
|
||||
value: dict,
|
||||
*,
|
||||
stale: bool,
|
||||
revalidating: bool,
|
||||
known_revisions: dict[str, int | None] | None = None,
|
||||
) -> dict:
|
||||
payload = dict(value)
|
||||
now = time.monotonic()
|
||||
section_freshness = {}
|
||||
@@ -1138,6 +1158,13 @@ def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> d
|
||||
"retry_in_seconds": min(retries, default=0),
|
||||
"sections": section_freshness,
|
||||
}
|
||||
payload["revisions"] = dict(_live_section_revisions)
|
||||
for section, known_revision in (known_revisions or {}).items():
|
||||
if known_revision is None or known_revision != _live_section_revisions[section]:
|
||||
continue
|
||||
payload.pop(section, None)
|
||||
if section == "notifications":
|
||||
payload.pop("notification_pagination", None)
|
||||
return payload
|
||||
|
||||
|
||||
@@ -1155,6 +1182,8 @@ def _remove_notification_from_live_snapshot(thread_id: int) -> None:
|
||||
for notification in retained_notifications
|
||||
if not isinstance(notification, dict) or notification.get("id") != thread_id
|
||||
]
|
||||
if updated["notifications"] != retained_notifications:
|
||||
_live_section_revisions["notifications"] += 1
|
||||
_live_snapshot_value = updated
|
||||
|
||||
|
||||
@@ -1180,9 +1209,18 @@ def _without_read_notifications(snapshot: dict) -> dict:
|
||||
|
||||
|
||||
@app.get("/api/v1/live")
|
||||
async def live_snapshot() -> JSONResponse:
|
||||
async def live_snapshot(
|
||||
context_revision: int | None = Query(default=None, ge=0),
|
||||
events_revision: int | None = Query(default=None, ge=0),
|
||||
notifications_revision: int | None = Query(default=None, ge=0),
|
||||
) -> JSONResponse:
|
||||
"""Return a freshness-bounded snapshot and share identical upstream loads."""
|
||||
global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at
|
||||
known_revisions = {
|
||||
"context": context_revision,
|
||||
"events": events_revision,
|
||||
"notifications": notifications_revision,
|
||||
}
|
||||
now = time.monotonic()
|
||||
due_sections = _due_live_sections(now)
|
||||
if (
|
||||
@@ -1197,19 +1235,28 @@ async def live_snapshot() -> JSONResponse:
|
||||
for state in _live_snapshot_value.get("sections", {}).values()
|
||||
),
|
||||
revalidating=False,
|
||||
known_revisions=known_revisions,
|
||||
)
|
||||
)
|
||||
task = _start_live_snapshot_refresh(due_sections)
|
||||
if _live_snapshot_value is not None:
|
||||
return JSONResponse(
|
||||
_live_snapshot_payload(
|
||||
_live_snapshot_value, stale=True, revalidating=True
|
||||
_live_snapshot_value,
|
||||
stale=True,
|
||||
revalidating=True,
|
||||
known_revisions=known_revisions,
|
||||
)
|
||||
)
|
||||
try:
|
||||
result = await asyncio.shield(task)
|
||||
return JSONResponse(
|
||||
_live_snapshot_payload(result, stale=False, revalidating=False)
|
||||
_live_snapshot_payload(
|
||||
result,
|
||||
stale=False,
|
||||
revalidating=False,
|
||||
known_revisions=known_revisions,
|
||||
)
|
||||
)
|
||||
except TimeoutError:
|
||||
return JSONResponse(
|
||||
|
||||
Reference in new issue
Block a user