From 0a3fb180edddf0de63f2bf8c4978f317f05e6729 Mon Sep 17 00:00:00 2001 From: timmy Date: Mon, 10 Aug 2026 01:14:11 +0000 Subject: [PATCH] perf: batch notification snapshot maintenance (#439) --- README.md | 6 ++++ src/live_snapshot_store.py | 20 +++++++---- src/main.py | 44 ++++++++++++----------- tests/test_live_snapshot_store.py | 23 ++++++++++++ tests/test_notification_read.py | 58 +++++++++++++++++++++++++++++++ 5 files changed, 124 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index c9c22fb..f9051f3 100644 --- a/README.md +++ b/README.md @@ -237,6 +237,12 @@ cannot suppress different content. The browser sends its known tokens on later p `/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. +Confirmed notification acknowledgements update this worker's live view immediately, then +maintain the shared snapshot outside the async request loop. Bulk acknowledgements remove +all successful notification IDs in one SQLite transaction and advance the notification +revision once. If shared-store lock admission fails after Gitea confirms the mutation, the +API keeps the confirmed result and process-local filtering rather than falsely reporting the +upstream write as failed; a later shared refresh reconciles the cache. ## Offline mobile shell diff --git a/src/live_snapshot_store.py b/src/live_snapshot_store.py index 1c16d76..5bf8807 100644 --- a/src/live_snapshot_store.py +++ b/src/live_snapshot_store.py @@ -217,13 +217,16 @@ class LiveSnapshotStore: connection.commit() return self.load() - def remove_notification(self, notification_id: int) -> LiveSnapshotState: - """Filter a confirmed read from every worker and subsequent stale refresh.""" + def remove_notifications(self, notification_ids: Iterable[int]) -> LiveSnapshotState: + """Filter confirmed reads from every worker and subsequent stale refreshes.""" + read_ids = set(notification_ids) + if not read_ids: + return self.load() with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") - connection.execute( + connection.executemany( "INSERT OR IGNORE INTO live_read_notification VALUES (?)", - (notification_id,), + ((notification_id,) for notification_id in read_ids), ) row = connection.execute( "SELECT value_json, revisions_json FROM live_snapshot WHERE singleton = 1" @@ -233,8 +236,9 @@ class LiveSnapshotStore: 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 + item + for item in previous + if not isinstance(item, dict) or item.get("id") not in read_ids ] if retained != previous: value["notifications"] = retained @@ -249,6 +253,10 @@ class LiveSnapshotStore: connection.commit() return self.load() + def remove_notification(self, notification_id: int) -> LiveSnapshotState: + """Filter one confirmed read from every worker and subsequent stale refresh.""" + return self.remove_notifications([notification_id]) + def release_refresh(self, owner: str) -> None: """Relinquish a lease that became unnecessary after a coherent recheck.""" with self._connect() as connection: diff --git a/src/main.py b/src/main.py index 6e9c568..a10aa9f 100644 --- a/src/main.py +++ b/src/main.py @@ -1735,30 +1735,30 @@ def _live_snapshot_payload( return payload -def _remove_notification_from_live_snapshot(thread_id: int) -> None: +async def _remove_notifications_from_live_snapshot(thread_ids: list[int]) -> None: global _live_snapshot_value, _read_notification_ids - _read_notification_ids = _read_notification_ids | {thread_id} + read_ids = set(thread_ids) + if not read_ids: + return + _read_notification_ids = _read_notification_ids | read_ids + if _live_snapshot_value is not None: + retained_notifications = _live_snapshot_value.get("notifications") + if isinstance(retained_notifications, list): + updated = dict(_live_snapshot_value) + updated["notifications"] = [ + notification + for notification in retained_notifications + if not isinstance(notification, dict) + or notification.get("id") not in read_ids + ] + if updated["notifications"] != retained_notifications: + _live_section_revisions["notifications"] += 1 + _live_snapshot_value = updated 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) + await asyncio.to_thread(_live_snapshot_store.remove_notifications, read_ids) 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") - if not isinstance(retained_notifications, list): - return - updated = dict(_live_snapshot_value) - updated["notifications"] = [ - notification - 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 def _without_read_notifications(snapshot: dict) -> dict: @@ -1986,7 +1986,6 @@ async def _mark_notification_read_result(thread_id: int) -> tuple[int, bool]: ) except Exception: return thread_id, False - _remove_notification_from_live_snapshot(thread_id) return thread_id, True @@ -2016,6 +2015,9 @@ async def read_notifications(batch: NotificationReadBatch) -> JSONResponse: for thread_id, succeeded in [task.result()] if succeeded } + await _remove_notifications_from_live_snapshot( + [thread_id for thread_id in thread_ids if thread_id in succeeded_ids] + ) failed = [thread_id for thread_id in thread_ids if thread_id not in succeeded_ids] return JSONResponse( { @@ -2116,7 +2118,7 @@ async def read_notification(thread_id: int = PathParam(gt=0)) -> JSONResponse: {"error": "The update could not be marked read. Please retry."}, status_code=503, ) - _remove_notification_from_live_snapshot(thread_id) + await _remove_notifications_from_live_snapshot([thread_id]) return JSONResponse({"id": thread_id, "status": "read"}) diff --git a/tests/test_live_snapshot_store.py b/tests/test_live_snapshot_store.py index 1ca6f2c..0b62586 100644 --- a/tests/test_live_snapshot_store.py +++ b/tests/test_live_snapshot_store.py @@ -102,6 +102,29 @@ def test_store_is_private_and_does_not_persist_upstream_token(tmp_path, monkeypa assert token.encode() not in path.read_bytes() +def test_remove_notifications_filters_batch_with_one_revision_advance(tmp_path): + store = LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: 100.0) + lease = store.try_acquire_refresh({"notifications"}, lease_seconds=5) + assert lease is not None + metadata = {section: None for section in ("context", "events", "notifications")} + published = store.publish_refresh( + lease, + value={ + "notifications": [{"id": 7}, {"id": 8}, {"id": 9}], + "sections": {"notifications": "fresh"}, + }, + created_at=metadata, + failure_count={section: 0 for section in metadata}, + retry_at=metadata, + changed_sections={"notifications"}, + ) + + removed = store.remove_notifications([7, 9, 7]) + + assert removed.value["notifications"] == [{"id": 8}] + assert removed.revisions["notifications"] == published.revisions["notifications"] + 1 + + def test_read_notification_filter_is_shared_with_future_publications(tmp_path): path = tmp_path / "live.sqlite3" first = LiveSnapshotStore(path, clock=lambda: 100.0) diff --git a/tests/test_notification_read.py b/tests/test_notification_read.py index 377d1eb..080bf5d 100644 --- a/tests/test_notification_read.py +++ b/tests/test_notification_read.py @@ -61,6 +61,37 @@ async def test_mark_notification_read_api_is_bounded_and_never_cacheable(monkeyp assert marked == [42] +@pytest.mark.anyio +async def test_snapshot_maintenance_does_not_block_the_event_loop(monkeypatch): + async def mark(_thread_id): + return None + + calls = [] + + class BlockingStore: + def remove_notification(self, thread_id): + calls.append([thread_id]) + time.sleep(0.08) + + def remove_notifications(self, thread_ids): + calls.append(list(thread_ids)) + time.sleep(0.08) + + monkeypatch.setattr(main, "mark_notification_read", mark) + monkeypatch.setattr(main, "_live_snapshot_store", BlockingStore()) + monkeypatch.setattr(main, "_live_snapshot_value", None) + started_at = asyncio.get_running_loop().time() + + request = asyncio.create_task(main.read_notification(42)) + await asyncio.sleep(0.01) + heartbeat_delay = asyncio.get_running_loop().time() - started_at + response = await request + + assert heartbeat_delay < 0.05 + assert response.status_code == 200 + assert calls == [[42]] + + @pytest.mark.anyio async def test_bulk_mark_read_reports_partial_progress_and_retains_only_failures(monkeypatch): calls = [] @@ -89,6 +120,33 @@ async def test_bulk_mark_read_reports_partial_progress_and_retains_only_failures assert main._live_snapshot_value == {"notifications": [{"id": 43}]} +@pytest.mark.anyio +async def test_bulk_mark_read_updates_shared_snapshot_once_for_successes(monkeypatch): + async def mark(thread_id): + if thread_id == 43: + raise httpx.HTTPError("upstream unavailable") + + calls = [] + + class RecordingStore: + def remove_notifications(self, thread_ids): + calls.append(tuple(sorted(thread_ids))) + + monkeypatch.setattr(main, "mark_notification_read", mark) + monkeypatch.setattr(main, "_live_snapshot_store", RecordingStore()) + monkeypatch.setattr( + main, + "_live_snapshot_value", + {"notifications": [{"id": 42}, {"id": 43}, {"id": 44}]}, + ) + + response = await main.read_notifications(main.NotificationReadBatch(ids=[42, 43, 44])) + + assert response.status_code == 200 + assert calls == [(42, 44)] + assert main._live_snapshot_value == {"notifications": [{"id": 43}]} + + @pytest.mark.anyio async def test_bulk_mark_read_limits_upstream_concurrency(monkeypatch): active = 0 -- 2.43.0