fix: isolate deadline reminder dispatch (Closes #711)

src/main.py

@@ -116,18 +116,24 @@ async def _push_poll_loop() -> None:
                 max_concurrency=max_concurrency,
                 max_individual_notifications=max_individual_notifications,
             )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            pass
+        try:
             await dispatch_deadline_reminders(
                 _push_subscription_store,
                 _push_configuration(),
                 gitea_proxy.assigned_issue_snapshot,
                 session_active=dashboard_auth.managed_session_active,
                 send_timeout_seconds=send_timeout,
+                lease_seconds=lease_seconds,
+                max_concurrency=max_concurrency,
             )
         except asyncio.CancelledError:
             raise
         except Exception:
-            # Gitea and push endpoints are external; one failed poll must not
+            # Gitea and push endpoints retry independently on the next poll.
-            # stop later delivery attempts.
             continue
 
 

src/push_notifications.py

@@ -62,6 +62,7 @@ async def dispatch_unread_updates(
     acquired = await asyncio.to_thread(
         store.acquire_dispatch_lease,
         owner,
+        channel="unread",
         now=time.time(),
         lease_seconds=lease_seconds,
     )
@@ -128,6 +129,7 @@ async def dispatch_unread_updates(
                     still_owner = await asyncio.to_thread(
                         store.acquire_dispatch_lease,
                         owner,
+                        channel="unread",
                         now=time.time(),
                         lease_seconds=lease_seconds,
                     )
@@ -174,6 +176,7 @@ async def dispatch_unread_updates(
                     still_owner = await asyncio.to_thread(
                         store.acquire_dispatch_lease,
                         owner,
+                        channel="unread",
                         now=time.time(),
                         lease_seconds=lease_seconds,
                     )
@@ -235,7 +238,7 @@ async def dispatch_unread_updates(
         )
         return sum(counts)
     finally:
-        await asyncio.to_thread(store.release_dispatch_lease, owner)
+        await asyncio.to_thread(store.release_dispatch_lease, owner, channel="unread")
 
 
 async def dispatch_deadline_reminders(
@@ -249,17 +252,22 @@ async def dispatch_deadline_reminders(
     acquired = await asyncio.to_thread(
         store.acquire_dispatch_lease,
         owner,
+        channel="deadline",
         now=time.time(),
-        lease_seconds=max(15.0, float(kwargs.get("send_timeout_seconds", 10.0)) + 5.0),
+        lease_seconds=max(
+            15.0,
+            float(kwargs.get("lease_seconds", 60.0)),
+            float(kwargs.get("send_timeout_seconds", 10.0)) + 5.0,
+        ),
     )
     if not acquired:
         return 0
     try:
         return await _dispatch_deadline_reminders_unlocked(
-            store, configuration, assigned, send, **kwargs
+            store, configuration, assigned, send, owner=owner, **kwargs
         )
     finally:
-        await asyncio.to_thread(store.release_dispatch_lease, owner)
+        await asyncio.to_thread(store.release_dispatch_lease, owner, channel="deadline")
 
 
 async def _dispatch_deadline_reminders_unlocked(
@@ -271,6 +279,9 @@ async def _dispatch_deadline_reminders_unlocked(
     now: datetime | None = None,
     session_active: Callable[[str], Awaitable[bool]] | None = None,
     send_timeout_seconds: float = 10.0,
+    lease_seconds: float = 60.0,
+    max_concurrency: int = 8,
+    owner: str,
 ) -> int:
     """Send one privacy-safe Agenda digest per eligible device and local day."""
     if not configuration.enabled:
@@ -297,18 +308,32 @@ async def _dispatch_deadline_reminders_unlocked(
             due_count += 1
     if not due_count:
         return 0
-    delivered = 0
+    semaphore = asyncio.Semaphore(max(1, max_concurrency))
-    for device in devices:
+
+    async def dispatch_device(device) -> int:
+        async with semaphore:
             try:
                 local_now = current.astimezone(ZoneInfo(device.timezone))
             except ZoneInfoNotFoundError:
-            continue
+                return 0
             local_day = local_now.date().isoformat()
-        if local_now.hour < device.reminder_hour or device.delivered_local_day == local_day:
+            if (
-            continue
+                local_now.hour < device.reminder_hour
+                or device.delivered_local_day == local_day
+            ):
+                return 0
             if session_active is not None and not await session_active(device.session_id):
                 await asyncio.to_thread(store.delete_session, device.session_id)
-            continue
+                return 0
+            still_owner = await asyncio.to_thread(
+                store.acquire_dispatch_lease,
+                owner,
+                channel="deadline",
+                now=time.time(),
+                lease_seconds=max(15.0, lease_seconds, send_timeout_seconds + 5.0),
+            )
+            if not still_owner:
+                return 0
             payload = json.dumps({
                 "title": f"{due_count} deadline{'s' if due_count != 1 else ''} need{'s' if due_count == 1 else ''} attention",
                 "body": f"Open Agenda to review or replan {'it' if due_count == 1 else 'them'}.",
@@ -324,9 +349,13 @@ async def _dispatch_deadline_reminders_unlocked(
                 )
                 await asyncio.wait_for(operation, timeout=send_timeout_seconds)
             except Exception:
-            continue
+                return 0
             await asyncio.to_thread(
                 store.mark_deadline_reminder_delivered, device.session_id, local_day
             )
-        delivered += 1
+            return 1
-    return delivered
+
+    results = await asyncio.gather(
+        *(dispatch_device(device) for device in devices), return_exceptions=True
+    )
+    return sum(result for result in results if isinstance(result, int))

src/push_subscription_store.py

@@ -75,7 +75,7 @@ class PushSubscriptionStore:
                         ON DELETE CASCADE
                 );
                 CREATE TABLE IF NOT EXISTS push_dispatch_lease (
-                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                    channel TEXT PRIMARY KEY,
                     owner TEXT NOT NULL,
                     expires_at REAL NOT NULL
                 );
@@ -113,6 +113,24 @@ class PushSubscriptionStore:
                 connection.execute(
                     "ALTER TABLE push_digest_pending ADD COLUMN revision TEXT NOT NULL DEFAULT '*'"
                 )
+            lease_columns = {
+                row[1] for row in connection.execute("PRAGMA table_info(push_dispatch_lease)")
+            }
+            if "singleton" in lease_columns:
+                connection.executescript(
+                    """
+                    ALTER TABLE push_dispatch_lease RENAME TO push_dispatch_lease_legacy;
+                    CREATE TABLE push_dispatch_lease (
+                        channel TEXT PRIMARY KEY,
+                        owner TEXT NOT NULL,
+                        expires_at REAL NOT NULL
+                    );
+                    INSERT INTO push_dispatch_lease(channel, owner, expires_at)
+                        SELECT 'unread', owner, expires_at
+                        FROM push_dispatch_lease_legacy WHERE singleton = 1;
+                    DROP TABLE push_dispatch_lease_legacy;
+                    """
+                )
         os.chmod(self.path, 0o600)
 
     def _connect(self):
@@ -121,29 +139,30 @@ class PushSubscriptionStore:
         return connection
 
     def acquire_dispatch_lease(
-        self, owner: str, *, now: float, lease_seconds: float
+        self, owner: str, *, channel: str = "unread", now: float, lease_seconds: float
     ) -> bool:
         with self._connect() as connection:
             connection.execute("BEGIN IMMEDIATE")
             current = connection.execute(
-                "SELECT owner, expires_at FROM push_dispatch_lease WHERE singleton = 1"
+                "SELECT owner, expires_at FROM push_dispatch_lease WHERE channel = ?",
+                (channel,),
             ).fetchone()
             if current is not None and current[0] != owner and current[1] > now:
                 return False
             connection.execute(
-                """INSERT INTO push_dispatch_lease(singleton, owner, expires_at)
+                """INSERT INTO push_dispatch_lease(channel, owner, expires_at)
-                   VALUES (1, ?, ?)
+                   VALUES (?, ?, ?)
-                   ON CONFLICT(singleton) DO UPDATE SET
+                   ON CONFLICT(channel) DO UPDATE SET
                        owner = excluded.owner, expires_at = excluded.expires_at""",
-                (owner, now + lease_seconds),
+                (channel, owner, now + lease_seconds),
             )
             return True
 
-    def release_dispatch_lease(self, owner: str) -> bool:
+    def release_dispatch_lease(self, owner: str, *, channel: str = "unread") -> bool:
         with self._connect() as connection:
             result = connection.execute(
-                "DELETE FROM push_dispatch_lease WHERE singleton = 1 AND owner = ?",
+                "DELETE FROM push_dispatch_lease WHERE channel = ? AND owner = ?",
-                (owner,),
+                (channel, owner),
             )
             return result.rowcount == 1
 

tests/test_deadline_reminders.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 from datetime import datetime, timezone
 
@@ -160,3 +161,43 @@ async def test_competing_workers_send_one_deadline_digest(tmp_path):
     assert competing == 0
     assert await active == 1
     assert len(sent) == 1
+
+
+@pytest.mark.anyio
+async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    for index in range(4):
+        session_id = f"device-{index}"
+        store.upsert(session_id, {
+            "endpoint": f"https://push.example/{session_id}",
+            "keys": {"p256dh": "public-key", "auth": "auth-secret"},
+        })
+        store.set_deadline_preferences(
+            session_id, enabled=True, timezone="UTC", reminder_hour=9
+        )
+    active = 0
+    peak = 0
+
+    async def assigned():
+        return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}
+
+    async def send(subscription, _payload):
+        nonlocal active, peak
+        active += 1
+        peak = max(peak, active)
+        await asyncio.sleep(0.01)
+        active -= 1
+        if subscription["endpoint"].endswith("device-1"):
+            raise RuntimeError("provider failed")
+
+    delivered = await dispatch_deadline_reminders(
+        store,
+        PushConfiguration("public", "private", "mailto:ops@example.com"),
+        assigned,
+        send,
+        now=datetime(2026, 8, 13, 10, 0, tzinfo=timezone.utc),
+        max_concurrency=2,
+    )
+
+    assert delivered == 3
+    assert peak == 2

tests/test_push_notifications.py

@@ -230,6 +230,51 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
     assert captured["max_individual_notifications"] == 4
 
 
+@pytest.mark.anyio
+async def test_push_poll_still_dispatches_deadlines_when_unread_dispatch_fails(monkeypatch):
+    calls = []
+
+    sleeps = 0
+
+    async def no_wait(_seconds):
+        nonlocal sleeps
+        sleeps += 1
+        if sleeps > 1:
+            raise asyncio.CancelledError
+
+    async def fail_unread(*_args, **_kwargs):
+        calls.append("unread")
+        raise RuntimeError("unread unavailable")
+
+    async def dispatch_deadlines(*_args, **_kwargs):
+        calls.append("deadline")
+
+    monkeypatch.setattr(main.asyncio, "sleep", no_wait)
+    monkeypatch.setattr(main, "dispatch_unread_updates", fail_unread)
+    monkeypatch.setattr(main, "dispatch_deadline_reminders", dispatch_deadlines)
+
+    with pytest.raises(asyncio.CancelledError):
+        await main._push_poll_loop()
+
+    assert calls == ["unread", "deadline"]
+
+
+def test_dispatch_leases_are_isolated_by_channel(tmp_path):
+    path = tmp_path / "push.sqlite3"
+    first = PushSubscriptionStore(path)
+    second = PushSubscriptionStore(path)
+
+    assert first.acquire_dispatch_lease(
+        "unread-owner", channel="unread", now=100, lease_seconds=30
+    ) is True
+    assert second.acquire_dispatch_lease(
+        "deadline-owner", channel="deadline", now=100, lease_seconds=30
+    ) is True
+    assert second.acquire_dispatch_lease(
+        "other-unread-owner", channel="unread", now=100, lease_seconds=30
+    ) is False
+
+
 def test_replacing_subscription_resets_delivery_cursor_and_revocation_removes_device(tmp_path):
     store = PushSubscriptionStore(tmp_path / "push.sqlite3")
     original = {