Make deadline reminder polling eligibility-aware #718
|
|
@ -106,6 +106,9 @@ async def _push_channel_loop(dispatch, *, interval: float) -> None:
|
||||||
|
|
||||||
async def _push_poll_loop() -> None:
|
async def _push_poll_loop() -> None:
|
||||||
interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")))
|
interval = max(5.0, float(os.getenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")))
|
||||||
|
deadline_interval = max(
|
||||||
|
60.0, float(os.getenv("STACKCHAIN_DEADLINE_POLL_SECONDS", "600"))
|
||||||
|
)
|
||||||
send_timeout = max(
|
send_timeout = max(
|
||||||
1.0, float(os.getenv("STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS", "10"))
|
1.0, float(os.getenv("STACKCHAIN_PUSH_SEND_TIMEOUT_SECONDS", "10"))
|
||||||
)
|
)
|
||||||
|
|
@ -145,7 +148,9 @@ async def _push_poll_loop() -> None:
|
||||||
|
|
||||||
channel_tasks = (
|
channel_tasks = (
|
||||||
asyncio.create_task(_push_channel_loop(dispatch_unread, interval=interval)),
|
asyncio.create_task(_push_channel_loop(dispatch_unread, interval=interval)),
|
||||||
asyncio.create_task(_push_channel_loop(dispatch_deadlines, interval=interval)),
|
asyncio.create_task(
|
||||||
|
_push_channel_loop(dispatch_deadlines, interval=deadline_interval)
|
||||||
|
),
|
||||||
)
|
)
|
||||||
try:
|
try:
|
||||||
done, _pending = await asyncio.wait(
|
done, _pending = await asyncio.wait(
|
||||||
|
|
|
||||||
|
|
@ -289,10 +289,23 @@ async def _dispatch_deadline_reminders_unlocked(
|
||||||
devices = await asyncio.to_thread(store.deadline_reminder_devices)
|
devices = await asyncio.to_thread(store.deadline_reminder_devices)
|
||||||
if not devices:
|
if not devices:
|
||||||
return 0
|
return 0
|
||||||
|
current = now or datetime.now(timezone.utc)
|
||||||
|
eligible_devices = []
|
||||||
|
for device in devices:
|
||||||
|
try:
|
||||||
|
local_now = current.astimezone(ZoneInfo(device.timezone))
|
||||||
|
except ZoneInfoNotFoundError:
|
||||||
|
continue
|
||||||
|
if (
|
||||||
|
local_now.hour >= device.reminder_hour
|
||||||
|
and device.delivered_local_day != local_now.date().isoformat()
|
||||||
|
):
|
||||||
|
eligible_devices.append(device)
|
||||||
|
if not eligible_devices:
|
||||||
|
return 0
|
||||||
snapshot = await assigned()
|
snapshot = await assigned()
|
||||||
if snapshot.get("complete") is False:
|
if snapshot.get("complete") is False:
|
||||||
return 0
|
return 0
|
||||||
current = now or datetime.now(timezone.utc)
|
|
||||||
due_cutoff = current + timedelta(hours=48)
|
due_cutoff = current + timedelta(hours=48)
|
||||||
due_count = 0
|
due_count = 0
|
||||||
for item in snapshot.get("items", []):
|
for item in snapshot.get("items", []):
|
||||||
|
|
@ -356,6 +369,6 @@ async def _dispatch_deadline_reminders_unlocked(
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
results = await asyncio.gather(
|
results = await asyncio.gather(
|
||||||
*(dispatch_device(device) for device in devices), return_exceptions=True
|
*(dispatch_device(device) for device in eligible_devices), return_exceptions=True
|
||||||
)
|
)
|
||||||
return sum(result for result in results if isinstance(result, int))
|
return sum(result for result in results if isinstance(result, int))
|
||||||
|
|
|
||||||
|
|
@ -87,6 +87,67 @@ async def test_deadline_reminder_fails_closed_for_incomplete_snapshot_and_before
|
||||||
assert sent == []
|
assert sent == []
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_deadline_reminder_skips_snapshot_until_a_device_reaches_its_local_hour(tmp_path):
|
||||||
|
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||||
|
store.upsert("device-a", {
|
||||||
|
"endpoint": "https://push.example/device-a",
|
||||||
|
"keys": {"p256dh": "public-key", "auth": "auth-secret"},
|
||||||
|
})
|
||||||
|
store.set_deadline_preferences(
|
||||||
|
"device-a", enabled=True, timezone="America/Los_Angeles", reminder_hour=9
|
||||||
|
)
|
||||||
|
snapshot_calls = 0
|
||||||
|
|
||||||
|
async def assigned():
|
||||||
|
nonlocal snapshot_calls
|
||||||
|
snapshot_calls += 1
|
||||||
|
return {"complete": True, "items": []}
|
||||||
|
|
||||||
|
delivered = await dispatch_deadline_reminders(
|
||||||
|
store,
|
||||||
|
PushConfiguration("public", "private", "mailto:ops@example.com"),
|
||||||
|
assigned,
|
||||||
|
now=datetime(2026, 8, 13, 15, 0, tzinfo=timezone.utc),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert delivered == 0
|
||||||
|
assert snapshot_calls == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_deadline_reminder_skips_snapshot_after_every_device_was_delivered_today(tmp_path):
|
||||||
|
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||||
|
for session_id, timezone_name in (
|
||||||
|
("device-a", "UTC"),
|
||||||
|
("device-b", "America/New_York"),
|
||||||
|
):
|
||||||
|
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=timezone_name, reminder_hour=9
|
||||||
|
)
|
||||||
|
store.mark_deadline_reminder_delivered(session_id, "2026-08-13")
|
||||||
|
snapshot_calls = 0
|
||||||
|
|
||||||
|
async def assigned():
|
||||||
|
nonlocal snapshot_calls
|
||||||
|
snapshot_calls += 1
|
||||||
|
return {"complete": True, "items": []}
|
||||||
|
|
||||||
|
delivered = await dispatch_deadline_reminders(
|
||||||
|
store,
|
||||||
|
PushConfiguration("public", "private", "mailto:ops@example.com"),
|
||||||
|
assigned,
|
||||||
|
now=datetime(2026, 8, 13, 15, 0, tzinfo=timezone.utc),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert delivered == 0
|
||||||
|
assert snapshot_calls == 0
|
||||||
|
|
||||||
|
|
||||||
def test_deadline_preferences_persist_on_the_existing_device_subscription(tmp_path):
|
def test_deadline_preferences_persist_on_the_existing_device_subscription(tmp_path):
|
||||||
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
store = PushSubscriptionStore(tmp_path / "push.sqlite3")
|
||||||
store.upsert("device-a", {
|
store.upsert("device-a", {
|
||||||
|
|
@ -177,6 +238,7 @@ async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_pa
|
||||||
)
|
)
|
||||||
active = 0
|
active = 0
|
||||||
peak = 0
|
peak = 0
|
||||||
|
two_active = asyncio.Event()
|
||||||
|
|
||||||
async def assigned():
|
async def assigned():
|
||||||
return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}
|
return {"complete": True, "items": [{"id": 1, "due_date": "2026-08-14T00:00:00Z"}]}
|
||||||
|
|
@ -185,7 +247,9 @@ async def test_deadline_reminders_bound_fanout_and_isolate_failed_devices(tmp_pa
|
||||||
nonlocal active, peak
|
nonlocal active, peak
|
||||||
active += 1
|
active += 1
|
||||||
peak = max(peak, active)
|
peak = max(peak, active)
|
||||||
await asyncio.sleep(0.01)
|
if active == 2:
|
||||||
|
two_active.set()
|
||||||
|
await asyncio.wait_for(two_active.wait(), timeout=0.5)
|
||||||
active -= 1
|
active -= 1
|
||||||
if subscription["endpoint"].endswith("device-1"):
|
if subscription["endpoint"].endswith("device-1"):
|
||||||
raise RuntimeError("provider failed")
|
raise RuntimeError("provider failed")
|
||||||
|
|
|
||||||
|
|
@ -295,6 +295,22 @@ async def test_push_poll_deadlines_continue_while_unread_dispatch_is_blocked(mon
|
||||||
await poll
|
await poll
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_push_poll_uses_a_lower_independent_deadline_cadence(monkeypatch):
|
||||||
|
intervals = []
|
||||||
|
|
||||||
|
async def capture_channel(_dispatch, *, interval):
|
||||||
|
intervals.append(interval)
|
||||||
|
|
||||||
|
monkeypatch.setenv("STACKCHAIN_PUSH_POLL_SECONDS", "30")
|
||||||
|
monkeypatch.setenv("STACKCHAIN_DEADLINE_POLL_SECONDS", "600")
|
||||||
|
monkeypatch.setattr(main, "_push_channel_loop", capture_channel)
|
||||||
|
|
||||||
|
await main._push_poll_loop()
|
||||||
|
|
||||||
|
assert sorted(intervals) == [30.0, 600.0]
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_push_channel_loop_skips_missed_ticks_without_overlapping(monkeypatch):
|
async def test_push_channel_loop_skips_missed_ticks_without_overlapping(monkeypatch):
|
||||||
starts = []
|
starts = []
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user