Wake deadline consumers on enqueue instead of polling every second

server/README.md

@@ -72,7 +72,7 @@ All configuration comes from environment variables (see `.env.example`):
 | `GAME_TTL_SECONDS` | `86400` | Sliding TTL of a live game in Redis |
 | `HAND_ACK_TIMEOUT_SECONDS` | `30` | Seconds the scopone between-hands scoring summary waits for acknowledgements (wired into the `ScoponeEngine`) |
 | `TURN_TIMEOUT_SECONDS` | `30` | Seconds a scopone player has to play before the server plays a random legal card for them (wired into the `ScoponeEngine`) |
-| `DEADLINE_HEARTBEAT_MS` | `1000` | Upper bound on the deadline consumer's poll interval (locally enqueued deadlines fire on time regardless) |
+| `DEADLINE_HEARTBEAT_MS` | `30000` | Backstop poll interval of the deadline consumer; enqueued deadlines are announced over pub/sub and wake consumers immediately, so this only bounds recovery from lost notifications or a dead worker |
 | `LOGGING_CONFIG` | unset | Path to a YAML logging configuration file (see below). Unset logs DEBUG to the console |
 | `CORS_ALLOW_ORIGINS` | unset | Comma-separated origins allowed for cross-origin requests, or `*` for any. CORS is disabled unless this or `CORS_ALLOW_ORIGIN_REGEX` is set |
 | `CORS_ALLOW_ORIGIN_REGEX` | unset | Regex (fullmatch) additionally matched against request origins, e.g. `https://tavolo-[a-z0-9-]+\.vercel\.app` |
@@ -176,6 +176,9 @@ aggregate with `max`, not `sum`.
   on any player being connected and survive the death of any worker
   (delivery is at-least-once; engines revalidate the entry's token
   against the live state before firing).
+- `tavolo:deadlines:events` — a pub/sub channel announcing newly enqueued
+  deadlines; it wakes every worker's consumer immediately, while the
+  heartbeat poll remains as a backstop for lost signals.
 
 ### Postgres (statistics, via Tortoise ORM + aerich migrations)
 

server/packages/tavolo-app/src/tavolo/config.py

@@ -97,10 +97,10 @@ class Settings:
     # card for them, covering disconnects and idle players (wired into the
     # ScoponeEngine; see tavolo.app).
     turn_timeout_seconds: int
-    # Upper bound on how long the deadline consumer sleeps between polls.
+    # Upper bound on how long the deadline consumer sleeps between polls
-    # Locally enqueued deadlines wake the consumer immediately; the
+    # when nothing else wakes it. Enqueued deadlines are announced over a
-    # heartbeat only bounds the discovery delay for deadlines enqueued by
+    # pub/sub channel, so this is only the backstop recovering lost
-    # other workers.
+    # notifications and deadlines orphaned by a dead worker.
     deadline_heartbeat_ms: int
     # Path to a YAML logging configuration file (logging.config.dictConfig
     # schema). Unset uses the built-in default: DEBUG to the console.
@@ -163,7 +163,7 @@ class Settings:
             static_dir=_env("STATIC_DIR", "web/dist"),
             hand_ack_timeout_seconds=int(_env("HAND_ACK_TIMEOUT_SECONDS", "30")),
             turn_timeout_seconds=int(_env("TURN_TIMEOUT_SECONDS", "30")),
-            deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "1000")),
+            deadline_heartbeat_ms=int(_env("DEADLINE_HEARTBEAT_MS", "30000")),
             logging_config=os.environ.get("LOGGING_CONFIG"),
             # CORS is disabled unless CORS_ALLOW_ORIGINS (a comma-separated
             # list of origins, or "*" for any) or CORS_ALLOW_ORIGIN_REGEX

server/packages/tavolo-platform/src/tavolo/platform/deadlines.py

@@ -11,8 +11,11 @@ machinery; the *meaning* of each deadline belongs to the game engine:
   token), and the scheduler enqueues it in the store's shared deadline
   queue (a Redis sorted set in production, see
   :mod:`tavolo.platform.store`); enqueueing is idempotent;
-* a background consumer running on **every** worker polls the queue for
+* enqueueing a deadline announces it on a shared pub/sub channel, so
-  due entries and hands them to the engine's
+  every worker's consumer wakes immediately instead of waiting for its
+  next poll; the consumer also wakes on a slow heartbeat, which is the
+  backstop recovering notifications lost to a reconnect or a worker
+  crash. Due entries are handed to the engine's
   :meth:`~tavolo.platform.engine.GameEngine.fire_deadline` under the
   per-game lock; the engine revalidates the token against the live state
   and raises :class:`~tavolo.platform.errors.GameError` when the entry
@@ -65,8 +68,8 @@ class DeadlineScheduler:
     """Enqueue and fire the deadlines game engines ask for.
 
     Holds the store and the registry so a single instance serves every
-    game type. One consumer task (and its wake-up event) is kept per
+    game type. One consumer task, its listener task and its wake-up
-    event loop — tests run each test on a fresh loop.
+    event are kept per event loop — tests run each test on a fresh loop.
 
     ``context_binder`` is an optional async callable the consumer awaits
     before persisting a finished match. The consumer is a long-lived task
@@ -80,14 +83,19 @@ class DeadlineScheduler:
         store: GameStore,
         registry: GameRegistry,
         heartbeat_ms: int = 1000,
+        retry_ms: int = 1000,
         context_binder: Optional[Callable[[], Awaitable[None]]] = None,
     ) -> None:
         self._store = store
         self._registry = registry
         self._heartbeat = heartbeat_ms / 1000
+        # Backoff for entries that failed to process and for listener
+        # reconnects; deliberately independent of the (slow) heartbeat.
+        self._retry_delay = retry_ms / 1000
         self._context_binder = context_binder
         self._consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
         self._wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
+        self._listeners: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
 
     async def sync_deadline(self, session: GameSession) -> None:
         """Enqueue the deadline the current state implies, if any.
@@ -109,6 +117,16 @@ class DeadlineScheduler:
         self.ensure_consumer()
         due_at = deadline.due_at.timestamp()
         await self._store.add_deadline(encode(entry), due_at)
+        # Announce the enqueue so consumers on other workers wake up too.
+        # The entry is already queued, so a lost notification only delays
+        # their discovery to the next poll backstop.
+        try:
+            await self._store.notify_deadline()
+        except Exception:
+            log.warning(
+                "deadline enqueue notification failed; relying on the poll backstop",
+                exc_info=True,
+            )
         wake = self._wake_events.get(asyncio.get_running_loop())
         if wake is not None:
             wake.set()
@@ -190,15 +208,22 @@ class DeadlineScheduler:
             if old.is_closed():
                 self._consumers.pop(old, None)
                 self._wake_events.pop(old, None)
+                self._listeners.pop(old, None)
         task = self._consumers.get(loop)
         if task is None or task.done():
             self._wake_events[loop] = asyncio.Event()
             self._consumers[loop] = loop.create_task(self._run(loop))
+            listener = self._listeners.get(loop)
+            if listener is None or listener.done():
+                self._listeners[loop] = loop.create_task(self._listen(loop))
             log.debug("deadline consumer started")
 
     def stop_consumer(self, loop: asyncio.AbstractEventLoop) -> None:
         task = self._consumers.pop(loop, None)
         self._wake_events.pop(loop, None)
+        listener = self._listeners.pop(loop, None)
+        if listener is not None:
+            listener.cancel()
         if task is not None:
             task.cancel()
 
@@ -228,21 +253,48 @@ class DeadlineScheduler:
                 log.exception("deadline consumer: poll failed; retrying")
                 failed = True
             if failed:
-                # A failed entry is still due: back off to the heartbeat
+                # A failed entry is still due: back off briefly instead of
-                # instead of polling it again immediately.
+                # polling it again immediately. Kept independent of the
-                delay = self._heartbeat
+                # heartbeat, which is only a slow backstop.
+                delay = self._retry_delay
             try:
                 await asyncio.wait_for(wake.wait(), timeout=delay)
             except asyncio.TimeoutError:
                 pass
 
+    async def _listen(self, loop: asyncio.AbstractEventLoop) -> None:
+        """Wake the consumer on deadline-enqueue notifications.
+
+        The subscription is best effort: a dropped connection or a lost
+        message is recovered by the consumer's heartbeat backstop, so
+        this task only has to reconnect and keep forwarding signals.
+        """
+        while True:
+            try:
+                async with self._store.subscribe_deadlines() as events:
+                    async for _ in events:
+                        wake = self._wake_events.get(loop)
+                        if wake is None:
+                            # Shut down: the consumer is gone.
+                            return
+                        wake.set()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                log.exception("deadline listener failed; reconnecting")
+            try:
+                await asyncio.sleep(self._retry_delay)
+            except asyncio.CancelledError:
+                raise
+
 
 class DeadlineSchedulerMixin(KayaMixin):
     """Run the deadline consumer for the whole app lifetime.
 
-    Every worker (and every pod) runs the same consumer; coordination
+    Every worker (and every pod) runs the same consumer and listener;
-    happens exclusively through the shared deadline queue and the
+    coordination happens exclusively through the shared deadline queue,
-    per-game locks, so any worker may fire any game's deadline.
+    its notification channel and the per-game locks, so any worker may
+    fire any game's deadline.
     """
 
     def __init__(self, scheduler: DeadlineScheduler) -> None:

server/packages/tavolo-platform/src/tavolo/platform/store.py

@@ -26,8 +26,9 @@ signal (never the state) means updated state reaches connections on
 every worker without leaking hidden information into the channel.
 
 Timeouts are driven by a shared delayed-deadline queue: producers
-enqueue an opaque ``member`` string with a due timestamp, and a consumer
+enqueue an opaque ``member`` string with a due timestamp and announce it
-on every worker polls for due entries (see
+on a shared "deadline enqueued" channel, and a consumer on every worker
+wakes on that signal to process due entries (see
 :mod:`tavolo.platform.deadlines`). Delivery is at-least-once — entries
 are removed only after they are processed — so a worker dying
 mid-processing cannot lose a deadline; engines revalidate entries
@@ -55,6 +56,9 @@ GAME_KEY_PREFIX = "tavolo:game:"
 CODE_KEY_PREFIX = "tavolo:code:"
 CHANNEL_PREFIX = "tavolo:game:"
 DEADLINES_KEY = "tavolo:deadlines"
+# Channel announcing newly enqueued deadline entries, so every worker's
+# consumer can wake immediately instead of polling on its heartbeat.
+DEADLINES_CHANNEL = "tavolo:deadlines:events"
 # Sorted set of running games (score = expiry timestamp), maintained by the
 # stores so the telemetry gauge can count ongoing matches without loading
 # every session. Finished games are removed on their final save.
@@ -165,6 +169,18 @@ class GameStore(ABC):
     async def remove_deadline(self, member: str) -> None:
         """Remove ``member`` from the queue; a no-op when absent."""
 
+    @abstractmethod
+    async def notify_deadline(self) -> None:
+        """Wake every worker's deadline consumer (best effort).
+
+        Notification is advisory: consumers stay correct when a signal
+        is lost, because their poll backstop recovers it.
+        """
+
+    @abstractmethod
+    def subscribe_deadlines(self) -> AsyncContextManager[AsyncIterator[None]]:
+        """Async context manager yielding deadline-enqueue signals."""
+
 
 def _channel(game_id: str) -> str:
     return f"{CHANNEL_PREFIX}{game_id}:events"
@@ -266,6 +282,20 @@ class RedisGameStore(GameStore):
     async def remove_deadline(self, member: str) -> None:
         await self._redis.zrem(DEADLINES_KEY, member)
 
+    async def notify_deadline(self) -> None:
+        await self._redis.publish(DEADLINES_CHANNEL, "deadline")
+
+    @contextlib.asynccontextmanager
+    async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
+        pubsub = self._redis.pubsub()
+        await pubsub.subscribe(DEADLINES_CHANNEL)
+        try:
+            yield _redis_events(pubsub)
+        finally:
+            with contextlib.suppress(Exception):
+                await pubsub.unsubscribe(DEADLINES_CHANNEL)
+                await pubsub.aclose()
+
 
 async def _redis_events(pubsub) -> AsyncIterator[None]:
     async for message in pubsub.listen():
@@ -284,6 +314,7 @@ class InMemoryGameStore(GameStore):
         self._locks: Dict[str, asyncio.Lock] = {}
         self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
         self._deadlines: Dict[str, float] = {}
+        self._deadline_subscribers: Set[asyncio.Queue] = set()
 
     def _lock_for(self, game_id: str) -> asyncio.Lock:
         lock = self._locks.get(game_id)
@@ -352,6 +383,19 @@ class InMemoryGameStore(GameStore):
     async def remove_deadline(self, member: str) -> None:
         self._deadlines.pop(member, None)
 
+    async def notify_deadline(self) -> None:
+        for queue in list(self._deadline_subscribers):
+            queue.put_nowait(_BUMP)
+
+    @contextlib.asynccontextmanager
+    async def subscribe_deadlines(self) -> AsyncIterator[AsyncIterator[None]]:
+        queue: asyncio.Queue = asyncio.Queue()
+        self._deadline_subscribers.add(queue)
+        try:
+            yield _queue_events(queue)
+        finally:
+            self._deadline_subscribers.discard(queue)
+
 
 async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]:
     while True:

server/packages/tavolo-platform/tests/helpers.py

@@ -220,6 +220,7 @@ class DummyEngine(GameEngine):
 
 def make_platform(
     engines: Sequence[GameEngine] = (DummyEngine(),),
+    heartbeat_ms: int = 50,
 ) -> Tuple[KayaApp, Platform, TortoiseMixin]:
     """Build a throwaway app + platform wired with in-memory stores."""
     registry = GameRegistry(engines)
@@ -251,7 +252,7 @@ def make_platform(
     scheduler = DeadlineScheduler(
         game_store,
         registry,
-        heartbeat_ms=50,
+        heartbeat_ms=heartbeat_ms,
         context_binder=tortoise_mixin.ensure_context,
     )
     platform = Platform(

server/packages/tavolo-platform/tests/test_deadlines.py

@@ -112,6 +112,38 @@ async def _finished(store, game_id: str) -> Optional[GameSession]:
     return None
 
 
+class NotificationTest(unittest.TestCase):
+    """Deadline-enqueue notifications wake consumers on other workers."""
+
+    @async_test
+    async def test_notification_wakes_a_long_heartbeat_consumer(self) -> None:
+        # The consumer's heartbeat is a minute away and nothing sets its
+        # local wake event: only the pub/sub notification can deliver the
+        # deadline in time. This is the cross-worker path.
+        _, platform, _ = make_platform(heartbeat_ms=60_000)
+        store = platform.game_store
+        scheduler = platform.scheduler
+        session = _started_session(deadline_in_seconds=0.1)
+        await store.save(session)
+        scheduler.ensure_consumer()
+        # Let the listener task subscribe before announcing.
+        await asyncio.sleep(0.05)
+
+        engine = platform.registry.require(session.game_type)
+        deadline = engine.next_deadline(session)
+        assert deadline is not None
+        member = encode({
+            "game_id": session.id,
+            "kind": deadline.kind,
+            "token": deadline.token,
+        })
+        await store.add_deadline(member, deadline.due_at.timestamp())
+        await store.notify_deadline()
+
+        result = await _wait_for(lambda: _plays_is(store, session.id, 1))
+        self.assertIsNotNone(result, "notification did not wake the consumer")
+
+
 class ProcessDueTest(unittest.TestCase):
     """Direct ``process_due`` behaviour: engine revalidation and idempotency."""
 

server/packages/tavolo-platform/tests/test_store.py

@@ -270,6 +270,17 @@ class InMemoryGameStoreTest(unittest.TestCase):
         self.assertEqual(60.0, await store.next_deadline())
         self.assertEqual(["b"], await store.due_deadlines(now=100.0))
 
+    @async_test
+    async def test_deadline_notifications_reach_subscribers(self) -> None:
+        store = InMemoryGameStore(_registry())
+        async with store.subscribe_deadlines() as events:
+            await store.notify_deadline()
+            signal = await asyncio.wait_for(anext(events), timeout=1.0)
+            self.assertIsNone(signal)
+        # The subscription is gone after the context exits: notifying
+        # again is a no-op rather than an error.
+        await store.notify_deadline()
+
 
 class RedisGameStoreTest(unittest.TestCase):
     @async_test