Persist matches finished by deadline timeouts
CI / Build and push docker image (push) Successful in 1m15s
CI / Build and push docker image (push) Successful in 1m15s
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:
@@ -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__":
|
||||
|
||||
Reference in New Issue
Block a user