diff --git a/README.md b/README.md index 0e584c7..32e5d45 100644 --- a/README.md +++ b/README.md @@ -273,15 +273,14 @@ 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. 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. + only to admitted users. Every member gets their **own session-scoped** token + (`authorization_details` on `POST /tokens`), bound at mint to the one session + they were seated on: it acts on that session and nothing else on the account, + so members sharing a slot hold distinct tokens rather than copies of one. +4. Issues short-lived tokens (default 60s). Because a token can be bound to a + session that already exists, it is re-minted on demand — the client refreshes + over the WebSocket via `request_token`, exposed as a standard `getJwt` + resolver for the Reactor SDK. 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` @@ -430,7 +429,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 floor (scoped member tokens always cover grace + session) | +| `RQ_TOKEN_TTL_SECONDS` | `tokenTtlSeconds` | `60` | Minted JWT lifetime | | `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 | @@ -611,15 +610,17 @@ createReactorQueueServer({ participants. The queued user still attaches with `connect({ sessionId })` and the queue still -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. +mints the JWT. Member tokens are always bound to the session they are issued +for, and a session can only be bound by **the same user** whose API key does the +minting — the same Reactor _account_ is not enough. So a custom-acquired session +must be created with `RQ_REACTOR_API_KEY`, or with another key belonging to that +same user. Source one from a different user and the mint is refused: members +receive `{ type: "error", message: "token_mint_failed" }` and the admin log +carries the Coordinator's `403`. The queue will not fall back to an unscoped +token to paper over it. + +If another client shares the session with the queued user, the platform must +allow more than one connection per session. ## Configuration (client) diff --git a/packages/queue/package.json b/packages/queue/package.json index 288546f..9213400 100644 --- a/packages/queue/package.json +++ b/packages/queue/package.json @@ -1,6 +1,6 @@ { "name": "@reactor-team/queue", - "version": "0.0.4", + "version": "0.0.5", "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 25d089d..a4912bc 100644 --- a/packages/queue/src/server/config.ts +++ b/packages/queue/src/server/config.ts @@ -45,10 +45,8 @@ export interface ReactorQueueServerConfig { /** Lead time for the `time_warning`, in ms. Env: `RQ_WARNING_BEFORE_MS`. */ warningBeforeMs?: number; /** - * 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: + * Requested lifetime for each minted Reactor JWT, in seconds. Keep it short: + * the client refreshes over `request_token` as needed. Env: * `RQ_TOKEN_TTL_SECONDS`. */ tokenTtlSeconds?: number; diff --git a/packages/queue/src/server/coordinator.ts b/packages/queue/src/server/coordinator.ts index 3a9e039..6394530 100644 --- a/packages/queue/src/server/coordinator.ts +++ b/packages/queue/src/server/coordinator.ts @@ -22,15 +22,24 @@ 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. + * {@link CoordinatorClient.mintToken}, the JWT is restricted to the named model + * and to the sessions its grant holds — nothing else on the account. */ export interface TokenScope { - /** Model the token may create sessions for (fully-qualified `org/model`). */ + /** Model the token is confined to (fully-qualified `org/model`). */ model: string; - /** How many sessions the token may ever create (spawned, not concurrent). */ - maxSessions: number; + /** + * Existing sessions the grant starts bound to. Each must still be open and + * owned by the API key doing the minting; the Coordinator answers `403` + * otherwise, and refuses a session whose model falls outside `model`. + */ + sessions?: string[]; + /** + * How many sessions the grant may hold over its lifetime. Left unset it + * resolves to the number of bound sessions, which leaves the token full on + * arrival: it operates what it was given and cannot create more. + */ + maxSessions?: number; } /** @@ -82,10 +91,10 @@ export class CoordinatorClient { /** * Exchange the API key for a JWT. `ttlSeconds` is passed as `expires_after`; * 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. + * carries session `authorization_details`: it is confined to `scope.model` + * and to the sessions on its grant, which `scope.sessions` can pre-populate + * with sessions that already exist. The bound set is server state rather than + * a claim, so it never appears in the token itself. */ async mintToken( ttlSeconds: number, @@ -105,8 +114,13 @@ export class CoordinatorClient { authorization_details: [ { type: "session", - resources: { models: { match: [scope.model] } }, - constraints: { max_sessions: scope.maxSessions }, + resources: { + models: { match: [scope.model] }, + ...(scope.sessions?.length ? { sessions: { bind: scope.sessions } } : {}), + }, + ...(scope.maxSessions === undefined + ? {} + : { constraints: { max_sessions: scope.maxSessions } }), }, ], } @@ -168,15 +182,9 @@ 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; - jwt?: string; - }): Promise { - const jwt = opts.jwt ?? (await this.getServerJwt()); + async createSession(opts: { model: string; webrtcVersion: string }): Promise { + const jwt = await this.getServerJwt(); const res = await fetch(`${this.baseUrl}/sessions`, { method: "POST", headers: { @@ -210,17 +218,14 @@ 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 - * 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). + * server-minted `connection_id`. This is a transport call, so it carries + * `Reactor-WebRTC-Version` rather than the API-version headers. * * 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, opts: { jwt?: string } = {}): Promise { - const jwt = opts.jwt ?? (await this.getServerJwt()); + async createConnection(sessionId: string): Promise { + const 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 3b60525..95089a2 100644 --- a/packages/queue/src/server/server.ts +++ b/packages/queue/src/server/server.ts @@ -73,18 +73,6 @@ 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. */ @@ -150,52 +138,18 @@ export function createReactorQueueServer( } /** - * 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. + * Mint one member's own token for `sessionId`: confined to the configured + * model and bound to that single session. Binding at mint is what lets a + * member operate a session the server created for them, so members sharing + * a slot hold distinct tokens rather than copies of one. The session limit + * is left to resolve to the bound count, which makes the grant full on + * arrival — the token drives the session it was given and can open none. */ - 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(), { + private mintMemberToken(sessionId: string): Promise<{ jwt: string; expiresAt: number }> { + return this.api.mintToken(this.config.tokenTtlSeconds, { model: this.config.model, - maxSessions: 1, + sessions: [sessionId], }); - // 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; } /** @@ -1030,10 +984,9 @@ export function createReactorQueueServer( connId, } ); - // 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. + // A member seated on a slot that is already running can be handed a + // token now; one holding a freshly reserved slot has no session to + // bind against yet and receives theirs at claim. const conn = this.room.getConnection(connId); if (conn) await this.sendMemberToken(conn); admittedAny = true; @@ -1084,22 +1037,12 @@ 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, - }); - } - } + // A member's token binds to exactly one session and the claim is what + // decides which — their own slot, a shared slot with room, or a fresh + // one — so it can only be minted here. Sent before session_ready so the + // SDK holds it the moment it is told what to attach to. + const conn = this.room.getConnection(connId); + if (conn) await this.sendMemberToken(conn); this.sendTo(connId, { type: "session_ready", @@ -1183,24 +1126,9 @@ export function createReactorQueueServer( if (!slot) return null; let sessionId = slot.sessionId; - let slotJwt = slot.jwt ?? null; if (!sessionId) { try { - 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); - } + sessionId = await this.runAcquire(this.config.model); } catch (err) { this.reportError("session_create_failed", err, { connId }); return null; @@ -1214,12 +1142,7 @@ export function createReactorQueueServer( for (let attempt = 1; attempt <= CONNECTION_MINT_ATTEMPTS; attempt++) { try { - // 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 } : {} - ); + const connectionId = await this.api.createConnection(sessionId); return { sessionId, connectionId }; } catch (err) { if (err instanceof CoordinatorError && err.status === 429) { @@ -1288,27 +1211,23 @@ export function createReactorQueueServer( } /** - * 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. + * Send a member a token for the session their slot holds. A member still in + * the admission grace has no session to bind to yet and is answered at + * claim instead; the SDK has nothing to attach to until then, and its + * `getJwt` resolver re-asks over `request_token` when it does. + * + * There is no unscoped path. A session an `acquireSession` override sourced + * from another user cannot be bound, and the Coordinator's refusal is + * reported as-is rather than quietly answered with an account-wide token. */ 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 }); + const member = await this.getMember(conn.id); + if (!member) return; + const slot = await this.getSlot(member.slotId); + if (!slot?.sessionId) return; + const token = await this.mintMemberToken(slot.sessionId); + this.send(conn, { type: "token", jwt: token.jwt, expiresAt: token.expiresAt }); } catch (err) { this.reportError("mintToken", err); this.send(conn, { type: "error", message: "token_mint_failed" }); diff --git a/packages/queue/test/coordinator.test.ts b/packages/queue/test/coordinator.test.ts index e670694..656503b 100644 --- a/packages/queue/test/coordinator.test.ts +++ b/packages/queue/test/coordinator.test.ts @@ -67,6 +67,40 @@ describe("CoordinatorClient.mintToken", () => { ]); }); + it("binds existing sessions and omits the limit so the grant arrives full", async () => { + const { calls } = stubFetch([{ status: 200, body: { jwt: "bound", expires_at: 500 } }]); + await new CoordinatorClient(opts).mintToken(60, { + model: "reactor/helios", + sessions: ["sess-1"], + }); + const body = JSON.parse(calls[0]!.init.body as string); + expect(body.authorization_details).toEqual([ + { + type: "session", + resources: { models: { match: ["reactor/helios"] }, sessions: { bind: ["sess-1"] } }, + }, + ]); + }); + + it("keeps an explicit limit alongside bound sessions", async () => { + const { calls } = stubFetch([{ status: 200, body: { jwt: "bound", expires_at: 500 } }]); + await new CoordinatorClient(opts).mintToken(60, { + model: "reactor/helios", + sessions: ["sess-1"], + maxSessions: 3, + }); + const detail = JSON.parse(calls[0]!.init.body as string).authorization_details[0]; + expect(detail.resources.sessions).toEqual({ bind: ["sess-1"] }); + expect(detail.constraints).toEqual({ max_sessions: 3 }); + }); + + it("omits an empty bind list rather than sending sessions: { bind: [] }", async () => { + const { calls } = stubFetch([{ status: 200, body: { jwt: "scoped", expires_at: 500 } }]); + await new CoordinatorClient(opts).mintToken(60, { model: "reactor/helios", sessions: [] }); + const detail = JSON.parse(calls[0]!.init.body as string).authorization_details[0]; + expect(detail.resources).toEqual({ models: { match: ["reactor/helios"] } }); + }); + 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); @@ -177,19 +211,6 @@ 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 } }, @@ -227,15 +248,6 @@ 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/server.test.ts b/packages/queue/test/server.test.ts index 7920e0d..a83e403 100644 --- a/packages/queue/test/server.test.ts +++ b/packages/queue/test/server.test.ts @@ -59,12 +59,11 @@ afterEach(() => { }); describe("admission, FIFO, and capacity", () => { - it("admits the first user and mints a token immediately", async () => { + it("admits the first user without touching the Coordinator", async () => { const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); const a = await h.join("a"); expect(a.types()).toContain("admitted"); - expect(a.types()).toContain("token"); - expect(a.ofType("token")[0]!.jwt).toBe("jwt-1"); + expect(h.api.calls).toHaveLength(0); }); it("queues the second user behind a full capacity ceiling, 1-based", async () => { @@ -240,103 +239,145 @@ describe("usersPerSession > 1", () => { }); }); +/** The `bind` list a `POST /tokens` call asked for, or undefined if unscoped. */ +function bindOf(call: { body: unknown }): string[] | undefined { + const details = (call.body as { authorization_details?: Array> }) + .authorization_details; + return (details?.[0] as unknown as { resources?: { sessions?: { bind?: string[] } } } | undefined) + ?.resources?.sessions?.bind; +} + +/** + * The `jwt-N` the mock hands back for each `POST /tokens`, paired with whether + * that mint asked for a scope — the mock numbers tokens in call order. + */ +function mintedTokens(api: CoordinatorMock): Array<{ jwt: string; scoped: boolean }> { + return api.callsTo("/tokens").map((call, i) => ({ + jwt: `jwt-${i + 1}`, + scoped: (call.body as Record).authorization_details !== undefined, + })); +} + 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"); + it("sends no token during grace, when there is no session to bind to yet", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); + const a = await h.join("a"); + expect(a.ofType("token")).toHaveLength(0); + expect(h.api.countTo("/tokens")).toBe(0); + }); + + it("binds the member's token to the session they claimed, before session_ready", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 1, tokenTtlSeconds: 60 }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); + + const sessionId = a.ofType("session_ready").at(-1)!.sessionId; const mint = h.api.callsTo("/tokens").at(-1)!; - const body = mint.body as { - expires_after: number; - authorization_details: unknown; - }; + const body = mint.body as { expires_after: number; authorization_details: unknown }; + expect(body.expires_after).toBe(60); + // No `constraints`: the limit resolves to the bound count, so the grant is + // full on arrival and the token cannot open a session of its own. expect(body.authorization_details).toEqual([ { type: "session", - resources: { models: { match: ["helios"] } }, - constraints: { max_sessions: 1 }, + resources: { models: { match: ["helios"] }, sessions: { bind: [sessionId] } }, }, ]); - // 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); + + const order = a.types(); + expect(order.indexOf("token")).toBeGreaterThanOrEqual(0); + expect(order.indexOf("token")).toBeLessThan(order.indexOf("session_ready")); }); - it("creates the session and connection with the slot token and re-delivers it", async () => { - const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); + it("gives members of one shared session distinct tokens", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); const a = await h.join("a"); - const admissionJwt = a.ofType("token").at(-1)!.jwt as string; + const b = await h.join("b"); await h.send(a, { type: "claim" }); + await h.send(b, { 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); + const sessionId = a.ofType("session_ready").at(-1)!.sessionId; + expect(b.ofType("session_ready").at(-1)!.sessionId).toBe(sessionId); + expect(a.ofType("token").at(-1)!.jwt).not.toBe(b.ofType("token").at(-1)!.jwt); + for (const mint of h.api.callsTo("/tokens").filter((c) => bindOf(c))) { + expect(bindOf(mint)).toEqual([sessionId]); + } }); - it("shares one slot token between members of a shared session", async () => { + it("hands a member seated on an already-running slot their token at admission", async () => { const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); const a = await h.join("a"); + await h.send(a, { type: "claim" }); const b = await h.join("b"); - expect(a.ofType("token").at(-1)!.jwt).toBe(b.ofType("token").at(-1)!.jwt); + + expect(bindOf(h.api.callsTo("/tokens").at(-1)!)).toEqual([ + a.ofType("session_ready").at(-1)!.sessionId, + ]); + expect(b.ofType("token")).toHaveLength(1); + }); + + it("re-mints on request_token rather than re-delivering the same token", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 1 }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); + const first = a.ofType("token").at(-1)!.jwt; + + const before = h.api.countTo("/tokens"); + await h.send(a, { type: "request_token" }); + expect(h.api.countTo("/tokens")).toBe(before + 1); + expect(a.ofType("token").at(-1)!.jwt).not.toBe(first); + expect(bindOf(h.api.callsTo("/tokens").at(-1)!)).toEqual([ + a.ofType("session_ready").at(-1)!.sessionId, + ]); }); - it("hands a spilled member the new slot's token before session_ready", async () => { + it("binds a spilled member's token to the session they actually landed on", 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. + // b to a brand-new slot with its own 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}`); + const sessionB = b.ofType("session_ready").at(-1)!.sessionId; + expect(sessionB).not.toBe(a.ofType("session_ready").at(-1)!.sessionId); + expect(bindOf(h.api.callsTo("/tokens").at(-1)!)).toEqual([sessionB]); }); - it("keeps unscoped tokens when acquireSession sources sessions externally", async () => { - const h = makeHarness({ - maxSessions: 1, - acquireSession: async () => "external-1", - }); + it("never authorizes a session or connection call with a member token", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); const a = await h.join("a"); - const mint = h.api.callsTo("/tokens").at(-1)!; - expect((mint.body as Record).authorization_details).toBeUndefined(); + const b = await h.join("b"); await h.send(a, { type: "claim" }); - expect(a.ofType("session_ready").at(-1)!.sessionId).toBe("external-1"); + await h.send(b, { type: "claim" }); + + const scoped = new Set( + mintedTokens(h.api) + .filter((t) => t.scoped) + .map((t) => `Bearer ${t.jwt}`) + ); + expect(scoped.size).toBeGreaterThan(0); + const serverCalls = h.api.calls.filter((c) => c.method !== "GET" && !c.url.endsWith("/tokens")); + expect(serverCalls.length).toBeGreaterThan(0); + for (const call of serverCalls) { + expect(scoped.has(call.headers["Authorization"]!)).toBe(false); + } }); - it("falls back to an unscoped token for a legacy slot with a grant-less session", async () => { + it("binds a token for a slot persisted before this version", 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. + // A slot written by an earlier version carries its own `jwt`; the session + // it created belongs to the same API key, so it binds like any other. await h.room.storage.put("slot:legacy", { slotId: "legacy", sessionId: "sess-old", members: ["a"], createdAt: Date.now(), + jwt: "stale-slot-jwt", + jwtExpiresAt: Math.floor(Date.now() / 1000) + 600, }); await h.room.storage.put("member:a", { slotId: "legacy", @@ -349,20 +390,87 @@ describe("session-scoped member tokens", () => { 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); + expect(bindOf(h.api.callsTo("/tokens").at(-1)!)).toEqual(["sess-old"]); + expect(a.ofType("token").at(-1)!.jwt).not.toBe("stale-slot-jwt"); + }); + + it("binds an acquireSession-sourced session rather than falling back to unscoped", async () => { + const h = makeHarness({ maxSessions: 1, acquireSession: async () => "external-1" }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); + + expect(a.ofType("session_ready").at(-1)!.sessionId).toBe("external-1"); + expect(bindOf(h.api.callsTo("/tokens").at(-1)!)).toEqual(["external-1"]); + }); + + it("never mints an unscoped token for a member", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); + const a = await h.join("a"); + const b = await h.join("b"); + await h.send(a, { type: "claim" }); + await h.send(b, { type: "claim" }); + await h.send(a, { type: "request_token" }); + + // The server's own JWT is the only unscoped mint; every token a member + // holds carries a bind. + const memberJwts = new Set( + [...a.ofType("token"), ...b.ofType("token")].map((m) => m.jwt as string) + ); + const bound = new Set( + mintedTokens(h.api) + .filter((t, i) => bindOf(h.api.callsTo("/tokens")[i]!) !== undefined && t.scoped) + .map((t) => t.jwt) + ); + expect(memberJwts.size).toBeGreaterThan(0); + for (const jwt of memberJwts) expect(bound.has(jwt)).toBe(true); }); }); describe("token mint failure", () => { - it("still admits but surfaces a token error to the client", async () => { + it("surfaces a token error when a member's mint fails", async () => { const h = makeHarness({ maxSessions: 1 }); - h.api.failTokensWith(500, "down"); const a = await h.join("a"); - expect(a.types()).toContain("admitted"); + await h.send(a, { type: "claim" }); + h.api.failTokensWith(500, "down"); + await h.send(a, { type: "request_token" }); expect(a.ofType("error").at(-1)).toMatchObject({ message: "token_mint_failed" }); }); + + it("reports the refusal when a session cannot be bound, rather than going unscoped", async () => { + const h = makeHarness({ + maxSessions: 1, + usersPerSession: 2, + acquireSession: async () => "someone-elses-session", + }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); // warms the server's own JWT + + const before = h.api.countTo("/tokens"); + h.api.failTokensWith(403, "requested session is not available to this API key"); + const b = await h.join("b"); + await h.send(b, { type: "claim" }); + + expect(b.ofType("error").at(-1)).toMatchObject({ message: "token_mint_failed" }); + expect(b.ofType("token")).toHaveLength(0); + // Every refused mint still asked to bind; none retried without the scope. + const refused = h.api.callsTo("/tokens").slice(before); + expect(refused.length).toBeGreaterThan(0); + for (const mint of refused) expect(bindOf(mint)).toEqual(["someone-elses-session"]); + }); + + it("still seats a member whose token mint fails, leaving getJwt to retry", async () => { + const h = makeHarness({ maxSessions: 1, usersPerSession: 2 }); + const a = await h.join("a"); + await h.send(a, { type: "claim" }); // warms the server JWT and the session + h.api.failTokensWith(500, "down"); + + const b = await h.join("b"); + await h.send(b, { type: "claim" }); + expect(b.ofType("error").at(-1)).toMatchObject({ message: "token_mint_failed" }); + expect(b.ofType("session_ready").at(-1)!.sessionId).toBe( + a.ofType("session_ready").at(-1)!.sessionId + ); + }); }); describe("alarm scheduling", () => { @@ -453,8 +561,8 @@ describe("admin mode", () => { it("persists warn/error events to history but not high-frequency info events", async () => { const h = makeHarness({ adminPassword: "secret", maxSessions: 1 }); - h.api.failTokensWith(500, "down"); // makes admission log an error (mint failed) - await h.join("a"); + h.api.failSessionsWith(500, "down"); // makes the claim log an error + await h.send(await h.join("a"), { type: "claim" }); const conn = await admin(h, "admin-1"); await h.send(conn, { type: "admin_auth", password: "secret" }); diff --git a/skills/building-reactor-queue-demos/SKILL.md b/skills/building-reactor-queue-demos/SKILL.md index 91f8f10..fe5a7d7 100644 --- a/skills/building-reactor-queue-demos/SKILL.md +++ b/skills/building-reactor-queue-demos/SKILL.md @@ -69,12 +69,14 @@ The PartyKit room is the **single source of truth**: `connections_per_session` cap) makes the queue spill the member to another or new session. 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. + resolver). Each member gets their own token (`authorization_details`, matched + to the configured model) bound at mint to the one session they were seated + on, so members sharing a slot hold distinct tokens. Binding works against a + session that already exists, so tokens stay short-lived and `request_token` + re-mints. A member still in the grace window has no session yet and receives + their token at claim. There is no unscoped fallback: an `acquireSession` + override must source sessions from the same **user** as the API key, or the + mint is refused and members get `token_mint_failed`. 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 @@ -113,9 +115,10 @@ You never call these directly — the queue does: - **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. + `authorization_details` carrying `resources.sessions.bind` — the member's JWT + is scoped to the configured model and bound to their one session → `{ 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 — @@ -269,27 +272,27 @@ 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` | 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) | +| 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) | Tuning intuition: throughput ≈ `(maxSessions × usersPerSession) / sessionDurationMs`. Shorter sessions / more slots move the line faster but cost more GPU. -`admissionGraceMs` must be < `tokenTtlSeconds` so the first token is still valid -when the user clicks Enter; both defaults satisfy this. +`tokenTtlSeconds` is independent of the grace window: a member's token is minted +against the session they claim, so it starts counting from claim, not admission. ### Hooks (observe-only; not env-configurable)