diff --git a/DECISIONS.md b/DECISIONS.md index 12310bb..cde25eb 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -134,14 +134,19 @@ The event log is the source of truth; finished games replay from it. SQLAlchemy 2.0 async + Alembic, one models module for SQLite (dev) and Postgres (later). Redis is never the source of truth for game state. -## 13. Model-swap-aware scheduling, RESOLVED 2026-09-12 (flagged) +## 13. Model-swap-aware scheduling, RESOLVED 2026-09-12, built Phase 16 Loading a model costs seconds; on a 6 GB card at most two 4B-class models fit at once and an 8B model sits alone. The turn scheduler treats model identity as a resource: group consecutive AI turns by model where turn order allows, prefer tables that share a model, and surface swap counts in metrics. Research (`docs/local-model-setup.md` §2.3) confirms keep-alive -tuning does not remove the swap cost, it only delays unloads. +tuning does not remove the swap cost, it only delays unloads. Built as +`app.llm.scheduler.ModelAffinityScheduler` -- one instance per LLM +base_url, transparently wrapping every client `ModelRegistry.client_for` +hands out; see `docs/local-model-setup.md` §2.3 for the full design and +what remains a heuristic (it tracks in-flight calls, not Ollama's own +real VRAM residency). ## 14. Observability, RESOLVED 2026-09-12 diff --git a/PROGRESS.md b/PROGRESS.md index cd47e0b..cb4442c 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -43,9 +43,9 @@ branch; only the phase branch opens a PR into `master`. | 11 | Chat and personas | done 2026-09-14 (PR #21) | | 12 | Leaderboard | done 2026-09-14 (PR #22) | | 13 | Observability | done 2026-09-14 (PR #23) | -| 14 | CI/CD (local dry-run; no server yet) | done 2026-09-14 (branch phase-14-cicd-dry-run, PR pending) | +| 14 | CI/CD (local dry-run; no server yet) | done 2026-09-14 (PR #24) | | 15 | Production deployment (deferred until local complete) | | -| 16 | Scale and polish | | +| 16 | Scale and polish | done 2026-09-16 (branch phase-16-scale-and-polish, PR pending) | ## Phase 7 — done 2026-09-13 (branch phase-7-rooms) @@ -1300,6 +1300,171 @@ Actions security linter, against every edited workflow file); the deploy dry-run steps above verified directly, outside prek, against a real throwaway Postgres and the real Docker build. +## Phase 16 — done 2026-09-16 (branch phase-16-scale-and-polish) + +Phase 15 ("Production deployment") needs a real server/domain purchase -- +not something buildable unattended, so deferred by owner decision; moved +straight to Phase 16 ("Scale and polish"). Two pieces, both flagged and +scoped with the owner before building rather than guessed at: the +model-affinity scheduler `docs/local-model-setup.md` had already named as +Phase 16's job (DECISIONS §13), and -- found during the "polish pass" +survey of every phase's own "Left for later" notes -- the join queue's +seat/room wiring, which `docs/rooms.md` and `docs/capacity-and-queueing.md` +had both explicitly documented as unbuilt since Phase 7. The join queue's +real scope turned out larger than first estimated (new response shapes, an +accept-offer flow, promotion logic, real frontend UI, not backend-only like +the scheduler) -- re-scoped with the owner mid-flight rather than +silently expanding; the owner chose the full build, backend and frontend +both. + +### Model-affinity scheduler (DECISIONS §13) + +- `app/llm/scheduler.py` (new): `ModelAffinityScheduler` -- one instance per + LLM `base_url`, bounded admission control. `acquire(model)` admits + immediately if that model already has an in-flight call or the active + set is under `LLM_MAX_LOADED_MODELS` (default 2); otherwise waits up to + `LLM_SCHEDULER_MAX_WAIT_SECONDS` (default 5.0) before force-admitting + anyway, logged. `release(model)` decrements and wakes one waiter. + Deliberately a heuristic, not a real VRAM model: "active" means "has an + in-flight call," not "Ollama actually has it loaded" (`/api/ps` is never + queried) -- documented as such in `docs/local-model-setup.md`'s new + §2.3.1, with `botpit_model_swaps_total` (Phase 13) named as the way to + empirically check its real-world effect later. +- `app/llm/registry.py`: `client_for()` now wraps every adapter in a new + `_ScheduledClient` (routes `complete()` through + `acquire`/`note_llm_call`/`release`) before caching it, one scheduler per + `base_url` via `self._schedulers`. `note_llm_call` (Phase 13's swap + counter) moved here from `AgentController.decide()`, so the metric + reflects real post-scheduling dispatch order rather than raw arrival + order. +- `DECISIONS.md` §13: `"RESOLVED ... (flagged)"` -> `"RESOLVED ..., + built Phase 16"`, naming the implementation. +- Tests: `tests/unit/llm/test_scheduler.py` (6, admission/wait/force-admit/ + release scenarios matching DECISIONS §13's own worked example) and + `tests/unit/llm/test_scheduled_client.py` (2, wrapper-level: a cold model + really does wait for a slot, `note_llm_call` fires at real dispatch + time). `tests/unit/llm/test_registry.py` updated for the `.inner` + indirection and per-`base_url` scheduler sharing. + +### Join queue, fully wired (docs/rooms.md, docs/capacity-and-queueing.md) + +Backend, both scopes: + +- **Room-seat scope** (`f"room:{id}"`): `RoomService.join_by_code` enqueues + instead of hard-refusing when a room is full (`RoomFullError`, 409, now + carrying `room_id` in the body too -- a caller who only had the join + code had no other way to know which room to keep polling). + `RoomService.leave` (now `async`) offers the freed seat to whoever's + next via `_offer_seat_to_next_queued`, which marks the seat `"offered"` + -- a new `RoomSeat.status` value (no migration; already an unconstrained + `str`) excluded from `_claim_open_seat`'s "any open seat" fallback so a + fresh joiner can't steal a seat that was actually offered to someone + specific. `_reclaim_expired_offers` sweeps and re-offers any offer whose + deadline passed unaccepted; `join_by_code` re-checks `queue.accept()` + immediately after a reclaim runs, since the very common case is the + offeree's own retry landing right after their own offer just expired and + got re-offered to them again -- a self-healing retry, not a second + failure. +- **Global scope** (`"global"`, DECISIONS §2's concurrent-game cap): + `RoomService.start` replaces the old hard-refusing capacity check with + queue-aware logic -- enqueues unless already at position 1, in which + case it pops itself off and proceeds; `create()`'s own capacity check is + deliberately unchanged (still hard-refuses -- a room can always be + created, only *starting* it queues). `_finish_room` tries to promote the + next globally-queued room once a slot frees (`_promote_next_queued_room`, + best-effort, suppressing `RoomNotWaitingError`/`RoomFullError`/ + `CapacityExceededError` -- it only ever suggests a room; `start`'s own + advisory-locked check is the real gate against a stale suggestion). +- **`app/rooms/queue.py`**: `sweep()`'s return type changed from the + requeued user ids to the reclaimed *seat numbers* (the caller's real + question after a sweep); added `pop_next`/`peek_head` for the global + scope's plain-FIFO needs (no offer/accept/TTL semantics -- the caller + acts immediately). `QueueStatus` gained `offered`/ + `offer_expires_in_seconds`, computed non-destructively in `snapshot()` + from the `offers` ZSET deadline (not the offer key's own PTTL -- see the + bug below) so a polling caller can tell "still waiting" apart from "your + turn came up, go accept." +- **`app/api/routes/rooms.py`**: `GET /{room_id}/queue` gained a + `scope=seat|global` query param (seat: is *this user* waiting for a + seat; global: is *this room* waiting for a slot) and a + `response_model=QueueStatus`; its 404 condition now also lets a live + offer through (previously any user with no active *queue* position + 404'd, even one sitting on a live offer). It also sweeps this room's + seat-scope queue on every poll, opportunistically reclaiming any offer + whose deadline just passed -- see "no standalone sweeper task" below. + +Two real bugs, both caught by the tests written for this feature rather +than assumed correct: + +- **A genuine pre-existing race in `JoinQueue`, never caught before + because no test exercised a real wall-clock TTL.** `_OFFER_NEXT`'s Redis + string key (`offer:{scope}:{user}`) had its own `SET NX PX` expiry set + to the *exact same* deadline the `offers` ZSET tracks for `sweep()`'s + explicit reclaim -- under a real wait (as opposed to every existing + `JoinQueue` test's synthetic `now_ms` override), Redis's own automatic + key expiry usually fires before or alongside `sweep()`'s own + `GETDEL`, which then finds nothing and silently assumes some other + worker already claimed it -- never reclaiming the seat. Fixed by giving + the key's own `PX` 4x headroom over the ZSET deadline: it's now just a + leak backstop for a crashed process, and `sweep()`'s explicit reclaim + always wins the race it's supposed to win. +- **`RoomService.start` left its Postgres advisory lock + (`pg_advisory_xact_lock`) held after raising `CapacityExceededError` or + the post-lock `RoomNotWaitingError`** -- xact-scoped locks only release + on commit/rollback, and neither exception path did either, so the lock + stayed held until whatever session owned it eventually closed. Harmless + within one short-lived request, but `_promote_next_queued_room` opens + its *own* session from inside `_finish_room` -- a caller that catches + the first exception and keeps working on the same session (exactly what + `test_finish_room_promotes_the_next_globally_queued_room` does, and + what triggered a genuine ~90s hang under a real `timeout` guard, not + just a slow test) deadlocks that second session against the lock the + first never released. Fixed with an explicit `rollback()` on both + exception paths, immediately after the check that decided to raise -- + nothing written before that point needed keeping (reads and the lock + call only). + +Frontend: + +- `components/Rooms/JoinByCode.tsx`: a 409-with-`room_id` response is + shown inline (position, length, ETA) instead of toasted as an error; + polls `GET /{room_id}/queue?scope=seat` every 3s while queued. A live + offer (`offered: true`) shows "Seat ready" with an "Accept seat" button + that just retries the same join call -- `join_by_code`'s `accept()` path + handles it. A 404 (dropped after too many missed offers) clears the + queued state. +- `routes/_layout/rooms.$roomId.tsx`: a 429 on Start shows "Waiting for a + game slot" with position/ETA (polling `scope=global`) instead of an + error toast; clears once the room's own status flips away from + `"waiting"` over the existing websocket -- that's the real signal the + wait ended, not a client-side guess. Also added a "Leave seat" button -- + there was no way for a human to leave a seat from the UI at all before + this, which meant the accept-offer flow had nothing to demonstrate it + end-to-end; a real, if small, product gap found while building this. +- `frontend/tests/rooms.spec.ts`: new E2E test, three real browser + contexts -- two fill a room, a third gets queued (position/ETA visible), + the creator leaves, the third's page shows "Seat ready" and accepts into + the freed seat. Run against the real Docker stack per this project's + standing gate; all 60 E2E tests pass (the other 59 unaffected). + +**Left for later, explicitly (not silently):** no standalone +leader-elected sweeper task, and no `seat_offer`/`queue_update` push +events -- both were Phase 7's original documented plan for this feature. +Built instead: opportunistic reclaim (a `join_by_code` retry sweeps as a +side effect; so does every `GET /queue` poll while a waiting caller's own +queue UI is open) and REST polling instead of a push transport. Covers +every realistic path -- the offeree is still watching, or someone else +eventually retries the room -- but a room whose only waiter closes their +tab, where nobody else ever touches that room's queue again, leaves that +one offer stale until something does. `docs/capacity-and-queueing.md` +documents this bound explicitly rather than claiming the original design +was built as originally planned. + +Full gate: backend suite (`uv run pytest tests`, all green, including new +tests for the scheduler, the queue wiring, and the two bugs above), +`prek run --all-files` clean, frontend `bun run build`/`lint` clean, full +Playwright E2E suite (60/60) against the real Docker stack. + ## Phase 9.7 — done 2026-09-14 (branch phase-9.7-infra-optimization) Infrastructure-only pass: Docker image build, Compose, CI/CD workflows, diff --git a/backend/app/agents/controller.py b/backend/app/agents/controller.py index a9ad086..2882f83 100644 --- a/backend/app/agents/controller.py +++ b/backend/app/agents/controller.py @@ -22,11 +22,7 @@ ) from app.games.base import AnyGame, PlayerView, Seat from app.llm.client import CompletionRequest, LLMClient, LLMError, Message, with_retries -from app.metrics import ( - AGENT_TURN_ERRORS_TOTAL, - AGENT_TURN_LATENCY_SECONDS, - note_llm_call, -) +from app.metrics import AGENT_TURN_ERRORS_TOTAL, AGENT_TURN_LATENCY_SECONDS from app.referee.seats import ControllerError, SeatKind, TurnContext from app.telemetry import current_event, get_logger @@ -99,7 +95,6 @@ async def decide(self, view: PlayerView, ctx: TurnContext) -> dict[str, Any]: ) call_start = time.perf_counter() try: - note_llm_call(self.client.base_url, self.model) result = await with_retries( functools.partial(self.client.complete, req) ) diff --git a/backend/app/api/routes/rooms.py b/backend/app/api/routes/rooms.py index 0c3b138..475e4f3 100644 --- a/backend/app/api/routes/rooms.py +++ b/backend/app/api/routes/rooms.py @@ -55,7 +55,7 @@ ) from app.ratelimit import check_rate, rate_limit_from_setting from app.rooms.guard import resolve_subscriber, subscription_guard -from app.rooms.queue import JoinQueue +from app.rooms.queue import JoinQueue, QueueStatus from app.rooms.service import ( NotSeatedError, NotYourTurnError, @@ -211,9 +211,14 @@ async def join_room( except RoomNotWaitingError as exc: raise HTTPException(status_code=409, detail=str(exc)) except RoomFullError as exc: - status = await queue.snapshot(f"room:{exc.room_id}", str(user.id)) + queue_status = await queue.snapshot(f"room:{exc.room_id}", str(user.id)) return JSONResponse( - {"detail": str(exc), "queue": status.model_dump()}, status_code=409 + { + "detail": str(exc), + "room_id": str(exc.room_id), + "queue": queue_status.model_dump(), + }, + status_code=409, ) return RoomSeatJoinPublic( **_seat_public(seat, reveal=True).model_dump(), room_id=seat.room_id @@ -221,11 +226,11 @@ async def join_room( @router.post("/{room_id}/leave", response_model=Message) -def leave_room( +async def leave_room( *, service: RoomServiceDep, user: CurrentUser, room_id: uuid.UUID ) -> Any: try: - service.leave(user, room_id) + await service.leave(user, room_id) except LookupError: raise HTTPException(status_code=404, detail="Room not found") return Message(message="left") @@ -297,14 +302,41 @@ async def stop_room( return _room_public(room, service.seats(room_id)) -@router.get("/{room_id}/queue") +@router.get("/{room_id}/queue", response_model=QueueStatus) async def room_queue( - *, queue: JoinQueueDep, user: CurrentUser, room_id: uuid.UUID + *, + service: RoomServiceDep, + queue: JoinQueueDep, + user: CurrentUser, + room_id: uuid.UUID, + scope: Literal["seat", "global"] = "seat", ) -> Any: - status = await queue.snapshot(f"room:{room_id}", str(user.id)) - if status.position is None: + # "seat" (default): is *this user* waiting for (or offered) a seat in + # this room. "global" (DECISIONS.md §2's concurrent-game cap): is *this + # room itself* waiting for a slot to start in -- same JoinQueue, same + # QueueStatus shape, just keyed by room id instead of user id. + # + # There is no standalone sweeper task (docs/capacity-and-queueing.md's + # "Left for later" -- reclaim happens opportunistically instead): a + # `join_by_code` retry sweeps stale offers as a side effect, and so does + # every poll of this route while the waiting party's own queue UI is + # open. A seat offer with nobody polling and nobody else ever retrying + # that exact room would stay stale until one of those two things + # happens -- a real, bounded gap, not a silent one. + if scope == "seat": + await service._reclaim_expired_offers(room_id) + key, member = ( + ("global", str(room_id)) + if scope == "global" + else ( + f"room:{room_id}", + str(user.id), + ) + ) + queue_status = await queue.snapshot(key, member) + if queue_status.position is None and not queue_status.offered: raise HTTPException(status_code=404, detail="not queued") - return status + return queue_status # -- live play (Phase 8: the minimal API a human plays through) ------------- diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 7bdc8a3..04dacc0 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -94,6 +94,16 @@ def emails_enabled(self) -> bool: # Off by default so CI, tests and a clone without Ollama still serve; rooms # refuse to seat an AI player while /health/llm is degraded (Phase 7). LLM_HEALTH_REQUIRED_AT_STARTUP: bool = False + # Model-affinity turn scheduling (Phase 16, DECISIONS.md §13): how many + # distinct models the backend behind one base_url can hold loaded at + # once without evicting -- matches `OLLAMA_MAX_LOADED_MODELS` on the + # Ollama server side (docs/local-model-setup.md §2.3's recommended + # value is 2 for a 6 GB card). A request for a model outside that set + # waits up to LLM_SCHEDULER_MAX_WAIT_SECONDS for a slot to free up + # before being admitted anyway -- bounded, so a turn's own deadline is + # never blocked forever by scheduling. + LLM_MAX_LOADED_MODELS: int = 2 + LLM_SCHEDULER_MAX_WAIT_SECONDS: float = 5.0 @field_validator("LLM_MODELS", mode="before") @classmethod diff --git a/backend/app/llm/registry.py b/backend/app/llm/registry.py index 8e9bd6f..460c138 100644 --- a/backend/app/llm/registry.py +++ b/backend/app/llm/registry.py @@ -7,9 +7,11 @@ import httpx from pydantic import field_validator -from app.llm.client import Frozen, LLMClient +from app.llm.client import CompletionRequest, CompletionResult, Frozen, LLMClient from app.llm.ollama import OllamaNativeAdapter from app.llm.openai_compat import OpenAICompatAdapter +from app.llm.scheduler import ModelAffinityScheduler +from app.metrics import note_llm_call if TYPE_CHECKING: from collections.abc import Iterable @@ -31,11 +33,64 @@ def _strip_slash(cls, value: str) -> str: return value.rstrip("/") +class _ScheduledClient: + """Wraps a real ``LLMClient`` with model-affinity admission control + (Phase 16, DECISIONS.md §13, ``app.llm.scheduler``) -- transparent to + every caller. ``AgentController`` and every existing test that + constructs a controller directly with a raw client (``FakeLLMClient`` + included) never need to know this exists; only clients handed out by + ``ModelRegistry.client_for`` go through it. + + Also where ``note_llm_call`` (the swap counter, Phase 13) actually + fires -- at real dispatch time, after the scheduler admits the call, + not when a caller merely *wants* to call. Firing it any earlier would + count a "swap" for every model-alternating request the scheduler + exists to reorder around, silently making the metric blind to whether + the scheduler is actually working. + + ``inner`` (the wrapped adapter) is a public attribute, not private -- + tests that need to reach past the wrapper (``isinstance(x.inner, + OllamaNativeAdapter)``, checking which shared ``httpx.AsyncClient`` an + adapter holds) are legitimate, so ``__getattr__`` also transparently + forwards any attribute this class doesn't define itself (``.client``, + ``.num_ctx``, ``.keep_alive``, ...) straight to ``inner``. + """ + + def __init__(self, inner: LLMClient, scheduler: ModelAffinityScheduler) -> None: + self.inner = inner + self._scheduler = scheduler + self.base_url = inner.base_url + + def __getattr__(self, name: str) -> Any: + return getattr(self.inner, name) + + async def complete(self, req: CompletionRequest) -> CompletionResult: + await self._scheduler.acquire(req.model) + note_llm_call(self.base_url, req.model) + try: + return await self.inner.complete(req) + finally: + await self._scheduler.release(req.model) + + async def aclose(self) -> None: + await self.inner.aclose() + + class ModelRegistry: - def __init__(self, configs: Iterable[ModelConfig], default_model: str) -> None: + def __init__( + self, + configs: Iterable[ModelConfig], + default_model: str, + *, + max_loaded_models: int = 2, + scheduler_max_wait_s: float = 5.0, + ) -> None: self._configs = {c.model: c for c in configs} self.default_model = default_model + self._max_loaded_models = max_loaded_models + self._scheduler_max_wait_s = scheduler_max_wait_s self._http: dict[str, httpx.AsyncClient] = {} + self._schedulers: dict[str, ModelAffinityScheduler] = {} self._clients: dict[str, LLMClient] = {} @classmethod @@ -52,7 +107,12 @@ def from_settings(cls, settings: Settings) -> ModelRegistry: **settings.LLM_MODEL_OVERRIDES.get(name, {}), } configs.append(ModelConfig(**fields)) - return cls(configs, settings.LLM_DEFAULT_MODEL) + return cls( + configs, + settings.LLM_DEFAULT_MODEL, + max_loaded_models=settings.LLM_MAX_LOADED_MODELS, + scheduler_max_wait_s=settings.LLM_SCHEDULER_MAX_WAIT_SECONDS, + ) @property def models(self) -> list[str]: @@ -71,7 +131,13 @@ def client_for(self, model: str) -> LLMClient: if model not in self._clients: cfg = self.config_for(model) http = self._http.setdefault(cfg.base_url, httpx.AsyncClient()) - self._clients[model] = ( + scheduler = self._schedulers.setdefault( + cfg.base_url, + ModelAffinityScheduler( + self._max_loaded_models, max_wait_s=self._scheduler_max_wait_s + ), + ) + inner: LLMClient = ( OllamaNativeAdapter( cfg.base_url, http, @@ -82,6 +148,7 @@ def client_for(self, model: str) -> LLMClient: if cfg.provider == "ollama_native" else OpenAICompatAdapter(cfg.base_url, http, cfg.timeout_s) ) + self._clients[model] = _ScheduledClient(inner, scheduler) return self._clients[model] async def aclose(self) -> None: diff --git a/backend/app/llm/scheduler.py b/backend/app/llm/scheduler.py new file mode 100644 index 0000000..8b8a942 --- /dev/null +++ b/backend/app/llm/scheduler.py @@ -0,0 +1,121 @@ +"""Model-affinity admission control (Phase 16, DECISIONS.md §13): "the turn +scheduler treats model identity as a resource: group consecutive AI turns +by model where turn order allows, prefer tables that share a model, and +surface swap counts in metrics." + +One instance per LLM base_url (a completely separate backend/VRAM budget +from any other base_url). Loading a model costs real seconds +(docs/local-model-setup.md §2.3); keep-alive tuning only delays an +eviction, it never removes the swap cost. This scheduler doesn't decide +turn *order* within one game -- that's the referee's job, fixed by each +game's own rules -- it decides *when a turn's already-due LLM call is +allowed to actually fire*, across every room sharing one backend, so a +request for a model with calls already in flight gets serviced ahead of +one that would grow the active set past capacity, wherever the referee +level leaves that choice open. That's what "prefer tables that share a +model" means at the layer this code can actually act on. + +Bounded, not a hard priority queue: a request for a model outside the +active set waits at most ``max_wait_s`` for room to open before being +admitted anyway. A turn's own deadline (``SeatSpec.timeout_s``) must never +be blocked forever by scheduling -- a liveness guarantee, not just an +optimization. + +Heuristic, not a perfect model of Ollama's own VRAM state: this tracks +"models with at least one in-flight call," not Ollama's actual keep-alive +residency (a model can still be resident with zero in-flight calls, for +up to ``LLM_KEEP_ALIVE``). Querying Ollama's own `/api/ps` for ground +truth would be more accurate but costs a network round trip per +admission decision; not built here -- this is a bias towards fewer swaps +under real concurrent load, not a guarantee. +""" + +from __future__ import annotations + +import asyncio +from collections import defaultdict + +from app.telemetry import get_logger + +log = get_logger("llm.scheduler") + + +class ModelAffinityScheduler: + def __init__(self, max_loaded: int, *, max_wait_s: float = 5.0) -> None: + self._max_loaded = max(1, max_loaded) + self._max_wait_s = max_wait_s + self._in_flight: dict[str, int] = {} + self._waiters: dict[str, list[asyncio.Event]] = defaultdict(list) + self._lock = asyncio.Lock() + + async def acquire(self, model: str) -> None: + """Blocks (briefly, boundedly) until ``model`` may make its call.""" + loop = asyncio.get_running_loop() + deadline: float | None = None + while True: + async with self._lock: + if model in self._in_flight or len(self._in_flight) < self._max_loaded: + self._in_flight[model] = self._in_flight.get(model, 0) + 1 + return + event = asyncio.Event() + self._waiters[model].append(event) + + if deadline is None: + deadline = loop.time() + self._max_wait_s + remaining = deadline - loop.time() + if remaining <= 0: + async with self._lock: + self._forget_waiter(model, event) + self._force_admit(model) + return + + try: + await asyncio.wait_for(event.wait(), timeout=remaining) + except TimeoutError: + async with self._lock: + self._forget_waiter(model, event) + self._force_admit(model) + return + # Woken because a slot actually freed (release() only wakes on + # that) -- loop back and re-check under the lock rather than + # blindly admitting: another waiter for a *different* model + # could already have taken it in the meantime. + async with self._lock: + self._forget_waiter(model, event) + + def _forget_waiter(self, model: str, event: asyncio.Event) -> None: + waiters = self._waiters.get(model) + if waiters and event in waiters: + waiters.remove(event) + + def _force_admit(self, model: str) -> None: + self._in_flight[model] = self._in_flight.get(model, 0) + 1 + log.info( + "model affinity scheduler: forced admit for %r after " + "%.1fs wait -- the active model set stayed full", + model, + self._max_wait_s, + ) + + async def release(self, model: str) -> None: + async with self._lock: + count = self._in_flight.get(model, 0) - 1 + if count > 0: + self._in_flight[model] = count + return + # The model dropped out of the active set entirely -- a slot is + # now actually free, unlike a call to a model that still has + # other in-flight calls (count > 0 above): waking a waiter then + # would let it self-admit into a set that never shrank, pushing + # the active set past max_loaded. Wake exactly one waiter (of + # any model waiting), not every one, to avoid a thundering herd + # all re-checking for the one slot that opened. + self._in_flight.pop(model, None) + for waiters in self._waiters.values(): + if waiters: + waiters.pop(0).set() + break + + def snapshot(self) -> dict[str, int]: + """Read-only view of in-flight counts per model, for tests/introspection.""" + return dict(self._in_flight) diff --git a/backend/app/rooms/queue.py b/backend/app/rooms/queue.py index 6a31377..abe8da8 100644 --- a/backend/app/rooms/queue.py +++ b/backend/app/rooms/queue.py @@ -27,12 +27,23 @@ # KEYS = queue, offers, scores; ARGV = now_ms, ttl_ms, seat, offer key prefix. # Returns {user, deadline_ms} or nil when the queue is empty. +# +# The offer string key's own PX gets 4x headroom over the ZSET deadline -- +# sweep() is what's supposed to reclaim an expired offer (ZRANGEBYSCORE then +# GETDEL), and it must always find the key still there to GETDEL. Giving the +# key the *same* TTL as the ZSET deadline made Redis's own automatic expiry +# race sweep()'s explicit reclaim for the same millisecond -- when Redis won, +# GETDEL found nothing and sweep() wrongly assumed some other worker already +# claimed it, silently dropping the seat instead of reclaiming it. The PX +# here is just a leak backstop (a crashed process's key still eventually +# vanishes); sweep() is the real reclaim path and must win the race. _OFFER_NEXT = """ local popped = redis.call('ZPOPMIN', KEYS[1]) if #popped == 0 then return nil end local user = popped[1] -local deadline = tonumber(ARGV[1]) + tonumber(ARGV[2]) -redis.call('SET', ARGV[4] .. user, ARGV[3], 'NX', 'PX', ARGV[2]) +local ttl = tonumber(ARGV[2]) +local deadline = tonumber(ARGV[1]) + ttl +redis.call('SET', ARGV[4] .. user, ARGV[3], 'NX', 'PX', ttl * 4) redis.call('ZADD', KEYS[2], deadline, user) redis.call('HSET', KEYS[3], user, popped[2]) return {user, deadline} @@ -60,6 +71,8 @@ class QueueStatus(BaseModel): position: int | None length: int eta_seconds: int | None + offered: bool = False + offer_expires_in_seconds: int | None = None class JoinQueue: @@ -132,11 +145,18 @@ async def accept(self, scope: str, user_id: str) -> int | None: log.info("queue accept key=offer:%s:%s seat=%s", scope, user_id, seat) return int(seat) - async def sweep(self, scope: str, now_ms: int | None = None) -> list[str]: - """Requeue users whose offer expired (at their original score); drop repeat no-shows.""" + async def sweep(self, scope: str, now_ms: int | None = None) -> list[int]: + """Reclaims every seat held by an expired offer -- requeuing that + user at their original score, or dropping them after + ``max_misses``. Returns the *seats* reclaimed (one per expired + offer, regardless of whether that offer's user was requeued or + dropped): the caller's real question after a sweep is "which + seats are free again," not who specifically missed one -- fairness + bookkeeping for the *user* side is already fully handled here and + needs no further action from the caller.""" now = _now_ms() if now_ms is None else now_ms offers = f"offers:{scope}" - requeued: list[str] = [] + reclaimed: list[int] = [] expired = cast( "list[str]", await self._redis.zrangebyscore(offers, "-inf", now) ) @@ -146,6 +166,7 @@ async def sweep(self, scope: str, now_ms: int | None = None) -> list[str]: ) if seat is None: continue # accepted (or already swept by another worker) meanwhile + reclaimed.append(int(seat)) score = await self._redis.hget(f"scores:{scope}", user_id) misses_key = f"misses:{scope}:{user_id}" misses = int(await self._redis.incrby(misses_key, 1)) @@ -159,11 +180,27 @@ async def sweep(self, scope: str, now_ms: int | None = None) -> list[str]: await self._redis.zadd( f"queue:{scope}", {user_id: float(score) if score else now}, nx=True ) - requeued.append(user_id) log.info( "queue requeue key=queue:%s user=%s misses=%s", scope, user_id, misses ) - return requeued + return reclaimed + + async def pop_next(self, scope: str) -> str | None: + """Plain FIFO pop, no offer/TTL/accept semantics -- for a scope + where nobody needs to explicitly accept (the caller acts + immediately on the popped id, e.g. a room auto-starting once its + turn comes up; DECISIONS.md §2's global concurrent-game cap).""" + popped = await self._redis.zpopmin(f"queue:{scope}") + if not popped: + return None + member, _score = cast("tuple[str, float]", popped[0]) + log.info("queue pop key=queue:%s user=%s", scope, member) + return str(member) + + async def peek_head(self, scope: str) -> str | None: + """Read the head of the queue without removing it.""" + head = cast("list[str]", await self._redis.zrange(f"queue:{scope}", 0, 0)) + return head[0] if head else None @staticmethod def eta_seconds( @@ -182,12 +219,31 @@ async def snapshot( self, scope: str, user_id: str, oldest_started_at: datetime | None = None ) -> QueueStatus: position = await self.position(scope, user_id) + offered = False + offer_expires_in_seconds = None + if position is None: + # Once offered, a user is popped out of `queue:{scope}` (so + # `position` above is already None) into a live offer -- the + # caller (a polling frontend) still needs to know "your turn + # came up, go accept" rather than reading plain silence as + # "still waiting" or "gave up." Both reads are non-destructive, + # unlike `accept()`/`sweep()`'s GETDEL. The deadline comes from + # the `offers` ZSET, not the offer key's own PTTL -- that key's + # PX is deliberately 4x the real deadline (see _OFFER_NEXT), so + # its PTTL alone would overstate how long the offer is actually + # good for. + deadline = await self._redis.zscore(f"offers:{scope}", user_id) + if deadline is not None: + offered = True + offer_expires_in_seconds = max(0, int((deadline - _now_ms()) / 1000)) return QueueStatus( position=position, length=await self.length(scope), eta_seconds=None if position is None else self.eta_seconds(position, oldest_started_at), + offered=offered, + offer_expires_in_seconds=offer_expires_in_seconds, ) async def _forget(self, scope: str, user_id: str) -> None: diff --git a/backend/app/rooms/service.py b/backend/app/rooms/service.py index 224f5ad..537ca5b 100644 --- a/backend/app/rooms/service.py +++ b/backend/app/rooms/service.py @@ -358,18 +358,46 @@ async def join_by_code(self, user: User, code: str) -> RoomSeat: if already is not None: return already + scope = f"room:{room.id}" + offered_seat = await self.queue.accept(scope, str(user.id)) + if offered_seat is None: + # A sweep here: an unresponsive offeree from this room's own + # queue must never permanently lock a seat `"offered"` -- any + # reclaimed seat goes back to `"open"` and is immediately + # re-offered to whoever's now at the head. That "whoever" is + # very often *this exact caller* (the common case: they missed + # their own offer's deadline and are retrying right after) -- + # so accept() must be re-checked, not just called once before + # the sweep ever had a chance to create a fresh offer for them. + await self._reclaim_expired_offers(room.id) + offered_seat = await self.queue.accept(scope, str(user.id)) + if offered_seat is not None: + target = self._fill_offered_seat(room.id, offered_seat, user.id) + if target is not None: + await self._publish_seat_filled(room.id, target, user) + return target + # The queue said this offer was live, but the seat itself + # wasn't (still) `"offered"` in the DB -- shouldn't normally + # happen since both sides move in lockstep, but fall through + # to a fresh join attempt rather than leaving the caller + # stuck with an offer Redis already consumed. + target = self._claim_open_seat(room.id, user.id) - if target is None: - raise RoomFullError(f"room {room.id} has no open seat", room_id=room.id) + if target is not None: + await self._publish_seat_filled(room.id, target, user) + return target + + await self.queue.enqueue(scope, str(user.id)) + raise RoomFullError(f"room {room.id} has no open seat", room_id=room.id) + async def _publish_seat_filled( + self, room_id: uuid.UUID, seat: RoomSeat, user: User + ) -> None: await self.bus.publish( make_event( - str(room.id), - SeatFilled(seat=target.seat, player=str(user.id)), - PUBLIC, + str(room_id), SeatFilled(seat=seat.seat, player=str(user.id)), PUBLIC ) ) - return target def _claim_open_seat( self, room_id: uuid.UUID, user_id: uuid.UUID @@ -433,7 +461,67 @@ def _claim_open_seat( # lost the race for this seat (or it just changed underfoot); retry return None - def leave(self, user: User, room_id: uuid.UUID) -> None: + def _fill_offered_seat( + self, room_id: uuid.UUID, seat_number: int, user_id: uuid.UUID + ) -> RoomSeat | None: + """Claims exactly the seat a live queue offer names -- mirrors + `_claim_open_seat`'s atomic UPDATE-with-WHERE pattern, but keyed + by seat number and `status == "offered"` rather than "any open + seat"; deliberately ignores `invited_user_id` entirely, since a + queue offer and a room invite are two independent reservation + mechanisms that never need to look at each other's state. Only + ever called right after `JoinQueue.accept()` already proved this + caller holds the live offer for it.""" + result = self.session.execute( # ty: ignore[deprecated] + update(RoomSeat) + .where( + col(RoomSeat.room_id) == room_id, + col(RoomSeat.seat) == seat_number, + col(RoomSeat.status) == "offered", + ) + .values(status="filled", user_id=user_id, invited_user_id=None) + ) + self.session.commit() + rowcount = cast("CursorResult[Any]", result).rowcount + if rowcount != 1: + return None + return next(s for s in self.seats(room_id) if s.seat == seat_number) + + async def _vacate_seat(self, room_id: uuid.UUID, seat: RoomSeat) -> None: + """Frees ``seat`` and, in one DB write, sets whichever status + actually applies: `"offered"` if this room's queue had someone + waiting (`queue.offer_next` already popped them and created their + live offer), `"open"` if nobody was. Checking the queue *before* + writing any status -- rather than writing `"open"` first and only + afterward re-marking it `"offered"`, as an earlier version of this + did -- closes a real window where a concurrent, non-queued + `join_by_code` could win the seat via `_claim_open_seat`'s "any + open seat" fallback out from under the very queue this feature + exists to make fair. `JoinQueue`'s own offer/accept/TTL bookkeeping + stays the sole source of truth for whether a specific offer is + still live; this DB status only protects the seat from *other* + callers in the meantime.""" + offer = await self.queue.offer_next(f"room:{room_id}", seat.seat) + seat.status = "offered" if offer is not None else "open" + seat.user_id = None + self.session.add(seat) + self.session.commit() + + async def _reclaim_expired_offers(self, room_id: uuid.UUID) -> None: + """Sweeps this room's queue for offers that expired unaccepted: + each reclaimed seat is freed via `_vacate_seat`, which re-offers it + to whoever's now at the head of the line (if anyone is) -- a seat + with an unresponsive offeree must not sit `"offered"` forever, nor + be silently handed to a random fresh joiner ahead of the next + person actually waiting for it.""" + scope = f"room:{room_id}" + for seat_number in await self.queue.sweep(scope): + seat = next((s for s in self.seats(room_id) if s.seat == seat_number), None) + if seat is None or seat.status != "offered": + continue # already changed underfoot; nothing to reclaim + await self._vacate_seat(room_id, seat) + + async def leave(self, user: User, room_id: uuid.UUID) -> None: room = self.get(room_id) if room.status != "waiting": return @@ -447,24 +535,34 @@ def leave(self, user: User, room_id: uuid.UUID) -> None: ) if seat is None: return - seat.status = "open" - seat.user_id = None - self.session.add(seat) - self.session.commit() + await self._vacate_seat(room_id, seat) # -- start / stop ------------------------------------------------------- async def force_start(self, user: User, room_id: uuid.UUID) -> Room: del user # authorised by the caller (route); nothing more to check here - # Staged, not committed: if `start()` below raises (room full some - # other way, capacity exceeded, ...), these human seats must not be - # left permanently converted to agent seats with the room stuck - # "waiting". `start()`'s own `self.seats(room_id)` query autoflushes - # this session's pending changes first, so it still sees the backfill; - # everything commits together in `start()`'s single final commit, or - # none of it does if `start()` raises and the session later rolls back. + # Committed immediately, not staged alongside `start()`'s own final + # commit: `start()` can now (Phase 16) *enqueue and return* on the + # global capacity cap instead of only ever raising outright, and + # `_promote_next_queued_room` later retries that exact room from a + # completely different session. An uncommitted backfill living only + # in this session's pending changes would be invisible to that + # retry -- its own fresh read of the seats would still see the + # original open human seats, fail the same "not all filled" check, + # and get silently swallowed by that retry's own exception + # suppression, permanently orphaning the room in the global queue + # with no further signal to anyone. Committing here means a later + # capacity-queue enqueue (or any other `start()` failure) leaves the + # backfill in place rather than losing it -- a moderator/promotion + # retry then just needs one more `start()` call to succeed, instead + # of never being able to succeed at all. for seat in self.seats(room_id): - if seat.kind == "human" and seat.status == "open": + # `!= "filled"`, not `== "open"`: also converts a seat mid-offer + # (see `start()`'s matching comment) -- force-starting means + # "right now, regardless," so a pending offer that never got + # accepted is superseded, not left to strand `start()` right + # back into `RoomFullError` on the very next line. + if seat.kind == "human" and seat.status != "filled": seat.kind = "agent" seat.model = self.registry.default_model seat.persona = "steady" @@ -472,6 +570,7 @@ async def force_start(self, user: User, room_id: uuid.UUID) -> Room: seat.user_id = None seat.invited_user_id = None self.session.add(seat) + self.session.commit() return await self.start(room_id) async def start(self, room_id: uuid.UUID) -> Room: @@ -479,7 +578,12 @@ async def start(self, room_id: uuid.UUID) -> Room: if room.status != "waiting": raise RoomNotWaitingError(f"room {room_id} is {room.status}") seats = self.seats(room_id) - if any(s.status == "open" for s in seats): + # `!= "filled"`, not `== "open"`: a seat mid-offer (queued for a + # specific user, not yet accepted -- see `_offer_seat_to_next_queued`) + # is not `"open"` either, but is just as unfilled as one that is; + # starting the game with it in that state would silently strand + # whoever the offer was made to. + if any(s.status != "filled" for s in seats): raise RoomFullError( f"room {room_id} still has an open seat; force_start or fill it first" ) @@ -516,22 +620,49 @@ async def start(self, room_id: uuid.UUID) -> Room: # Serialises the check-capacity-then-flip-to-"in_progress" step across # every worker process (a Postgres advisory lock, not a Python one, so # it holds even if this app ever runs with more than one worker). - # `capacity_check()` alone (also used by `create()`) only reads the - # current count; without this lock two concurrent `start()` calls - # could both pass that read before either commits, overshooting - # `max_concurrent_games`. Xact-scoped: released automatically at the - # commit/rollback that ends this method. + # A plain read of the current count alone (what `create()`'s own + # `capacity_check()` still does, deliberately unchanged -- a room can + # always be *created* regardless of global capacity, only *starting* + # it queues) is not enough on its own; without this lock two + # concurrent `start()` calls could both pass that read before either + # commits, overshooting `max_concurrent_games`. Xact-scoped: released + # automatically at the commit/rollback that ends this method. # `session.exec()` is SQLModel's wrapper for `select()` returning model # rows; a bare `text()` DB call has no model to return, so the plain # SQLAlchemy `execute()` is the only option here. self.session.execute( # ty: ignore[deprecated] text("SELECT pg_advisory_xact_lock(:key)"), {"key": _START_LOCK_KEY} ) - await self.capacity_check() + # Phase 16: queue-aware, not a hard refusal. `position != 1` (queued, + # but not yet its turn) enqueues even if capacity happens to be free + # right now -- jumping the line just because the caller got lucky on + # timing would break the fairness the queue exists to provide. + # `position == 1` pops the room off the queue itself before + # proceeding (whether this call arrived via a real retry or via + # `_promote_next_queued_room`'s suggestion after a slot freed). + limit = settings_store.max_concurrent_games(self.session) + count = self.session.exec( + select(func.count()).select_from(Room).where(Room.status == "in_progress") + ).one() + position = await self.queue.position("global", str(room_id)) + if count >= limit or (position is not None and position != 1): + await self.queue.enqueue("global", str(room_id)) + status = await self.queue.snapshot("global", str(room_id)) + # `pg_advisory_xact_lock` only releases on commit/rollback -- an + # uncommitted session left holding it (e.g. a caller that catches + # this and keeps working on the same session, as tests do) would + # otherwise starve every other connection's start() call on this + # same lock key until that session eventually closes. Nothing + # written above needs keeping: only reads and the lock call itself. + self.session.rollback() + raise CapacityExceededError("global", status) + if position == 1: + await self.queue.pop_next("global") # Re-check status: something else could have started or stopped this # exact room while we were building specs or waiting for the lock. self.session.refresh(room) if room.status != "waiting": + self.session.rollback() # see the CapacityExceededError rollback above raise RoomNotWaitingError(f"room {room_id} is {room.status}") try: @@ -594,6 +725,49 @@ async def _finish_room(self, room_id: str, outcome: str) -> None: except Exception: log.exception("failed to archive game history for room %s", room_id) + try: + await self._promote_next_queued_room(session) + except Exception: + log.exception( + "failed to promote the next globally-queued room after " + "room %s finished", + room_id, + ) + + async def _promote_next_queued_room(self, session: Session) -> None: + """A concurrent-game slot just freed (only ever called from + ``_finish_room``) -- try to start whatever room is at the head of + the global queue, if any. This only ever *suggests* a room to try; + the actual dequeue happens inside ``start()``'s own + advisory-locked capacity check, not here, so if the real check + finds the suggestion is stale (no longer at the head, already + started/stopped some other way, or capacity is gone again by the + time the lock is acquired) it re-enqueues or no-ops harmlessly + instead of double-starting anything.""" + next_room_id = await self.queue.peek_head("global") + if next_room_id is None: + return + temp_service = RoomService( + session, self.runner, self.queue, self.registry, self.bus + ) + try: + await temp_service.start(uuid.UUID(next_room_id)) + except RoomNotWaitingError, RoomFullError, CapacityExceededError: + # Expected when the suggestion was stale (see the docstring + # above) -- but also what a room stuck with an unfilled seat + # (e.g. an interrupted `force_start`) looks like from here, and + # that case never resolves itself: it stays at the head of the + # global queue and this same suppressed failure repeats every + # time a slot frees, forever. Logged (not re-raised -- this is + # still best-effort) so a room that's actually stuck is + # observable instead of silently retried forever. + log.info( + "promotion of globally-queued room %s did not start it " + "(stale suggestion, or the room itself is not actually " + "startable right now)", + next_room_id, + ) + async def _archive_game(self, session: Session, room: Room, outcome: str) -> None: """Copies this game's event log from the live ``EventBus`` into ``game_record``/``game_event`` -- the durable record DECISIONS.md §12 diff --git a/backend/tests/rooms/test_room_service.py b/backend/tests/rooms/test_room_service.py index dd4dff0..0f08ab3 100644 --- a/backend/tests/rooms/test_room_service.py +++ b/backend/tests/rooms/test_room_service.py @@ -12,6 +12,7 @@ from typing import Literal import pytest +from redis.asyncio import Redis from sqlmodel import Session from app import settings_store @@ -19,10 +20,12 @@ from app.authz.permissions import P from app.core.config import settings from app.events import InMemoryEventBus +from app.llm import ModelRegistry from app.metrics import ACTIVE_GAMES, MOVES_TOTAL from app.models import RoomCreate, RoomSeatSpec from app.referee import GameRunner from app.rooms.codes import is_valid_code +from app.rooms.queue import JoinQueue from app.rooms.service import ( CapacityExceededError, RoomFullError, @@ -283,7 +286,7 @@ async def test_leave_frees_the_seat(db: Session, service: RoomService) -> None: room = await service.create(creator, _two_human()) joiner = create_random_user(db) await service.join_by_code(joiner, room.code) - service.leave(joiner, room.id) + await service.leave(joiner, room.id) seat = next(s for s in service.seats(room.id) if s.seat == 0) assert seat.status == "open" assert seat.user_id is None @@ -294,7 +297,7 @@ async def test_leave_is_a_noop_when_not_seated( ) -> None: creator = create_random_user(db) room = await service.create(creator, _two_human()) - service.leave(create_random_user(db), room.id) # must not raise + await service.leave(create_random_user(db), room.id) # must not raise async def test_leave_blocked_once_in_progress( @@ -304,7 +307,7 @@ async def test_leave_blocked_once_in_progress( room = await service.create(creator, _two_agent()) await service.start(room.id) try: - service.leave(creator, room.id) # no-op: room is in_progress + await service.leave(creator, room.id) # no-op: room is in_progress seats = service.seats(room.id) assert all(s.status == "filled" for s in seats) finally: @@ -648,3 +651,191 @@ async def test_finish_room_does_not_overwrite_an_already_stopped_room( db.expire(room) assert service.get(room.id).status == "stopped" + + +# -- queueing (Phase 16) ------------------------------------------------- + + +async def test_join_by_code_enqueues_when_room_is_full( + db: Session, service: RoomService +) -> None: + creator = create_random_user(db) + room = await service.create(creator, _two_human()) + for _ in range(2): + await service.join_by_code(create_random_user(db), room.code) + + latecomer = create_random_user(db) + with pytest.raises(RoomFullError): + await service.join_by_code(latecomer, room.code) + assert await service.queue.position(f"room:{room.id}", str(latecomer.id)) == 1 + + +async def test_leave_offers_the_freed_seat_and_a_random_joiner_cannot_steal_it( + db: Session, service: RoomService +) -> None: + creator = create_random_user(db) + room = await service.create(creator, _two_human()) + occupants = [create_random_user(db) for _ in range(2)] + seats = [await service.join_by_code(u, room.code) for u in occupants] + + waiter = create_random_user(db) + with pytest.raises(RoomFullError): + await service.join_by_code(waiter, room.code) + + await service.leave(occupants[0], room.id) + freed_seat = seats[0].seat + reloaded = next(s for s in service.seats(room.id) if s.seat == freed_seat) + assert reloaded.status == "offered" # held for the waiter, not generally open + assert reloaded.user_id is None + + # A fresh joiner (never queued) cannot steal an "offered" seat -- only + # the waiter's own accept can claim it. + other = create_random_user(db) + with pytest.raises(RoomFullError): + await service.join_by_code(other, room.code) + + target = await service.join_by_code(waiter, room.code) + assert target.seat == freed_seat + assert target.user_id == waiter.id + assert target.status == "filled" + + +async def test_expired_offer_is_reclaimed_and_reoffered_to_the_next_waiter( + db: Session, + runner: GameRunner, + registry: ModelRegistry, + bus: InMemoryEventBus, + rds: Redis, +) -> None: + """A real, short TTL (not a `now_ms` override -- that's already + covered at the `JoinQueue` level in `tests/unit/test_join_queue.py`; + this proves the *RoomService*-level reclaim-and-reoffer end to end).""" + fast_queue = JoinQueue(rds, offer_ttl_s=1) + service = RoomService(db, runner, fast_queue, registry, bus) + + creator = create_random_user(db) + room = await service.create(creator, _two_human()) + a, b = create_random_user(db), create_random_user(db) + await service.join_by_code(a, room.code) + await service.join_by_code(b, room.code) + + first_waiter = create_random_user(db) + second_waiter = create_random_user(db) + with pytest.raises(RoomFullError): + await service.join_by_code(first_waiter, room.code) + with pytest.raises(RoomFullError): + await service.join_by_code(second_waiter, room.code) + + await service.leave(a, room.id) # offers a's seat to first_waiter + await asyncio.sleep(1.2) # let the 1s offer genuinely expire + + # first_waiter's own retry sweeps first (finds no live offer -- it + # just expired), reclaiming the seat; since first_waiter is still at + # the head of the line (this was their only miss so far), it's + # immediately re-offered and re-accepted within this same call -- + # a self-healing retry, not a second failure. + target = await service.join_by_code(first_waiter, room.code) + assert target.user_id == first_waiter.id + + # second_waiter was never touched by any of this -- still queued, + # now at the head. + assert await fast_queue.position(f"room:{room.id}", str(second_waiter.id)) == 1 + + +async def test_finish_room_promotes_the_next_globally_queued_room( + db: Session, service: RoomService, runner: GameRunner +) -> None: + """A concurrent-game slot freeing (via `_finish_room`) tries to start + whatever room is at the head of the global queue -- the join-queue + wiring (Phase 16), not the model-swap scheduler (also Phase 16, but a + separate concern).""" + settings_store.set_value(db, "max_concurrent_games", 1, None) + room_b = None + try: + creator = create_random_user(db) + room_a = await service.create(creator, _two_agent()) + room_b = await service.create(creator, _two_agent()) # both still "waiting" + # Simulate a running game without a real GameRunner task -- matches + # `test_finish_room_writes_finished_and_finished_at`'s own pattern + # above; capacity accounting only ever reads `Room.status` from the + # DB, so this alone is enough to make room_b's start() see count>=limit. + # (Must happen *after* both creates: `create()` itself is still + # gated by the unchanged, hard-refusing `capacity_check()`.) + room_a.status = "in_progress" + db.add(room_a) + db.commit() + + with pytest.raises(CapacityExceededError): + await service.start(room_b.id) + assert await service.queue.position("global", str(room_b.id)) == 1 + + await service._finish_room(str(room_a.id), "finished") + db.expire_all() + + # Promotion runs `start()` on room_b for real, from inside + # `_finish_room`'s own fresh session -- poll rather than assume + # it's already committed and visible to this session. + for _ in range(200): + if service.get(room_b.id).status == "in_progress": + break + await asyncio.sleep(0.01) + assert service.get(room_b.id).status == "in_progress" + assert runner.get(str(room_b.id)) is not None + assert await service.queue.position("global", str(room_b.id)) is None + finally: + if room_b is not None: + await service.stop(room_b.id, "test cleanup") + settings_store.set_value( + db, "max_concurrent_games", settings.MAX_CONCURRENT_GAMES, None + ) + + +async def test_force_start_backfill_survives_a_capacity_queue_rollback( + db: Session, service: RoomService, runner: GameRunner +) -> None: + """`force_start`'s human->agent seat backfill must survive `start()` + hitting the global capacity cap: `start()` now (Phase 16) enqueues and + rolls back on `CapacityExceededError` instead of only ever raising + outright, and `_promote_next_queued_room` later retries this exact room + from a completely different session. If the backfill were only staged + (not committed) on this session, that retry's own fresh read of the + seats would still see the original open human seats, fail the same + "not all filled" check `start()` itself enforces, and get silently + swallowed -- permanently orphaning the room in the global queue with no + way to ever succeed.""" + settings_store.set_value(db, "max_concurrent_games", 1, None) + room_b = None + try: + creator = create_random_user(db) + room_a = await service.create(creator, _two_agent()) + room_b = await service.create(creator, _two_human()) # unfilled human seats + room_a.status = "in_progress" + db.add(room_a) + db.commit() + + with pytest.raises(CapacityExceededError): + await service.force_start(creator, room_b.id) + + # The backfill committed despite `start()`'s rollback -- not staged + # and lost with it. + db.expire_all() + assert all( + s.kind == "agent" and s.status == "filled" for s in service.seats(room_b.id) + ) + assert await service.queue.position("global", str(room_b.id)) == 1 + + await service._finish_room(str(room_a.id), "finished") + db.expire_all() + + for _ in range(200): + if service.get(room_b.id).status == "in_progress": + break + await asyncio.sleep(0.01) + assert service.get(room_b.id).status == "in_progress" + assert runner.get(str(room_b.id)) is not None + finally: + if room_b is not None: + await service.stop(room_b.id, "test cleanup") + settings_store.set_value( + db, "max_concurrent_games", settings.MAX_CONCURRENT_GAMES, None + ) diff --git a/backend/tests/rooms/test_rooms_api.py b/backend/tests/rooms/test_rooms_api.py index 3031d70..879681f 100644 --- a/backend/tests/rooms/test_rooms_api.py +++ b/backend/tests/rooms/test_rooms_api.py @@ -129,7 +129,41 @@ def test_join_full_room_is_409_with_queue_body( r = client.post(f"{API}/rooms/join", headers=latecomer, json={"code": code}) assert r.status_code == 409 body = r.json() - assert set(body["queue"]) == {"position", "length", "eta_seconds"} + assert body["room_id"] + assert set(body["queue"]) == { + "position", + "length", + "eta_seconds", + "offered", + "offer_expires_in_seconds", + } + + +def test_room_queue_reports_position_then_404s_once_not_queued( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + r = client.post(f"{API}/rooms/", headers=superuser_token_headers, json=_body()) + room_id = r.json()["id"] + code = r.json()["code"] + for _ in range(2): + headers = _new_user_headers(client) + joined = client.post(f"{API}/rooms/join", headers=headers, json={"code": code}) + assert joined.status_code == 200, joined.text + + latecomer = _new_user_headers(client) + joined = client.post(f"{API}/rooms/join", headers=latecomer, json={"code": code}) + assert joined.status_code == 409 + + r = client.get(f"{API}/rooms/{room_id}/queue", headers=latecomer) + assert r.status_code == 200, r.text + assert r.json()["position"] == 1 + assert r.json()["offered"] is False + + # A caller with no queue membership at all (never tried to join this + # room) gets a plain 404, not a stale/empty status. + stranger = _new_user_headers(client) + r = client.get(f"{API}/rooms/{room_id}/queue", headers=stranger) + assert r.status_code == 404 def test_start_needs_creator_or_force_stop( @@ -184,7 +218,13 @@ def test_capacity_exceeded_is_429_with_queue_body( r2 = client.post(f"{API}/rooms/", headers=h, json=_body(agent=True)) assert r2.status_code == 429 - assert set(r2.json()["queue"]) == {"position", "length", "eta_seconds"} + assert set(r2.json()["queue"]) == { + "position", + "length", + "eta_seconds", + "offered", + "offer_expires_in_seconds", + } finally: if room_id is not None: client.post(f"{API}/rooms/{room_id}/stop", headers=h) diff --git a/backend/tests/unit/llm/test_registry.py b/backend/tests/unit/llm/test_registry.py index 2a778f7..645efaf 100644 --- a/backend/tests/unit/llm/test_registry.py +++ b/backend/tests/unit/llm/test_registry.py @@ -43,9 +43,9 @@ async def test_one_http_client_per_base_url(settings: Settings) -> None: }, ) a, b, c = (reg.client_for(m) for m in ("a:1b", "b:1b", "c:1b")) - assert isinstance(a, OllamaNativeAdapter) - assert isinstance(b, OllamaNativeAdapter) - assert isinstance(c, OpenAICompatAdapter) + assert isinstance(a.inner, OllamaNativeAdapter) + assert isinstance(b.inner, OllamaNativeAdapter) + assert isinstance(c.inner, OpenAICompatAdapter) assert a.client is b.client assert a.client is not c.client assert a.num_ctx == settings.LLM_NUM_CTX @@ -54,6 +54,9 @@ async def test_one_http_client_per_base_url(settings: Settings) -> None: assert reg.config_for("c:1b").num_ctx == 1 assert reg.client_for("a:1b") is a + assert a._scheduler is b._scheduler # same base_url, shared affinity scheduler + assert a._scheduler is not c._scheduler + await reg.aclose() assert a.client.is_closed assert c.client.is_closed diff --git a/backend/tests/unit/llm/test_scheduled_client.py b/backend/tests/unit/llm/test_scheduled_client.py new file mode 100644 index 0000000..09e6ca5 --- /dev/null +++ b/backend/tests/unit/llm/test_scheduled_client.py @@ -0,0 +1,79 @@ +"""``_ScheduledClient`` (Phase 16, DECISIONS.md §13): the transparent +wrapper ``ModelRegistry.client_for`` hands out, gating real calls through +a ``ModelAffinityScheduler`` and firing the swap counter (Phase 13) at +actual dispatch time.""" + +from __future__ import annotations + +import asyncio + +from app.llm.client import CompletionRequest, CompletionResult, Message +from app.llm.registry import _ScheduledClient +from app.llm.scheduler import ModelAffinityScheduler +from app.metrics import MODEL_SWAPS_TOTAL + + +class _FakeInner: + """Minimal LLMClient: holds `complete()` open until released, so a + test can observe the wrapper's admission gating directly.""" + + def __init__(self, base_url: str) -> None: + self.base_url = base_url + self.release_event = asyncio.Event() + self.calls: list[str] = [] + + async def complete(self, req: CompletionRequest) -> CompletionResult: + self.calls.append(req.model) + await self.release_event.wait() + return CompletionResult( + text="{}", + model=req.model, + latency_ms=1, + prompt_tokens=1, + completion_tokens=1, + ) + + async def aclose(self) -> None: + return None + + +def _req(model: str) -> CompletionRequest: + return CompletionRequest( + model=model, messages=(Message(role="user", content="hi"),) + ) + + +async def test_scheduled_client_delays_a_cold_model_until_a_slot_frees() -> None: + base_url = "http://fake-scheduled-client-test:1" + inner = _FakeInner(base_url) + scheduler = ModelAffinityScheduler(max_loaded=1, max_wait_s=5.0) + client = _ScheduledClient(inner, scheduler) + + task_a = asyncio.create_task(client.complete(_req("model-a"))) + await asyncio.sleep(0.02) + assert inner.calls == ["model-a"] + + task_b = asyncio.create_task(client.complete(_req("model-b"))) + await asyncio.sleep(0.02) + assert inner.calls == ["model-a"], "model-b should be gated, not yet dispatched" + + inner.release_event.set() + await asyncio.wait_for(task_a, timeout=1.0) + await asyncio.wait_for(task_b, timeout=1.0) + assert inner.calls == ["model-a", "model-b"] + + +async def test_note_llm_call_fires_at_dispatch_time_through_the_wrapper() -> None: + base_url = "http://fake-scheduled-client-swap-test:1" + inner = _FakeInner(base_url) + inner.release_event.set() # complete immediately, no gating needed + scheduler = ModelAffinityScheduler(max_loaded=2, max_wait_s=5.0) + client = _ScheduledClient(inner, scheduler) + + before = MODEL_SWAPS_TOTAL.labels(base_url=base_url)._value.get() + await client.complete(_req("model-a")) + await client.complete(_req("model-a")) # same model: not a swap + assert MODEL_SWAPS_TOTAL.labels(base_url=base_url)._value.get() == before + + await client.complete(_req("model-b")) # different model: a swap + assert MODEL_SWAPS_TOTAL.labels(base_url=base_url)._value.get() == before + 1 diff --git a/backend/tests/unit/llm/test_scheduler.py b/backend/tests/unit/llm/test_scheduler.py new file mode 100644 index 0000000..4cd913c --- /dev/null +++ b/backend/tests/unit/llm/test_scheduler.py @@ -0,0 +1,111 @@ +"""ModelAffinityScheduler (Phase 16, DECISIONS.md §13): bounded admission +control biasing concurrent LLM traffic towards fewer distinct +simultaneously-active models.""" + +from __future__ import annotations + +import asyncio + +from app.llm.scheduler import ModelAffinityScheduler + + +async def test_acquire_admits_up_to_max_loaded_immediately() -> None: + sched = ModelAffinityScheduler(max_loaded=2, max_wait_s=5.0) + await asyncio.wait_for(sched.acquire("a"), timeout=0.1) + await asyncio.wait_for(sched.acquire("b"), timeout=0.1) + assert sched.snapshot() == {"a": 1, "b": 1} + + +async def test_repeated_acquire_of_an_already_loaded_model_never_blocks() -> None: + """A model already in the active set never counts as a *new* one -- + unbounded concurrent calls to the same model (however many turns are + using it right now) never queue behind capacity.""" + sched = ModelAffinityScheduler(max_loaded=1, max_wait_s=5.0) + await asyncio.wait_for(sched.acquire("a"), timeout=0.1) + await asyncio.wait_for(sched.acquire("a"), timeout=0.1) + await asyncio.wait_for(sched.acquire("a"), timeout=0.1) + assert sched.snapshot() == {"a": 3} + + +async def test_acquire_for_a_new_model_waits_at_capacity_then_admits_on_release() -> ( + None +): + sched = ModelAffinityScheduler(max_loaded=1, max_wait_s=5.0) + await sched.acquire("a") + + task = asyncio.create_task(sched.acquire("b")) + await asyncio.sleep(0.05) + assert not task.done(), "b should be waiting -- a already holds the one slot" + + await sched.release("a") + await asyncio.wait_for(task, timeout=1.0) + assert sched.snapshot() == {"b": 1} + + +async def test_acquire_force_admits_after_max_wait_timeout() -> None: + """Never starves a turn forever: if the active set stays full past + max_wait_s, admit anyway rather than block indefinitely.""" + sched = ModelAffinityScheduler(max_loaded=1, max_wait_s=0.05) + await sched.acquire("a") # never released during this test + + start = asyncio.get_event_loop().time() + await asyncio.wait_for(sched.acquire("b"), timeout=1.0) + elapsed = asyncio.get_event_loop().time() - start + + assert elapsed >= 0.05 + assert sched.snapshot() == {"a": 1, "b": 1} + + +async def test_release_drops_a_model_from_the_active_set_at_zero_in_flight() -> None: + sched = ModelAffinityScheduler(max_loaded=2, max_wait_s=5.0) + await sched.acquire("a") + await sched.acquire("a") + assert sched.snapshot() == {"a": 2} + await sched.release("a") + assert sched.snapshot() == {"a": 1} + await sched.release("a") + assert sched.snapshot() == {} + + +async def test_releasing_one_of_several_in_flight_calls_does_not_wake_a_waiter() -> ( + None +): + """A model with more than one concurrent call still occupies its one + active-set slot after *one* of those calls finishes -- releasing it + must not wake a different model's waiter, or the active set grows past + max_loaded (the exact invariant this scheduler exists to hold).""" + sched = ModelAffinityScheduler(max_loaded=1, max_wait_s=5.0) + await sched.acquire("a") + await sched.acquire("a") # two in-flight calls to "a" + + task_b = asyncio.create_task(sched.acquire("b")) + await asyncio.sleep(0.05) + assert not task_b.done() + + await sched.release("a") # only one of a's two calls finished + await asyncio.sleep(0.05) + assert not task_b.done(), "b must still wait -- a still holds the one slot" + assert sched.snapshot() == {"a": 1} + + await sched.release("a") # now a's active-set slot actually frees + await asyncio.wait_for(task_b, timeout=1.0) + assert sched.snapshot() == {"b": 1} + + +async def test_three_models_at_max_loaded_two_forces_a_wait_not_a_deadlock() -> None: + """The scenario DECISIONS.md §13 actually describes: three concurrent + tables wanting three different models against a backend that can only + hold two loaded. The third waits (biasing traffic towards the + already-active two) rather than all three firing simultaneously.""" + sched = ModelAffinityScheduler(max_loaded=2, max_wait_s=5.0) + await sched.acquire("a") + await sched.acquire("b") + + task_c = asyncio.create_task(sched.acquire("c")) + await asyncio.sleep(0.05) + assert not task_c.done() + assert sched.snapshot() == {"a": 1, "b": 1} + + await sched.release("b") + await asyncio.wait_for(task_c, timeout=1.0) + assert sched.snapshot() == {"a": 1, "c": 1} diff --git a/backend/tests/unit/test_join_queue.py b/backend/tests/unit/test_join_queue.py index 834d105..cb2e1c5 100644 --- a/backend/tests/unit/test_join_queue.py +++ b/backend/tests/unit/test_join_queue.py @@ -44,7 +44,10 @@ async def test_offer_and_accept(q: JoinQueue, rds: Redis, scope: str) -> None: user, deadline = offered assert user == "a" ttl = int(await rds.pttl(f"offer:{scope}:a")) - assert 0 < ttl <= 30_000 + # The key's own PX is 4x the offer TTL (see _OFFER_NEXT's comment) -- + # sweep()'s explicit ZSET-deadline reclaim, not Redis's own auto-expiry, + # is what's supposed to actually free the seat. + assert 0 < ttl <= 30_000 * 4 assert await q.length(scope) == 1 assert await q.accept(scope, "a") == 3 assert await q.accept(scope, "a") is None @@ -62,14 +65,14 @@ async def test_sweep_requeues_at_original_score_then_drops( deadline = offered[1] await q.enqueue(scope, "c") assert await q.sweep(scope, now_ms=deadline - 1) == [] - assert await q.sweep(scope, now_ms=deadline) == ["a"] + assert await q.sweep(scope, now_ms=deadline) == [1] # seat 1 reclaimed assert await q.position(scope, "a") == 1 assert await q.position(scope, "c") == 3 # second miss: dropped offered = await q.offer_next(scope, 1) assert offered is not None assert offered[0] == "a" - assert await q.sweep(scope, now_ms=offered[1]) == [] + assert await q.sweep(scope, now_ms=offered[1]) == [1] # seat reclaimed either way assert await q.position(scope, "a") is None assert await q.accept(scope, "a") is None assert await q.position(scope, "b") == 1 @@ -94,11 +97,31 @@ async def test_accept_fails_once_sweep_claims_it_first( await q.enqueue(scope, "b") offered = await q.offer_next(scope, 1) assert offered is not None - assert await q.sweep(scope, now_ms=offered[1]) == ["a"] # sweep wins first + assert await q.sweep(scope, now_ms=offered[1]) == [1] # sweep wins first assert await q.accept(scope, "a") is None # too late: nothing left to claim assert await q.position(scope, "a") == 1 # requeued, not silently dropped +async def test_pop_next_is_a_plain_fifo_pop(q: JoinQueue, scope: str) -> None: + assert await q.pop_next(scope) is None + await q.enqueue(scope, "a") + await q.enqueue(scope, "b") + assert await q.pop_next(scope) == "a" + assert await q.length(scope) == 1 + assert await q.position(scope, "b") == 1 + assert await q.pop_next(scope) == "b" + assert await q.pop_next(scope) is None + + +async def test_peek_head_does_not_remove(q: JoinQueue, scope: str) -> None: + assert await q.peek_head(scope) is None + await q.enqueue(scope, "a") + await q.enqueue(scope, "b") + assert await q.peek_head(scope) == "a" + assert await q.peek_head(scope) == "a" # still there + assert await q.length(scope) == 2 + + def test_eta_arithmetic() -> None: assert JoinQueue.eta_seconds(1, None) is None started = datetime.now(UTC) - timedelta(seconds=300) @@ -127,4 +150,26 @@ async def test_snapshot(q: JoinQueue, scope: str) -> None: "position": None, "length": 2, "eta_seconds": None, + "offered": False, + "offer_expires_in_seconds": None, } + + +async def test_snapshot_reports_a_live_offer_without_consuming_it( + q: JoinQueue, scope: str +) -> None: + await q.enqueue(scope, "a") + offered = await q.offer_next(scope, 7) + assert offered is not None + + snap = await q.snapshot(scope, "a") + assert snap.position is None # popped out of queue: into offers: + assert snap.offered is True + assert snap.offer_expires_in_seconds is not None + assert 0 < snap.offer_expires_in_seconds <= 30 + + # Non-destructive: a second read sees the exact same live offer, and the + # offer itself is still acceptable afterwards. + snap_again = await q.snapshot(scope, "a") + assert snap_again.offered is True + assert await q.accept(scope, "a") == 7 diff --git a/docs/capacity-and-queueing.md b/docs/capacity-and-queueing.md index ef14dd8..a6f55c5 100644 --- a/docs/capacity-and-queueing.md +++ b/docs/capacity-and-queueing.md @@ -106,13 +106,32 @@ places wait one typical game each: `max(0, typical - elapsed) + (position - 1) * estimate on purpose; refine with a measured median game length once games persist durations. -## What Phase 7 rooms wire on top +## What rooms wire on top - `POST /rooms` with `rate_limit_from_setting("room_create", ...)`, `POST /rooms/{code}/join` with `room_join`; chat via `check_rate_ws`. - Global cap: a room may start only if running games < `MAX_CONCURRENT_GAMES`, - else the room enqueues on `"global"` and starts on its offer. + else the room enqueues on `"global"` and starts on its offer -- `start` + itself enqueues (or, at position 1, pops itself off) rather than + `create`'s own hard refusal, and `RoomService._finish_room` tries to + promote the next globally-queued room once a slot frees. - Seat cap: a full room enqueues joiners on `f"room:{id}"`; a leaving human triggers `offer_next(scope, seat)`. -- One sweeper task per process, leader-elected, calling `sweep` for every scope - with live offers; `seat_offer` / `queue_update` events on the room stream. +- **No standalone sweeper task (Phase 7's original plan for this).** Phase 16 + wired the queue fully but reclaims expired offers opportunistically + instead of via a leader-elected background task: `join_by_code` sweeps as + a side effect of a caller's own retry, and `GET /{room_id}/queue` sweeps + on every poll while a waiting caller's own queue UI is open. Between + those two, the realistic paths are covered (either the offeree is still + watching, or someone else eventually tries the room again) -- but a + room whose only waiter closes their tab, where nobody else ever touches + that room's queue, leaves that one offer stale until something does. A + real background sweeper would close this fully; not built, since neither + path it would add coverage for is the common case. +- **No `seat_offer` / `queue_update` push events either.** The frontend + polls `GET /{room_id}/queue?scope=seat|global` (`QueueStatus`, including + `offered`/`offer_expires_in_seconds` for a live seat offer) rather than + receiving a push notification over the room's event stream -- simpler, + and the room already flips to `"in_progress"` over the existing + websocket once a promoted/accepted seat actually starts a game, so the + wait itself is the only thing polling covers. diff --git a/docs/local-model-setup.md b/docs/local-model-setup.md index a4de457..68cb7af 100644 --- a/docs/local-model-setup.md +++ b/docs/local-model-setup.md @@ -70,12 +70,42 @@ names from config. at once, and the botpit-side turn scheduler grouping consecutive AI turns by model where turn order allows. -**Conflict with DECISIONS.md §13 (flag for review once the file is -available):** if §13 assumes keep-alive tuning alone makes multi-model -tables cheap, that assumption is wrong on a 6 GB card. The practical rule -is "at most two 4B-class models resident; an 8B model is alone." The -scheduler in Phase 16 should treat model identity as a resource, not just -a label. +**Resolved -- DECISIONS.md §13 agrees with this, not in conflict with it.** +The practical rule is "at most two 4B-class models resident; an 8B model +is alone," matching `OLLAMA_MAX_LOADED_MODELS=2` above. + +### 2.3.1 The scheduler, built (Phase 16) + +`app.llm.scheduler.ModelAffinityScheduler` -- one instance per LLM +base_url (created lazily inside `ModelRegistry`, never shared across +different backends), wrapping every client `ModelRegistry.client_for` +hands out via a transparent `_ScheduledClient`. Before a real completion +call is allowed to fire, it must `acquire()` its model: immediate if that +model already has a call in flight, or if the active set is below +`LLM_MAX_LOADED_MODELS`; otherwise it waits (up to +`LLM_SCHEDULER_MAX_WAIT_SECONDS`, then is admitted anyway -- a turn's own +deadline must never be blocked forever by scheduling). This is exactly +what "prefer tables that share a model" and "group consecutive AI turns +by model" mean at the one layer this code can actually act on: which +already-due call gets to fire next across every room sharing one +backend, not which seat's turn comes next within one game (that's fixed +by the game's own rules). + +Deliberately a heuristic, not a perfect model of Ollama's real VRAM +state: "active" here means "has at least one in-flight call," not "Ollama +still has it loaded under keep_alive." A model with zero in-flight calls +drops out of the tracked set immediately, even if Ollama itself hasn't +evicted it yet -- so under bursty traffic the scheduler can still admit a +third model sooner than ideal. Querying Ollama's own `/api/ps` for ground +truth would close that gap but costs a real network round trip per +admission decision; not built, a documented tradeoff. + +`botpit_model_swaps_total` (Phase 13's metric) is exactly what would show +whether this is actually reducing swaps in practice: it fires at real +dispatch time (after the scheduler admits a call), so comparing its rate +before/after this phase -- or under different `LLM_MAX_LOADED_MODELS` +values -- is the intended way to validate the scheduler's real effect +once real concurrent multi-model traffic exists to measure. ### 2.4 The API surface botpit will use diff --git a/docs/rooms.md b/docs/rooms.md index 10141da..6acd124 100644 --- a/docs/rooms.md +++ b/docs/rooms.md @@ -229,17 +229,22 @@ seen event's own `stream_id` to resume after it. Full detail: even the creator) — an admin/spectator view across every seat at once, built on top of `/ws`'s privileged feed, is still Phase 10 work in progress (DECISIONS §10). -- **The join queue is read-only in this phase.** `capacity_check` reads - `queue.snapshot` for its error payload, and the `/rooms/join` route reads - it for a full room's 409 body, but neither `create` nor `join_by_code` - ever calls `queue.enqueue`/`offer_next`: nobody actually waits in line - yet. `docs/capacity-and-queueing.md`'s "a room may start only if running - games < `MAX_CONCURRENT_GAMES`, else it enqueues on `global`" is not - implemented as queueing — Phase 7 instead refuses `create` outright - (`CapacityExceededError`) once at capacity. Wiring real enqueue-and-offer - behavior on top of the now-tested `JoinQueue` primitives is future work. -- **`leave` does not trigger `offer_next`.** Freeing a seat does not offer - it to anyone waiting (nobody is waiting yet, per the point above). +- **The join queue is fully wired (Phase 16).** `join_by_code` enqueues on + `"room:{id}"` when a room is full (`RoomFullError`, 409, with the queue + snapshot and the room's own id in the body so a caller who only had the + join code can still poll `GET /{room_id}/queue`); `leave` calls + `offer_next` on the seat it just freed, and `RoomService._reclaim_expired_offers` + sweeps and re-offers any offer whose deadline passed unaccepted. + `start` is queue-aware on `"global"` (DECISIONS §2's concurrent-game cap): + it enqueues rather than refusing outright unless the room is already at + position 1, in which case it pops itself off; `_finish_room` tries to + promote the next globally-queued room once a slot frees + (`_promote_next_queued_room`, best-effort — `start`'s own advisory-locked + check is the real gate against a stale suggestion). `create`'s own + capacity check is deliberately unchanged: a room can always be created, + only *starting* it queues. `GET /{room_id}/queue?scope=seat|global` + reports position/ETA either way, plus `offered`/`offer_expires_in_seconds` + for a live seat offer (`QueueStatus`, `app/rooms/queue.py`). ## `subscription_guard` (`app/rooms/guard.py`) diff --git a/frontend/src/client/index.ts b/frontend/src/client/index.ts index 83b9695..6cf321f 100644 --- a/frontend/src/client/index.ts +++ b/frontend/src/client/index.ts @@ -1,4 +1,4 @@ // This file is auto-generated by @hey-api/openapi-ts export { AdminService, FlagsService, GamesService, HealthService, ItemsService, LeaderboardService, LoginService, MetricsService, type Options, PrivateService, RolesService, RoomsService, UsersService, UtilsService } from './sdk.gen'; -export type { adminCreateOverrideData, adminCreateOverrideError, adminCreateOverrideErrors, adminCreateOverrideResponse, adminCreateOverrideResponses, adminReadFlagsData, adminReadFlagsResponse, adminReadFlagsResponses, adminReadSettingsData, adminReadSettingsResponse, adminReadSettingsResponses, adminRemoveOverrideData, adminRemoveOverrideError, adminRemoveOverrideErrors, adminRemoveOverrideResponse, adminRemoveOverrideResponses, adminUpdateFlagData, adminUpdateFlagError, adminUpdateFlagErrors, adminUpdateFlagResponse, adminUpdateFlagResponses, adminUpdateSettingData, adminUpdateSettingError, adminUpdateSettingErrors, adminUpdateSettingResponse, adminUpdateSettingResponses, Body_login_login_access_token, ChatSend, ClientOptions, FeatureFlagOverrideCreate, FeatureFlagOverridePublic, FeatureFlagPublic, FeatureFlagUpdate, flagsReadMyFlagsData, flagsReadMyFlagsResponse, flagsReadMyFlagsResponses, GameRecordPublic, gamesGameEventsData, gamesGameEventsError, gamesGameEventsErrors, gamesGameEventsResponses, gamesGetGameRecordData, gamesGetGameRecordError, gamesGetGameRecordErrors, gamesGetGameRecordResponse, gamesGetGameRecordResponses, gamesReplayData, gamesReplayError, gamesReplayErrors, gamesReplayResponses, healthHealthData, healthHealthLlmData, healthHealthLlmResponse, healthHealthLlmResponses, healthHealthResponse, healthHealthResponses, HealthReport, HTTPValidationError, ItemCreate, ItemPublic, itemsCreateItemData, itemsCreateItemError, itemsCreateItemErrors, itemsCreateItemResponse, itemsCreateItemResponses, itemsDeleteItemData, itemsDeleteItemError, itemsDeleteItemErrors, itemsDeleteItemResponse, itemsDeleteItemResponses, ItemsPublic, itemsReadItemData, itemsReadItemError, itemsReadItemErrors, itemsReadItemResponse, itemsReadItemResponses, itemsReadItemsData, itemsReadItemsError, itemsReadItemsErrors, itemsReadItemsResponse, itemsReadItemsResponses, itemsUpdateItemData, itemsUpdateItemError, itemsUpdateItemErrors, itemsUpdateItemResponse, itemsUpdateItemResponses, ItemUpdate, leaderboardAccountsLeaderboardData, leaderboardAccountsLeaderboardError, leaderboardAccountsLeaderboardErrors, leaderboardAccountsLeaderboardResponse, leaderboardAccountsLeaderboardResponses, LeaderboardEntry, leaderboardModelsLeaderboardData, leaderboardModelsLeaderboardError, leaderboardModelsLeaderboardErrors, leaderboardModelsLeaderboardResponse, leaderboardModelsLeaderboardResponses, LLMHealth, loginLoginAccessTokenData, loginLoginAccessTokenError, loginLoginAccessTokenErrors, loginLoginAccessTokenResponse, loginLoginAccessTokenResponses, loginRecoverPasswordData, loginRecoverPasswordError, loginRecoverPasswordErrors, loginRecoverPasswordHtmlContentData, loginRecoverPasswordHtmlContentError, loginRecoverPasswordHtmlContentErrors, loginRecoverPasswordHtmlContentResponse, loginRecoverPasswordHtmlContentResponses, loginRecoverPasswordResponse, loginRecoverPasswordResponses, loginResetPasswordData, loginResetPasswordError, loginResetPasswordErrors, loginResetPasswordResponse, loginResetPasswordResponses, loginTestTokenData, loginTestTokenResponse, loginTestTokenResponses, Message, metricsMetricsData, metricsMetricsResponses, MoveSubmission, NewPassword, PermissionsPublic, privateCreateUserData, privateCreateUserError, privateCreateUserErrors, privateCreateUserResponse, privateCreateUserResponses, PrivateUserCreate, RoleAssignment, RolePublic, rolesGrantUserRoleData, rolesGrantUserRoleError, rolesGrantUserRoleErrors, rolesGrantUserRoleResponse, rolesGrantUserRoleResponses, rolesReadPermissionsData, rolesReadPermissionsResponse, rolesReadPermissionsResponses, rolesReadRolesData, rolesReadRolesResponse, rolesReadRolesResponses, rolesReadUserRolesData, rolesReadUserRolesError, rolesReadUserRolesErrors, rolesReadUserRolesResponse, rolesReadUserRolesResponses, rolesRevokeUserRoleData, rolesRevokeUserRoleError, rolesRevokeUserRoleErrors, rolesRevokeUserRoleResponse, rolesRevokeUserRoleResponses, RoomCreate, RoomJoinRequest, RoomPublic, roomsCreateRoomData, roomsCreateRoomError, roomsCreateRoomErrors, roomsCreateRoomResponse, roomsCreateRoomResponses, RoomSeatJoinPublic, RoomSeatPublic, RoomSeatSpec, roomsForceStartRoomData, roomsForceStartRoomError, roomsForceStartRoomErrors, roomsForceStartRoomResponse, roomsForceStartRoomResponses, roomsGetRoomData, roomsGetRoomError, roomsGetRoomErrors, roomsGetRoomResponse, roomsGetRoomResponses, roomsJoinRoomData, roomsJoinRoomError, roomsJoinRoomErrors, roomsJoinRoomResponse, roomsJoinRoomResponses, roomsLeaveRoomData, roomsLeaveRoomError, roomsLeaveRoomErrors, roomsLeaveRoomResponse, roomsLeaveRoomResponses, roomsListRoomsData, roomsListRoomsError, roomsListRoomsErrors, roomsListRoomsResponse, roomsListRoomsResponses, roomsMovesSchemaData, roomsMovesSchemaError, roomsMovesSchemaErrors, roomsMovesSchemaResponses, RoomsPublic, roomsRoomEventsData, roomsRoomEventsError, roomsRoomEventsErrors, roomsRoomEventsResponses, roomsRoomQueueData, roomsRoomQueueError, roomsRoomQueueErrors, roomsRoomQueueResponses, roomsRoomViewData, roomsRoomViewError, roomsRoomViewErrors, roomsRoomViewResponses, roomsSendChatData, roomsSendChatError, roomsSendChatErrors, roomsSendChatResponse, roomsSendChatResponses, roomsStartRoomData, roomsStartRoomError, roomsStartRoomErrors, roomsStartRoomResponse, roomsStartRoomResponses, roomsStopRoomData, roomsStopRoomError, roomsStopRoomErrors, roomsStopRoomResponse, roomsStopRoomResponses, roomsSubmitMoveData, roomsSubmitMoveError, roomsSubmitMoveErrors, roomsSubmitMoveResponse, roomsSubmitMoveResponses, roomsWsTicketData, roomsWsTicketError, roomsWsTicketErrors, roomsWsTicketResponse, roomsWsTicketResponses, RuntimeSettingPublic, RuntimeSettingsPublic, RuntimeSettingUpdate, Token, UpdatePassword, UserCreate, UserPublic, UserRegister, UserRolesPublic, usersCreateUserData, usersCreateUserError, usersCreateUserErrors, usersCreateUserResponse, usersCreateUserResponses, usersDeleteUserData, usersDeleteUserError, usersDeleteUserErrors, usersDeleteUserMeData, usersDeleteUserMeResponse, usersDeleteUserMeResponses, usersDeleteUserResponse, usersDeleteUserResponses, UsersPublic, usersReadUserByIdData, usersReadUserByIdError, usersReadUserByIdErrors, usersReadUserByIdResponse, usersReadUserByIdResponses, usersReadUserMeData, usersReadUserMePermissionsData, usersReadUserMePermissionsResponse, usersReadUserMePermissionsResponses, usersReadUserMeResponse, usersReadUserMeResponses, usersReadUsersData, usersReadUsersError, usersReadUsersErrors, usersReadUsersResponse, usersReadUsersResponses, usersRegisterUserData, usersRegisterUserError, usersRegisterUserErrors, usersRegisterUserResponse, usersRegisterUserResponses, usersUpdatePasswordMeData, usersUpdatePasswordMeError, usersUpdatePasswordMeErrors, usersUpdatePasswordMeResponse, usersUpdatePasswordMeResponses, usersUpdateUserData, usersUpdateUserError, usersUpdateUserErrors, usersUpdateUserMeData, usersUpdateUserMeError, usersUpdateUserMeErrors, usersUpdateUserMeResponse, usersUpdateUserMeResponses, usersUpdateUserResponse, usersUpdateUserResponses, UserUpdate, UserUpdateMe, utilsHealthCheckData, utilsHealthCheckResponse, utilsHealthCheckResponses, utilsTestEmailData, utilsTestEmailError, utilsTestEmailErrors, utilsTestEmailResponse, utilsTestEmailResponses, ValidationError } from './types.gen'; +export type { adminCreateOverrideData, adminCreateOverrideError, adminCreateOverrideErrors, adminCreateOverrideResponse, adminCreateOverrideResponses, adminReadFlagsData, adminReadFlagsResponse, adminReadFlagsResponses, adminReadSettingsData, adminReadSettingsResponse, adminReadSettingsResponses, adminRemoveOverrideData, adminRemoveOverrideError, adminRemoveOverrideErrors, adminRemoveOverrideResponse, adminRemoveOverrideResponses, adminUpdateFlagData, adminUpdateFlagError, adminUpdateFlagErrors, adminUpdateFlagResponse, adminUpdateFlagResponses, adminUpdateSettingData, adminUpdateSettingError, adminUpdateSettingErrors, adminUpdateSettingResponse, adminUpdateSettingResponses, Body_login_login_access_token, ChatSend, ClientOptions, FeatureFlagOverrideCreate, FeatureFlagOverridePublic, FeatureFlagPublic, FeatureFlagUpdate, flagsReadMyFlagsData, flagsReadMyFlagsResponse, flagsReadMyFlagsResponses, GameRecordPublic, gamesGameEventsData, gamesGameEventsError, gamesGameEventsErrors, gamesGameEventsResponses, gamesGetGameRecordData, gamesGetGameRecordError, gamesGetGameRecordErrors, gamesGetGameRecordResponse, gamesGetGameRecordResponses, gamesReplayData, gamesReplayError, gamesReplayErrors, gamesReplayResponses, healthHealthData, healthHealthLlmData, healthHealthLlmResponse, healthHealthLlmResponses, healthHealthResponse, healthHealthResponses, HealthReport, HTTPValidationError, ItemCreate, ItemPublic, itemsCreateItemData, itemsCreateItemError, itemsCreateItemErrors, itemsCreateItemResponse, itemsCreateItemResponses, itemsDeleteItemData, itemsDeleteItemError, itemsDeleteItemErrors, itemsDeleteItemResponse, itemsDeleteItemResponses, ItemsPublic, itemsReadItemData, itemsReadItemError, itemsReadItemErrors, itemsReadItemResponse, itemsReadItemResponses, itemsReadItemsData, itemsReadItemsError, itemsReadItemsErrors, itemsReadItemsResponse, itemsReadItemsResponses, itemsUpdateItemData, itemsUpdateItemError, itemsUpdateItemErrors, itemsUpdateItemResponse, itemsUpdateItemResponses, ItemUpdate, leaderboardAccountsLeaderboardData, leaderboardAccountsLeaderboardError, leaderboardAccountsLeaderboardErrors, leaderboardAccountsLeaderboardResponse, leaderboardAccountsLeaderboardResponses, LeaderboardEntry, leaderboardModelsLeaderboardData, leaderboardModelsLeaderboardError, leaderboardModelsLeaderboardErrors, leaderboardModelsLeaderboardResponse, leaderboardModelsLeaderboardResponses, LLMHealth, loginLoginAccessTokenData, loginLoginAccessTokenError, loginLoginAccessTokenErrors, loginLoginAccessTokenResponse, loginLoginAccessTokenResponses, loginRecoverPasswordData, loginRecoverPasswordError, loginRecoverPasswordErrors, loginRecoverPasswordHtmlContentData, loginRecoverPasswordHtmlContentError, loginRecoverPasswordHtmlContentErrors, loginRecoverPasswordHtmlContentResponse, loginRecoverPasswordHtmlContentResponses, loginRecoverPasswordResponse, loginRecoverPasswordResponses, loginResetPasswordData, loginResetPasswordError, loginResetPasswordErrors, loginResetPasswordResponse, loginResetPasswordResponses, loginTestTokenData, loginTestTokenResponse, loginTestTokenResponses, Message, metricsMetricsData, metricsMetricsResponses, MoveSubmission, NewPassword, PermissionsPublic, privateCreateUserData, privateCreateUserError, privateCreateUserErrors, privateCreateUserResponse, privateCreateUserResponses, PrivateUserCreate, QueueStatus, RoleAssignment, RolePublic, rolesGrantUserRoleData, rolesGrantUserRoleError, rolesGrantUserRoleErrors, rolesGrantUserRoleResponse, rolesGrantUserRoleResponses, rolesReadPermissionsData, rolesReadPermissionsResponse, rolesReadPermissionsResponses, rolesReadRolesData, rolesReadRolesResponse, rolesReadRolesResponses, rolesReadUserRolesData, rolesReadUserRolesError, rolesReadUserRolesErrors, rolesReadUserRolesResponse, rolesReadUserRolesResponses, rolesRevokeUserRoleData, rolesRevokeUserRoleError, rolesRevokeUserRoleErrors, rolesRevokeUserRoleResponse, rolesRevokeUserRoleResponses, RoomCreate, RoomJoinRequest, RoomPublic, roomsCreateRoomData, roomsCreateRoomError, roomsCreateRoomErrors, roomsCreateRoomResponse, roomsCreateRoomResponses, RoomSeatJoinPublic, RoomSeatPublic, RoomSeatSpec, roomsForceStartRoomData, roomsForceStartRoomError, roomsForceStartRoomErrors, roomsForceStartRoomResponse, roomsForceStartRoomResponses, roomsGetRoomData, roomsGetRoomError, roomsGetRoomErrors, roomsGetRoomResponse, roomsGetRoomResponses, roomsJoinRoomData, roomsJoinRoomError, roomsJoinRoomErrors, roomsJoinRoomResponse, roomsJoinRoomResponses, roomsLeaveRoomData, roomsLeaveRoomError, roomsLeaveRoomErrors, roomsLeaveRoomResponse, roomsLeaveRoomResponses, roomsListRoomsData, roomsListRoomsError, roomsListRoomsErrors, roomsListRoomsResponse, roomsListRoomsResponses, roomsMovesSchemaData, roomsMovesSchemaError, roomsMovesSchemaErrors, roomsMovesSchemaResponses, RoomsPublic, roomsRoomEventsData, roomsRoomEventsError, roomsRoomEventsErrors, roomsRoomEventsResponses, roomsRoomQueueData, roomsRoomQueueError, roomsRoomQueueErrors, roomsRoomQueueResponse, roomsRoomQueueResponses, roomsRoomViewData, roomsRoomViewError, roomsRoomViewErrors, roomsRoomViewResponses, roomsSendChatData, roomsSendChatError, roomsSendChatErrors, roomsSendChatResponse, roomsSendChatResponses, roomsStartRoomData, roomsStartRoomError, roomsStartRoomErrors, roomsStartRoomResponse, roomsStartRoomResponses, roomsStopRoomData, roomsStopRoomError, roomsStopRoomErrors, roomsStopRoomResponse, roomsStopRoomResponses, roomsSubmitMoveData, roomsSubmitMoveError, roomsSubmitMoveErrors, roomsSubmitMoveResponse, roomsSubmitMoveResponses, roomsWsTicketData, roomsWsTicketError, roomsWsTicketErrors, roomsWsTicketResponse, roomsWsTicketResponses, RuntimeSettingPublic, RuntimeSettingsPublic, RuntimeSettingUpdate, Token, UpdatePassword, UserCreate, UserPublic, UserRegister, UserRolesPublic, usersCreateUserData, usersCreateUserError, usersCreateUserErrors, usersCreateUserResponse, usersCreateUserResponses, usersDeleteUserData, usersDeleteUserError, usersDeleteUserErrors, usersDeleteUserMeData, usersDeleteUserMeResponse, usersDeleteUserMeResponses, usersDeleteUserResponse, usersDeleteUserResponses, UsersPublic, usersReadUserByIdData, usersReadUserByIdError, usersReadUserByIdErrors, usersReadUserByIdResponse, usersReadUserByIdResponses, usersReadUserMeData, usersReadUserMePermissionsData, usersReadUserMePermissionsResponse, usersReadUserMePermissionsResponses, usersReadUserMeResponse, usersReadUserMeResponses, usersReadUsersData, usersReadUsersError, usersReadUsersErrors, usersReadUsersResponse, usersReadUsersResponses, usersRegisterUserData, usersRegisterUserError, usersRegisterUserErrors, usersRegisterUserResponse, usersRegisterUserResponses, usersUpdatePasswordMeData, usersUpdatePasswordMeError, usersUpdatePasswordMeErrors, usersUpdatePasswordMeResponse, usersUpdatePasswordMeResponses, usersUpdateUserData, usersUpdateUserError, usersUpdateUserErrors, usersUpdateUserMeData, usersUpdateUserMeError, usersUpdateUserMeErrors, usersUpdateUserMeResponse, usersUpdateUserMeResponses, usersUpdateUserResponse, usersUpdateUserResponses, UserUpdate, UserUpdateMe, utilsHealthCheckData, utilsHealthCheckResponse, utilsHealthCheckResponses, utilsTestEmailData, utilsTestEmailError, utilsTestEmailErrors, utilsTestEmailResponse, utilsTestEmailResponses, ValidationError } from './types.gen'; diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 566a4eb..b0dd54d 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -419,6 +419,32 @@ export type PrivateUserCreate = { is_verified?: boolean; }; +/** + * QueueStatus + */ +export type QueueStatus = { + /** + * Position + */ + position: number | null; + /** + * Length + */ + length: number; + /** + * Eta Seconds + */ + eta_seconds: number | null; + /** + * Offered + */ + offered?: boolean; + /** + * Offer Expires In Seconds + */ + offer_expires_in_seconds?: number | null; +}; + /** * RoleAssignment */ @@ -2086,7 +2112,12 @@ export type roomsRoomQueueData = { */ room_id: string; }; - query?: never; + query?: { + /** + * Scope + */ + scope?: 'seat' | 'global'; + }; url: '/api/v1/rooms/{room_id}/queue'; }; @@ -2101,13 +2132,13 @@ export type roomsRoomQueueError = roomsRoomQueueErrors[keyof roomsRoomQueueError export type roomsRoomQueueResponses = { /** - * Response Rooms-Room Queue - * * Successful Response */ - 200: unknown; + 200: QueueStatus; }; +export type roomsRoomQueueResponse = roomsRoomQueueResponses[keyof roomsRoomQueueResponses]; + export type roomsRoomViewData = { body?: never; path: { diff --git a/frontend/src/components/Rooms/JoinByCode.tsx b/frontend/src/components/Rooms/JoinByCode.tsx index 3a17723..fe2ca24 100644 --- a/frontend/src/components/Rooms/JoinByCode.tsx +++ b/frontend/src/components/Rooms/JoinByCode.tsx @@ -1,51 +1,148 @@ -import { useMutation, useQueryClient } from "@tanstack/react-query" +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" import { useNavigate } from "@tanstack/react-router" -import { useState } from "react" +import { AxiosError } from "axios" +import { useEffect, useState } from "react" import { RoomsService } from "@/client" +import { Badge } from "@/components/ui/badge" import { Button } from "@/components/ui/button" import { Input } from "@/components/ui/input" import useCustomToast from "@/hooks/useCustomToast" import { handleError } from "@/utils" +function formatEta(seconds: number | null | undefined): string | null { + if (seconds === null || seconds === undefined) return null + const minutes = Math.ceil(seconds / 60) + return minutes <= 1 ? "~1 min" : `~${minutes} min` +} + export function JoinByCode() { const [code, setCode] = useState("") + const [queued, setQueued] = useState<{ roomId: string; code: string } | null>( + null, + ) const navigate = useNavigate() const queryClient = useQueryClient() const { showErrorToast } = useCustomToast() - const mutation = useMutation({ + const joinMutation = useMutation({ mutationFn: (joinCode: string) => RoomsService.joinRoom({ body: { code: joinCode } }), onSuccess: (res) => { if (res.data) { + setQueued(null) navigate({ to: "/rooms/$roomId", params: { roomId: res.data.room_id } }) } }, - onError: handleError.bind(showErrorToast), + onError: (err, joinCode) => { + // A 409 with a `room_id` body means "no open seat, you've been + // enqueued" (see `RoomFullError` in app/api/routes/rooms.py) -- that's + // an expected state to show inline, not an error to toast. + if (err instanceof AxiosError && err.response?.status === 409) { + const roomId = (err.response.data as { room_id?: string } | undefined) + ?.room_id + if (roomId) { + setQueued({ roomId, code: joinCode }) + return + } + } + handleError.call(showErrorToast, err) + }, onSettled: () => { queryClient.invalidateQueries({ queryKey: ["rooms"] }) }, }) + const queueQuery = useQuery({ + queryKey: ["room-queue", queued?.roomId], + queryFn: () => + RoomsService.roomQueue({ + path: { room_id: queued?.roomId ?? "" }, + query: { scope: "seat" }, + }), + enabled: queued !== null, + refetchInterval: 3000, + retry: false, + }) + + // A 404 means the queue itself let go of us (dropped after too many + // missed offers) -- nothing left to poll or accept. + useEffect(() => { + if ( + queued && + queueQuery.error instanceof AxiosError && + queueQuery.error.response?.status === 404 + ) { + setQueued(null) + } + }, [queued, queueQuery.error]) + + const status = queueQuery.data?.data + const eta = formatEta(status?.eta_seconds) + return ( -
{ - e.preventDefault() - if (code.trim()) mutation.mutate(code.trim()) - }} - > - setCode(e.target.value)} - className="max-w-40" - /> - -
+
+
{ + e.preventDefault() + if (code.trim()) joinMutation.mutate(code.trim()) + }} + > + setCode(e.target.value)} + className="max-w-40" + disabled={queued !== null} + /> + +
+ + {queued ? ( +
+ {status?.offered ? ( + <> +
+ Seat ready + + A seat opened up + {status.offer_expires_in_seconds != null + ? ` — accept within ${status.offer_expires_in_seconds}s` + : ""} + . + +
+ + + ) : ( +
+ Queued + + Position {status?.position ?? "…"} of {status?.length ?? "…"} + {eta ? ` — ETA ${eta}` : ""} + +
+ )} +
+ ) : null} +
) } diff --git a/frontend/src/routes/_layout/rooms.$roomId.tsx b/frontend/src/routes/_layout/rooms.$roomId.tsx index 936167b..5463220 100644 --- a/frontend/src/routes/_layout/rooms.$roomId.tsx +++ b/frontend/src/routes/_layout/rooms.$roomId.tsx @@ -1,6 +1,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query" import { createFileRoute } from "@tanstack/react-router" -import { useEffect } from "react" +import { AxiosError } from "axios" +import { useEffect, useState } from "react" import { toast } from "sonner" import { RoomsService } from "@/client" @@ -54,13 +55,47 @@ function RoomDetail() { } }, [live.seatsVersion, queryClient, roomId]) + const [awaitingSlot, setAwaitingSlot] = useState(false) + const startMutation = useMutation({ mutationFn: () => RoomsService.startRoom({ path: { room_id: roomId } }), - onError: handleError.bind(showErrorToast), - onSuccess: () => - queryClient.invalidateQueries({ queryKey: ["rooms", roomId] }), + onError: (err) => { + // A 429 here (`CapacityExceededError`, app/rooms/service.py) means + // this room got enqueued on the global concurrent-game cap rather + // than refused outright -- show the wait, don't toast an error. + if (err instanceof AxiosError && err.response?.status === 429) { + setAwaitingSlot(true) + return + } + handleError.call(showErrorToast, err) + }, + onSuccess: () => { + setAwaitingSlot(false) + queryClient.invalidateQueries({ queryKey: ["rooms", roomId] }) + }, }) + const globalQueueQuery = useQuery({ + queryKey: ["room-queue-global", roomId], + queryFn: () => + RoomsService.roomQueue({ + path: { room_id: roomId }, + query: { scope: "global" }, + }), + enabled: awaitingSlot, + refetchInterval: 5000, + retry: false, + }) + + // The room itself flips to "in_progress" (via WS, `live.roomStatus`) once + // `_promote_next_queued_room` actually starts it -- that is the real + // signal this wait is over, not a client-side guess. + useEffect(() => { + if (awaitingSlot && live.roomStatus && live.roomStatus !== "waiting") { + setAwaitingSlot(false) + } + }, [awaitingSlot, live.roomStatus]) + const forceStartMutation = useMutation({ mutationFn: () => RoomsService.forceStartRoom({ path: { room_id: roomId } }), @@ -75,6 +110,13 @@ function RoomDetail() { onError: handleError.bind(showErrorToast), }) + const leaveMutation = useMutation({ + mutationFn: () => RoomsService.leaveRoom({ path: { room_id: roomId } }), + onError: handleError.bind(showErrorToast), + onSuccess: () => + queryClient.invalidateQueries({ queryKey: ["rooms", roomId] }), + }) + if (roomQuery.isLoading || !room) { return

Loading room...

} @@ -115,24 +157,54 @@ function RoomDetail() { {seat.persona ? ( {seat.persona} ) : null} + {seat.user_id === currentUser?.id ? ( + + ) : null} ))} {isPrivileged ? ( -
- - +
+
+ + +
+ {awaitingSlot ? ( +
+ Waiting for a game slot + + Position {globalQueueQuery.data?.data.position ?? "…"} of{" "} + {globalQueueQuery.data?.data.length ?? "…"} rooms in line + {globalQueueQuery.data?.data.eta_seconds != null + ? ` — ETA ~${Math.ceil( + globalQueueQuery.data.data.eta_seconds / 60, + )} min` + : ""} + +
+ ) : null}
) : null}
diff --git a/frontend/tests/rooms.spec.ts b/frontend/tests/rooms.spec.ts index 86681ae..91b0486 100644 --- a/frontend/tests/rooms.spec.ts +++ b/frontend/tests/rooms.spec.ts @@ -103,6 +103,67 @@ test.describe("Create and play a room", () => { await joinerContext.close() }) + test("A third joiner is queued, then accepts a seat once someone leaves", async ({ + browser, + }) => { + const creatorEmail = randomEmail() + const creatorPassword = randomPassword() + const joinerEmail = randomEmail() + const joinerPassword = randomPassword() + const latecomerEmail = randomEmail() + const latecomerPassword = randomPassword() + await createUser({ email: creatorEmail, password: creatorPassword }) + await createUser({ email: joinerEmail, password: joinerPassword }) + await createUser({ email: latecomerEmail, password: latecomerPassword }) + + const creatorContext = await browser.newContext() + const joinerContext = await browser.newContext() + const latecomerContext = await browser.newContext() + const creatorPage = await creatorContext.newPage() + const joinerPage = await joinerContext.newPage() + const latecomerPage = await latecomerContext.newPage() + + await logInUser(creatorPage, creatorEmail, creatorPassword) + await logInUser(joinerPage, joinerEmail, joinerPassword) + await logInUser(latecomerPage, latecomerEmail, latecomerPassword) + + await creatorPage.getByRole("button", { name: "Create room" }).click() + await creatorPage + .getByRole("button", { name: "Create", exact: true }) + .click() + + const codeText = await creatorPage.getByText(/^Code /).textContent() + const code = codeText?.replace("Code ", "").split(" ")[0] ?? "" + expect(code).not.toBe("") + + // Fill both seats. + await joinerPage.goto("/") + await joinerPage.getByPlaceholder("Join code").fill(code) + await joinerPage.getByRole("button", { name: "Join" }).click() + await joinerPage.waitForURL(/\/rooms\/.+/) + + // A third joiner hits a full room -- enqueued, not rejected outright. + await latecomerPage.goto("/") + await latecomerPage.getByPlaceholder("Join code").fill(code) + await latecomerPage.getByRole("button", { name: "Join" }).click() + await expect(latecomerPage.getByText("Queued")).toBeVisible() + await expect(latecomerPage.getByText(/Position 1 of 1/)).toBeVisible() + + // The creator leaves their own seat, freeing it -- the join queue offers + // it straight to the latecomer waiting in line. + await creatorPage.getByRole("button", { name: "Leave seat" }).click() + + await expect(latecomerPage.getByText("Seat ready")).toBeVisible({ + timeout: 10_000, + }) + await latecomerPage.getByRole("button", { name: "Accept seat" }).click() + await latecomerPage.waitForURL(/\/rooms\/.+/) + + await creatorContext.close() + await joinerContext.close() + await latecomerContext.close() + }) + test("Two human players can exchange chat, seen live on both pages", async ({ browser, }) => {