feat: make push delivery multi-worker safe (Closes #551)

src/main.py

@@ -80,6 +80,13 @@ async def _drain_authored_action_operations() -> None:
 
 async def _push_poll_loop() -> None:
     interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")))
+    send_timeout = max(
+        1.0, float(os.getenv("STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS", "10"))
+    )
+    lease_seconds = max(
+        send_timeout + 5.0,
+        float(os.getenv("STACKCHAIN_PUSH_LEASE_SECONDS", "60")),
+    )
     while True:
         await asyncio.sleep(interval)
         try:
@@ -87,6 +94,8 @@ async def _push_poll_loop() -> None:
                 _push_subscription_store,
                 _push_configuration(),
                 notifications,
+                lease_seconds=lease_seconds,
+                send_timeout_seconds=send_timeout,
             )
         except asyncio.CancelledError:
             raise

src/push_notifications.py

@@ -1,5 +1,7 @@
 import asyncio
 import json
+import secrets
+import time
 from dataclasses import dataclass
 from typing import Awaitable, Callable
 
@@ -37,38 +39,68 @@ async def dispatch_unread_updates(
     configuration: PushConfiguration,
     unread: Callable[[], Awaitable[dict]],
     send: Callable[[dict, str], Awaitable[None]] | None = None,
+    *,
+    lease_seconds: float = 60.0,
+    send_timeout_seconds: float = 10.0,
 ) -> int:
     if not configuration.enabled:
         return 0
-    page = await unread()
-    thread_ids = {
-        int(item["notification_id"])
-        for item in page.get("items", [])
-        if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
-    }
-    count = 0
-    for delivery in store.claim_unseen(thread_ids):
-        for thread_id in delivery.thread_ids:
-            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:
-                    await send_web_push(delivery.subscription, payload, configuration)
-                else:
-                    await send(delivery.subscription, payload)
-            except Exception as error:
-                status = getattr(getattr(error, "response", None), "status_code", None)
-                if status in {404, 410}:
-                    store.delete_session(delivery.session_id)
-                    break
-                raise
-            store.mark_delivered(delivery.session_id, (thread_id,))
-            count += 1
-    return count
+    owner = secrets.token_urlsafe(18)
+    acquired = await asyncio.to_thread(
+        store.acquire_dispatch_lease,
+        owner,
+        now=time.time(),
+        lease_seconds=lease_seconds,
+    )
+    if not acquired:
+        return 0
+    try:
+        page = await unread()
+        thread_ids = {
+            int(item["notification_id"])
+            for item in page.get("items", [])
+            if isinstance(item, dict) and str(item.get("notification_id", "")).isdigit()
+        }
+        count = 0
+        for delivery in await asyncio.to_thread(store.claim_unseen, thread_ids):
+            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
+                        )
+                    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)
+                        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
+    finally:
+        await asyncio.to_thread(store.release_dispatch_lease, owner)

src/push_subscription_store.py

@@ -1,4 +1,5 @@
 import json
+import os
 import sqlite3
 from dataclasses import dataclass
 from pathlib import Path
@@ -18,6 +19,7 @@ class PushSubscriptionStore:
     def __init__(self, path: str | Path):
         self.path = Path(path)
         self.path.parent.mkdir(parents=True, exist_ok=True)
+        os.chmod(self.path.parent, 0o700)
         with self._connect() as connection:
             connection.executescript(
                 """
@@ -33,14 +35,47 @@ class PushSubscriptionStore:
                     FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                         ON DELETE CASCADE
                 );
+                CREATE TABLE IF NOT EXISTS push_dispatch_lease (
+                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                    owner TEXT NOT NULL,
+                    expires_at REAL NOT NULL
+                );
                 """
             )
+        os.chmod(self.path, 0o600)
 
     def _connect(self):
         connection = sqlite3.connect(self.path, timeout=2)
         connection.execute("PRAGMA foreign_keys = ON")
         return connection
 
+    def acquire_dispatch_lease(
+        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 singleton = 1"
+            ).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)
+                   VALUES (1, ?, ?)
+                   ON CONFLICT(singleton) DO UPDATE SET
+                       owner = excluded.owner, expires_at = excluded.expires_at""",
+                (owner, now + lease_seconds),
+            )
+            return True
+
+    def release_dispatch_lease(self, owner: str) -> bool:
+        with self._connect() as connection:
+            result = connection.execute(
+                "DELETE FROM push_dispatch_lease WHERE singleton = 1 AND owner = ?",
+                (owner,),
+            )
+            return result.rowcount == 1
+
     def upsert(self, session_id: str, subscription: dict) -> None:
         endpoint = subscription["endpoint"]
         encoded = json.dumps(subscription, separators=(",", ":"), sort_keys=True)