perf: bound Web Push device fan-out (Closes #557)

README.md

@@ -193,10 +193,12 @@ export STACKCHAIN_VAPID_PUBLIC_KEY='<url-safe-public-key>'
 export STACKCHAIN_VAPID_PRIVATE_KEY='<private-key-from-secret-manager>'
 export STACKCHAIN_VAPID_SUBJECT='mailto:ops@example.com'
 # Optional; defaults to a 30-second poll, 10-second endpoint deadline,
-# 60-second renewable cross-worker lease, and
-# STACKCHAIN_STATE_DIR/push-subscriptions.sqlite3.
+# 8 concurrently dispatched devices, 60-second renewable cross-worker lease,
+# and STACKCHAIN_STATE_DIR/push-subscriptions.sqlite3. Threads stay ordered
+# within each device; one slow device does not delay healthy devices behind it.
 export STACKCHAIN_PUSH_POLL_SECONDS=30
 export STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS=10
+export STACKCHAIN_PUSH_MAX_CONCURRENCY=8
 export STACKCHAIN_PUSH_LEASE_SECONDS=60
 export STACKCHAIN_PUSH_DB='/var/lib/stackchain-dashboard/push-subscriptions.sqlite3'
 uvicorn src.main:app --host 127.0.0.1 --port 8000

src/main.py

@@ -87,6 +87,9 @@ async def _push_poll_loop() -> None:
         send_timeout + 5.0,
         float(os.getenv("STACKCHAIN_PUSH_LEASE_SECONDS", "60")),
     )
+    max_concurrency = max(
+        1, int(os.getenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "8"))
+    )
     while True:
         await asyncio.sleep(interval)
         try:
@@ -97,6 +100,7 @@ async def _push_poll_loop() -> None:
                 session_active=dashboard_auth.managed_session_active,
                 lease_seconds=lease_seconds,
                 send_timeout_seconds=send_timeout,
+                max_concurrency=max_concurrency,
             )
         except asyncio.CancelledError:
             raise

src/push_notifications.py

@@ -43,6 +43,7 @@ async def dispatch_unread_updates(
     session_active: Callable[[str], Awaitable[bool]] | None = None,
     lease_seconds: float = 60.0,
     send_timeout_seconds: float = 10.0,
+    max_concurrency: int = 8,
 ) -> int:
     if not configuration.enabled:
         return 0
@@ -78,46 +79,62 @@ async def dispatch_unread_updates(
                 for delivery, active in zip(deliveries, authorized)
                 if active
             ]
-        count = 0
-        for delivery in deliveries:
-            for thread_id in delivery.thread_ids:
-                still_owner = await asyncio.to_thread(
-                    store.acquire_dispatch_lease,
-                    owner,
-                    now=time.time(),
-                    lease_seconds=lease_seconds,
-                )
-                if not still_owner:
-                    return count
-                payload = json.dumps(
-                    {
-                        "title": "New work update",
-                        "body": "Tap to review it in Stackchain.",
-                        "route": f"#/my-work/update/{thread_id}",
-                        "tag": f"stackchain-update-{thread_id}",
-                    },
-                    separators=(",", ":"),
-                )
-                try:
-                    if send is None:
-                        operation = send_web_push(
-                            delivery.subscription, payload, configuration
+        semaphore = asyncio.Semaphore(max(1, max_concurrency))
+        ownership_lost = asyncio.Event()
+
+        async def dispatch_device(delivery) -> int:
+            async with semaphore:
+                if ownership_lost.is_set():
+                    return 0
+                count = 0
+                for thread_id in delivery.thread_ids:
+                    still_owner = await asyncio.to_thread(
+                        store.acquire_dispatch_lease,
+                        owner,
+                        now=time.time(),
+                        lease_seconds=lease_seconds,
+                    )
+                    if not still_owner:
+                        ownership_lost.set()
+                        return count
+                    payload = json.dumps(
+                        {
+                            "title": "New work update",
+                            "body": "Tap to review it in Stackchain.",
+                            "route": f"#/my-work/update/{thread_id}",
+                            "tag": f"stackchain-update-{thread_id}",
+                        },
+                        separators=(",", ":"),
+                    )
+                    try:
+                        if send is None:
+                            operation = send_web_push(
+                                delivery.subscription, payload, configuration
+                            )
+                        else:
+                            operation = send(delivery.subscription, payload)
+                        await asyncio.wait_for(operation, timeout=send_timeout_seconds)
+                    except Exception as error:
+                        status = getattr(
+                            getattr(error, "response", None), "status_code", None
                         )
-                    else:
-                        operation = send(delivery.subscription, payload)
-                    await asyncio.wait_for(operation, timeout=send_timeout_seconds)
-                except Exception as error:
-                    status = getattr(getattr(error, "response", None), "status_code", None)
-                    if status in {404, 410}:
-                        await asyncio.to_thread(store.delete_session, delivery.session_id)
+                        if status in {404, 410}:
+                            await asyncio.to_thread(
+                                store.delete_session, delivery.session_id
+                            )
+                            break
+                        # Leave this device's transient failures unseen for a later
+                        # poll instead of paying the endpoint deadline repeatedly.
                         break
-                    # A transient provider failure belongs to this endpoint;
-                    # leave it unseen for a later poll and continue fan-out.
-                    continue
-                await asyncio.to_thread(
-                    store.mark_delivered, delivery.session_id, (thread_id,)
-                )
-                count += 1
-        return count
+                    await asyncio.to_thread(
+                        store.mark_delivered, delivery.session_id, (thread_id,)
+                    )
+                    count += 1
+                return count
+
+        counts = await asyncio.gather(
+            *(dispatch_device(delivery) for delivery in deliveries)
+        )
+        return sum(counts)
     finally:
         await asyncio.to_thread(store.release_dispatch_lease, owner)

tests/test_push_notifications.py

@@ -83,6 +83,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
         captured.update(kwargs)
         raise asyncio.CancelledError
 
+    monkeypatch.setenv("STACKCHAIN_PUSH_MAX_CONCURRENCY", "3")
     monkeypatch.setattr(main.asyncio, "sleep", no_wait)
     monkeypatch.setattr(main, "dispatch_unread_updates", stop_after_capture)
 
@@ -90,6 +91,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
         await main._push_poll_loop()
 
     assert captured["session_active"] is dashboard_auth.managed_session_active
+    assert captured["max_concurrency"] == 3
 
 
 def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
@@ -224,6 +226,38 @@ async def test_transient_endpoint_failure_does_not_block_healthy_devices(tmp_pat
     assert store.claim_unseen({8})[0].session_id == "session-a"
 
 
+@pytest.mark.anyio
+async def test_transient_failure_stops_that_device_until_the_next_poll(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    for session_id in ("failing-device", "healthy-device"):
+        store.upsert(session_id, {
+            "endpoint": f"https://push.example/{session_id}",
+            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
+        })
+    attempts = []
+
+    async def unread():
+        return {"items": [
+            {"notification_id": 8},
+            {"notification_id": 9},
+        ]}
+
+    async def send(subscription, payload):
+        endpoint = subscription["endpoint"]
+        thread_id = int(json.loads(payload)["route"].rsplit("/", 1)[1])
+        attempts.append((endpoint, thread_id))
+        if endpoint.endswith("failing-device"):
+            raise RuntimeError("provider unavailable")
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    assert await dispatch_unread_updates(store, config, unread, send) == 2
+    assert [item for item in attempts if item[0].endswith("failing-device")] == [
+        ("https://push.example/failing-device", 8)
+    ]
+    remaining = {item.session_id: item.thread_ids for item in store.claim_unseen({8, 9})}
+    assert remaining == {"failing-device": (8, 9)}
+
+
 @pytest.mark.anyio
 async def test_timed_out_endpoint_does_not_stall_push_fanout(tmp_path):
     store = PushSubscriptionStore(tmp_path / "push.sqlite3")
@@ -254,6 +288,45 @@ async def test_timed_out_endpoint_does_not_stall_push_fanout(tmp_path):
     assert sent == ["https://push.example/session-b"]
 
 
+@pytest.mark.anyio
+async def test_dispatches_devices_concurrently_with_a_strict_bound(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    for index in range(4):
+        store.upsert(f"session-{index}", {
+            "endpoint": f"https://push.example/session-{index}",
+            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
+        })
+    active = 0
+    peak = 0
+    bound_reached = asyncio.Event()
+    release = asyncio.Event()
+
+    async def unread():
+        return {"items": [{"notification_id": 13}]}
+
+    async def send(_subscription, _payload):
+        nonlocal active, peak
+        active += 1
+        peak = max(peak, active)
+        if active == 2:
+            bound_reached.set()
+        await release.wait()
+        active -= 1
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    dispatch = asyncio.create_task(dispatch_unread_updates(
+        store, config, unread, send, max_concurrency=2
+    ))
+    await asyncio.wait_for(bound_reached.wait(), timeout=0.2)
+
+    assert peak == 2
+    assert dispatch.done() is False
+
+    release.set()
+    assert await dispatch == 4
+    assert peak == 2
+
+
 @pytest.mark.anyio
 async def test_dispatch_removes_inactive_sessions_without_blocking_active_devices(tmp_path):
     store = PushSubscriptionStore(tmp_path / "push.sqlite3")