Persist matches finished by deadline timeouts
The deadline consumer runs in a long-lived task outside any request, so a timeout that ended the match raised 'No TortoiseContext is currently active' in save_match_result before the state was saved: the game stayed stuck on the last turn and the entry retried forever. It only surfaced on the match-deciding turn; ordinary timeouts and human plays were fine. Bind the Tortoise context before persisting a finished match (optional context_binder wired to TortoiseMixin.ensure_context), and back off to the heartbeat when a due entry fails instead of hot-looping on it.
This commit is contained in:
@@ -163,6 +163,9 @@ scheduler = DeadlineScheduler(
|
||||
game_store,
|
||||
registry,
|
||||
heartbeat_ms=settings.deadline_heartbeat_ms,
|
||||
# The consumer task has no request context; bind the database context
|
||||
# before it persists a match that a timeout finished.
|
||||
context_binder=tortoise_mixin.ensure_context,
|
||||
)
|
||||
platform = Platform(
|
||||
registry=registry,
|
||||
|
||||
@@ -31,7 +31,7 @@ import json
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from logging import getLogger
|
||||
from typing import Any, Dict, Optional
|
||||
from typing import Any, Awaitable, Callable, Dict, Optional
|
||||
|
||||
from kaya.core import KayaApp, KayaMixin
|
||||
|
||||
@@ -67,6 +67,12 @@ class DeadlineScheduler:
|
||||
Holds the store and the registry so a single instance serves every
|
||||
game type. One consumer task (and its wake-up event) is kept per
|
||||
event loop — tests run each test on a fresh loop.
|
||||
|
||||
``context_binder`` is an optional async callable the consumer awaits
|
||||
before persisting a finished match. The consumer is a long-lived task
|
||||
of its own, outside any request, so it must bind whatever ambient
|
||||
context the database layer needs (see
|
||||
:meth:`~tavolo.platform.tortoise_mixin.TortoiseMixin.ensure_context`).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -74,10 +80,12 @@ class DeadlineScheduler:
|
||||
store: GameStore,
|
||||
registry: GameRegistry,
|
||||
heartbeat_ms: int = 1000,
|
||||
context_binder: Optional[Callable[[], Awaitable[None]]] = None,
|
||||
) -> None:
|
||||
self._store = store
|
||||
self._registry = registry
|
||||
self._heartbeat = heartbeat_ms / 1000
|
||||
self._context_binder = context_binder
|
||||
self._consumers: Dict[asyncio.AbstractEventLoop, asyncio.Task] = {}
|
||||
self._wake_events: Dict[asyncio.AbstractEventLoop, asyncio.Event] = {}
|
||||
|
||||
@@ -117,6 +125,10 @@ class DeadlineScheduler:
|
||||
if engine.is_finished(session):
|
||||
if session.finished_at is None:
|
||||
session.finished_at = datetime.now(timezone.utc)
|
||||
# A deadline can finish the match in the consumer task, which
|
||||
# has no request context: bind the database context first.
|
||||
if self._context_binder is not None:
|
||||
await self._context_binder()
|
||||
await save_match_result(session, engine)
|
||||
log.info("game %s (%s) finished", session.id, session.game_type)
|
||||
await self._store.save(session)
|
||||
@@ -196,6 +208,7 @@ class DeadlineScheduler:
|
||||
# Clear before polling so an enqueue racing the poll re-wakes us.
|
||||
wake.clear()
|
||||
delay = self._heartbeat
|
||||
failed = False
|
||||
try:
|
||||
for member in await self._store.due_deadlines(time.time()):
|
||||
try:
|
||||
@@ -205,6 +218,7 @@ class DeadlineScheduler:
|
||||
except Exception:
|
||||
# Left in the queue; retried on the next pass.
|
||||
log.exception("deadline consumer: failed to process %r", member)
|
||||
failed = True
|
||||
next_due = await self._store.next_deadline()
|
||||
if next_due is not None:
|
||||
delay = max(0.0, min(self._heartbeat, next_due - time.time()))
|
||||
@@ -212,6 +226,11 @@ class DeadlineScheduler:
|
||||
raise
|
||||
except Exception:
|
||||
log.exception("deadline consumer: poll failed; retrying")
|
||||
failed = True
|
||||
if failed:
|
||||
# A failed entry is still due: back off to the heartbeat
|
||||
# instead of polling it again immediately.
|
||||
delay = self._heartbeat
|
||||
try:
|
||||
await asyncio.wait_for(wake.wait(), timeout=delay)
|
||||
except asyncio.TimeoutError:
|
||||
|
||||
@@ -19,6 +19,9 @@ This mixin therefore:
|
||||
(``before_websocket`` hook), because the match-result write happens at
|
||||
the end of a WebSocket match. The long-lived connection task keeps the
|
||||
context for its whole lifetime.
|
||||
4. Exposes :meth:`ensure_context` for background work that belongs to no
|
||||
request: the deadline scheduler calls it before persisting a match that
|
||||
a timeout finished, binding the context to its own long-lived task.
|
||||
|
||||
It deliberately avoids the global-fallback singleton
|
||||
(``_enable_global_fallback``), which Tortoise only allows to be set once
|
||||
@@ -115,6 +118,17 @@ class TortoiseMixin(KayaMixin):
|
||||
log.info("sqlite schemas generated")
|
||||
return ctx
|
||||
|
||||
async def ensure_context(self) -> None:
|
||||
"""Bind this mixin's context to the calling task.
|
||||
|
||||
Background work with no request of its own (the deadline
|
||||
scheduler's finished-match write) calls this before touching the
|
||||
database. It is cheap when the context is already built for the
|
||||
running loop and idempotent for callers that already have it
|
||||
bound (HTTP requests, WebSocket connections).
|
||||
"""
|
||||
await self._bind()
|
||||
|
||||
async def _bind(self) -> None:
|
||||
loop = get_running_loop()
|
||||
if self._init_loop is not loop:
|
||||
|
||||
@@ -248,7 +248,12 @@ def make_platform(
|
||||
spec_path="/api/openapi.json",
|
||||
docs_path="/api/docs",
|
||||
)
|
||||
scheduler = DeadlineScheduler(game_store, registry, heartbeat_ms=50)
|
||||
scheduler = DeadlineScheduler(
|
||||
game_store,
|
||||
registry,
|
||||
heartbeat_ms=50,
|
||||
context_binder=tortoise_mixin.ensure_context,
|
||||
)
|
||||
platform = Platform(
|
||||
registry=registry,
|
||||
game_store=game_store,
|
||||
|
||||
@@ -75,6 +75,28 @@ class ConnectionIndependenceTest(unittest.TestCase):
|
||||
await platform.scheduler.sync_deadline(session)
|
||||
self.assertIsNone(await platform.game_store.next_deadline())
|
||||
|
||||
@async_test
|
||||
async def test_tick_finishing_match_persists_result_with_no_connections(self) -> None:
|
||||
# A deadline that ends the match must persist the result from the
|
||||
# consumer task, which has no request context of its own.
|
||||
from tavolo.platform.models import Match
|
||||
|
||||
_, platform, tortoise_mixin = make_platform()
|
||||
store = platform.game_store
|
||||
scheduler = platform.scheduler
|
||||
session = _started_session(target=1, deadline_in_seconds=0.05)
|
||||
await store.save(session)
|
||||
await scheduler.sync_deadline(session)
|
||||
|
||||
# Nobody ever connects: the consumer must finish the match and
|
||||
# write it to Postgres.
|
||||
result = await _wait_for(lambda: _finished(store, session.id))
|
||||
self.assertIsNotNone(result, "deadline never finished the match")
|
||||
|
||||
ctx = await use_db(tortoise_mixin)
|
||||
with ctx:
|
||||
self.assertEqual(1, await Match.all().count())
|
||||
|
||||
|
||||
async def _plays_is(store, game_id: str, count: int) -> Optional[GameSession]:
|
||||
session = await store.load(game_id)
|
||||
@@ -83,6 +105,13 @@ async def _plays_is(store, game_id: str, count: int) -> Optional[GameSession]:
|
||||
return None
|
||||
|
||||
|
||||
async def _finished(store, game_id: str) -> Optional[GameSession]:
|
||||
session = await store.load(game_id)
|
||||
if session is not None and session.state["finished"]:
|
||||
return session
|
||||
return None
|
||||
|
||||
|
||||
class ProcessDueTest(unittest.TestCase):
|
||||
"""Direct ``process_due`` behaviour: engine revalidation and idempotency."""
|
||||
|
||||
@@ -163,10 +192,11 @@ class ProcessDueTest(unittest.TestCase):
|
||||
@async_test
|
||||
async def test_finished_match_is_persisted_on_tick(self) -> None:
|
||||
# A tick that completes the match writes the result to Postgres.
|
||||
# No context is bound in this task: the scheduler must bind its
|
||||
# own, exactly like its consumer task at app startup.
|
||||
from tavolo.platform.models import Match
|
||||
|
||||
_, platform, tortoise_mixin = make_platform()
|
||||
ctx = await use_db(tortoise_mixin)
|
||||
scheduler = platform.scheduler
|
||||
store = platform.game_store
|
||||
session = _started_session(target=1, deadline_in_seconds=3600)
|
||||
@@ -178,9 +208,10 @@ class ProcessDueTest(unittest.TestCase):
|
||||
"kind": deadline.kind,
|
||||
"token": deadline.token,
|
||||
})
|
||||
await scheduler.process_due(member)
|
||||
ctx = await use_db(tortoise_mixin)
|
||||
with ctx:
|
||||
await scheduler.process_due(member)
|
||||
self.assertEqual(1, await Match.all().count())
|
||||
self.assertEqual(1, await Match.all().count())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -12,10 +12,11 @@ import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
from tavolo.app import game_store, platform, scheduler
|
||||
from tavolo.app import game_store, platform, scheduler, tortoise_mixin
|
||||
from tavolo.platform import GameSession, Seat
|
||||
from tavolo.platform.deadlines import encode
|
||||
from tavolo.scopone.state import PlayerState, ScoponeState
|
||||
from tavolo.platform.models import Match
|
||||
from tavolo.scopone.state import Card, PlayerState, ScoponeState
|
||||
from tests.helpers import async_test
|
||||
|
||||
PLAYERS = ("alice", "bob", "carol", "dave")
|
||||
@@ -124,6 +125,41 @@ class ConnectionIndependenceTest(unittest.TestCase):
|
||||
self.assertEqual(2, result.state.hand_number)
|
||||
self.assertEqual([], result.state.acked)
|
||||
|
||||
@async_test
|
||||
async def test_turn_timeout_finishing_match_persists_result(self) -> None:
|
||||
# The match-deciding last turn auto-played by the deadline consumer
|
||||
# runs in a task with no request context: the result must still
|
||||
# reach Postgres. Previously the write raised and the deadline
|
||||
# retried forever, leaving the game stuck on the last turn.
|
||||
session = _started_session("dl-final-1", "DLF001", turn_timeout=1)
|
||||
state = session.state
|
||||
state.players[0].hand = [Card.parse("02D")]
|
||||
for player in state.players[1:]:
|
||||
player.hand = []
|
||||
state.table = [Card.parse("02C")]
|
||||
state.turn = 0
|
||||
state.target_score = 1
|
||||
state.turn_deadline = (
|
||||
datetime.now(timezone.utc) + timedelta(seconds=1)
|
||||
).isoformat()
|
||||
|
||||
await game_store.save(session)
|
||||
await scheduler.sync_deadline(session)
|
||||
|
||||
# Nobody plays: the consumer auto-plays the last card, which ends
|
||||
# the hand and the match.
|
||||
result = await _wait_for(lambda: _phase_is(session.id, "finished"))
|
||||
self.assertIsNotNone(result, "turn deadline never finished the match")
|
||||
assert result is not None
|
||||
self.assertEqual(0, result.state.winner)
|
||||
self.assertTrue(result.stats_saved)
|
||||
|
||||
await tortoise_mixin.ensure_context()
|
||||
ctx = tortoise_mixin._ctx
|
||||
assert ctx is not None
|
||||
with ctx:
|
||||
self.assertEqual(1, await Match.all().count())
|
||||
|
||||
|
||||
async def _turn_is(game_id: str, turn: int) -> Optional[GameSession]:
|
||||
session = await game_store.load(game_id)
|
||||
|
||||
Reference in New Issue
Block a user