From be133191f55c9024496da12ec4709a26643960b4 Mon Sep 17 00:00:00 2001 From: Conner Ruhl Date: Sun, 19 Jul 2026 17:54:50 -0700 Subject: [PATCH] feat(server): start the session countdown on an explicit session_started signal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit By default a user's turn is counted from claim(), so any client-side loading after the claim (asset download, world build, model warm-up) is spent out of the user's play time — the countdown can run low, or the time_warning can fire, while the user is still on a loading screen. The claim itself can't be deferred: it's what creates the session the client needs in order to connect and load. Add an opt-in startTimerOnSessionStart server option (env RQ_START_TIMER_ON_SESSION_START, default false) and a session_started client message (ReactorQueueClient.sessionStarted()). When enabled, claim still creates the session but the member gets a short loading deadline (admissionGraceMs) so an abandoned claim frees its slot; the full sessionDurationMs countdown starts only when the client reports playback. A timerStarted member flag gates the time_warning during loading and makes a duplicate session_started a no-op so a client cannot extend its turn. Default off preserves today's behavior exactly. Co-authored-by: Cursor --- README.md | 102 +++++++++++++++++++++++----- packages/queue/src/client.ts | 12 ++++ packages/queue/src/protocol.ts | 16 ++++- packages/queue/src/server/config.ts | 22 ++++++ packages/queue/src/server/server.ts | 51 +++++++++++++- packages/queue/test/config.test.ts | 13 ++++ packages/queue/test/server.test.ts | 77 +++++++++++++++++++++ 7 files changed, 271 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 0e584c7..6b2b10c 100644 --- a/README.md +++ b/README.md @@ -420,24 +420,25 @@ also reads from an env var that **overrides** the code value, so a deploy can be retuned without a code change. Values resolve **default → `createReactorQueueServer({...})` → env var** (env wins). The API key must come from a secret. -| Env var | Config key | Default | Purpose | -| -------------------------------- | --------------------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------ | -| `RQ_REACTOR_API_KEY` | `apiKey` | — (**required**, secret) | Mints JWTs; creates/stops sessions | -| `RQ_MODEL` | `model` | — (**required**) | Model for `POST /sessions` | -| `RQ_MAX_SESSIONS` | `maxSessions` | `1` | Concurrent Reactor sessions (GPU ceiling) | -| `RQ_USERS_PER_SESSION` | `usersPerSession` | `1` | Members per session (fill-in before create) | -| `RQ_WEBRTC_VERSION` | `webrtcVersion` | `1.0` | WebRTC version in session create body | -| `RQ_SESSION_DURATION_MS` | `sessionDurationMs` | `120000` | Session budget after claim | -| `RQ_ADMISSION_GRACE_MS` | `admissionGraceMs` | `45000` | Time to claim a reserved slot | -| `RQ_WARNING_BEFORE_MS` | `warningBeforeMs` | `30000` | Lead time for `time_warning` | -| `RQ_TOKEN_TTL_SECONDS` | `tokenTtlSeconds` | `60` | Minted JWT lifetime floor (scoped member tokens always cover grace + session) | -| `RQ_POLL_INTERVAL_MS` | `pollIntervalMs` | `15000` | Session reconciliation cadence | -| `RQ_COORDINATOR_URL` | `coordinatorUrl` | `https://api.reactor.inc` | Reactor API base URL | -| `RQ_API_VERSION` | `apiVersion` | `1` | `Reactor-API-Version` header | -| `RQ_STOP_SESSIONS` | `stopSessionsOnExpiry` | `true` | `DELETE` session on expiry | -| `RQ_ADMIN_PASSWORD` | `adminPassword` | — (off) | Password for admin dashboard connections | -| `RQ_ALLOW_DUPLICATE_CONNECTIONS` | `allowDuplicateConnections` | `false` | Allow the same browser to hold multiple connections (disables the duplicate-tab `rejected`) | -| `RQ_ALLOWED_ORIGINS` | `allowedOrigins` | — (allow all) | Comma-separated `Origin` allow-list for WebSocket connections; unset accepts any origin, `*` allows all explicitly | +| Env var | Config key | Default | Purpose | +| --------------------------------- | --------------------------- | ------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `RQ_REACTOR_API_KEY` | `apiKey` | — (**required**, secret) | Mints JWTs; creates/stops sessions | +| `RQ_MODEL` | `model` | — (**required**) | Model for `POST /sessions` | +| `RQ_MAX_SESSIONS` | `maxSessions` | `1` | Concurrent Reactor sessions (GPU ceiling) | +| `RQ_USERS_PER_SESSION` | `usersPerSession` | `1` | Members per session (fill-in before create) | +| `RQ_WEBRTC_VERSION` | `webrtcVersion` | `1.0` | WebRTC version in session create body | +| `RQ_SESSION_DURATION_MS` | `sessionDurationMs` | `120000` | Session budget after claim | +| `RQ_ADMISSION_GRACE_MS` | `admissionGraceMs` | `45000` | Time to claim a reserved slot | +| `RQ_WARNING_BEFORE_MS` | `warningBeforeMs` | `30000` | Lead time for `time_warning` | +| `RQ_TOKEN_TTL_SECONDS` | `tokenTtlSeconds` | `60` | Minted JWT lifetime floor (scoped member tokens always cover grace + session) | +| `RQ_POLL_INTERVAL_MS` | `pollIntervalMs` | `15000` | Session reconciliation cadence | +| `RQ_COORDINATOR_URL` | `coordinatorUrl` | `https://api.reactor.inc` | Reactor API base URL | +| `RQ_API_VERSION` | `apiVersion` | `1` | `Reactor-API-Version` header | +| `RQ_STOP_SESSIONS` | `stopSessionsOnExpiry` | `true` | `DELETE` session on expiry | +| `RQ_START_TIMER_ON_SESSION_START` | `startTimerOnSessionStart` | `false` | Start each turn's countdown when the client reports playback (`session_started`) instead of at claim — see [Not counting load time](#not-counting-load-time-against-a-turn) | +| `RQ_ADMIN_PASSWORD` | `adminPassword` | — (off) | Password for admin dashboard connections | +| `RQ_ALLOW_DUPLICATE_CONNECTIONS` | `allowDuplicateConnections` | `false` | Allow the same browser to hold multiple connections (disables the duplicate-tab `rejected`) | +| `RQ_ALLOWED_ORIGINS` | `allowedOrigins` | — (allow all) | Comma-separated `Origin` allow-list for WebSocket connections; unset accepts any origin, `*` allows all explicitly | `acquireSession` / `releaseSession` are code-only overrides (functions, not env) — see [Overriding session lifecycle](#overriding-session-lifecycle). @@ -557,6 +558,71 @@ account's concurrent-session quota is lower than the queue's ceiling — the adm log shows it plainly (`status: 403`, the quota message in `body`). The browser client only gets a generic `error` with `session_create_failed`. +## Not counting load time against a turn + +By default a user's turn is `sessionDurationMs` long counted **from `claim()`** — +the moment the queue creates their Reactor session. For most demos that's fine. +But if your app does noticeable work _after_ claiming and _before_ anything plays +— downloading assets, generating or building a world, warming up a model — that +loading time is spent out of the user's turn. In the worst case the countdown can +run low, or the "time's almost up" warning can fire, while the user is still +staring at a loading screen. + +You can't just delay the `claim()` to fix this: the claim is what creates the +session the client needs in order to connect and start loading in the first place. + +Set `startTimerOnSessionStart` (env `RQ_START_TIMER_ON_SESSION_START`) to move the +start of the countdown from claim to the moment playback actually begins: + +```ts +export default createReactorQueueServer({ + model: "helios", + sessionDurationMs: 120_000, + startTimerOnSessionStart: true, // countdown starts on session_started, not claim +}); +``` + +With it on: + +1. **Claim** still creates the session immediately, so the client can connect and + load — but instead of the full `sessionDurationMs`, the member gets a short + deadline of `admissionGraceMs`. This is a safety cap: a client that claims and + then never starts playing (a stall, a closed tab mid-load) still frees its slot + for the next person instead of holding a GPU for the whole turn. +2. **When playback begins**, your client calls `sessionStarted()`. The server then + starts the real `sessionDurationMs` countdown from that instant, and the + `time_warning` is measured against it — so no warning fires during loading. + +Call `sessionStarted()` at the point that means "the user's turn has really +begun" for your app — typically the first rendered video frame: + +```ts +// vanilla client +queue.claim(); +const { sessionId, connectionId } = queue.getState(); +await reactor.connect(queue.getJwt, { sessionId: sessionId!, connectionId: connectionId! }); +// …once your app renders the first frame / playback is live: +queue.sessionStarted(); +``` + +```tsx +// React: reach the underlying client via the escape-hatch hook +import { useReactorQueueClient } from "@reactor-team/queue/react"; + +const client = useReactorQueueClient(); +// in your "first frame" / "playback started" handler: +client.sessionStarted(); +``` + +Calling `sessionStarted()` more than once is harmless — only the first call after +a claim starts the countdown, so a client can't use it to extend its own turn. On +a server that doesn't have the option enabled, the call is simply ignored. + +> **Sizing note.** While loading, the member's deadline is `admissionGraceMs` — +> the same value used for the claim-grace window. If your load can legitimately +> take longer than that grace window, raise `admissionGraceMs` (or it will expire +> mid-load), since the two share one value today. + ## Overriding session lifecycle How a session is obtained on `claim()`, and what happens when a user leaves, are diff --git a/packages/queue/src/client.ts b/packages/queue/src/client.ts index 43edc8e..e82aaed 100644 --- a/packages/queue/src/client.ts +++ b/packages/queue/src/client.ts @@ -165,6 +165,18 @@ export class ReactorQueueClient { } } + /** + * Tell the server that playback has begun. When the server runs with + * `startTimerOnSessionStart`, the user's `sessionDurationMs` countdown + * starts on this call instead of at `claim()` — call it the moment your app + * actually starts streaming (e.g. on the first video frame), so time spent + * loading doesn't count against the user's turn. A no-op on servers without + * the option. + */ + sessionStarted(): void { + this.send({ type: "session_started" }); + } + /** * The Reactor session ended client-side (e.g. the user quit the turn): free * the slot so the queue slides, and return to `idle` so the app can show its diff --git a/packages/queue/src/protocol.ts b/packages/queue/src/protocol.ts index 0c02d6e..4d91638 100644 --- a/packages/queue/src/protocol.ts +++ b/packages/queue/src/protocol.ts @@ -166,12 +166,26 @@ export interface SessionEndedMessage { type: "session_ended"; } +/** + * Playback has begun on the client. Only meaningful when the server runs with + * `startTimerOnSessionStart`: the full `sessionDurationMs` countdown starts + * when this message arrives instead of at claim. Ignored otherwise. + */ +export interface SessionStartedMessage { + type: "session_started"; +} + /** Leave the queue / release the slot without intending to rejoin. */ export interface LeaveMessage { type: "leave"; } -export type ClientMessage = ClaimMessage | RequestTokenMessage | SessionEndedMessage | LeaveMessage; +export type ClientMessage = + | ClaimMessage + | RequestTokenMessage + | SessionEndedMessage + | SessionStartedMessage + | LeaveMessage; // ───────────────────────────────────────────────────────────────────────────── // Admin mode (server → admin client) diff --git a/packages/queue/src/server/config.ts b/packages/queue/src/server/config.ts index 25d089d..5b48634 100644 --- a/packages/queue/src/server/config.ts +++ b/packages/queue/src/server/config.ts @@ -86,6 +86,24 @@ export interface ReactorQueueServerConfig { */ allowDuplicateConnections?: boolean; + /** + * Start each user's session countdown when their client reports that + * playback has begun (the `session_started` message), instead of at + * `claim()`. + * + * By default the `sessionDurationMs` countdown starts the moment a claim + * succeeds, so any client-side loading after the claim (downloading assets, + * building a world, model warm-up) eats into the user's play time. With this + * option on, `claim()` still creates the session — the client needs it to + * connect — but the countdown waits until the client calls + * `ReactorQueueClient.sessionStarted()`. Until then the member has a short + * deadline of `admissionGraceMs`, so a client that claims but never starts + * playing cannot hold the slot indefinitely. + * + * Default false. Env: `RQ_START_TIMER_ON_SESSION_START`. + */ + startTimerOnSessionStart?: boolean; + /** * Cross-origin allow-list for incoming WebSocket connections. Each entry is an * exact `Origin` header value (scheme + host + optional port, no path), e.g. @@ -156,6 +174,8 @@ export interface ResolvedConfig { adminPassword: string | null; /** When true, the duplicate-tab (same `clientId`) rejection is disabled. */ allowDuplicateConnections: boolean; + /** When true, the session countdown starts on `session_started` rather than at claim. */ + startTimerOnSessionStart: boolean; /** Allowed `Origin` values for WebSocket connections. Empty = allow all. `["*"]` = allow all explicitly. */ allowedOrigins: string[]; /** Custom session acquisition, or null to create via the Reactor API. */ @@ -281,6 +301,8 @@ export function resolveConfig(config: ReactorQueueServerConfig, env: Env): Resol adminPassword: envStr(env, "RQ_ADMIN_PASSWORD") ?? config.adminPassword ?? null, allowDuplicateConnections: envBool(env, "RQ_ALLOW_DUPLICATE_CONNECTIONS") ?? config.allowDuplicateConnections ?? false, + startTimerOnSessionStart: + envBool(env, "RQ_START_TIMER_ON_SESSION_START") ?? config.startTimerOnSessionStart ?? false, allowedOrigins: envList(env, "RQ_ALLOWED_ORIGINS") ?? config.allowedOrigins ?? [], acquireSession: config.acquireSession ?? null, releaseSession: config.releaseSession ?? null, diff --git a/packages/queue/src/server/server.ts b/packages/queue/src/server/server.ts index 3b60525..a8228a8 100644 --- a/packages/queue/src/server/server.ts +++ b/packages/queue/src/server/server.ts @@ -57,6 +57,13 @@ interface MemberData { expiresAt: number; warned: boolean; claimed: boolean; + /** + * True once the full `sessionDurationMs` countdown is running. Set at claim, + * unless `startTimerOnSessionStart` is on — then it stays false (and + * `expiresAt` is a short loading deadline) until the client sends + * `session_started`. + */ + timerStarted: boolean; /** True while a claim is mid-flight (session being acquired) to reject duplicates. */ claiming?: boolean; } @@ -823,6 +830,30 @@ export function createReactorQueueServer( break; } + case "session_started": { + // Start the full session countdown from now. Ignored before claim + // and once the timer is already running, so a client cannot send + // this repeatedly to extend its turn. + const member = await this.getMember(sender.id); + if (member && member.claimed && !member.timerStarted) { + const expiresAt = Date.now() + this.config.sessionDurationMs; + await this.setMember(sender.id, { + ...member, + expiresAt, + warned: false, + timerStarted: true, + }); + await this.scheduleNextAlarm(); + this.log( + "info", + "session_timer_started", + "Client reported playback start; session countdown running", + { connId: sender.id, data: { expiresAt } } + ); + } + break; + } + case "request_token": { const member = await this.getMember(sender.id); if (member) await this.sendMemberToken(sender); @@ -878,7 +909,14 @@ export function createReactorQueueServer( continue; } - if (data.claimed && !data.warned && now >= data.expiresAt - this.config.warningBeforeMs) { + // No time_warning while the timer hasn't started (e.g. still loading + // under startTimerOnSessionStart) — the warning describes play time. + if ( + data.claimed && + data.timerStarted && + !data.warned && + now >= data.expiresAt - this.config.warningBeforeMs + ) { this.sendTo(connId, { type: "time_warning", secondsLeft: Math.round((data.expiresAt - now) / 1000), @@ -948,7 +986,7 @@ export function createReactorQueueServer( for (const [, data] of members) { earliest = Math.min(earliest, data.expiresAt); - if (data.claimed && !data.warned) { + if (data.claimed && data.timerStarted && !data.warned) { earliest = Math.min(earliest, data.expiresAt - this.config.warningBeforeMs); } } @@ -1012,6 +1050,7 @@ export function createReactorQueueServer( expiresAt: Date.now() + this.config.admissionGraceMs, warned: false, claimed: false, + timerStarted: false, }); const activeCount = await this.getActiveMemberCount(); @@ -1072,7 +1111,12 @@ export function createReactorQueueServer( } const { slotId, sessionId, connectionId } = placement; - const expiresAt = Date.now() + this.config.sessionDurationMs; + // Under startTimerOnSessionStart the countdown waits for the client's + // `session_started`; until then the member gets a short loading deadline + // (admissionGraceMs) so an abandoned claim still frees the slot. + const deferTimer = this.config.startTimerOnSessionStart; + const expiresAt = + Date.now() + (deferTimer ? this.config.admissionGraceMs : this.config.sessionDurationMs); await this.setMember(connId, { ...member, slotId, @@ -1082,6 +1126,7 @@ export function createReactorQueueServer( warned: false, claimed: true, claiming: false, + timerStarted: !deferTimer, }); // The claim may have seated the member on a different slot than the one diff --git a/packages/queue/test/config.test.ts b/packages/queue/test/config.test.ts index 64baa79..f398414 100644 --- a/packages/queue/test/config.test.ts +++ b/packages/queue/test/config.test.ts @@ -35,6 +35,7 @@ describe("resolveConfig — defaults", () => { expect(c.coordinatorUrl).toBe("https://api.reactor.inc"); expect(c.stopSessionsOnExpiry).toBe(true); expect(c.allowDuplicateConnections).toBe(false); + expect(c.startTimerOnSessionStart).toBe(false); expect(c.adminPassword).toBeNull(); expect(c.allowedOrigins).toEqual([]); expect(c.acquireSession).toBeNull(); @@ -103,6 +104,18 @@ describe("resolveConfig — booleans", () => { const c = resolveConfig({ ...base, stopSessionsOnExpiry: true }, { RQ_STOP_SESSIONS: "false" }); expect(c.stopSessionsOnExpiry).toBe(false); }); + + it("resolves startTimerOnSessionStart from config and env", () => { + expect( + resolveConfig({ ...base, startTimerOnSessionStart: true }, {}).startTimerOnSessionStart + ).toBe(true); + expect( + resolveConfig( + { ...base, startTimerOnSessionStart: true }, + { RQ_START_TIMER_ON_SESSION_START: "false" } + ).startTimerOnSessionStart + ).toBe(false); + }); }); describe("resolveConfig — allowedOrigins list", () => { diff --git a/packages/queue/test/server.test.ts b/packages/queue/test/server.test.ts index 7920e0d..9a74766 100644 --- a/packages/queue/test/server.test.ts +++ b/packages/queue/test/server.test.ts @@ -210,6 +210,83 @@ describe("alarm: grace, expiry, warning, liveness, and reconciliation", () => { }); }); +describe("startTimerOnSessionStart: countdown starts on session_started, not at claim", () => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const member = async (h: Harness) => (await h.room.storage.get("member:a")) as any; + + it("gives a claimed member a short loading deadline until session_started starts the countdown", async () => { + const h = makeHarness({ + maxSessions: 1, + startTimerOnSessionStart: true, + admissionGraceMs: 50, + sessionDurationMs: 100_000, + pollIntervalMs: 1_000_000, + }); + const a = await h.join("a"); + + const before = Date.now(); + await h.send(a, { type: "claim" }); + const claimed = await member(h); + // The session exists (the client can connect) but the countdown has not + // started: the deadline is the short loading deadline, not the full duration. + expect(a.types()).toContain("session_ready"); + expect(claimed.timerStarted).toBe(false); + expect(claimed.expiresAt).toBeLessThan(before + 10_000); + + await h.send(a, { type: "session_started" }); + const started = await member(h); + expect(started.timerStarted).toBe(true); + expect(started.expiresAt).toBeGreaterThan(before + 50_000); + }); + + it("expires at the loading deadline when the client never reports playback", async () => { + const h = makeHarness({ + maxSessions: 1, + startTimerOnSessionStart: true, + admissionGraceMs: 10, + sessionDurationMs: 1_000_000, + pollIntervalMs: 1_000_000, + }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); + await sleep(25); + await h.server.onAlarm(); + expect(a.ofType("expired").at(-1)).toMatchObject({ reason: "timeout" }); + }); + + it("ignores a duplicate session_started (cannot extend the turn)", async () => { + const h = makeHarness({ + maxSessions: 1, + startTimerOnSessionStart: true, + admissionGraceMs: 50, + sessionDurationMs: 100_000, + pollIntervalMs: 1_000_000, + }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); + await h.send(a, { type: "session_started" }); + const firstDeadline = (await member(h)).expiresAt; + await sleep(10); + await h.send(a, { type: "session_started" }); + expect((await member(h)).expiresAt).toBe(firstDeadline); + }); + + it("default (off) starts the full countdown at claim, unchanged", async () => { + const h = makeHarness({ + maxSessions: 1, + admissionGraceMs: 50, + sessionDurationMs: 100_000, + pollIntervalMs: 1_000_000, + }); + const a = await h.join("a"); + const before = Date.now(); + await h.send(a, { type: "claim" }); + const claimed = await member(h); + expect(claimed.timerStarted).toBe(true); + expect(claimed.expiresAt).toBeGreaterThan(before + 50_000); + }); +}); + describe("usersPerSession > 1", () => { it("packs members into one session, each with a distinct connection", async () => { const h = makeHarness({ maxSessions: 1, usersPerSession: 2 });