src/push_notifications.py

@@ -219,8 +219,17 @@ async def dispatch_unread_updates(
                     count += 1
                 return count
 
+        async def dispatch_device_safely(delivery) -> int:
+            try:
+                return await dispatch_device(delivery)
+            except Exception:
+                # Device-specific validation, persistence, or provider failures
+                # must not cancel healthy siblings. Leave any uncheckpointed
+                # revisions unseen so a later poll can retry them.
+                return 0
+
         counts = await asyncio.gather(
-            *(dispatch_device(delivery) for delivery in deliveries)
+            *(dispatch_device_safely(delivery) for delivery in deliveries)
         )
         return sum(counts)
     finally:

tests/test_push_notifications.py

@@ -2,6 +2,7 @@ import json
 import asyncio
 import os
 import sqlite3
+import time
 from types import SimpleNamespace
 
 import httpx
@@ -567,6 +568,75 @@ 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_unexpected_device_failure_waits_for_siblings_and_retries_only_that_device(tmp_path):
+    path = tmp_path / "push.sqlite3"
+    store = PushSubscriptionStore(path)
+    competitor = PushSubscriptionStore(path)
+    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"},
+        })
+    healthy_started = asyncio.Event()
+    release_healthy = asyncio.Event()
+    sent = []
+
+    async def unread():
+        return {"items": [{"id": 8, "updated_at": "r1"}], "complete": True}
+
+    async def validate(endpoint):
+        if endpoint.endswith("failing-device"):
+            raise TimeoutError("endpoint validation unavailable")
+        return endpoint
+
+    async def send(subscription, _payload):
+        sent.append(subscription["endpoint"])
+        healthy_started.set()
+        await release_healthy.wait()
+
+    config = PushConfiguration("public", "private", "mailto:ops@example.com")
+    dispatch = asyncio.create_task(dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        send,
+        endpoint_validator=validate,
+    ))
+    await asyncio.wait_for(healthy_started.wait(), timeout=0.2)
+    await asyncio.sleep(0)
+
+    assert dispatch.done() is False
+    assert competitor.acquire_dispatch_lease(
+        "competing-worker", now=time.time(), lease_seconds=60
+    ) is False
+
+    release_healthy.set()
+    assert await dispatch == 1
+    remaining = store.claim_unseen({8: "r1"})
+    assert [(item.session_id, item.thread_ids) for item in remaining] == [
+        ("failing-device", (8,))
+    ]
+
+    retried = []
+
+    async def validate_recovered(endpoint):
+        return endpoint
+
+    async def send_retry(subscription, _payload):
+        retried.append(subscription["endpoint"])
+
+    assert await dispatch_unread_updates(
+        store,
+        config,
+        unread,
+        send_retry,
+        endpoint_validator=validate_recovered,
+    ) == 1
+    assert retried == ["https://push.example/failing-device"]
+    assert sent == ["https://push.example/healthy-device"]
+
+
 @pytest.mark.anyio
 async def test_transient_failure_stops_that_device_until_the_next_poll(tmp_path):
     store = PushSubscriptionStore(tmp_path / "push.sqlite3")