diff --git a/src/gitea_proxy.py b/src/gitea_proxy.py index 57b2e45..6594a47 100644 --- a/src/gitea_proxy.py +++ b/src/gitea_proxy.py @@ -709,6 +709,41 @@ async def notifications() -> dict: return await notification_page(1) +async def unread_notification_snapshot( + *, + limit: int = 50, + max_pages: int = 20, + max_concurrency: int = 4, + deadline_seconds: float = 5.0, +) -> dict: + """Load one complete, bounded snapshot of unread notification threads.""" + async with asyncio.timeout(deadline_seconds): + first = await notification_page(1, limit) + page_count = max(1, (first["total"] + limit - 1) // limit) + if page_count > max_pages: + raise ValueError("Unread notification snapshot exceeds the scan limit") + semaphore = asyncio.Semaphore(max(1, max_concurrency)) + + async def load(page: int) -> dict: + async with semaphore: + return await notification_page(page, limit) + + remaining = await asyncio.gather( + *(load(page) for page in range(2, page_count + 1)) + ) + pages = [first, *remaining] + if any(page["total"] != first["total"] for page in pages[1:]): + raise ValueError("Unread notification pagination changed during the scan") + items = [item for page in pages for item in page["items"]] + thread_ids = [ + item.get("id") for item in items + if isinstance(item, dict) and isinstance(item.get("id"), int) and item["id"] > 0 + ] + if len(items) != first["total"] or len(set(thread_ids)) != first["total"]: + raise ValueError("Unread notification snapshot has an incomplete thread set") + return {"items": items, "total": first["total"], "complete": True} + + async def notification_page(page: int, limit: int = 50) -> dict: response = await _get_client().get( f"/api/v1/notifications?status-types=unread&limit={limit}&page={page}", diff --git a/src/main.py b/src/main.py index 5198239..c0f5ba9 100644 --- a/src/main.py +++ b/src/main.py @@ -100,7 +100,7 @@ async def _push_poll_loop() -> None: await dispatch_unread_updates( _push_subscription_store, _push_configuration(), - notifications, + gitea_proxy.unread_notification_snapshot, session_active=dashboard_auth.managed_session_active, lease_seconds=lease_seconds, send_timeout_seconds=send_timeout, @@ -1735,7 +1735,9 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request): payload.model_dump(), ) try: - current = await asyncio.wait_for(notifications(), NOTIFICATION_PAGE_TIMEOUT_SECONDS) + current = await gitea_proxy.unread_notification_snapshot( + deadline_seconds=NOTIFICATION_PAGE_TIMEOUT_SECONDS + ) except Exception as error: await asyncio.to_thread(_push_subscription_store.delete_session, device_id) raise HTTPException( @@ -1743,13 +1745,13 @@ async def subscribe_push(payload: PushSubscriptionPayload, request: Request): detail="Unread updates are temporarily unavailable", headers={"Retry-After": "1"}, ) from error - existing_ids = { - int(item["id"]) + existing_revisions = { + int(item["id"]): str(item.get("updated_at") or "") for item in current.get("items", []) if isinstance(item, dict) and str(item.get("id", "")).isdigit() } await asyncio.to_thread( - _push_subscription_store.mark_delivered, device_id, existing_ids + _push_subscription_store.mark_delivered, device_id, existing_revisions ) return {"subscribed": True} diff --git a/src/push_notifications.py b/src/push_notifications.py index 920a935..128a52d 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -59,11 +59,14 @@ async def dispatch_unread_updates( return 0 try: page = await unread() + if page.get("complete") is False: + return 0 thread_revisions = { int(item["id"]): str(item.get("updated_at") or "") for item in page.get("items", []) if isinstance(item, dict) and str(item.get("id", "")).isdigit() } + await asyncio.to_thread(store.reconcile_unread, thread_revisions) deliveries = await asyncio.to_thread(store.claim_unseen, thread_revisions) if session_active is not None: try: diff --git a/src/push_subscription_store.py b/src/push_subscription_store.py index ce4edc1..9e9139f 100644 --- a/src/push_subscription_store.py +++ b/src/push_subscription_store.py @@ -223,6 +223,33 @@ class PushSubscriptionStore: ) return deliveries + def reconcile_unread(self, thread_ids: Iterable[int]) -> None: + """Prune per-device checkpoints that are absent from a complete snapshot.""" + unread_ids = tuple(sorted({int(value) for value in thread_ids if int(value) > 0})) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + "CREATE TEMP TABLE current_unread(thread_id INTEGER PRIMARY KEY)" + ) + connection.executemany( + "INSERT INTO current_unread(thread_id) VALUES (?)", + ((thread_id,) for thread_id in unread_ids), + ) + connection.execute( + """DELETE FROM push_deliveries + WHERE NOT EXISTS ( + SELECT 1 FROM current_unread + WHERE current_unread.thread_id = push_deliveries.thread_id + )""" + ) + connection.execute( + """DELETE FROM push_digest_pending + WHERE NOT EXISTS ( + SELECT 1 FROM current_unread + WHERE current_unread.thread_id = push_digest_pending.thread_id + )""" + ) + def mark_digest_pending( self, session_id: str, diff --git a/tests/test_gitea_notifications.py b/tests/test_gitea_notifications.py index e90bf86..74e9dd5 100644 --- a/tests/test_gitea_notifications.py +++ b/tests/test_gitea_notifications.py @@ -40,6 +40,123 @@ async def test_notification_page_preserves_upstream_total_without_loading_other_ assert [item["id"] for item in result["items"]] == [42] +@pytest.mark.anyio +async def test_unread_notification_snapshot_loads_every_page_with_bounded_concurrency(): + active = 0 + peak = 0 + requested_pages = [] + + async def upstream(request): + nonlocal active, peak + page = int(request.url.params["page"]) + requested_pages.append(page) + active += 1 + peak = max(peak, active) + await __import__("asyncio").sleep(0) + active -= 1 + start = (page - 1) * 50 + 1 + end = min(start + 50, 126) + return httpx.Response( + 200, + headers={"X-Total-Count": "125"}, + json=[{ + "id": thread_id, + "updated_at": f"2026-08-11T12:{thread_id % 60:02d}:00Z", + "subject": {}, + "repository": {}, + } for thread_id in range(start, end)], + ) + + gitea_proxy.start_client(transport=httpx.MockTransport(upstream)) + try: + result = await gitea_proxy.unread_notification_snapshot( + max_pages=3, max_concurrency=2, deadline_seconds=1 + ) + finally: + await gitea_proxy.stop_client() + + assert requested_pages[0] == 1 + assert sorted(requested_pages) == [1, 2, 3] + assert peak == 2 + assert result["complete"] is True + assert result["total"] == 125 + assert [item["id"] for item in result["items"]] == list(range(1, 126)) + + +@pytest.mark.anyio +async def test_unread_notification_snapshot_rejects_pagination_that_changes_mid_scan(): + def upstream(request): + page = int(request.url.params["page"]) + total = 51 if page == 1 else 50 + start = (page - 1) * 50 + 1 + return httpx.Response( + 200, + headers={"X-Total-Count": str(total)}, + json=[{ + "id": thread_id, + "subject": {}, + "repository": {}, + } for thread_id in range(start, min(start + 50, total + 1))], + ) + + gitea_proxy.start_client(transport=httpx.MockTransport(upstream)) + try: + with pytest.raises(ValueError, match="changed during the scan"): + await gitea_proxy.unread_notification_snapshot(deadline_seconds=1) + finally: + await gitea_proxy.stop_client() + + +@pytest.mark.anyio +async def test_unread_notification_snapshot_rejects_missing_or_duplicate_threads(): + def upstream(_request): + return httpx.Response( + 200, + headers={"X-Total-Count": "3"}, + json=[ + {"id": 7, "subject": {}, "repository": {}}, + {"id": 7, "subject": {}, "repository": {}}, + "malformed", + ], + ) + + gitea_proxy.start_client(transport=httpx.MockTransport(upstream)) + try: + with pytest.raises(ValueError, match="incomplete thread set"): + await gitea_proxy.unread_notification_snapshot(deadline_seconds=1) + finally: + await gitea_proxy.stop_client() + + +@pytest.mark.anyio +async def test_unread_notification_snapshot_fails_before_exceeding_the_scan_limit(): + requested_pages = [] + + def upstream(request): + page = int(request.url.params["page"]) + requested_pages.append(page) + start = (page - 1) * 50 + 1 + return httpx.Response( + 200, + headers={"X-Total-Count": "101"}, + json=[ + {"id": thread_id, "subject": {}, "repository": {}} + for thread_id in range(start, min(start + 50, 102)) + ], + ) + + gitea_proxy.start_client(transport=httpx.MockTransport(upstream)) + try: + with pytest.raises(ValueError, match="scan limit"): + await gitea_proxy.unread_notification_snapshot( + max_pages=2, deadline_seconds=1 + ) + finally: + await gitea_proxy.stop_client() + + assert requested_pages == [1] + + @pytest.mark.anyio async def test_unread_notifications_are_normalized_for_mobile_handoff(): result = gitea_proxy._normalize_notifications( diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 7c9a725..4966971 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -84,6 +84,67 @@ def test_subscription_store_does_not_realert_for_an_older_snapshot_revision(tmp_ assert store.claim_unseen({42: "2026-08-11T12:00:00Z"}) == [] +def test_complete_unread_snapshot_reconciles_delivery_and_digest_checkpoints(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("session-a", { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + revisions = {1: "r1", 2: "r2", 3: "r3"} + store.mark_delivered("session-a", revisions) + store.mark_digest_pending("session-a", {3: "r3"}) + + store.reconcile_unread({2}) + delivery = store.claim_unseen(revisions)[0] + + assert delivery.thread_ids == (1, 3) + assert delivery.digest_ids == () + + +@pytest.mark.anyio +async def test_dispatch_reconciles_stale_checkpoints_after_a_complete_snapshot(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("session-a", { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + store.mark_delivered("session-a", {1: "r1"}) + + async def unread(): + return {"items": [{"id": 2, "updated_at": "r2"}], "complete": True} + + async def send(_subscription, _payload): + return None + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates(store, config, unread, send) == 1 + + remaining = store.claim_unseen({1: "r1", 2: "r2"}) + assert remaining[0].thread_ids == (1,) + + +@pytest.mark.anyio +async def test_dispatch_sends_nothing_and_preserves_checkpoints_for_an_incomplete_snapshot(tmp_path): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + store.upsert("session-a", { + "endpoint": "https://push.example/device-a", + "keys": {"p256dh": "public-key", "auth": "auth-secret"}, + }) + store.mark_delivered("session-a", {1: "r1"}) + sent = [] + + async def unread(): + return {"items": [{"id": 2, "updated_at": "r2"}], "complete": False} + + async def send(_subscription, payload): + sent.append(payload) + + config = PushConfiguration("public", "private", "mailto:ops@example.com") + assert await dispatch_unread_updates(store, config, unread, send) == 0 + assert sent == [] + assert store.claim_unseen({1: "r1"}) == [] + + def test_subscription_store_migrates_legacy_delivery_without_replaying_it(tmp_path): path = tmp_path / "push.sqlite3" subscription = { @@ -140,7 +201,8 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch async def no_wait(_seconds): return None - async def stop_after_capture(*_args, **kwargs): + async def stop_after_capture(*args, **kwargs): + captured["unread"] = args[2] captured.update(kwargs) raise asyncio.CancelledError @@ -152,6 +214,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch with pytest.raises(asyncio.CancelledError): await main._push_poll_loop() + assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot assert captured["session_active"] is dashboard_auth.managed_session_active assert captured["max_concurrency"] == 3 assert captured["max_individual_notifications"] == 4 @@ -638,10 +701,10 @@ async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe( monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id) - async def no_unread(): - return {"items": []} + async def no_unread(**_kwargs): + return {"items": [], "complete": True} - monkeypatch.setattr(main, "notifications", no_unread) + monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", no_unread) request = SimpleNamespace(state=SimpleNamespace( dashboard_session=SimpleNamespace(session_id="session-a") )) @@ -659,6 +722,50 @@ async def test_authenticated_device_can_subscribe_report_status_and_unsubscribe( assert (await main.push_status(request))["subscribed"] is False +@pytest.mark.anyio +async def test_subscription_baselines_every_revision_from_the_complete_snapshot(tmp_path, monkeypatch): + store = PushSubscriptionStore(tmp_path / "push.sqlite3") + monkeypatch.setattr(main, "_push_subscription_store", store) + monkeypatch.setattr( + main, + "_push_configuration", + lambda: PushConfiguration("public", "private", "mailto:ops@example.com"), + ) + + async def management_id(_session): + return "device-a" + + revisions = { + thread_id: f"2026-08-11T12:{thread_id % 60:02d}:00Z" + for thread_id in range(1, 126) + } + + async def complete_snapshot(**_kwargs): + return { + "items": [ + {"id": thread_id, "updated_at": revision} + for thread_id, revision in revisions.items() + ], + "total": 125, + "complete": True, + } + + async def first_page_must_not_be_used(): + raise AssertionError("subscription used the first-page inbox endpoint") + + monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id) + monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", complete_snapshot) + monkeypatch.setattr(main, "notifications", first_page_must_not_be_used) + request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object())) + payload = main.PushSubscriptionPayload( + endpoint="https://push.example/device-a", + keys={"p256dh": "public-key", "auth": "auth-secret"}, + ) + + assert await main.subscribe_push(payload, request) == {"subscribed": True} + assert store.claim_unseen(revisions) == [] + + @pytest.mark.anyio async def test_subscription_baselines_production_notification_ids(tmp_path, monkeypatch): store = PushSubscriptionStore(tmp_path / "push.sqlite3") @@ -672,11 +779,11 @@ async def test_subscription_baselines_production_notification_ids(tmp_path, monk async def management_id(_session): return "device-a" - async def current_unread(): - return {"items": [{"id": 42}]} + async def current_unread(**_kwargs): + return {"items": [{"id": 42}], "complete": True} monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id) - monkeypatch.setattr(main, "notifications", current_unread) + monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", current_unread) request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object())) payload = main.PushSubscriptionPayload( endpoint="https://push.example/device-a", @@ -700,11 +807,11 @@ async def test_subscription_fails_closed_when_existing_unread_baseline_is_unavai async def management_id(_session): return "device-a" - async def unavailable(): + async def unavailable(**_kwargs): raise RuntimeError("Gitea unavailable") monkeypatch.setattr(main.dashboard_auth, "session_management_id", management_id) - monkeypatch.setattr(main, "notifications", unavailable) + monkeypatch.setattr(main.gitea_proxy, "unread_notification_snapshot", unavailable) request = SimpleNamespace(state=SimpleNamespace(dashboard_session=object())) payload = main.PushSubscriptionPayload( endpoint="https://push.example/device-a",