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.
This commit is contained in:
+12
-2
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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))
|
||||
|
||||
@@ -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:
|
||||
|
||||
+17
-133
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user