feat: make start-day reminders lease-safe (Closes #1168)

src/main.py

@@ -180,6 +180,7 @@ async def _push_poll_loop() -> None:
                 session_statuses=dashboard_auth.managed_session_statuses,
                 send_timeout_seconds=send_timeout,
                 lease_seconds=lease_seconds,
+                max_concurrency=max_concurrency,
             )
 
     channel_tasks = (

src/push_notifications.py

@@ -366,6 +366,7 @@ async def dispatch_start_day_reminders(
     session_statuses: Callable[[list[str]], Awaitable[dict[str, str]]] | None = None,
     send_timeout_seconds: float = 10.0,
     lease_seconds: float = 60.0,
+    max_concurrency: int = 8,
 ) -> int:
     if not configuration.enabled:
         return 0
@@ -426,34 +427,50 @@ async def dispatch_start_day_reminders(
             "tag": f"stackchain-start-day-{plan_date}",
             "plan_date": plan_date,
         }, separators=(",", ":"))
-        delivered = 0
-        for device, _local_day in due_devices:
-            try:
-                operation = (
-                    send(device.subscription, payload) if send is not None
-                    else send_web_push(device.subscription, payload, configuration)
+        semaphore = asyncio.Semaphore(max(1, max_concurrency))
+
+        async def dispatch_device(device) -> int:
+            async with semaphore:
+                still_owner = await asyncio.to_thread(
+                    store.acquire_dispatch_lease,
+                    owner,
+                    channel="start-day",
+                    now=time.time(),
+                    lease_seconds=max(15.0, lease_seconds, send_timeout_seconds + 5.0),
                 )
-                await asyncio.wait_for(operation, timeout=send_timeout_seconds)
-            except Exception as error:
-                status = getattr(getattr(error, "response", None), "status_code", None)
-                if isinstance(error, UnsafePushEndpoint) or status in {404, 410}:
-                    await asyncio.to_thread(store.delete_session, device.session_id)
-                else:
-                    await asyncio.to_thread(
-                        store.mark_delivery_failed,
-                        device.session_id,
-                        "start-day",
-                        _delivery_failure_reason(error),
+                if not still_owner:
+                    return 0
+                try:
+                    operation = (
+                        send(device.subscription, payload) if send is not None
+                        else send_web_push(device.subscription, payload, configuration)
                     )
-                continue
-            await asyncio.to_thread(
-                store.mark_delivery_succeeded, device.session_id, "start-day"
-            )
-            await asyncio.to_thread(
-                store.mark_start_day_reminder_delivered, device.session_id, plan_date
-            )
-            delivered += 1
-        return delivered
+                    await asyncio.wait_for(operation, timeout=send_timeout_seconds)
+                except Exception as error:
+                    status = getattr(getattr(error, "response", None), "status_code", None)
+                    if isinstance(error, UnsafePushEndpoint) or status in {404, 410}:
+                        await asyncio.to_thread(store.delete_session, device.session_id)
+                    else:
+                        await asyncio.to_thread(
+                            store.mark_delivery_failed,
+                            device.session_id,
+                            "start-day",
+                            _delivery_failure_reason(error),
+                        )
+                    return 0
+                await asyncio.to_thread(
+                    store.mark_delivery_succeeded, device.session_id, "start-day"
+                )
+                await asyncio.to_thread(
+                    store.mark_start_day_reminder_delivered, device.session_id, plan_date
+                )
+                return 1
+
+        results = await asyncio.gather(
+            *(dispatch_device(device) for device, _local_day in due_devices),
+            return_exceptions=True,
+        )
+        return sum(result for result in results if isinstance(result, int))
     finally:
         await asyncio.to_thread(store.release_dispatch_lease, owner, channel="start-day")