Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 20 additions & 19 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand Down Expand Up @@ -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 |
Expand Down Expand Up @@ -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)

Expand Down
2 changes: 1 addition & 1 deletion packages/queue/package.json
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
6 changes: 2 additions & 4 deletions packages/queue/src/server/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
59 changes: 32 additions & 27 deletions packages/queue/src/server/coordinator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}

/**
Expand Down Expand Up @@ -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,
Expand All @@ -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 } }),
},
],
}
Expand Down Expand Up @@ -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<string> {
const jwt = opts.jwt ?? (await this.getServerJwt());
async createSession(opts: { model: string; webrtcVersion: string }): Promise<string> {
const jwt = await this.getServerJwt();
const res = await fetch(`${this.baseUrl}/sessions`, {
method: "POST",
headers: {
Expand Down Expand Up @@ -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<number> {
const jwt = opts.jwt ?? (await this.getServerJwt());
async createConnection(sessionId: string): Promise<number> {
const jwt = await this.getServerJwt();
const endpoint = `/sessions/${encodeURIComponent(sessionId)}/transport/webrtc/connections`;
const res = await fetch(`${this.baseUrl}${endpoint}`, {
method: "POST",
Expand Down
Loading