Drive timeouts from a shared Redis deadline queue

Turn auto-play and hand-end auto-continue were process-local asyncio
tasks armed only by client connects and state broadcasts: with no
sockets connected the next turn's timer was never armed, a hand-end
timer died with its worker, and neither survived a pod restart.

Deadlines are now driven by the absolute timestamps persisted on the
game state and enqueued in a shared Redis sorted set. Every worker runs
a consumer that fires due entries under the per-game lock after
revalidating them against the live state, so timeouts no longer depend
on any player being connected and survive the death of any worker.
Delivery is at-least-once: entries are removed only after processing,
and revalidation makes duplicate deliveries no-ops.

Queue entries carry the deadline as integer epoch milliseconds, which
also serves as the revalidation token, and the score derives from the
same value.

server/README.md

@@ -60,6 +60,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 between-hands scoring summary waits for acknowledgements |
 | `TURN_TIMEOUT_SECONDS` | `30` | Seconds a player has to play before the server plays a random legal card for them |
+| `DEADLINE_HEARTBEAT_MS` | `1000` | Upper bound on the deadline consumer's poll interval (locally enqueued deadlines fire on time regardless) |
 | `LOGGING_CONFIG` | unset | Path to a YAML logging configuration file (see below). Unset logs DEBUG to the console |
 | `APP_HOST` / `APP_PORT` | `0.0.0.0` / `8080` | Bind address |
 
@@ -111,6 +112,12 @@ loggers:
 - `tavolo:game:<uuid>:events` — a pub/sub channel carrying "state changed"
   signals; every open WebSocket reloads the state and pushes the
   personalized view to its player.
+- `tavolo:deadlines` — a sorted set (score = due timestamp) of pending
+  timeouts: turn auto-plays and hand-end auto-continues. Every worker runs
+  a consumer that fires due entries under the per-game lock, so timeouts
+  do not depend on any player being connected and survive the death of
+  any worker (delivery is at-least-once; entries are revalidated against
+  the live state before firing).
 
 ### Postgres (statistics, via Tortoise ORM + aerich migrations)
 
@@ -181,7 +188,9 @@ player on turn does not move before it, the server plays a random legal
 card for them (picking one of the legal captures at random when a capture
 is required), so a disconnected or idle player cannot stall the match. The
 timeout is `TURN_TIMEOUT_SECONDS` (default 30); the auto-played move is
-broadcast like any other.
+broadcast like any other. Deadlines fire from the shared `tavolo:deadlines`
+queue (see above), not from timers tied to client connections, so the
+match keeps progressing even with every player disconnected.
 
 ### Hand-end summary
 
@@ -257,7 +266,8 @@ src/tavolo/
 ├── aerich_config.py    # aerich CLI configuration
 ├── models.py           # Match, MatchPlayer (Postgres)
 ├── stats.py            # finished match -> Postgres persistence
-├── store.py            # Redis / in-memory live-game store
+├── store.py            # Redis / in-memory live-game store (+ deadline queue)
+├── deadlines.py        # connection-independent timeout scheduler
 ├── ws.py               # WebSocket live-play endpoint
 ├── game/
 │   ├── state.py        # GameState / PlayerState / Card, JSON (de)serialization

server/src/tavolo/app.py

@@ -28,6 +28,7 @@ from kaya.session.redis import RedisSessionStore
 from redis.asyncio import Redis
 
 from .config import settings
+from .deadlines import DeadlineSchedulerMixin
 from .logging_config import configure_logging
 from .store import GameStore, InMemoryGameStore, RedisGameStore
 from .tortoise_mixin import TortoiseMixin
@@ -76,7 +77,8 @@ tortoise_mixin = TortoiseMixin(
     skip_paths=frozenset({"/api/health", "/api/docs", "/api/openapi.json"}),
 )
 
-app = KayaApp(mixins=[session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin])
+app = KayaApp(mixins=[session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin,
+                      DeadlineSchedulerMixin(game_store)])
 log.debug(
     "timeouts: hand_ack=%ds turn=%ds",
     settings.hand_ack_timeout_seconds,

server/src/tavolo/config.py

@@ -76,6 +76,11 @@ class Settings:
     # Seconds a player has to play before the server plays a random legal
     # card for them (covering disconnects and idle players).
     turn_timeout_seconds: int
+    # Upper bound on how long the deadline consumer sleeps between polls.
+    # Locally enqueued deadlines wake the consumer immediately; the
+    # heartbeat only bounds the discovery delay for deadlines enqueued by
+    # other workers.
+    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.
     logging_config: Optional[str]
@@ -113,6 +118,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")),
             logging_config=os.environ.get("LOGGING_CONFIG"),
         )
 

server/src/tavolo/deadlines.py

@@ -0,0 +1,290 @@
+"""Deadline-driven timeouts, independent of player connections.
+
+Both in-match timeouts — the per-turn auto-play (``turn_deadline``) and
+the hand-end summary auto-continue (``hand_end_deadline``) — are driven by
+the absolute deadlines persisted on the game state, never by which players
+(or whether any players) are connected.
+
+Every mutation that sets a deadline enqueues an entry in the store's
+shared deadline queue (a Redis sorted set in production, see
+:mod:`tavolo.store`), and a background consumer running on **every**
+worker polls the queue for due entries. An entry records the phase, hand,
+turn and deadline (as integer epoch milliseconds) it was enqueued for;
+before acting, the consumer revalidates all of it against the live state
+under the per-game lock, so entries that were overtaken by events (a play
+landed in time, the hand was acknowledged, the deadline moved) are simply
+discarded.
+
+Delivery is at-least-once: an entry is removed from the queue only after
+it has been processed. If a worker dies mid-processing, the entry stays in
+Redis and another worker's consumer picks it up — the lock plus
+revalidation make the duplicate delivery a no-op. Entries whose game has
+expired are dropped the first time they fire, so the queue is
+self-cleaning.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from datetime import datetime
+from logging import getLogger
+from typing import Any, Dict, Optional
+
+from kaya.core import KayaApp, KayaMixin
+
+from .config import settings
+from .game import engine
+from .game.errors import GameError
+from .game.state import PHASE_HAND_END, PHASE_PLAYING, PHASE_FINISHED, GameState
+from .stats import save_match_result
+from .store import GameStore
+
+log = getLogger(__name__)
+
+# Entry kinds enqueued in the deadline queue.
+KIND_TURN = "turn"
+KIND_HAND_END = "hand_end"
+
+# One consumer task and its wake-up event per event loop (tests run each
+# test on a fresh loop).
+_consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
+_wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
+
+
+def encode(entry: Dict[str, Any]) -> str:
+    """Canonical queue-member encoding for a deadline entry."""
+    return json.dumps(entry, sort_keys=True)
+
+
+def _decode(member: Any) -> Optional[Dict[str, Any]]:
+    if isinstance(member, bytes):
+        member = member.decode("utf-8")
+    if not isinstance(member, str):
+        return None
+    try:
+        entry = json.loads(member)
+    except ValueError:
+        return None
+    return entry if isinstance(entry, dict) else None
+
+
+def _deadline_ms(iso: Optional[str]) -> Optional[int]:
+    """Epoch milliseconds for an ISO-8601 deadline, ``None`` when absent
+    or unparseable. Queue entries carry this integer (never the ISO
+    string) as their revalidation token."""
+    if not iso:
+        return None
+    try:
+        return int(datetime.fromisoformat(iso).timestamp() * 1000)
+    except ValueError:
+        return None
+
+
+async def sync_deadline(store: GameStore, state: GameState) -> None:
+    """Enqueue the deadline the current state carries, if any.
+
+    Called after every mutation that can set a deadline (plays, acks, game
+    start) and as a backstop when a client connects. Enqueueing is
+    idempotent: an identical entry is already queued with the same due
+    time, so re-adding it changes nothing.
+    """
+    entry: Optional[Dict[str, Any]] = None
+    due_ms: Optional[int] = None
+    if state.phase == PHASE_PLAYING and state.turn_deadline:
+        due_ms = _deadline_ms(state.turn_deadline)
+        entry = {
+            "game_id": state.id,
+            "kind": KIND_TURN,
+            "hand": state.hand_number,
+            "turn": state.turn,
+            "deadline": due_ms,
+        }
+    elif state.phase == PHASE_HAND_END and state.hand_end_deadline:
+        due_ms = _deadline_ms(state.hand_end_deadline)
+        entry = {
+            "game_id": state.id,
+            "kind": KIND_HAND_END,
+            "hand": state.hand_number,
+            "deadline": due_ms,
+        }
+    if entry is None or due_ms is None:
+        if entry is not None:
+            log.warning("game %s: unparseable deadline", state.id)
+        return
+    ensure_consumer(store)
+    # The score derives from the same value carried in the member, so the
+    # two can never disagree.
+    await store.add_deadline(encode(entry), due_ms / 1000)
+    wake = _wake_events.get(asyncio.get_running_loop())
+    if wake is not None:
+        wake.set()
+
+
+async def finalize_mutation(store: GameStore, state: GameState) -> None:
+    """Persist a successful mutation, notify subscribers and enqueue the
+    next deadline.
+
+    Callers must hold the per-game lock. Handles the terminal transition:
+    the match result is written to Postgres once (guarded by
+    ``stats_saved``).
+    """
+    if state.phase == PHASE_FINISHED:
+        await save_match_result(state)
+        log.info(
+            "game %s finished: team %s wins %d-%d",
+            state.id,
+            "A" if state.winner == 0 else "B",
+            state.scores[0],
+            state.scores[1],
+        )
+    await store.save(state)
+    await store.publish(state.id)
+    await sync_deadline(store, state)
+
+
+async def process_due(store: GameStore, member: Any) -> None:
+    """Fire a single due deadline entry.
+
+    Revalidates the entry against the live state under the per-game lock;
+    stale or foreign entries are discarded without effect. The entry is
+    removed from the queue once handled (including "nothing to do"); if
+    handling fails (e.g. the lock cannot be acquired), the entry is left
+    in the queue so another consumer retries it.
+    """
+    entry = _decode(member)
+    if entry is None:
+        log.warning("deadline consumer: dropping malformed entry %r", member)
+        await store.remove_deadline(member)
+        return
+    game_id = entry.get("game_id")
+    kind = entry.get("kind")
+    if not isinstance(game_id, str):
+        await store.remove_deadline(member)
+        return
+    async with store.lock(game_id):
+        state = await store.load(game_id)
+        if state is not None:
+            if kind == KIND_TURN:
+                await _fire_turn(store, state, entry)
+            elif kind == KIND_HAND_END:
+                await _fire_hand_end(store, state, entry)
+    await store.remove_deadline(member)
+
+
+async def _fire_turn(store: GameStore, state: GameState, entry: Dict[str, Any]) -> None:
+    if (
+        state.phase != PHASE_PLAYING
+        or state.hand_number != entry.get("hand")
+        or state.turn != entry.get("turn")
+        or _deadline_ms(state.turn_deadline) != entry.get("deadline")
+    ):
+        return
+    seat = state.turn
+    try:
+        engine.auto_play(state)
+    except GameError:
+        return
+    log.info(
+        "game %s: auto-played for %s (turn timeout, hand %d)",
+        state.id,
+        state.players[seat].sub if seat < len(state.players) else "?",
+        entry.get("hand"),
+    )
+    await finalize_mutation(store, state)
+
+
+async def _fire_hand_end(store: GameStore, state: GameState, entry: Dict[str, Any]) -> None:
+    if (
+        state.phase != PHASE_HAND_END
+        or state.hand_number != entry.get("hand")
+        or _deadline_ms(state.hand_end_deadline) != entry.get("deadline")
+    ):
+        return
+    for player in state.players:
+        engine.acknowledge_hand(state, player.sub)
+    log.info(
+        "game %s: hand %d auto-advanced after the acknowledgement timeout",
+        state.id,
+        entry.get("hand"),
+    )
+    await finalize_mutation(store, state)
+
+
+# --- consumer lifecycle -------------------------------------------------------
+
+
+def ensure_consumer(store: GameStore) -> None:
+    """Start the deadline consumer on the running loop if not yet running.
+
+    Called lazily whenever a deadline is enqueued (the ASGI test transport
+    never fires the lifespan hooks, so the mixin's ``setup`` alone is not
+    enough) and on application startup.
+    """
+    loop = asyncio.get_running_loop()
+    for old in list(_consumers):
+        if old.is_closed():
+            _consumers.pop(old, None)
+            _wake_events.pop(old, None)
+    task = _consumers.get(loop)
+    if task is None or task.done():
+        _wake_events[loop] = asyncio.Event()
+        _consumers[loop] = loop.create_task(_run(store, loop))
+        log.debug("deadline consumer started")
+
+
+def stop_consumer(loop: asyncio.AbstractEventLoop) -> None:
+    task = _consumers.pop(loop, None)
+    _wake_events.pop(loop, None)
+    if task is not None:
+        task.cancel()
+
+
+async def _run(store: GameStore, loop: asyncio.AbstractEventLoop) -> None:
+    wake = _wake_events[loop]
+    heartbeat = settings.deadline_heartbeat_ms / 1000
+    while True:
+        # Clear before polling so an enqueue racing the poll re-wakes us.
+        wake.clear()
+        delay = heartbeat
+        try:
+            for member in await store.due_deadlines(time.time()):
+                try:
+                    await process_due(store, member)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    # Left in the queue; retried on the next pass.
+                    log.exception("deadline consumer: failed to process %r", member)
+            next_due = await store.next_deadline()
+            if next_due is not None:
+                delay = max(0.0, min(heartbeat, next_due - time.time()))
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            log.exception("deadline consumer: poll failed; retrying")
+        try:
+            await asyncio.wait_for(wake.wait(), timeout=delay)
+        except asyncio.TimeoutError:
+            pass
+
+
+class DeadlineSchedulerMixin(KayaMixin):
+    """Run the deadline consumer for the whole app lifetime.
+
+    Every worker (and every pod) runs the same consumer; coordination
+    happens exclusively through the shared deadline queue and the per-game
+    locks, so any worker may fire any game's deadline.
+    """
+
+    def __init__(self, store: GameStore) -> None:
+        self._store = store
+
+    def apply(self, app: KayaApp) -> None:
+        pass
+
+    def setup(self, loop: asyncio.AbstractEventLoop) -> None:
+        ensure_consumer(self._store)
+
+    def shutdown(self, loop: asyncio.AbstractEventLoop) -> None:
+        stop_consumer(loop)

server/src/tavolo/routes/games.py

@@ -16,7 +16,7 @@ from typing import Any, Dict, Optional
 from kaya.core import HttpContext
 from kaya.openapi import operation
 
-from .. import auth
+from .. import auth, deadlines
 from ..app import app, game_store, oidc_mixin
 from ..auth import require_auth
 from ..config import settings
@@ -222,6 +222,9 @@ async def join_game(ctx: HttpContext) -> None:
             return
         await game_store.save(state)
         await game_store.publish(state.id)
+        # When the fourth join started the match, the first turn deadline
+        # was armed; queue it so it fires even if nobody ever connects.
+        await deadlines.sync_deadline(game_store, state)
     seat = next(p.seat for p in state.players if p.sub == user.sub)
     if state.phase == PHASE_LOBBY:
         log.info("%s joined game %s (seat %d, %d/4 players)", user.sub, state.id, seat, len(state.players))

server/src/tavolo/store.py

@@ -18,6 +18,14 @@ channel as a simple "something changed" signal; every open websocket
 reloads the state and renders the personalized view. Publishing only a
 signal (never the state) means updated state reaches connections on every
 worker without leaking hidden hands into the channel.
+
+Timeouts (turn auto-play, hand-end auto-continue) are driven by a shared
+delayed-deadline queue: producers enqueue an opaque ``member`` string with
+a due timestamp, and a consumer on every worker polls for due entries.
+Delivery is at-least-once — entries are removed only after they are
+processed — so a worker dying mid-processing cannot lose a deadline;
+consumers revalidate entries against the live state under the per-game
+lock, which makes duplicate deliveries harmless.
 """
 from __future__ import annotations
 
@@ -26,7 +34,7 @@ import contextlib
 import json
 from abc import ABC, abstractmethod
 from logging import getLogger
-from typing import AsyncContextManager, AsyncIterator, Dict, Optional, Set
+from typing import AsyncContextManager, AsyncIterator, Dict, List, Optional, Set, cast
 
 from redis.asyncio import Redis
 
@@ -37,6 +45,7 @@ log = getLogger(__name__)
 GAME_KEY_PREFIX = "tavolo:game:"
 CODE_KEY_PREFIX = "tavolo:code:"
 CHANNEL_PREFIX = "tavolo:game:"
+DEADLINES_KEY = "tavolo:deadlines"
 
 # Sentinel pushed into in-memory subscriber queues to signal a change.
 _BUMP = b"update"
@@ -69,6 +78,26 @@ class GameStore(ABC):
     async def publish(self, game_id: str) -> None:
         """Signal that the state of ``game_id`` changed."""
 
+    @abstractmethod
+    async def add_deadline(self, member: str, due_at: float) -> None:
+        """Enqueue ``member`` to fire at ``due_at`` (epoch seconds).
+
+        Idempotent for identical members: re-adding an existing member only
+        updates its due time.
+        """
+
+    @abstractmethod
+    async def due_deadlines(self, now: float, limit: int = 32) -> List[str]:
+        """Return up to ``limit`` enqueued members due at or before ``now``."""
+
+    @abstractmethod
+    async def next_deadline(self) -> Optional[float]:
+        """Return the earliest pending due time (epoch seconds), if any."""
+
+    @abstractmethod
+    async def remove_deadline(self, member: str) -> None:
+        """Remove ``member`` from the queue; a no-op when absent."""
+
 
 def _channel(game_id: str) -> str:
     return f"{CHANNEL_PREFIX}{game_id}:events"
@@ -128,6 +157,25 @@ class RedisGameStore(GameStore):
         await self._redis.publish(_channel(game_id), "update")
         log.debug("redis publish %s", game_id)
 
+    async def add_deadline(self, member: str, due_at: float) -> None:
+        await self._redis.zadd(DEADLINES_KEY, {member: due_at})
+
+    async def due_deadlines(self, now: float, limit: int = 32) -> List[str]:
+        members = cast(
+            list,
+            await self._redis.zrangebyscore(
+                DEADLINES_KEY, "-inf", now, start=0, num=limit
+            ),
+        )
+        return [m.decode("utf-8") if isinstance(m, bytes) else m for m in members]
+
+    async def next_deadline(self) -> Optional[float]:
+        earliest = await self._redis.zrange(DEADLINES_KEY, 0, 0, withscores=True)
+        return float(earliest[0][1]) if earliest else None
+
+    async def remove_deadline(self, member: str) -> None:
+        await self._redis.zrem(DEADLINES_KEY, member)
+
 
 async def _redis_events(pubsub) -> AsyncIterator[None]:
     async for message in pubsub.listen():
@@ -143,6 +191,7 @@ class InMemoryGameStore(GameStore):
         self._codes: Dict[str, str] = {}
         self._locks: Dict[str, asyncio.Lock] = {}
         self._subscribers: Dict[str, Set[asyncio.Queue]] = {}
+        self._deadlines: Dict[str, float] = {}
 
     def _lock_for(self, game_id: str) -> asyncio.Lock:
         lock = self._locks.get(game_id)
@@ -187,6 +236,20 @@ class InMemoryGameStore(GameStore):
         for queue in list(self._subscribers.get(game_id, ())):
             queue.put_nowait(_BUMP)
 
+    async def add_deadline(self, member: str, due_at: float) -> None:
+        self._deadlines[member] = due_at
+
+    async def due_deadlines(self, now: float, limit: int = 32) -> List[str]:
+        due = [m for m, due_at in self._deadlines.items() if due_at <= now]
+        due.sort(key=self._deadlines.__getitem__)
+        return due[:limit]
+
+    async def next_deadline(self) -> Optional[float]:
+        return min(self._deadlines.values(), default=None)
+
+    async def remove_deadline(self, member: str) -> None:
+        self._deadlines.pop(member, None)
+
 
 async def _queue_events(queue: asyncio.Queue) -> AsyncIterator[None]:
     while True:

server/src/tavolo/ws.py

@@ -30,29 +30,28 @@ state is saved to Redis and a change signal is published. Every connected
 websocket is subscribed to that signal and re-renders the state, so all
 players see the move immediately (and consistently across workers).
 
-If a player does not move before the per-game ``turn_timeout``, the server
+Timeouts do not depend on anyone being connected: both the per-turn
-plays a random card (with a random legal capture when one is required) for
+auto-play and the hand-end auto-continue are driven by the absolute
-them, so a disconnected or idle player cannot stall the match. The timer is
+deadlines persisted on the game state, via the shared deadline queue
-re-armed by every client connection and state broadcast, and fires
+drained by a consumer on every worker (see :mod:`tavolo.deadlines`). A
-immediately when a reconnect finds the deadline already past.
+disconnected or idle player therefore cannot stall the match, and a
+worker dying cannot either.
 """
 from __future__ import annotations
 
 import asyncio
 import json
 from contextlib import suppress
-from datetime import datetime, timezone
 from logging import getLogger
-from typing import Any, Awaitable, Callable, Dict, Optional
+from typing import Any, Awaitable, Callable, Dict
 
 from kaya.core import WebSocket
 
-from . import auth
+from . import auth, deadlines
 from .app import app, game_store
 from .game import engine
 from .game.errors import GameError
-from .game.state import PHASE_FINISHED, PHASE_HAND_END, PHASE_PLAYING, GameState
+from .game.state import PHASE_FINISHED, GameState
-from .stats import save_match_result
 
 log = getLogger(__name__)
 
@@ -95,7 +94,10 @@ async def game_socket(ws: WebSocket, game_id: str) -> None:
             await ws.send_text(json.dumps(payload))
 
     await send(_state_message(state, user.sub))
-    schedule_turn_timer(game_id, state)
+    # Backstop: make sure the current phase's deadline is queued even if
+    # its entry was lost (e.g. the queue was flushed while the game lived
+    # on thanks to its sliding TTL).
+    await deadlines.sync_deadline(game_store, state)
 
     async with game_store.subscribe(game_id) as events:
         forward = asyncio.create_task(
@@ -126,7 +128,6 @@ async def _forward(
         state = await game_store.load(game_id)
         if state is None:
             return
-        schedule_turn_timer(game_id, state)
         await send(_state_message(state, sub))
         if state.phase == PHASE_FINISHED:
             await send(
@@ -166,10 +167,6 @@ async def _handle_message(send: Send, game_id: str, sub: str, raw: str) -> None:
 
 # --- hand-end acknowledgement ------------------------------------------------
 
-# Running auto-continue timers, keyed by (game_id, hand_number), so a hand's
-# timeout is scheduled only once even when several clients are connected.
-_hand_end_timers: Dict[tuple, asyncio.Task] = {}
-
 
 async def _handle_ack(send: Send, game_id: str, sub: str) -> None:
     async with game_store.lock(game_id):
@@ -185,122 +182,9 @@ async def _handle_ack(send: Send, game_id: str, sub: str) -> None:
         log.debug("game %s: %s acknowledged hand %d", game_id, sub, state.hand_number)
         await game_store.save(state)
         await game_store.publish(game_id)
-
+        # The fourth ack deals the next hand, which arms a new turn
-
+        # deadline; earlier acks change nothing and this is a no-op.
-def schedule_hand_end_timer(game_id: str, hand_number: int, timeout: int) -> None:
+        await deadlines.sync_deadline(game_store, state)
-    """Deal the next hand after the acknowledgement timeout, even if not
-    everyone has clicked. Fizzles if the hand already advanced."""
-    key = (game_id, hand_number)
-    if key in _hand_end_timers:
-        return
-
-    async def _auto_advance() -> None:
-        try:
-            await asyncio.sleep(timeout)
-            async with game_store.lock(game_id):
-                state = await game_store.load(game_id)
-                if (
-                    state is None
-                    or state.phase != engine.PHASE_HAND_END
-                    or state.hand_number != hand_number
-                ):
-                    return
-                for player in state.players:
-                    engine.acknowledge_hand(state, player.sub)
-                await game_store.save(state)
-                await game_store.publish(game_id)
-                log.info(
-                    "game %s: hand %d auto-advanced after the acknowledgement timeout",
-                    game_id,
-                    hand_number,
-                )
-        finally:
-            _hand_end_timers.pop(key, None)
-
-    _hand_end_timers[key] = asyncio.create_task(_auto_advance())
-
-
-# --- auto-play on turn timeout ------------------------------------------------
-
-# Running turn timers, keyed by (game_id, hand_number, turn, deadline), so a
-# turn's timeout is scheduled only once even when several clients are
-# connected. Including the deadline means a re-arm after a reconnect cannot
-# duplicate a timer for a turn that was already auto-played.
-_turn_timers: Dict[tuple, asyncio.Task] = {}
-
-
-def schedule_turn_timer(game_id: str, state: GameState) -> None:
-    """Auto-play a random legal card if the player on turn misses the
-    deadline. Fizzles if the turn already advanced."""
-    if state.phase != PHASE_PLAYING or not state.turn_deadline:
-        return
-    key = (game_id, state.hand_number, state.turn, state.turn_deadline)
-    if key in _turn_timers:
-        return
-
-    hand_number = state.hand_number
-    turn = state.turn
-    deadline_raw = state.turn_deadline
-    try:
-        deadline = datetime.fromisoformat(deadline_raw)
-    except ValueError:
-        return
-
-    async def _auto_play() -> None:
-        try:
-            delay = (deadline - datetime.now(timezone.utc)).total_seconds()
-            await asyncio.sleep(max(delay, 0))
-            async with game_store.lock(game_id):
-                state = await game_store.load(game_id)
-                if (
-                    state is None
-                    or state.phase != PHASE_PLAYING
-                    or state.hand_number != hand_number
-                    or state.turn != turn
-                    or state.turn_deadline != deadline_raw
-                ):
-                    # The turn moved on (or the game ended) without this
-                    # timer firing: make sure the current turn is armed.
-                    if state is not None:
-                        schedule_turn_timer(game_id, state)
-                    return
-                try:
-                    engine.auto_play(state)
-                except GameError:
-                    return
-                log.info(
-                    "game %s: auto-played for %s (turn timeout, hand %d)",
-                    game_id,
-                    state.players[turn].sub if turn < len(state.players) else "?",
-                    hand_number,
-                )
-                await _after_play(state, game_id)
-        finally:
-            _turn_timers.pop(key, None)
-
-    _turn_timers[key] = asyncio.create_task(_auto_play())
-
-
-async def _after_play(state: GameState, game_id: str) -> None:
-    """Persist a successful move and notify every connected player.
-
-    Callers must hold the per-game lock. Handles the two terminal
-    transitions: the match result is written to Postgres once, and a
-    hand-end summary schedules the auto-continue timeout.
-    """
-    if state.phase == PHASE_FINISHED:
-        await save_match_result(state)
-        log.info(
-            "game %s finished: team %s wins %d-%d",
-            game_id,
-            "A" if state.winner == 0 else "B",
-            state.scores[0],
-            state.scores[1],
-        )
-    elif state.phase == PHASE_HAND_END:
-        schedule_hand_end_timer(game_id, state.hand_number, state.hand_ack_timeout)
-    await game_store.save(state)
-    await game_store.publish(game_id)
 
 
 async def _handle_play(
@@ -335,4 +219,4 @@ async def _handle_play(
             return
 
         log.debug("game %s: %s played %s (capture: %s)", game_id, sub, card, capture or "-")
-        await _after_play(state, game_id)
+        await deadlines.finalize_mutation(game_store, state)

server/tests/test_deadlines.py

@@ -0,0 +1,191 @@
+"""Deadline-queue timeout tests.
+
+Timeouts must be driven by the persisted deadlines and the shared queue,
+not by connected sockets: these tests seed games, queue their deadlines
+and let the background consumer fire them without a single websocket.
+"""
+from __future__ import annotations
+
+import asyncio
+import unittest
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from pwo import async_test
+
+from tavolo import deadlines
+from tavolo.app import game_store
+from tavolo.game import engine
+from tavolo.game.state import GameState, PlayerState
+
+PLAYERS = ("alice", "bob", "carol", "dave")
+
+
+def _ms(iso: str) -> int:
+    """Epoch milliseconds for an ISO-8601 timestamp (the queue-entry form)."""
+    return int(datetime.fromisoformat(iso).timestamp() * 1000)
+
+
+def _hand_end_state(game_id: str, deadline: str) -> GameState:
+    """A game paused on the hand-end summary, waiting for acks."""
+    state = GameState(
+        id=game_id,
+        join_code="DLhend",
+        creator_sub="alice",
+        target_score=11,
+        phase="hand_end",
+        # Long turn timeout: the next hand's auto-play must not interfere
+        # with later tests sharing this store.
+        turn_timeout=3600,
+    )
+    state.players = [
+        PlayerState(sub=name, name=name.capitalize(), seat=i)
+        for i, name in enumerate(PLAYERS)
+    ]
+    state.hand_end_deadline = deadline
+    return state
+
+
+async def _wait_for(predicate, timeout: float = 5.0) -> Optional[GameState]:
+    """Poll the store until ``predicate`` holds for the loaded state."""
+    deadline = asyncio.get_running_loop().time() + timeout
+    while asyncio.get_running_loop().time() < deadline:
+        state = await predicate()
+        if state is not None:
+            return state
+        await asyncio.sleep(0.05)
+    return None
+
+
+class ConnectionIndependenceTest(unittest.TestCase):
+    @async_test
+    async def test_turn_timeout_fires_with_no_connections(self) -> None:
+        state = engine.create_game(
+            "dl-turn-1", "DLT001", "alice", "Alice",
+            target_score=11, turn_timeout=1,
+        )
+        for name in PLAYERS[1:]:
+            engine.join_game(state, name, name.capitalize())
+        assert state.turn_deadline is not None
+        await game_store.save(state)
+        await deadlines.sync_deadline(game_store, state)
+
+        # Nobody ever connects: the consumer must still auto-play for Bob
+        # (seat 1, first to act).
+        result = await _wait_for(
+            lambda: _turn_is(state.id, 2),
+        )
+        self.assertIsNotNone(result, "turn deadline never fired")
+        assert result is not None
+        self.assertEqual(1, result.last_move.seat if result.last_move else None)
+
+        # Defuse the follow-on turn deadlines so this game cannot keep
+        # auto-playing while later tests run.
+        result.turn_timeout = 3600
+        await game_store.save(result)
+
+    @async_test
+    async def test_hand_end_timeout_fires_with_no_connections(self) -> None:
+        deadline = (datetime.now(timezone.utc) + timedelta(seconds=1)).isoformat()
+        state = _hand_end_state("dl-handend-1", deadline)
+        await game_store.save(state)
+        await deadlines.sync_deadline(game_store, state)
+
+        # Nobody acks (nobody is even connected): the deadline must deal
+        # the next hand.
+        result = await _wait_for(
+            lambda: _phase_is("dl-handend-1", "playing"),
+        )
+        self.assertIsNotNone(result, "hand-end deadline never fired")
+        assert result is not None
+        self.assertEqual(2, result.hand_number)
+        self.assertEqual([], result.acked)
+
+
+async def _turn_is(game_id: str, turn: int) -> Optional[GameState]:
+    state = await game_store.load(game_id)
+    return state if state is not None and state.turn == turn else None
+
+
+async def _phase_is(game_id: str, phase: str) -> Optional[GameState]:
+    state = await game_store.load(game_id)
+    return state if state is not None and state.phase == phase else None
+
+
+class ProcessDueTest(unittest.TestCase):
+    """Direct ``process_due`` behaviour: revalidation and idempotency."""
+
+    @async_test
+    async def test_processing_twice_is_a_no_op(self) -> None:
+        # Simulates a worker dying after firing but before removing the
+        # entry: another worker re-delivers the same entry.
+        deadline = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
+        state = _hand_end_state("dl-idem-1", deadline)
+        await game_store.save(state)
+        member = deadlines.encode({
+            "game_id": state.id,
+            "kind": deadlines.KIND_HAND_END,
+            "hand": state.hand_number,
+            "deadline": _ms(deadline),
+        })
+
+        await deadlines.process_due(game_store, member)
+        await deadlines.process_due(game_store, member)
+
+        result = await game_store.load(state.id)
+        assert result is not None
+        # Advanced exactly once: hand 2, not hand 3.
+        self.assertEqual("playing", result.phase)
+        self.assertEqual(2, result.hand_number)
+
+    @async_test
+    async def test_stale_entry_is_discarded(self) -> None:
+        # A turn entry enqueued before a play landed in time: the state's
+        # deadline has moved, so the entry must not fire.
+        state = engine.create_game(
+            "dl-stale-1", "DLS001", "alice", "Alice",
+            target_score=11, turn_timeout=3600,
+        )
+        for name in PLAYERS[1:]:
+            engine.join_game(state, name, name.capitalize())
+        await game_store.save(state)
+        member = deadlines.encode({
+            "game_id": state.id,
+            "kind": deadlines.KIND_TURN,
+            "hand": state.hand_number,
+            "turn": state.turn,
+            # Not the live deadline (epoch milliseconds).
+            "deadline": 946684800000,
+        })
+        await game_store.add_deadline(member, due_at=0.0)
+
+        await deadlines.process_due(game_store, member)
+
+        result = await game_store.load(state.id)
+        assert result is not None
+        self.assertEqual(state.turn, result.turn)
+        # The entry was removed after processing.
+        self.assertNotIn(member, await game_store.due_deadlines(float("inf")))
+
+    @async_test
+    async def test_entry_for_expired_game_is_dropped(self) -> None:
+        member = deadlines.encode({
+            "game_id": "dl-gone",
+            "kind": deadlines.KIND_TURN,
+            "hand": 1,
+            "turn": 0,
+            "deadline": 946684800000,
+        })
+        await game_store.add_deadline(member, due_at=0.0)
+        await deadlines.process_due(game_store, member)
+        self.assertNotIn(member, await game_store.due_deadlines(float("inf")))
+
+    @async_test
+    async def test_malformed_entry_is_dropped(self) -> None:
+        await game_store.add_deadline("not json", due_at=0.0)
+        await deadlines.process_due(game_store, "not json")
+        self.assertNotIn("not json", await game_store.due_deadlines(float("inf")))
+
+
+if __name__ == "__main__":
+    unittest.main()

server/tests/test_store.py

@@ -108,6 +108,29 @@ class InMemoryGameStoreTest(unittest.TestCase):
             ["holder-enter", "holder-exit", "contender"], order
         )
 
+    @async_test
+    async def test_deadline_queue(self) -> None:
+        store = InMemoryGameStore()
+        self.assertIsNone(await store.next_deadline())
+        self.assertEqual([], await store.due_deadlines(now=100.0))
+
+        await store.add_deadline("b", due_at=50.0)
+        await store.add_deadline("a", due_at=10.0)
+        await store.add_deadline("c", due_at=200.0)
+        # Re-adding an existing member only updates its due time.
+        await store.add_deadline("b", due_at=60.0)
+
+        self.assertEqual(10.0, await store.next_deadline())
+        self.assertEqual(["a"], await store.due_deadlines(now=10.0))
+        self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0))
+        # Due entries come out in due-time order and stay queued until removed.
+        self.assertEqual(["a", "b"], await store.due_deadlines(now=100.0))
+
+        await store.remove_deadline("a")
+        await store.remove_deadline("a")  # removing twice is a no-op
+        self.assertEqual(60.0, await store.next_deadline())
+        self.assertEqual(["b"], await store.due_deadlines(now=100.0))
+
 
 if __name__ == "__main__":
     unittest.main()