From 79309196b8e5edf16c16cef2a4eea196c472bbea Mon Sep 17 00:00:00 2001 From: timmy Date: Tue, 11 Aug 2026 08:54:48 +0000 Subject: [PATCH] perf: bound Web Push device fan-out (Closes #557) --- README.md | 6 +- src/main.py | 4 ++ src/push_notifications.py | 95 +++++++++++++++++++------------- tests/test_push_notifications.py | 73 ++++++++++++++++++++++++ 4 files changed, 137 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index 265ccc7..4266ab1 100644 --- a/README.md +++ b/README.md @@ -193,10 +193,12 @@ export STACKCHAIN_VAPID_PUBLIC_KEY='' export STACKCHAIN_VAPID_PRIVATE_KEY='' 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 diff --git a/src/main.py b/src/main.py index 0a7e13d..e29fc24 100644 --- a/src/main.py +++ b/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 diff --git a/src/push_notifications.py b/src/push_notifications.py index 5656dd6..a8cafde 100644 --- a/src/push_notifications.py +++ b/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) diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 4c76ee1..7eef745 100644 --- a/tests/test_push_notifications.py +++ b/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") -- 2.43.0