diff --git a/README.md b/README.md index f1d697c..0e584c7 100644 --- a/README.md +++ b/README.md @@ -252,7 +252,7 @@ Anyone building on Reactor who needs to meter live access to a model: ┌─────────────────────┐ WebSocket ┌────────────────────────────┐ REST ┌──────────────────────┐ │ @reactor-team/queue │◀───────────▶│ @reactor-team/queue/server │────────▶│ POST /tokens │ │ • partysocket │ queue + │ • FIFO queue + session cap │ │ POST /sessions │ - │ • getJwt() resolver│ tokens │ • mints 60s Reactor JWTs │ │ GET /sessions/{id} │ + │ • getJwt() resolver│ tokens │ • mints scoped Reactor JWTs│ │ GET /sessions/{id} │ │ • sessionId on claim│ │ • creates sessions on claim│ │ DELETE /sessions/{id}│ │ • zustand store │ │ • per-user session timer │ └──────────────────────┘ └──────────┬──────────┘ │ • stops + reaps sessions │ @@ -273,10 +273,15 @@ The queue server is the single source of truth. It: every connection; the client only adopts them. Creating nothing during grace means an abandoned admission never orphans a GPU session. 3. Mints the Reactor JWT server-side (the API key is a server secret) and sends it - only to admitted users. -4. Issues short-lived tokens (default 60s). The client refreshes them on demand - over the WebSocket via a `request_token` command, exposed as a standard `getJwt` - resolver for the Reactor SDK. + only to admitted users. Each slot gets its own **session-scoped** token + (`authorization_details` on `POST /tokens`): it can create one session for the + configured model and act only on that session — a leaked token exposes nothing + else on the account. The slot's session is created _with_ that token, which is + what binds the two together. +4. Issues each member their slot's token. A scoped token is the session bond and + cannot be refreshed mid-session, so it is minted to outlive the slot (grace + + session budget; `tokenTtlSeconds` only raises that floor). `request_token` / + the `getJwt` resolver re-deliver the stored token. 5. Gives each admitted user a bounded session (default 120s), then calls `DELETE /sessions/{id}` to stop the GPU session when time runs out. 6. Frees a slot the instant a member leaves — via an explicit `session_ended` @@ -425,7 +430,7 @@ retuned without a code change. Values resolve **default → `createReactorQueueS | `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 | +| `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 | @@ -610,6 +615,12 @@ mints the JWT, so a custom-acquired session must belong to the same Reactor account as the queue's API key. If another client shares the session with the queued user, the platform must allow more than one connection per session. +With an `acquireSession` override the queue hands out **unscoped** member +tokens: the externally acquired session is not bound to any token the queue +holds, and a session-scoped token can only act on sessions its own grant +created, so scoping would lock members out. The default session source is what +enables scoped tokens. + ## Configuration (client) `ReactorQueueClientOptions` / `` props: `host` (required), diff --git a/packages/queue/package.json b/packages/queue/package.json index 33c6803..288546f 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -1,6 +1,6 @@ { "name": "@reactor-team/queue", - "version": "0.0.3", + "version": "0.0.4", "description": "Drop-in waiting room for Reactor demos: a browser client (framework-agnostic + React) and a PartyKit server that gate access to a capacity-limited app and admit waiting users in order.", "license": "Apache-2.0", "homepage": "https://github.com/reactor-team/reactor-queue#readme", diff --git a/packages/queue/src/server/config.ts b/packages/queue/src/server/config.ts index 41be8df..25d089d 100644 --- a/packages/queue/src/server/config.ts +++ b/packages/queue/src/server/config.ts @@ -44,7 +44,13 @@ export interface ReactorQueueServerConfig { admissionGraceMs?: number; /** Lead time for the `time_warning`, in ms. Env: `RQ_WARNING_BEFORE_MS`. */ warningBeforeMs?: number; - /** Requested lifetime for each minted Reactor JWT, in seconds. Env: `RQ_TOKEN_TTL_SECONDS`. */ + /** + * Requested lifetime floor for each minted Reactor JWT, in seconds. Member + * tokens are session-scoped and cannot be refreshed mid-session, so they are + * minted to cover at least the admission grace plus the full session budget + * regardless of this value; it only raises that floor. Env: + * `RQ_TOKEN_TTL_SECONDS`. + */ tokenTtlSeconds?: number; /** How often to reconcile tracked sessions with Reactor, in ms. Env: `RQ_POLL_INTERVAL_MS`. */ pollIntervalMs?: number; diff --git a/packages/queue/src/server/coordinator.ts b/packages/queue/src/server/coordinator.ts index 1b1c9a4..3a9e039 100644 --- a/packages/queue/src/server/coordinator.ts +++ b/packages/queue/src/server/coordinator.ts @@ -20,17 +20,34 @@ export class CoordinatorError extends Error { } } +/** + * Session authorization scope for a minted JWT. When passed to + * {@link CoordinatorClient.mintToken}, the JWT is restricted to creating + * sessions for the named model (at most `maxSessions` of them) and to + * operating the sessions it created — nothing else on the account. + */ +export interface TokenScope { + /** Model the token may create sessions for (fully-qualified `org/model`). */ + model: string; + /** How many sessions the token may ever create (spawned, not concurrent). */ + maxSessions: number; +} + /** * Thin server-side client for the Reactor Coordinator REST API. From inside the * trusted PartyKit server it: * * 1. mints short-lived client JWTs from the API key (`POST /tokens`), + * optionally scoped to one model via `authorization_details`, * 2. creates sessions (`POST /sessions`), * 3. reads a session's state (`GET /sessions/{id}/runtime`), and * 4. stops a session (`DELETE /sessions/{id}`). * - * (2)–(4) need a Bearer JWT, so the client keeps its own cached "server JWT" - * (minted with a longer TTL) and reuses it across calls. + * (2)–(4) need a Bearer JWT. By default the client keeps its own cached + * "server JWT" (unscoped, minted with a longer TTL) and reuses it across + * calls; `createSession`/`createConnection` also accept an explicit `jwt` so + * a session can be created *by* a scoped token, binding it to that token's + * grant. */ export class CoordinatorClient { private readonly baseUrl: string; @@ -64,9 +81,16 @@ export class CoordinatorClient { /** * Exchange the API key for a JWT. `ttlSeconds` is passed as `expires_after`; - * the Coordinator caps it at its server maximum. + * the Coordinator caps it at its server maximum. With a `scope`, the JWT + * carries session `authorization_details`: it can only create sessions for + * `scope.model` (at most `scope.maxSessions`) and act on the sessions it + * created. Bound sessions live on the token's server-side grant, so a + * scoped token cannot be re-minted for an existing session. */ - async mintToken(ttlSeconds: number): Promise<{ jwt: string; expiresAt: number }> { + async mintToken( + ttlSeconds: number, + scope?: TokenScope + ): Promise<{ jwt: string; expiresAt: number }> { const res = await fetch(`${this.baseUrl}/tokens`, { method: "POST", headers: { @@ -74,7 +98,20 @@ export class CoordinatorClient { "Reactor-API-Key": this.apiKey, ...this.versionHeaders(), }, - body: JSON.stringify({ expires_after: Math.max(1, Math.floor(ttlSeconds)) }), + body: JSON.stringify({ + expires_after: Math.max(1, Math.floor(ttlSeconds)), + ...(scope + ? { + authorization_details: [ + { + type: "session", + resources: { models: { match: [scope.model] } }, + constraints: { max_sessions: scope.maxSessions }, + }, + ], + } + : {}), + }), }); if (!res.ok) { @@ -131,9 +168,15 @@ export class CoordinatorClient { /** * Create a Reactor session for the configured model. Returns the new * `session_id`. Runs billing/quota checks against the server's API key. + * With `opts.jwt` the session is created by that token — for a scoped + * token, this is the step that binds the session to its grant. */ - async createSession(opts: { model: string; webrtcVersion: string }): Promise { - const jwt = await this.getServerJwt(); + async createSession(opts: { + model: string; + webrtcVersion: string; + jwt?: string; + }): Promise { + const jwt = opts.jwt ?? (await this.getServerJwt()); const res = await fetch(`${this.baseUrl}/sessions`, { method: "POST", headers: { @@ -169,14 +212,15 @@ export class CoordinatorClient { * Register a WebRTC connection under an existing session and return the * server-minted `connection_id`. This is a transport call (carries * `Reactor-WebRTC-Version`, not the API-version headers) and must use a JWT - * for the session's owning user — the server JWT, minted from the same API - * key that created the session, satisfies that. + * allowed to act on the session: pass `opts.jwt` for the scoped token whose + * grant owns the session, or omit it to use the server JWT (same-user + * ownership, minted from the same API key). * * A {@link CoordinatorError} with `status === 429` means the session hit its * `connections_per_session` cap; the caller falls back to another/new session. */ - async createConnection(sessionId: string): Promise { - const jwt = await this.getServerJwt(); + async createConnection(sessionId: string, opts: { jwt?: string } = {}): Promise { + const jwt = opts.jwt ?? (await this.getServerJwt()); const endpoint = `/sessions/${encodeURIComponent(sessionId)}/transport/webrtc/connections`; const res = await fetch(`${this.baseUrl}${endpoint}`, { method: "POST", diff --git a/packages/queue/src/server/server.ts b/packages/queue/src/server/server.ts index c9d7a51..3b60525 100644 --- a/packages/queue/src/server/server.ts +++ b/packages/queue/src/server/server.ts @@ -73,6 +73,18 @@ interface SlotRecord { members: string[]; createdAt: number; lastPollAt?: number; + /** + * The slot's session-scoped JWT: minted once per slot, used to create the + * slot's session (binding the session to the token's grant), and handed to + * every member seated here — it is the only token the Coordinator lets act + * on this session besides same-user unscoped ones. Never re-minted: a fresh + * scoped token would have an empty grant and could not attach. Absent on + * slots created before scoped tokens existed and when `acquireSession` + * sources sessions externally; those slots fall back to unscoped tokens. + */ + jwt?: string | null; + /** Unix seconds when `jwt` expires (from the mint response). */ + jwtExpiresAt?: number | null; } /** Maps a stable browser id → its current connection, for duplicate-tab eviction. */ @@ -137,6 +149,55 @@ export function createReactorQueueServer( return this.api.createSession({ model, webrtcVersion: this.config.webrtcVersion }); } + /** + * Whether members get session-scoped tokens (the default). Requires the + * default session source: an `acquireSession` override creates sessions + * under a grant this server does not hold, which a scoped token could + * never attach to — so those deployments keep unscoped member tokens. + */ + private get scopedTokens(): boolean { + return !this.config.acquireSession; + } + + /** + * TTL for a slot's scoped token. The token is the session bond and cannot + * be refreshed (a re-mint starts a new, empty grant), so it must outlive + * the slot: admission grace + the full session budget + a margin. The + * configured `tokenTtlSeconds` still acts as a floor. + */ + private slotTokenTtlSeconds(): number { + const coverMs = this.config.admissionGraceMs + this.config.sessionDurationMs; + return Math.max(this.config.tokenTtlSeconds, Math.ceil(coverMs / 1000) + 120); + } + + /** + * The slot's scoped token, minted and persisted on first use. Returns null + * when the slot vanished mid-mint, or for a legacy slot whose session + * already exists without a stored grant (created before scoped tokens): + * that session only accepts unscoped same-user tokens, so callers fall + * back to the unscoped mint. + */ + private async ensureSlotToken( + slotId: string + ): Promise<{ jwt: string; expiresAt: number } | null> { + const slot = await this.getSlot(slotId); + if (!slot) return null; + if (slot.jwt) return { jwt: slot.jwt, expiresAt: slot.jwtExpiresAt ?? 0 }; + if (slot.sessionId) return null; + + const minted = await this.api.mintToken(this.slotTokenTtlSeconds(), { + model: this.config.model, + maxSessions: 1, + }); + // Re-read before persisting: the mint is a network call and the input + // gate was open, so the slot may have been mutated or deleted. + const fresh = await this.getSlot(slotId); + if (!fresh) return null; + if (fresh.jwt) return { jwt: fresh.jwt, expiresAt: fresh.jwtExpiresAt ?? 0 }; + await this.setSlot({ ...fresh, jwt: minted.jwt, jwtExpiresAt: minted.expiresAt }); + return minted; + } + /** * A user left a session. Uses the configured `releaseSession` override, or by * default deletes the session via the Reactor API once the last member leaves @@ -764,7 +825,7 @@ export function createReactorQueueServer( case "request_token": { const member = await this.getMember(sender.id); - if (member) await this.mintAndSend(sender); + if (member) await this.sendMemberToken(sender); break; } @@ -969,10 +1030,12 @@ export function createReactorQueueServer( connId, } ); - // Token can be minted now (it's not session-scoped); the SDK uses it - // after claim. getJwt/request_token keep it fresh. + // Hand the member their slot's token right away so the SDK has it + // before claim. In scoped mode this is the token that will create + // (and therefore own) the slot's session; if the claim spills them + // to a different slot, claimMember re-sends that slot's token. const conn = this.room.getConnection(connId); - if (conn) await this.mintAndSend(conn); + if (conn) await this.sendMemberToken(conn); admittedAny = true; } } @@ -1021,6 +1084,23 @@ export function createReactorQueueServer( claiming: false, }); + // The claim may have seated the member on a different slot than the one + // whose token they received at admission (spill / new slot). The attach + // token must be the one whose grant owns the session, so re-send the + // final slot's token before session_ready. Reads the stored token — no + // extra mint — and skips grant-less (legacy/override) sessions, whose + // members keep their unscoped admission token. + if (this.scopedTokens) { + const finalSlot = await this.getSlot(slotId); + if (finalSlot?.jwt) { + this.sendTo(connId, { + type: "token", + jwt: finalSlot.jwt, + expiresAt: finalSlot.jwtExpiresAt ?? 0, + }); + } + } + this.sendTo(connId, { type: "session_ready", sessionId, @@ -1103,9 +1183,24 @@ export function createReactorQueueServer( if (!slot) return null; let sessionId = slot.sessionId; + let slotJwt = slot.jwt ?? null; if (!sessionId) { try { - sessionId = await this.runAcquire(this.config.model); + if (this.scopedTokens) { + // Create the session *with the slot's scoped token* — this is the + // binding step: the Coordinator adds the session to that token's + // grant, making it the token members attach with. + const token = await this.ensureSlotToken(slotId); + if (!token) return null; + slotJwt = token.jwt; + sessionId = await this.api.createSession({ + model: this.config.model, + webrtcVersion: this.config.webrtcVersion, + jwt: slotJwt, + }); + } else { + sessionId = await this.runAcquire(this.config.model); + } } catch (err) { this.reportError("session_create_failed", err, { connId }); return null; @@ -1119,7 +1214,12 @@ export function createReactorQueueServer( for (let attempt = 1; attempt <= CONNECTION_MINT_ATTEMPTS; attempt++) { try { - const connectionId = await this.api.createConnection(sessionId); + // A session created by the slot token must also be operated with it; + // grant-less (legacy/override) sessions use the server JWT as before. + const connectionId = await this.api.createConnection( + sessionId, + slotJwt ? { jwt: slotJwt } : {} + ); return { sessionId, connectionId }; } catch (err) { if (err instanceof CoordinatorError && err.status === 429) { @@ -1187,8 +1287,26 @@ export function createReactorQueueServer( return new Promise((resolve) => setTimeout(resolve, ms)); } - private async mintAndSend(conn: Party.Connection): Promise { + /** + * Send a member the token their slot's session accepts. Scoped mode hands + * out the slot's own token (minting it on first use); a legacy slot with a + * grant-less session, and deployments with an `acquireSession` override, + * get the old unscoped mint instead. + */ + private async sendMemberToken(conn: Party.Connection): Promise { try { + if (this.scopedTokens) { + const member = await this.getMember(conn.id); + if (!member) return; + const token = await this.ensureSlotToken(member.slotId); + if (token) { + this.send(conn, { type: "token", jwt: token.jwt, expiresAt: token.expiresAt }); + return; + } + const slot = await this.getSlot(member.slotId); + if (!slot) return; + // Legacy slot: fall through to the unscoped mint below. + } const { jwt, expiresAt } = await this.api.mintToken(this.config.tokenTtlSeconds); this.send(conn, { type: "token", jwt, expiresAt }); } catch (err) { diff --git a/packages/queue/test/coordinator.test.ts b/packages/queue/test/coordinator.test.ts index 06fd3b7..e670694 100644 --- a/packages/queue/test/coordinator.test.ts +++ b/packages/queue/test/coordinator.test.ts @@ -49,6 +49,31 @@ describe("CoordinatorClient.mintToken", () => { expect(JSON.parse(init.body as string)).toEqual({ expires_after: 60 }); }); + it("carries session authorization_details when a scope is passed", async () => { + const { calls } = stubFetch([{ status: 200, body: { jwt: "scoped", expires_at: 500 } }]); + const res = await new CoordinatorClient(opts).mintToken(300, { + model: "reactor/helios", + maxSessions: 1, + }); + expect(res).toEqual({ jwt: "scoped", expiresAt: 500 }); + const body = JSON.parse(calls[0]!.init.body as string); + expect(body.expires_after).toBe(300); + expect(body.authorization_details).toEqual([ + { + type: "session", + resources: { models: { match: ["reactor/helios"] } }, + constraints: { max_sessions: 1 }, + }, + ]); + }); + + it("omits authorization_details without a scope", async () => { + const { calls } = stubFetch([{ status: 200, body: { jwt: "j", expires_at: 1 } }]); + await new CoordinatorClient(opts).mintToken(60); + const body = JSON.parse(calls[0]!.init.body as string); + expect("authorization_details" in body).toBe(false); + }); + it("falls back to a `token` field and derives expiresAt when absent", async () => { stubFetch([{ status: 200, body: { token: "legacy" } }]); const before = Math.floor(Date.now() / 1000); @@ -152,6 +177,19 @@ describe("CoordinatorClient.createSession", () => { expect(body.supported_transports).toEqual([{ protocol: "webrtc", version: "1.0" }]); }); + it("uses an explicit jwt without minting a server JWT", async () => { + const { calls } = stubFetch([{ status: 201, body: { session_id: "sess-9" } }]); + const id = await new CoordinatorClient(opts).createSession({ + model: "helios", + webrtcVersion: "1.0", + jwt: "slot-jwt", + }); + expect(id).toBe("sess-9"); + expect(calls.some((c) => c.url.endsWith("/tokens"))).toBe(false); + const headers = calls[0]!.init.headers as Record; + expect(headers["Authorization"]).toBe("Bearer slot-jwt"); + }); + it("throws when the response has no session_id", async () => { stubFetch([ { status: 200, body: { jwt: "j", expires_at: Math.floor(Date.now() / 1000) + 600 } }, @@ -189,6 +227,15 @@ describe("CoordinatorClient.createConnection", () => { expect((mint.init.headers as Record)["Reactor-WebRTC-Version"]).toBe("1.0"); }); + it("uses an explicit jwt without minting a server JWT", async () => { + const { calls } = stubFetch([{ status: 201, body: { connection_id: 3 } }]); + const id = await new CoordinatorClient(opts).createConnection("sess-1", { jwt: "slot-jwt" }); + expect(id).toBe(3); + expect(calls.some((c) => c.url.endsWith("/tokens"))).toBe(false); + const headers = calls[0]!.init.headers as Record; + expect(headers["Authorization"]).toBe("Bearer slot-jwt"); + }); + it("raises a 429 CoordinatorError when the session is at its connection cap", async () => { stubFetch([ { status: 200, body: { jwt: "j", expires_at: Math.floor(Date.now() / 1000) + 600 } }, diff --git a/packages/queue/test/helpers/coordinator-fetch.ts b/packages/queue/test/helpers/coordinator-fetch.ts index 36c6cb3..9cf903e 100644 --- a/packages/queue/test/helpers/coordinator-fetch.ts +++ b/packages/queue/test/helpers/coordinator-fetch.ts @@ -27,6 +27,8 @@ export interface CoordinatorMock { failSessionsWith(status: number, body?: string): void; /** Force `POST .../connections` to fail with this status/body until reset. */ failConnectionsWith(status: number, body?: string): void; + /** Fail only the next `n` `POST .../connections` calls, then succeed again. */ + failNextConnections(n: number, status: number, body?: string): void; /** Force `POST /tokens` to fail with this status/body until reset. */ failTokensWith(status: number, body?: string): void; /** State returned by `GET .../runtime` (default "ACTIVE"). */ @@ -72,6 +74,7 @@ export function installCoordinatorFetch(): CoordinatorMock { tokenSeq: 0, sessionFailure: null as Failure | null, connectionFailure: null as Failure | null, + connectionFailuresLeft: 0, tokenFailure: null as Failure | null, }; @@ -103,8 +106,15 @@ export function installCoordinatorFetch(): CoordinatorMock { } if (method === "POST" && url.endsWith("/connections")) { - if (state.connectionFailure) - return textResponse(state.connectionFailure.status, state.connectionFailure.body); + if (state.connectionFailure) { + const failure = state.connectionFailure; + // failuresLeft 0 means "fail forever"; a positive count burns down. + if (state.connectionFailuresLeft > 0) { + state.connectionFailuresLeft -= 1; + if (state.connectionFailuresLeft === 0) state.connectionFailure = null; + } + return textResponse(failure.status, failure.body); + } const id = state.nextConnectionId; state.nextConnectionId += 1; return jsonResponse(201, { connection_id: id }); @@ -135,7 +145,14 @@ export function installCoordinatorFetch(): CoordinatorMock { callsTo: (fragment) => state.calls.filter((c) => c.url.includes(fragment)), countTo: (fragment) => state.calls.filter((c) => c.url.includes(fragment)).length, failSessionsWith: (status, b = "") => (state.sessionFailure = { status, body: b }), - failConnectionsWith: (status, b = "") => (state.connectionFailure = { status, body: b }), + failConnectionsWith: (status, b = "") => { + state.connectionFailure = { status, body: b }; + state.connectionFailuresLeft = 0; + }, + failNextConnections: (n, status, b = "") => { + state.connectionFailure = { status, body: b }; + state.connectionFailuresLeft = n; + }, failTokensWith: (status, b = "") => (state.tokenFailure = { status, body: b }), get runtimeState() { return state.runtimeState; diff --git a/packages/queue/test/server.test.ts b/packages/queue/test/server.test.ts index ba6ab1a..7920e0d 100644 --- a/packages/queue/test/server.test.ts +++ b/packages/queue/test/server.test.ts @@ -240,6 +240,121 @@ describe("usersPerSession > 1", () => { }); }); +describe("session-scoped member tokens", () => { + it("mints the slot token with authorization_details covering grace + session", async () => { + const h = makeHarness({ + maxSessions: 1, + usersPerSession: 1, + admissionGraceMs: 45_000, + sessionDurationMs: 120_000, + tokenTtlSeconds: 60, + }); + await h.join("a"); + const mint = h.api.callsTo("/tokens").at(-1)!; + const body = mint.body as { + expires_after: number; + authorization_details: unknown; + }; + expect(body.authorization_details).toEqual([ + { + type: "session", + resources: { models: { match: ["helios"] } }, + constraints: { max_sessions: 1 }, + }, + ]); + // The slot token is the session bond and cannot be refreshed, so its TTL + // must cover the grace window plus the full session budget. + expect(body.expires_after).toBeGreaterThanOrEqual((45_000 + 120_000) / 1000); + }); + + it("creates the session and connection with the slot token and re-delivers it", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); + const a = await h.join("a"); + const admissionJwt = a.ofType("token").at(-1)!.jwt as string; + await h.send(a, { type: "claim" }); + + const create = h.api.calls.find((c) => c.method === "POST" && c.url.endsWith("/sessions"))!; + expect(create.headers["Authorization"]).toBe(`Bearer ${admissionJwt}`); + const conn = h.api.callsTo("/connections").at(-1)!; + expect(conn.headers["Authorization"]).toBe(`Bearer ${admissionJwt}`); + + // Claim re-sends the slot token (same jwt) and request_token re-delivers + // it without a second mint — a fresh scoped token would have an empty + // grant and could not attach to the existing session. + expect(a.ofType("token").at(-1)!.jwt).toBe(admissionJwt); + const mintsBefore = h.api.countTo("/tokens"); + await h.send(a, { type: "request_token" }); + expect(a.ofType("token").at(-1)!.jwt).toBe(admissionJwt); + expect(h.api.countTo("/tokens")).toBe(mintsBefore); + }); + + it("shares one slot token between members of a shared session", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); + const a = await h.join("a"); + const b = await h.join("b"); + expect(a.ofType("token").at(-1)!.jwt).toBe(b.ofType("token").at(-1)!.jwt); + }); + + it("hands a spilled member the new slot's token before session_ready", async () => { + const h = makeHarness({ maxSessions: 2, usersPerSession: 2 }); + const a = await h.join("a"); + const b = await h.join("b"); // same slot as a + await h.send(a, { type: "claim" }); + const slot1Jwt = a.ofType("token").at(-1)!.jwt as string; + + // b's connection mint on the shared session hits the cap once, spilling + // b to a brand-new slot with its own token and session. + h.api.failNextConnections(1, 429, "connections_per_session"); + await h.send(b, { type: "claim" }); + + const readyB = b.ofType("session_ready").at(-1)!; + expect(readyB.sessionId).not.toBe(a.ofType("session_ready").at(-1)!.sessionId); + const bJwt = b.ofType("token").at(-1)!.jwt as string; + expect(bJwt).not.toBe(slot1Jwt); + // The new session was created by b's (new) slot token. + const creates = h.api.calls.filter((c) => c.method === "POST" && c.url.endsWith("/sessions")); + expect(creates.at(-1)!.headers["Authorization"]).toBe(`Bearer ${bJwt}`); + }); + + it("keeps unscoped tokens when acquireSession sources sessions externally", async () => { + const h = makeHarness({ + maxSessions: 1, + acquireSession: async () => "external-1", + }); + const a = await h.join("a"); + const mint = h.api.callsTo("/tokens").at(-1)!; + expect((mint.body as Record).authorization_details).toBeUndefined(); + await h.send(a, { type: "claim" }); + expect(a.ofType("session_ready").at(-1)!.sessionId).toBe("external-1"); + }); + + it("falls back to an unscoped token for a legacy slot with a grant-less session", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); + // Plant pre-upgrade state: a slot whose session exists but carries no jwt, + // as persisted by a version before scoped tokens. + await h.room.storage.put("slot:legacy", { + slotId: "legacy", + sessionId: "sess-old", + members: ["a"], + createdAt: Date.now(), + }); + await h.room.storage.put("member:a", { + slotId: "legacy", + sessionId: "sess-old", + connectionId: 1, + expiresAt: Date.now() + 60_000, + warned: false, + claimed: true, + }); + const a = h.room.register(new FakeConnection("a")); + await h.server.onMessage(JSON.stringify({ type: "request_token" }), a); + + const mint = h.api.callsTo("/tokens").at(-1)!; + expect((mint.body as Record).authorization_details).toBeUndefined(); + expect(a.ofType("token")).toHaveLength(1); + }); +}); + describe("token mint failure", () => { it("still admits but surfaces a token error to the client", async () => { const h = makeHarness({ maxSessions: 1 }); diff --git a/skills/building-reactor-queue-demos/SKILL.md b/skills/building-reactor-queue-demos/SKILL.md index b703fa5..91f8f10 100644 --- a/skills/building-reactor-queue-demos/SKILL.md +++ b/skills/building-reactor-queue-demos/SKILL.md @@ -68,7 +68,13 @@ The PartyKit room is the **single source of truth**: GPU session. The mint is the real capacity test: a 429 (session at its `connections_per_session` cap) makes the queue spill the member to another or new session. -3. **Mints short-lived JWTs** only for admitted members (the `getJwt` resolver). +3. **Mints session-scoped JWTs** only for admitted members (the `getJwt` + resolver). Each slot has one scoped token (`authorization_details`, + `max_sessions: 1`, matched to the configured model); the slot's session is + created _with_ it, binding the two, and every member seated on the slot + receives that token. It cannot be refreshed mid-session (the grant is the + bond), so it is minted to cover grace + the full session budget. With an + `acquireSession` override members get legacy unscoped tokens instead. 4. **Bounded turns** (default 120s after claim). On the last member leaving, the session is released (default: `DELETE /sessions/{id}`). The **server is the sole stopper** — adopting clients never `DELETE`, so a tab closing leaves the @@ -105,9 +111,11 @@ import pull them in. You never call these directly — the queue does: -- **Token minting**: `POST /tokens` with `Reactor-API-Key: rk_...` and - `{"expires_after": }` → `{ jwt, expires_at }`. The API key is a - **server secret**; the browser only ever sees minted JWTs. +- **Token minting**: `POST /tokens` with `Reactor-API-Key: rk_...`, + `{"expires_after": }`, and (for member tokens) session + `authorization_details` scoping the JWT to one session for the configured + model → `{ jwt, expires_at }`. The API key is a **server secret**; the + browser only ever sees minted, scoped JWTs. - **Session lifecycle**: with a queue the **server** owns the session and the SDK **attaches** via `connect(jwt, { sessionId, connectionId })`. States `CREATED → PENDING → WAITING → ACTIVE → INACTIVE → CLOSED`. No webhook for end — @@ -261,22 +269,22 @@ from an env var that overrides the code value (per-deploy tuning). Resolution order: **built-in default → `createReactorQueueServer({...})` → env var** (env wins). The API key MUST come from a secret, never code. -| Env var | Config key | Default | Purpose | -| -------------------------------- | --------------------------- | ------------------------- | -------------------------------------------- | -| `RQ_REACTOR_API_KEY` | `apiKey` | — (**required**, secret) | mint JWTs; create/stop sessions | -| `RQ_MODEL` | `model` | — (**required**) | model for `POST /sessions` | -| `RQ_MAX_SESSIONS` | `maxSessions` | `1` | concurrent Reactor sessions | -| `RQ_USERS_PER_SESSION` | `usersPerSession` | `1` | members per session | -| `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 (keep short) | -| `RQ_POLL_INTERVAL_MS` | `pollIntervalMs` | `15000` | session reconciliation cadence | -| `RQ_COORDINATOR_URL` | `coordinatorUrl` | `https://api.reactor.inc` | Reactor API base | -| `RQ_API_VERSION` | `apiVersion` | `1` | `Reactor-API-Version` header | -| `RQ_STOP_SESSIONS` | `stopSessionsOnExpiry` | `true` | `DELETE` session on expiry (default release) | -| `RQ_ALLOW_DUPLICATE_CONNECTIONS` | `allowDuplicateConnections` | `false` | allow same browser in multiple tabs | -| `RQ_ADMIN_PASSWORD` | `adminPassword` | — (off) | enables admin mode (see below) | +| Env var | Config key | Default | Purpose | +| -------------------------------- | --------------------------- | ------------------------- | -------------------------------------------------------- | +| `RQ_REACTOR_API_KEY` | `apiKey` | — (**required**, secret) | mint JWTs; create/stop sessions | +| `RQ_MODEL` | `model` | — (**required**) | model for `POST /sessions` | +| `RQ_MAX_SESSIONS` | `maxSessions` | `1` | concurrent Reactor sessions | +| `RQ_USERS_PER_SESSION` | `usersPerSession` | `1` | members per session | +| `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` | JWT lifetime floor (scoped tokens cover grace + session) | +| `RQ_POLL_INTERVAL_MS` | `pollIntervalMs` | `15000` | session reconciliation cadence | +| `RQ_COORDINATOR_URL` | `coordinatorUrl` | `https://api.reactor.inc` | Reactor API base | +| `RQ_API_VERSION` | `apiVersion` | `1` | `Reactor-API-Version` header | +| `RQ_STOP_SESSIONS` | `stopSessionsOnExpiry` | `true` | `DELETE` session on expiry (default release) | +| `RQ_ALLOW_DUPLICATE_CONNECTIONS` | `allowDuplicateConnections` | `false` | allow same browser in multiple tabs | +| `RQ_ADMIN_PASSWORD` | `adminPassword` | — (off) | enables admin mode (see below) | Tuning intuition: throughput ≈ `(maxSessions × usersPerSession) / sessionDurationMs`. Shorter sessions / more slots move the line faster but cost more GPU.