From 72842ecbcdb4f1f0aaeca90de43a9edcfc7b6cc3 Mon Sep 17 00:00:00 2001 From: timmy Date: Thu, 13 Aug 2026 05:55:50 +0000 Subject: [PATCH] fix: isolate deadline reminder dispatch (Closes #711) --- src/main.py | 10 ++- src/push_notifications.py | 101 ++++++++++++++++++++----------- src/push_subscription_store.py | 39 +++++++++--- tests/test_deadline_reminders.py | 41 +++++++++++++ tests/test_push_notifications.py | 45 ++++++++++++++ 5 files changed, 188 insertions(+), 48 deletions(-) diff --git a/src/main.py b/src/main.py index 19a7605..8aadeee 100644 --- a/src/main.py +++ b/src/main.py @@ -116,18 +116,24 @@ 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 are external; one failed poll must not - # stop later delivery attempts. + # Gitea and push endpoints retry independently on the next poll. continue diff --git a/src/push_notifications.py b/src/push_notifications.py index 14fb463..649f616 100644 --- a/src/push_notifications.py +++ b/src/push_notifications.py @@ -62,6 +62,7 @@ async def dispatch_unread_updates( acquired = await asyncio.to_thread( store.acquire_dispatch_lease, owner, + channel="unread", now=time.time(), lease_seconds=lease_seconds, ) @@ -128,6 +129,7 @@ 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, ) @@ -174,6 +176,7 @@ 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, ) @@ -235,7 +238,7 @@ async def dispatch_unread_updates( ) return sum(counts) finally: - await asyncio.to_thread(store.release_dispatch_lease, owner) + await asyncio.to_thread(store.release_dispatch_lease, owner, channel="unread") async def dispatch_deadline_reminders( @@ -249,17 +252,22 @@ async def dispatch_deadline_reminders( acquired = await asyncio.to_thread( store.acquire_dispatch_lease, owner, + channel="deadline", now=time.time(), - lease_seconds=max(15.0, float(kwargs.get("send_timeout_seconds", 10.0)) + 5.0), + lease_seconds=max( + 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, **kwargs + store, configuration, assigned, send, owner=owner, **kwargs ) finally: - await asyncio.to_thread(store.release_dispatch_lease, owner) + await asyncio.to_thread(store.release_dispatch_lease, owner, channel="deadline") async def _dispatch_deadline_reminders_unlocked( @@ -271,6 +279,9 @@ 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: @@ -297,36 +308,54 @@ async def _dispatch_deadline_reminders_unlocked( due_count += 1 if not due_count: return 0 - delivered = 0 - for device in devices: - try: - local_now = current.astimezone(ZoneInfo(device.timezone)) - except ZoneInfoNotFoundError: - continue - local_day = local_now.date().isoformat() - if local_now.hour < device.reminder_hour or device.delivered_local_day == local_day: - continue - if session_active is not None and not await session_active(device.session_id): - await asyncio.to_thread(store.delete_session, device.session_id) - continue - 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'}.", - "route": "#/my-work/agenda", - "tag": f"stackchain-deadline-digest-{local_day}", - "deadline_count": due_count, - }, separators=(",", ":")) - 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: + try: + local_now = current.astimezone(ZoneInfo(device.timezone)) + except ZoneInfoNotFoundError: + return 0 + local_day = local_now.date().isoformat() + if ( + local_now.hour < device.reminder_hour + 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 + 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), ) - await asyncio.wait_for(operation, timeout=send_timeout_seconds) - except Exception: - continue - await asyncio.to_thread( - store.mark_deadline_reminder_delivered, device.session_id, local_day - ) - delivered += 1 - return delivered + 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'}.", + "route": "#/my-work/agenda", + "tag": f"stackchain-deadline-digest-{local_day}", + "deadline_count": due_count, + }, separators=(",", ":")) + try: + operation = ( + send(device.subscription, payload) + if send is not None + else send_web_push(device.subscription, payload, configuration) + ) + await asyncio.wait_for(operation, timeout=send_timeout_seconds) + except Exception: + return 0 + await asyncio.to_thread( + store.mark_deadline_reminder_delivered, device.session_id, local_day + ) + return 1 + + 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)) diff --git a/src/push_subscription_store.py b/src/push_subscription_store.py index 2784915..7ce5098 100644 --- a/src/push_subscription_store.py +++ b/src/push_subscription_store.py @@ -75,7 +75,7 @@ class PushSubscriptionStore: ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS push_dispatch_lease ( - singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + channel TEXT PRIMARY KEY, owner TEXT NOT NULL, expires_at REAL NOT NULL ); @@ -113,6 +113,24 @@ 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): @@ -121,29 +139,30 @@ class PushSubscriptionStore: return connection def acquire_dispatch_lease( - self, owner: str, *, now: float, lease_seconds: float + self, owner: str, *, channel: str = "unread", 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" + "SELECT owner, expires_at FROM push_dispatch_lease WHERE channel = ?", + (channel,), ).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 + """INSERT INTO push_dispatch_lease(channel, owner, expires_at) + VALUES (?, ?, ?) + ON CONFLICT(channel) DO UPDATE SET owner = excluded.owner, expires_at = excluded.expires_at""", - (owner, now + lease_seconds), + (channel, owner, now + lease_seconds), ) return True - def release_dispatch_lease(self, owner: str) -> bool: + def release_dispatch_lease(self, owner: str, *, channel: str = "unread") -> bool: with self._connect() as connection: result = connection.execute( - "DELETE FROM push_dispatch_lease WHERE singleton = 1 AND owner = ?", - (owner,), + "DELETE FROM push_dispatch_lease WHERE channel = ? AND owner = ?", + (channel, owner), ) return result.rowcount == 1 diff --git a/tests/test_deadline_reminders.py b/tests/test_deadline_reminders.py index 7695423..faeb90f 100644 --- a/tests/test_deadline_reminders.py +++ b/tests/test_deadline_reminders.py @@ -1,3 +1,4 @@ +import asyncio import json from datetime import datetime, timezone @@ -160,3 +161,43 @@ 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 diff --git a/tests/test_push_notifications.py b/tests/test_push_notifications.py index dd2105d..ac68596 100644 --- a/tests/test_push_notifications.py +++ b/tests/test_push_notifications.py @@ -230,6 +230,51 @@ 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 = { -- 2.43.0