src/main.py

@@ -116,24 +116,18 @@ 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 retry independently on the next poll.
+            # Gitea and push endpoints are external; one failed poll must not
+            # stop later delivery attempts.
             continue
 
 

src/push_notifications.py

@@ -62,7 +62,6 @@ async def dispatch_unread_updates(
     acquired = await asyncio.to_thread(
         store.acquire_dispatch_lease,
         owner,
-        channel="unread",
         now=time.time(),
         lease_seconds=lease_seconds,
     )
@@ -129,7 +128,6 @@ 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,
                     )
@@ -176,7 +174,6 @@ 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,
                     )
@@ -238,7 +235,7 @@ async def dispatch_unread_updates(
         )
         return sum(counts)
     finally:
-        await asyncio.to_thread(store.release_dispatch_lease, owner, channel="unread")
+        await asyncio.to_thread(store.release_dispatch_lease, owner)
 
 
 async def dispatch_deadline_reminders(
@@ -252,22 +249,17 @@ async def dispatch_deadline_reminders(
     acquired = await asyncio.to_thread(
         store.acquire_dispatch_lease,
         owner,
-        channel="deadline",
         now=time.time(),
-        lease_seconds=max(
+        lease_seconds=max(15.0, float(kwargs.get("send_timeout_seconds", 10.0)) + 5.0),
-            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, owner=owner, **kwargs
+            store, configuration, assigned, send, **kwargs
         )
     finally:
-        await asyncio.to_thread(store.release_dispatch_lease, owner, channel="deadline")
+        await asyncio.to_thread(store.release_dispatch_lease, owner)
 
 
 async def _dispatch_deadline_reminders_unlocked(
@@ -279,9 +271,6 @@ 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:
@@ -308,32 +297,18 @@ async def _dispatch_deadline_reminders_unlocked(
             due_count += 1
     if not due_count:
         return 0
-    semaphore = asyncio.Semaphore(max(1, max_concurrency))
+    delivered = 0
-
+    for device in devices:
-    async def dispatch_device(device) -> int:
-        async with semaphore:
         try:
             local_now = current.astimezone(ZoneInfo(device.timezone))
         except ZoneInfoNotFoundError:
-                return 0
+            continue
         local_day = local_now.date().isoformat()
-            if (
+        if local_now.hour < device.reminder_hour or device.delivered_local_day == local_day:
-                local_now.hour < device.reminder_hour
+            continue
-                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)
-                return 0
+            continue
-            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'}.",
@@ -349,13 +324,9 @@ async def _dispatch_deadline_reminders_unlocked(
             )
             await asyncio.wait_for(operation, timeout=send_timeout_seconds)
         except Exception:
-                return 0
+            continue
         await asyncio.to_thread(
             store.mark_deadline_reminder_delivered, device.session_id, local_day
         )
-            return 1
+        delivered += 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 (
-                    channel TEXT PRIMARY KEY,
+                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
                     owner TEXT NOT NULL,
                     expires_at REAL NOT NULL
                 );
@@ -113,24 +113,6 @@ 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):
@@ -139,30 +121,29 @@ class PushSubscriptionStore:
         return connection
 
     def acquire_dispatch_lease(
-        self, owner: str, *, channel: str = "unread", now: float, lease_seconds: float
+        self, owner: str, *, 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 channel = ?",
+                "SELECT owner, expires_at FROM push_dispatch_lease WHERE singleton = 1"
-                (channel,),
             ).fetchone()
             if current is not None and current[0] != owner and current[1] > now:
                 return False
             connection.execute(
-                """INSERT INTO push_dispatch_lease(channel, owner, expires_at)
+                """INSERT INTO push_dispatch_lease(singleton, owner, expires_at)
-                   VALUES (?, ?, ?)
+                   VALUES (1, ?, ?)
-                   ON CONFLICT(channel) DO UPDATE SET
+                   ON CONFLICT(singleton) DO UPDATE SET
                        owner = excluded.owner, expires_at = excluded.expires_at""",
-                (channel, owner, now + lease_seconds),
+                (owner, now + lease_seconds),
             )
             return True
 
-    def release_dispatch_lease(self, owner: str, *, channel: str = "unread") -> bool:
+    def release_dispatch_lease(self, owner: str) -> bool:
         with self._connect() as connection:
             result = connection.execute(
-                "DELETE FROM push_dispatch_lease WHERE channel = ? AND owner = ?",
+                "DELETE FROM push_dispatch_lease WHERE singleton = 1 AND owner = ?",
-                (channel, owner),
+                (owner,),
             )
             return result.rowcount == 1
 

tests/test_deadline_reminders.py

@@ -1,4 +1,3 @@
-import asyncio
 import json
 from datetime import datetime, timezone
 
@@ -161,43 +160,3 @@ 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,51 +230,6 @@ 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 = {