Merge pull request 'Bulk-authorize push devices before notification fan-out' (#726) from timmy/725-bulk-push-authorization into main

src/dashboard_auth.py

@@ -244,6 +244,14 @@ async def managed_session_active(management_id: str) -> bool:
     return status == "active"
 
 
+async def managed_session_statuses(management_ids) -> dict[str, str]:
+    return await asyncio.to_thread(
+        _session_store().managed_statuses,
+        management_ids,
+        idle_timeout_seconds=idle_timeout_seconds(),
+    )
+
+
 async def touch_session(session: Session) -> bool:
     return await asyncio.to_thread(
         _session_store().touch,

src/main.py

@@ -128,7 +128,7 @@ async def _push_poll_loop() -> None:
                 _push_subscription_store,
                 _push_configuration(),
                 gitea_proxy.unread_notification_snapshot,
-                session_active=dashboard_auth.managed_session_active,
+                session_statuses=dashboard_auth.managed_session_statuses,
                 lease_seconds=lease_seconds,
                 send_timeout_seconds=send_timeout,
                 max_concurrency=max_concurrency,
@@ -140,7 +140,7 @@ async def _push_poll_loop() -> None:
                 _push_subscription_store,
                 _push_configuration(),
                 gitea_proxy.assigned_issue_snapshot,
-                session_active=dashboard_auth.managed_session_active,
+                session_statuses=dashboard_auth.managed_session_statuses,
                 send_timeout_seconds=send_timeout,
                 lease_seconds=lease_seconds,
                 max_concurrency=max_concurrency,

src/push_notifications.py

@@ -49,7 +49,7 @@ async def dispatch_unread_updates(
     unread: Callable[[], Awaitable[dict]],
     send: Callable[[dict, str], Awaitable[None]] | None = None,
     *,
-    session_active: Callable[[str], Awaitable[bool]] | None = None,
+    session_statuses: Callable[[list[str]], Awaitable[dict[str, str]]] | None = None,
     lease_seconds: float = 60.0,
     send_timeout_seconds: float = 10.0,
     max_concurrency: int = 8,
@@ -79,20 +79,22 @@ async def dispatch_unread_updates(
         }
         await asyncio.to_thread(store.reconcile_unread, thread_revisions)
         deliveries = await asyncio.to_thread(store.claim_unseen, thread_revisions)
-        if session_active is not None:
+        if session_statuses is not None:
             try:
-                authorized = [await session_active(item.session_id) for item in deliveries]
+                statuses = await session_statuses(
+                    [item.session_id for item in deliveries]
+                )
             except Exception:
                 # Authorization state is mandatory for delivery. Preserve subscriptions
                 # so a temporary registry failure can be retried safely.
                 return 0
-            for delivery, active in zip(deliveries, authorized):
-                if not active:
+            for delivery in deliveries:
+                if statuses.get(delivery.session_id) != "active":
                     await asyncio.to_thread(store.delete_session, delivery.session_id)
             deliveries = [
                 delivery
-                for delivery, active in zip(deliveries, authorized)
-                if active
+                for delivery in deliveries
+                if statuses.get(delivery.session_id) == "active"
             ]
         semaphore = asyncio.Semaphore(max(1, max_concurrency))
         ownership_lost = asyncio.Event()
@@ -277,7 +279,7 @@ async def _dispatch_deadline_reminders_unlocked(
     send: Callable[[dict, str], Awaitable[None]] | None = None,
     *,
     now: datetime | None = None,
-    session_active: Callable[[str], Awaitable[bool]] | None = None,
+    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,
@@ -318,6 +320,35 @@ async def _dispatch_deadline_reminders_unlocked(
         due_days.append(due_day)
     if not due_days:
         return 0
+    due_counts = {}
+    for device in eligible_devices:
+        local_now = current.astimezone(ZoneInfo(device.timezone))
+        local_cutoff = local_now.date() + timedelta(days=2)
+        due_count = sum(due_day <= local_cutoff for due_day in due_days)
+        if due_count:
+            due_counts[device.session_id] = due_count
+    eligible_devices = [
+        device for device in eligible_devices if device.session_id in due_counts
+    ]
+    if not eligible_devices:
+        return 0
+    if session_statuses is not None:
+        try:
+            statuses = await session_statuses(
+                [device.session_id for device in eligible_devices]
+            )
+        except Exception:
+            return 0
+        for device in eligible_devices:
+            if statuses.get(device.session_id) != "active":
+                await asyncio.to_thread(store.delete_session, device.session_id)
+        eligible_devices = [
+            device
+            for device in eligible_devices
+            if statuses.get(device.session_id) == "active"
+        ]
+        if not eligible_devices:
+            return 0
     semaphore = asyncio.Semaphore(max(1, max_concurrency))
 
     async def dispatch_device(device) -> int:
@@ -332,13 +363,7 @@ async def _dispatch_deadline_reminders_unlocked(
                 or device.delivered_local_day == local_day
             ):
                 return 0
-            local_cutoff = local_now.date() + timedelta(days=2)
-            due_count = sum(due_day <= local_cutoff for due_day in due_days)
-            if not due_count:
-                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
+            due_count = due_counts[device.session_id]
             still_owner = await asyncio.to_thread(
                 store.acquire_dispatch_lease,
                 owner,

src/session_store.py

@@ -273,21 +273,40 @@ class SessionStore:
 
     def managed_status(self, management_id: str, *, idle_timeout_seconds: int) -> str:
         """Return the authorization state for a durable device identifier."""
+        return self.managed_statuses(
+            [management_id], idle_timeout_seconds=idle_timeout_seconds
+        )[management_id]
+
+    def managed_statuses(
+        self, management_ids, *, idle_timeout_seconds: int
+    ) -> dict[str, str]:
+        """Return authorization states for durable device identifiers in one read."""
+        requested = set(management_ids)
+        if not requested:
+            return {}
         now = int(self.clock())
         try:
             with self._connect() as connection:
-                row = connection.execute(
-                    "SELECT expires_at, last_active_at FROM active_sessions "
-                    "WHERE management_id = ?",
-                    (management_id,),
-                ).fetchone()
+                placeholders = ",".join("?" for _ in requested)
+                rows = connection.execute(
+                    "SELECT management_id, expires_at, last_active_at "
+                    f"FROM active_sessions WHERE management_id IN ({placeholders})",
+                    tuple(requested),
+                ).fetchall()
         except (OSError, sqlite3.Error) as exc:
             raise SessionStoreError("Session registry is temporarily unavailable") from exc
-        if row is None or row[0] <= now:
-            return "revoked"
-        if row[1] + max(1, idle_timeout_seconds) <= now:
-            return "idle"
-        return "active"
+        known = {row[0]: row[1:] for row in rows if row[0] in requested}
+        idle_timeout = max(1, idle_timeout_seconds)
+        statuses = {}
+        for management_id in requested:
+            row = known.get(management_id)
+            if row is None or row[0] <= now:
+                statuses[management_id] = "revoked"
+            elif row[1] + idle_timeout <= now:
+                statuses[management_id] = "idle"
+            else:
+                statuses[management_id] = "active"
+        return statuses
 
     def revoke_managed(self, management_id: str) -> bool:
         try:

tests/test_deadline_reminders.py

@@ -296,3 +296,45 @@ async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_pa
 
     assert delivered == 3
     assert peak == 2
+
+
+@pytest.mark.anyio
+async def test_deadline_reminders_bulk_authorize_due_devices_and_prune_inactive(tmp_path):
+    store = PushSubscriptionStore(tmp_path / "push.sqlite3")
+    for session_id in ("active-device", "revoked-device"):
+        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
+        )
+    authorization_calls = []
+    sent = []
+
+    async def assigned():
+        return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}
+
+    async def session_statuses(management_ids):
+        authorization_calls.append(list(management_ids))
+        return {
+            management_id: "active" if management_id == "active-device" else "revoked"
+            for management_id in management_ids
+        }
+
+    async def send(subscription, _payload):
+        sent.append(subscription["endpoint"])
+
+    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),
+        session_statuses=session_statuses,
+    )
+
+    assert delivered == 1
+    assert authorization_calls == [["active-device", "revoked-device"]]
+    assert sent == ["https://push.example/active-device"]
+    assert store.is_subscribed("revoked-device") is False

tests/test_push_notifications.py

@@ -204,6 +204,24 @@ async def test_managed_session_active_applies_configured_idle_deadline(monkeypat
     assert calls == [("active-device", 321), ("idle-device", 321)]
 
 
+@pytest.mark.anyio
+async def test_managed_session_statuses_apply_configured_idle_deadline_in_one_call(monkeypatch):
+    calls = []
+
+    class Store:
+        def managed_statuses(self, management_ids, *, idle_timeout_seconds):
+            calls.append((list(management_ids), idle_timeout_seconds))
+            return {management_id: "active" for management_id in set(management_ids)}
+
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_IDLE_TIMEOUT_SECONDS", "321")
+    monkeypatch.setattr(dashboard_auth, "_session_store", lambda: Store())
+
+    assert await dashboard_auth.managed_session_statuses(
+        ["device-b", "device-a", "device-b"]
+    ) == {"device-a": "active", "device-b": "active"}
+    assert calls == [(["device-b", "device-a", "device-b"], 321)]
+
+
 @pytest.mark.anyio
 async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch):
     captured = {}
@@ -225,7 +243,7 @@ async def test_push_poll_authorizes_delivery_against_managed_session(monkeypatch
         await main._push_poll_loop()
 
     assert captured["unread"] is main.gitea_proxy.unread_notification_snapshot
-    assert captured["session_active"] is dashboard_auth.managed_session_active
+    assert captured["session_statuses"] is dashboard_auth.managed_session_statuses
     assert captured["max_concurrency"] == 3
     assert captured["max_individual_notifications"] == 4
 
@@ -881,19 +899,26 @@ async def test_dispatch_removes_inactive_sessions_without_blocking_active_device
     async def unread():
         return {"items": [{"id": 21}]}
 
-    async def session_active(management_id):
-        return management_id == "active-device"
+    authorization_calls = []
+
+    async def session_statuses(management_ids):
+        authorization_calls.append(list(management_ids))
+        return {
+            management_id: "active" if management_id == "active-device" else "revoked"
+            for management_id in management_ids
+        }
 
     async def send(subscription, _payload):
         sent.append(subscription["endpoint"])
 
     config = PushConfiguration("public", "private", "mailto:ops@example.com")
     assert await dispatch_unread_updates(
-        store, config, unread, send, session_active=session_active
+        store, config, unread, send, session_statuses=session_statuses
     ) == 1
     assert store.is_subscribed("expired-device") is False
     assert store.is_subscribed("active-device") is True
     assert sent == ["https://push.example/active-device"]
+    assert authorization_calls == [["active-device", "expired-device"]]
 
 
 @pytest.mark.anyio
@@ -908,7 +933,7 @@ async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_regi
     async def unread():
         return {"items": [{"id": 22}]}
 
-    async def registry_unavailable(_management_id):
+    async def registry_unavailable(_management_ids):
         raise RuntimeError("session registry unavailable")
 
     async def send(subscription, _payload):
@@ -916,7 +941,7 @@ async def test_dispatch_fails_closed_and_retains_subscriptions_when_session_regi
 
     config = PushConfiguration("public", "private", "mailto:ops@example.com")
     assert await dispatch_unread_updates(
-        store, config, unread, send, session_active=registry_unavailable
+        store, config, unread, send, session_statuses=registry_unavailable
     ) == 0
     assert store.is_subscribed("active-device") is True
     assert sent == []

tests/test_session_store.py

@@ -102,6 +102,45 @@ def test_managed_session_status_enforces_absolute_and_idle_expiry(tmp_path):
     assert store.managed_status("missing-device", idle_timeout_seconds=900) == "revoked"
 
 
+def test_managed_session_statuses_classify_a_batch_with_one_read(tmp_path, monkeypatch):
+    now = [1_000.0]
+    store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])
+    store.activate("active-session", 3_000, management_id="active-device")
+    store.activate("idle-session", 3_000, management_id="idle-device")
+    store.activate("expired-session", 1_500, management_id="expired-device")
+    now[0] = 2_000.0
+    store.touch("active-session", 3_000, idle_timeout_seconds=1_500)
+    statements = []
+    connect = sqlite3.connect
+
+    def traced_connect(*args, **kwargs):
+        connection = connect(*args, **kwargs)
+        connection.set_trace_callback(statements.append)
+        return connection
+
+    monkeypatch.setattr(session_store.sqlite3, "connect", traced_connect)
+
+    assert store.managed_statuses(
+        ["active-device", "idle-device", "expired-device", "missing-device", "active-device"],
+        idle_timeout_seconds=900,
+    ) == {
+        "active-device": "active",
+        "idle-device": "idle",
+        "expired-device": "revoked",
+        "missing-device": "revoked",
+    }
+    assert [statement for statement in statements if statement.lstrip().upper().startswith("SELECT")] == [
+        next(statement for statement in statements if "FROM active_sessions" in statement)
+    ]
+
+
+def test_managed_session_statuses_skips_database_for_an_empty_batch(tmp_path, monkeypatch):
+    store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: 1_000.0)
+    monkeypatch.setattr(store, "_connect", lambda *args, **kwargs: pytest.fail("database opened"))
+
+    assert store.managed_statuses([], idle_timeout_seconds=900) == {}
+
+
 def test_touch_extends_only_the_matching_live_session(tmp_path):
     now = [1_000.0]
     store = SessionStore(tmp_path / "sessions.sqlite3", clock=lambda: now[0])