perf: batch notification snapshot maintenance (#439)

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)

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