feat: bound bulk notification fan-out (#145)

tests/test_notification_read.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 
 import httpx
 import pytest
@@ -77,6 +78,83 @@ 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_limits_upstream_concurrency(monkeypatch):
+    active = 0
+    peak_active = 0
+    started = asyncio.Event()
+    release = asyncio.Event()
+    calls = []
+
+    monkeypatch.setattr(main, "_read_notification_ids", set())
+    monkeypatch.setattr(main, "_live_snapshot_value", None)
+
+    async def mark(thread_id):
+        nonlocal active, peak_active
+        calls.append(thread_id)
+        active += 1
+        peak_active = max(peak_active, active)
+        if active == 5:
+            started.set()
+        try:
+            await release.wait()
+        finally:
+            active -= 1
+
+    monkeypatch.setattr(main, "mark_notification_read", mark)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        request = asyncio.create_task(
+            client.patch("/api/v1/notifications/read", json={"ids": list(range(1, 51))})
+        )
+        await asyncio.wait_for(started.wait(), timeout=1)
+        await asyncio.sleep(0)
+        assert peak_active == 5
+        assert len(calls) == 5
+        release.set()
+        response = await request
+
+    assert response.status_code == 200
+    assert response.json() == {"marked": list(range(1, 51)), "failed": []}
+    assert sorted(calls) == list(range(1, 51))
+
+
+@pytest.mark.anyio
+async def test_bulk_mark_read_cancels_unfinished_work_at_batch_deadline(monkeypatch):
+    active = 0
+    cancelled = []
+
+    async def mark(thread_id):
+        nonlocal active
+        active += 1
+        try:
+            await asyncio.sleep(0.08)
+        except asyncio.CancelledError:
+            cancelled.append(thread_id)
+            raise
+        finally:
+            active -= 1
+
+    monkeypatch.setattr(main, "mark_notification_read", mark)
+    monkeypatch.setattr(main, "BULK_NOTIFICATION_DEADLINE_SECONDS", 0.01, raising=False)
+    monkeypatch.setattr(main, "_read_notification_ids", set())
+    monkeypatch.setattr(main, "_live_snapshot_value", None)
+    transport = httpx.ASGITransport(app=main.app)
+    started_at = time.monotonic()
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.patch(
+            "/api/v1/notifications/read", json={"ids": list(range(1, 11))}
+        )
+    elapsed = time.monotonic() - started_at
+
+    assert elapsed < 0.06
+    assert response.status_code == 200
+    assert response.json() == {"marked": [], "failed": list(range(1, 11))}
+    assert response.headers["retry-after"] == "1"
+    assert active == 0
+    assert sorted(cancelled) == list(range(1, 6))
+
+
 @pytest.mark.anyio
 async def test_bulk_mark_read_rejects_empty_invalid_and_oversized_batches(monkeypatch):
     marked = []