diff --git a/server/README.md b/server/README.md index 434922f..80fb845 100644 --- a/server/README.md +++ b/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::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 diff --git a/server/src/tavolo/app.py b/server/src/tavolo/app.py index 3aaae64..c2fd73c 100644 --- a/server/src/tavolo/app.py +++ b/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, diff --git a/server/src/tavolo/config.py b/server/src/tavolo/config.py index 4b48d61..c456dfc 100644 --- a/server/src/tavolo/config.py +++ b/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"), ) diff --git a/server/src/tavolo/deadlines.py b/server/src/tavolo/deadlines.py new file mode 100644 index 0000000..4d238a8 --- /dev/null +++ b/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) diff --git a/server/src/tavolo/routes/games.py b/server/src/tavolo/routes/games.py index b67f05e..40037f3 100644 --- a/server/src/tavolo/routes/games.py +++ b/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)) diff --git a/server/src/tavolo/store.py b/server/src/tavolo/store.py index c7066cc..9b16873 100644 --- a/server/src/tavolo/store.py +++ b/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: diff --git a/server/src/tavolo/ws.py b/server/src/tavolo/ws.py index 234eaca..38df213 100644 --- a/server/src/tavolo/ws.py +++ b/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 -plays a random card (with a random legal capture when one is required) for -them, so a disconnected or idle player cannot stall the match. The timer is -re-armed by every client connection and state broadcast, and fires -immediately when a reconnect finds the deadline already past. +Timeouts do not depend on anyone being connected: both the per-turn +auto-play and the hand-end auto-continue are driven by the absolute +deadlines persisted on the game state, via the shared deadline queue +drained by a consumer on every worker (see :mod:`tavolo.deadlines`). A +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 .stats import save_match_result +from .game.state import PHASE_FINISHED, GameState 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) - - -def schedule_hand_end_timer(game_id: str, hand_number: int, timeout: int) -> None: - """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) + # The fourth ack deals the next hand, which arms a new turn + # deadline; earlier acks change nothing and this is a no-op. + await deadlines.sync_deadline(game_store, state) 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) diff --git a/server/tests/test_deadlines.py b/server/tests/test_deadlines.py new file mode 100644 index 0000000..ad61e4c --- /dev/null +++ b/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() diff --git a/server/tests/test_store.py b/server/tests/test_store.py index 6511e37..35c050e 100644 --- a/server/tests/test_store.py +++ b/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()