eb2781dd2f
A restart-safe cron worker runs outside the gateway, queues its final send in cron/deliveries.db and polls that row once a second. The gateway drained the queue only as a housekeeping chore at the 60 s tick, so a reminder whose turn ended in 8 s reached the user a minute later (the issue's 60.86 s silence). The housekeeping thread now watches the queue file (and its WAL) between ticks and drains as soon as a stamp moved; the tick's drain stays as the fallback. The watch never resets itself after draining: the drain's own status write costs one empty pass, a worker that enqueued during the drain still wakes it, and an empty pass writes nothing, so the stamp settles. Fixes #117307
79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
"""Wake the restart-safe cron delivery drain when a worker writes the queue (#117307).
|
|
|
|
A restart-safe worker runs its job outside the gateway, queues the final send in
|
|
``cron/deliveries.db`` and polls that row once a second; the gateway drained the
|
|
queue only on its housekeeping tick, so a turn that ended in 8 s reached the user
|
|
a minute later. Between ticks the housekeeping thread now stats the queue file
|
|
(and its WAL) and drains as soon as either changed. The tick's own drain stays
|
|
as the fallback for anything the stat cannot see.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, Iterable, Optional
|
|
|
|
# How often the housekeeping thread looks at the queue file between its ticks.
|
|
DELIVERY_QUEUE_WATCH_SECONDS = 1.0
|
|
DRAIN_LABEL = "Cron durable delivery queue drain"
|
|
|
|
|
|
class DeliveryQueueWatch:
|
|
"""Stamp-compare every served profile's queue file; ``drain`` when a stamp moved.
|
|
|
|
The watch never resets itself after draining: the drain's own status writes
|
|
move the stamp once more and cost one empty pass, after which the stamp is
|
|
stable, while a worker that enqueued during the drain still moves it again.
|
|
"""
|
|
|
|
def __init__(self, homes: Callable[[], Iterable[Optional[Path]]], drain: Callable[[], None]) -> None:
|
|
self._homes = homes
|
|
self.drain = drain
|
|
self._served: list = list(homes())
|
|
self._seen = self._stamp()
|
|
|
|
def refresh_homes(self) -> None:
|
|
"""Re-resolve the served profiles once per tick (a directory read under multiplex),
|
|
not on every poll; a profile that appears mid-interval is drained by the tick itself."""
|
|
self._served = list(self._homes())
|
|
|
|
def _stamp(self) -> tuple:
|
|
from cron.delivery_queue import queue_path
|
|
|
|
stamps = []
|
|
for home in self._served:
|
|
path = queue_path(home)
|
|
for candidate in (path, path.with_name(path.name + "-wal")):
|
|
try:
|
|
info = os.stat(candidate)
|
|
except OSError:
|
|
continue
|
|
stamps.append((str(candidate), info.st_mtime_ns, info.st_size))
|
|
return tuple(stamps)
|
|
|
|
def changed(self) -> bool:
|
|
current = self._stamp()
|
|
if current == self._seen:
|
|
return False
|
|
self._seen = current
|
|
return True
|
|
|
|
|
|
def wait_for_next_tick(stop_event, interval: float, watch: Optional[DeliveryQueueWatch], chore) -> None:
|
|
"""Sleep one housekeeping interval, draining the worker queue the moment it changes."""
|
|
if watch is None:
|
|
stop_event.wait(timeout=interval)
|
|
return
|
|
watch.refresh_homes()
|
|
deadline = time.monotonic() + interval
|
|
while True:
|
|
remaining = max(0.0, deadline - time.monotonic())
|
|
if stop_event.wait(timeout=min(remaining, DELIVERY_QUEUE_WATCH_SECONDS)):
|
|
return
|
|
if watch.changed():
|
|
chore(DRAIN_LABEL, watch.drain)
|
|
if time.monotonic() >= deadline:
|
|
return
|