diff --git a/src/dashboard_auth.py b/src/dashboard_auth.py index 576943e..8341078 100644 --- a/src/dashboard_auth.py +++ b/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, diff --git a/src/main.py b/src/main.py index 5ce5496..687689f 100644 --- a/src/main.py +++ b/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, diff --git a/src/push_notifications.py b/src/push_notifications.py index f9306e5..cf68741 100644 --- a/src/push_notifications.py +++ b/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, diff --git a/src/session_store.py b/src/session_store.py index 44c1c36..a9226cf 100644 --- a/src/session_store.py +++ b/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: diff --git a/tests/test_deadline_reminders.py b/tests/test_deadline_reminders.py index ab10680..ae6905c 100644 --- a/tests/test_deadline_reminders.py +++ b/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 diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index 44a698b..fa40c49 100644 --- a/tests/test_push_notifications.py +++ b/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 == [] diff --git a/tests/test_session_store.py b/tests/test_session_store.py index bea7ae4..bde13d6 100644 --- a/tests/test_session_store.py +++ b/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])