diff --git a/src/backend/server.ts b/src/backend/server.ts index 43e97b6a4..a350b5f14 100644 --- a/src/backend/server.ts +++ b/src/backend/server.ts @@ -1633,7 +1633,11 @@ export class PuterServer { */ async #runHook( order: 'bottomUp' | 'topDown', - hook: 'onServerStart' | 'onServerPrepareShutdown' | 'onServerShutdown', + hook: + | 'onServerStart' + | 'onServerDrainStart' + | 'onServerPrepareShutdown' + | 'onServerShutdown', ) { const layers = [ this.clients, @@ -1664,6 +1668,11 @@ export class PuterServer { async prepareShutdown() { if (this.#server) { + try { + await this.#runHook('bottomUp', 'onServerDrainStart'); + } catch (err) { + console.error('[server] onServerDrainStart failed', err); + } this.#server.close(async () => { console.log( 'PuterServer has stopped accepting new connections', diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts index b730dee95..6e079804e 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.test.ts @@ -9,9 +9,12 @@ import { setupTestServer, type TestUserCredentials, } from '../../testUtil'; +import { kv } from '../../util/kvSingleton'; import { REALTIME_PATH, + realtimeAuthKey, realtimeTokenFrom, + realtimeTranscriptionRows, realtimeUsageRows, } from './OpenAIRealtimeRelayService'; @@ -87,6 +90,58 @@ describe('realtimeUsageRows', () => { }); }); +describe('realtimeTranscriptionRows', () => { + it('prices token usage by transcription model', () => { + const rows = realtimeTranscriptionRows('gpt-4o-mini-transcribe', { + type: 'tokens', + input_tokens: 30, + input_token_details: { text_tokens: 10, audio_tokens: 20 }, + output_tokens: 5, + }); + expect(rows.map((r) => [r.usageType, r.costOverride])).toEqual([ + [ + 'openai:gpt-4o-mini-transcribe:realtime:transcription_text_input', + 10 * 125, + ], + [ + 'openai:gpt-4o-mini-transcribe:realtime:transcription_audio_input', + 20 * 300, + ], + [ + 'openai:gpt-4o-mini-transcribe:realtime:transcription_text_output', + 5 * 500, + ], + ]); + }); + + it('prices duration usage per second', () => { + expect( + realtimeTranscriptionRows('whisper-1', { + type: 'duration', + seconds: 2.5, + }), + ).toEqual([ + { + usageType: 'openai:whisper-1:realtime:transcription_seconds', + usageAmount: 2.5, + costOverride: 25_000, + }, + ]); + }); + + it('bills an unknown model at the highest known rate', () => { + const [row] = realtimeTranscriptionRows(undefined, { + type: 'tokens', + input_tokens: 10, + }); + expect(row).toEqual({ + usageType: 'openai:unknown:realtime:transcription_audio_input', + usageAmount: 10, + costOverride: 10 * 600, + }); + }); +}); + describe('OpenAIRealtimeRelayService relay', () => { let server: PuterServer; let upstream: WebSocketServer; @@ -116,6 +171,47 @@ describe('OpenAIRealtimeRelayService relay', () => { } const event = JSON.parse(data.toString()); ws.send(JSON.stringify({ type: 'echo', event })); + if (event.type === 'transcribe') { + ws.send( + JSON.stringify({ + type: 'session.updated', + session: { + audio: { + input: { + transcription: { + model: 'gpt-4o-transcribe', + }, + }, + }, + }, + }), + ); + ws.send( + JSON.stringify({ + type: 'conversation.item.input_audio_transcription.completed', + usage: { + type: 'tokens', + input_tokens: 4, + input_token_details: { audio_tokens: 4 }, + output_tokens: 2, + }, + }), + ); + } + // Holds a response open so the drain has one in flight. + if (event.type === 'slow.response') { + ws.send(JSON.stringify({ type: 'response.created' })); + setTimeout( + () => + ws.send( + JSON.stringify({ + type: 'response.done', + response: {}, + }), + ), + 300, + ); + } if (event.type === 'response.create') { ws.send( JSON.stringify({ @@ -279,6 +375,70 @@ describe('OpenAIRealtimeRelayService relay', () => { expect(relay.stats().upstreamFailed).toBe(before + 1); }); + it('meters input transcription at the session transcription model', async () => { + const spy = vi.spyOn(server.services.metering, 'batchIncrementUsages'); + spy.mockClear(); + const { ws, waitFor } = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + await new Promise((r) => ws.once('open', r)); + ws.send(JSON.stringify({ type: 'transcribe' })); + await waitFor('conversation.item.input_audio_transcription.completed'); + await vi.waitFor(() => expect(spy).toHaveBeenCalled()); + const [, rows] = spy.mock.calls.at(-1)!; + expect(rows.map((r) => [r.usageType, r.costOverride])).toEqual([ + [ + 'openai:gpt-4o-transcribe:realtime:transcription_audio_input', + 4 * 600, + ], + [ + 'openai:gpt-4o-transcribe:realtime:transcription_text_output', + 2 * 1000, + ], + ]); + ws.close(); + }); + + it('closes a session whose credential stops authenticating', async () => { + const relay = server.services.openaiRealtime; + const auth = vi.spyOn(server.services.auth, 'authenticate'); + const { ws, closed, waitFor } = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + await waitFor('session.created'); + + // Within the cache window nothing is re-checked. + auth.mockClear(); + await relay.recheckSessions(); + expect(auth).not.toHaveBeenCalled(); + expect(ws.readyState).toBe(WebSocket.OPEN); + + auth.mockResolvedValue({ invalid: true } as never); + kv.del(realtimeAuthKey(user.token)); + await relay.recheckSessions(); + const error = (await waitFor('error')) as { + error: { code: string }; + }; + expect(error.error.code).toBe('invalid_api_key'); + expect((await closed).code).toBe(1008); + auth.mockRestore(); + }); + + it('keeps a session when the re-check lookup itself fails', async () => { + const relay = server.services.openaiRealtime; + const auth = vi.spyOn(server.services.auth, 'authenticate'); + const { ws, waitFor } = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + await waitFor('session.created'); + auth.mockRejectedValue(new Error('db down')); + kv.del(realtimeAuthKey(user.token)); + await relay.recheckSessions(); + expect(ws.readyState).toBe(WebSocket.OPEN); + auth.mockRestore(); + ws.close(); + }); + it('reports counters, event-loop delay and memory', async () => { const relay = server.services.openaiRealtime; await vi.waitFor(() => expect(relay.stats().active).toBe(0)); @@ -295,4 +455,30 @@ describe('OpenAIRealtimeRelayService relay', () => { expect(Number.isFinite(stats[key])).toBe(true); } }); + + // Last: the relay stays in drain mode afterwards. + it('drains: refuses new sessions, finishes in-flight responses, then closes with 1012', async () => { + const relay = server.services.openaiRealtime; + const live = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + await live.waitFor('session.created'); + live.ws.send(JSON.stringify({ type: 'slow.response' })); + await live.waitFor('response.created'); + + relay.onServerDrainStart(); + const late = open( + `${relayUrl}?model=gpt-realtime-mini&auth_token=${user.token}`, + ); + const error = (await late.waitFor('error')) as { + error: { code: string }; + }; + expect(error.error.code).toBe('server_restarting'); + expect((await late.closed).code).toBe(1012); + expect(live.ws.readyState).toBe(WebSocket.OPEN); + + await relay.onServerPrepareShutdown(); + expect((await live.closed).code).toBe(1012); + expect(live.events.some((e) => e.type === 'response.done')).toBe(true); + }); }); diff --git a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts index 747fa3b82..0f2fc1a27 100644 --- a/src/backend/services/realtime/OpenAIRealtimeRelayService.ts +++ b/src/backend/services/realtime/OpenAIRealtimeRelayService.ts @@ -17,6 +17,7 @@ * along with this program. If not, see . */ +import { createHash } from 'node:crypto'; import type { IncomingMessage, Server as HttpServer } from 'node:http'; import type { Duplex } from 'node:stream'; import { @@ -31,7 +32,9 @@ import { CONCURRENT_SLOT_TTL_MS, acquireConcurrent, } from '../../core/http/middleware/rateLimit.js'; +import { SPEECH_TO_TEXT_COSTS } from '../../drivers/ai-speech2txt/costs.js'; import { readProviderKey } from '../../drivers/util/providerRegistry.js'; +import { kv } from '../../util/kvSingleton.js'; import type { UsageInput } from '../metering/types.js'; import { decideSocketAuth } from '../socket/SocketService.js'; import { PuterService } from '../types.js'; @@ -98,6 +101,29 @@ export interface RealtimeUsage { output_token_details?: { text_tokens?: number; audio_tokens?: number }; } +/** Micro-cents per token for input transcription billed in tokens. */ +const TRANSCRIPTION_TOKEN_PRICES: Record< + string, + { textIn: number; audioIn: number; textOut: number } +> = { + 'gpt-4o-transcribe': { textIn: 250, audioIn: 600, textOut: 1000 }, + 'gpt-4o-mini-transcribe': { textIn: 125, audioIn: 300, textOut: 500 }, +}; +/** Micro-cents per second for input transcription billed by duration. */ +const TRANSCRIPTION_SECOND_PRICES: Record = { + 'whisper-1': 10_000, + 'gpt-transcribe': SPEECH_TO_TEXT_COSTS['openai:gpt-transcribe:second'], +}; + +/** The `usage` on `conversation.item.input_audio_transcription.completed`. */ +export interface RealtimeTranscriptionUsage { + type?: 'tokens' | 'duration'; + input_tokens?: number; + output_tokens?: number; + input_token_details?: { text_tokens?: number; audio_tokens?: number }; + seconds?: number; +} + const OPENAI_KEY_PROTOCOL = 'openai-insecure-api-key.'; const MAX_CLIENT_PAYLOAD = 16 * 1024 * 1024; // Client reads pause past this much buffered before the upstream opens. @@ -112,6 +138,16 @@ const HEARTBEAT_MS = 30_000; const UPSTREAM_HANDSHAKE_MS = 10_000; const STATS_INTERVAL_MS = 30_000; const DEFAULT_MAX_PER_USER = 20; +// Credentials are re-authenticated once their kv entry expires. +const AUTH_RECHECK_SEC = 60; +const AUTH_SWEEP_MS = 15_000; +// At shutdown, how long a session may finish its in-flight response. +const DRAIN_RESPONSE_MS = 5_000; +// RFC 6455 "Service Restart": the client should reconnect. +const CLOSE_RESTART = 1012; +const TYPE_PREFIX = /^\{\s*"type"\s*:\s*"([^"]+)"/; +const TRANSCRIPTION_DONE = + 'conversation.item.input_audio_transcription.completed'; /** * The credential a realtime client presents, in the places OpenAI's own clients @@ -179,6 +215,74 @@ export const realtimeUsageRows = ( })); }; +/** + * Usage rows for one completed input transcription. A model missing from the + * tables is billed at the highest known rate, so it is never free. + */ +export const realtimeTranscriptionRows = ( + model: string | undefined, + usage: RealtimeTranscriptionUsage | undefined, +): UsageInput[] => { + if (!usage) return []; + const n = (v: number | undefined) => + typeof v === 'number' && Number.isFinite(v) && v > 0 ? v : 0; + const known = + model && + (TRANSCRIPTION_TOKEN_PRICES[model] || + TRANSCRIPTION_SECOND_PRICES[model]) + ? model + : 'unknown'; + const usageType = (kind: string) => + `openai:${known}:realtime:transcription_${kind}`; + if (usage.type === 'duration') { + const seconds = n(usage.seconds); + const ucents = + TRANSCRIPTION_SECOND_PRICES[known] ?? + Math.max(...Object.values(TRANSCRIPTION_SECOND_PRICES)); + if (seconds === 0) return []; + return [ + { + usageType: usageType('seconds'), + usageAmount: seconds, + costOverride: Math.ceil(seconds * ucents), + }, + ]; + } + const price = + TRANSCRIPTION_TOKEN_PRICES[known] ?? + TRANSCRIPTION_TOKEN_PRICES['gpt-4o-transcribe']; + const details = usage.input_token_details; + const audioIn = details ? n(details.audio_tokens) : n(usage.input_tokens); + const counts: [string, number, number][] = [ + ['text_input', n(details?.text_tokens), price.textIn], + ['audio_input', audioIn, price.audioIn], + ['text_output', n(usage.output_tokens), price.textOut], + ]; + return counts + .filter(([, amount]) => amount > 0) + .map(([kind, amount, ucents]) => ({ + usageType: usageType(kind), + usageAmount: amount, + costOverride: amount * ucents, + })); +}; + +/** The transcription model a `session.created`/`session.updated` configures. */ +const transcriptionModelOf = (session: unknown): string | undefined => { + const s = session as { + audio?: { input?: { transcription?: { model?: unknown } | null } }; + input_audio_transcription?: { model?: unknown } | null; + } | null; + const model = + s?.audio?.input?.transcription?.model ?? + s?.input_audio_transcription?.model; + return typeof model === 'string' ? model : undefined; +}; + +/** The kv key whose presence means `token` was verified recently. */ +export const realtimeAuthKey = (token: string): string => + `realtime:auth:${createHash('sha256').update(token).digest('base64url')}`; + /** An error event in OpenAI's realtime shape. */ const errorEvent = (code: string, message: string): string => JSON.stringify({ @@ -212,8 +316,14 @@ export class OpenAIRealtimeRelayService extends PuterService { #sessions = new Set<{ close: (code: number, reason: string) => void; heartbeat: () => void; + recheck: () => Promise; + drain: () => Promise; }>(); #heartbeatTimer: ReturnType | null = null; + #authTimer: ReturnType | null = null; + // One re-authentication per token at a time, shared by its sessions. + #authChecks = new Map>(); + #draining = false; #stats = { opened: 0, closed: 0, @@ -270,15 +380,37 @@ export class OpenAIRealtimeRelayService extends PuterService { for (const session of this.#sessions) session.heartbeat(); }, HEARTBEAT_MS); this.#heartbeatTimer.unref?.(); + this.#authTimer = setInterval( + () => void this.recheckSessions(), + AUTH_SWEEP_MS, + ); + this.#authTimer.unref?.(); } - override onServerPrepareShutdown(): void { + /** Re-authenticates every session whose token's kv entry has expired. */ + async recheckSessions(): Promise { + await Promise.all([...this.#sessions].map((s) => s.recheck())); + } + + /** New sessions are refused; open ones keep running through the drain. */ + override onServerDrainStart(): void { + this.#draining = true; + console.info( + `[realtime] draining; ${this.#sessions.size} sessions stay open`, + ); + } + + /** + * Each session gets a moment to finish its in-flight response, then closes + * with 1012 so the client reconnects to another node. + */ + override async onServerPrepareShutdown(): Promise { + this.#draining = true; if (this.#statsTimer) clearInterval(this.#statsTimer); if (this.#heartbeatTimer) clearInterval(this.#heartbeatTimer); + if (this.#authTimer) clearInterval(this.#authTimer); this.#loopDelay?.disable(); - for (const session of this.#sessions) { - session.close(1001, 'server shutting down'); - } + await Promise.all([...this.#sessions].map((s) => s.drain())); this.#wss?.close(); } @@ -336,6 +468,14 @@ export class OpenAIRealtimeRelayService extends PuterService { let renewTimer: ReturnType | null = null; let budgetTimer: ReturnType | null = null; let stallTimer: ReturnType | null = null; + const token = realtimeTokenFrom(req, url); + let actor: Actor | null = null; + let rechecking = false; + // Responses started upstream and not yet done, for the drain. + let inFlight = 0; + let onIdle: (() => void) | null = null; + // Decoded upstream state the meter needs. + const decoded: { transcriptionModel?: string } = {}; // Client reads stay paused while any reason holds. const holds = new Set<'pending' | 'upstream' | 'budget'>(); @@ -358,6 +498,7 @@ export class OpenAIRealtimeRelayService extends PuterService { if (renewTimer) clearInterval(renewTimer); if (budgetTimer) clearTimeout(budgetTimer); if (stallTimer) clearTimeout(stallTimer); + onIdle?.(); void Promise.all(slots.map((s) => s.release())); if (client.readyState === WebSocket.OPEN) { client.close(code, reason); @@ -380,19 +521,63 @@ export class OpenAIRealtimeRelayService extends PuterService { alive = false; if (client.readyState === WebSocket.OPEN) client.ping(); }, + // Called every AUTH_SWEEP_MS; re-authenticates once the token's kv + // entry has expired. A failed lookup is retried on the next sweep. + recheck: async () => { + if (!actor || !token || closed || rechecking) return; + if (kv.get(realtimeAuthKey(token)) !== undefined) return; + rechecking = true; + try { + const next = await this.#authenticateCached(token, req); + if (closed) return; + if (!next || next.user?.uuid !== actor.user?.uuid) { + reject( + 'invalid_api_key', + 'Credentials are no longer valid', + ); + return; + } + actor = next; + } catch (err) { + console.warn('[realtime] credential recheck failed', err); + } finally { + rechecking = false; + } + }, + drain: async () => { + if (inFlight > 0 && !closed) { + await new Promise((resolve) => { + const timer = setTimeout(resolve, DRAIN_RESPONSE_MS); + onIdle = () => { + clearTimeout(timer); + resolve(); + }; + }); + } + session.close(CLOSE_RESTART, 'server restarting'); + }, }; this.#sessions.add(session); this.#stats.opened++; client.on('pong', () => (alive = true)); - const reject = (code: string, message: string) => { + const reject = (code: string, message: string, closeCode = 1008) => { this.#stats.rejected++; if (client.readyState === WebSocket.OPEN) { client.send(errorEvent(code, message)); } - session.close(1008, code); + session.close(closeCode, code); }; + if (this.#draining) { + reject( + 'server_restarting', + 'Server is restarting; reconnect', + CLOSE_RESTART, + ); + return; + } + // Token bucket over client bytes: one max-size frame of burst, then // reads pause until the sustained rate catches up. const rate = @@ -457,9 +642,11 @@ export class OpenAIRealtimeRelayService extends PuterService { return; } - const actor = await this.#authenticate(req, url).catch( - (): null => null, - ); + try { + actor = token ? await this.#authenticateCached(token, req) : null; + } catch { + actor = null; + } if (closed) return; if (!actor) { reject('invalid_api_key', 'Authentication failed'); @@ -558,9 +745,31 @@ export class OpenAIRealtimeRelayService extends PuterService { SLOW_CLIENT_MS, ); } - if (!isBinary && buf.subarray(0, 64).includes('"response.done"')) { - this.#meter(actor, model, buf, () => - reject('insufficient_funds', 'No usage left for request'), + if (isBinary) return; + // Audio deltas dominate; only the type prefix is read per frame. + const type = TYPE_PREFIX.exec( + buf.subarray(0, 96).toString('latin1'), + )?.[1]; + if (type === 'response.created') inFlight++; + if (type === 'response.done') { + inFlight = Math.max(0, inFlight - 1); + if (inFlight === 0) onIdle?.(); + } + if ( + type === 'response.done' || + type === TRANSCRIPTION_DONE || + type === 'session.created' || + type === 'session.updated' + ) { + // Parsed and billed after the frame is forwarded, in arrival + // order, so metering never holds up the relay. + setImmediate(() => + this.#decode(type, buf, model, decoded, actor!, () => + reject( + 'insufficient_funds', + 'No usage left for request', + ), + ), ); } }); @@ -594,34 +803,78 @@ export class OpenAIRealtimeRelayService extends PuterService { }); } - #meter( - actor: Actor, - model: string, + async #decode( + type: string, buf: Buffer, + model: string, + decoded: { transcriptionModel?: string }, + actor: Actor, onExhausted: () => void, - ): void { - let usage: RealtimeUsage | undefined; + ): Promise { + let event: { + session?: unknown; + response?: { usage?: RealtimeUsage }; + usage?: RealtimeTranscriptionUsage; + }; try { - usage = JSON.parse(buf.toString('utf8'))?.response?.usage; + event = JSON.parse(buf.toString('utf8')); } catch { return; } - const rows = realtimeUsageRows(model, usage); + if (type === 'session.created' || type === 'session.updated') { + decoded.transcriptionModel = transcriptionModelOf(event?.session); + return; + } + const rows = + type === 'response.done' + ? realtimeUsageRows(model, event?.response?.usage) + : realtimeTranscriptionRows( + decoded.transcriptionModel, + event?.usage, + ); if (rows.length === 0) return; - void this.services.metering - .batchIncrementUsages(actor, rows) - .then(() => this.services.metering.hasEnoughCredits(actor, 1)) - .then((ok) => { - if (!ok) onExhausted(); - }) - .catch((err: unknown) => - console.warn('[realtime] metering failed', err), - ); + try { + await this.services.metering.batchIncrementUsages(actor, rows); + if (!(await this.services.metering.hasEnoughCredits(actor, 1))) { + onExhausted(); + } + } catch (err) { + console.warn('[realtime] metering failed', err); + } } - async #authenticate(req: IncomingMessage, url: URL): Promise { - const token = realtimeTokenFrom(req, url); - if (!token) return null; + /** + * Authenticates `token`, caching success in kv for AUTH_RECHECK_SEC so + * sessions sharing a token re-check it once per expiry. Throws on lookup + * errors, which callers treat as "unknown" rather than "revoked". + */ + #authenticateCached( + token: string, + req: IncomingMessage, + ): Promise { + const key = realtimeAuthKey(token); + let check = this.#authChecks.get(key); + if (!check) { + check = (async () => { + try { + const actor = await this.#authenticate(token, req); + if (actor) { + kv.set(key, actor.user!.uuid, { EX: AUTH_RECHECK_SEC }); + } + return actor; + } finally { + this.#authChecks.delete(key); + } + })(); + this.#authChecks.set(key, check); + } + return check; + } + + async #authenticate( + token: string, + req: IncomingMessage, + ): Promise { const ua = req.headers['user-agent']; const result = await this.services.auth.authenticate(token, { ip: req.socket.remoteAddress, diff --git a/src/backend/services/realtime/README.md b/src/backend/services/realtime/README.md index ac7d0e6ed..e7c11e525 100644 --- a/src/backend/services/realtime/README.md +++ b/src/backend/services/realtime/README.md @@ -166,7 +166,7 @@ client ──ws──▶ api./puterai/openai/v1/realtime?model=… 2. The client socket is **accepted before authenticating** (see [Quirks](#quirks)). 3. The relay checks the model, then the credential, then that the account has at least 1 cent of usage left, then takes a per-user concurrency slot. Any failure sends an OpenAI-shaped `error` event and closes with `1008`. 4. It dials OpenAI with the server key. Client frames sent while that is in flight are buffered and replayed in order, keeping their text/binary type. -5. From then on frames pass through untouched in both directions. Only `response.done` is parsed, to meter its `usage`. +5. From then on frames pass through untouched in both directions. Only `response.done`, input transcription completions and `session.created`/`session.updated` are parsed, after forwarding, to meter usage. 6. Either side closing tears down the other. ### Credentials @@ -191,6 +191,8 @@ Prices are per 1M tokens, taken from OpenAI's pricing page on 2026-10-15. `REALT Each `response.done` records usage rows `openai::realtime:`, where `kind` is one of `text_input`, `text_cached`, `audio_input`, `audio_cached`, `image_input`, `image_cached`, `text_output` or `audio_output`. Each row carries a `costOverride`. Cached tokens are subtracted from their modality's input count, so nothing is billed twice. +Each `conversation.item.input_audio_transcription.completed` records `openai::realtime:transcription_` rows (`text_input`, `audio_input`, `text_output`, or `seconds` for duration-billed models). The transcription model is read from the session's `session.created`/`session.updated`. Token prices: `gpt-4o-transcribe` $2.50 / $6 / $10, `gpt-4o-mini-transcribe` $1.25 / $3 / $5 (text in / audio in / text out per 1M); `whisper-1` $0.006/min. A model not in these tables is billed as `unknown` at the highest known rate. + ## Configuration ```jsonc @@ -221,6 +223,8 @@ Each `response.done` records usage rows `openai::realtime:`, where | Client missing a ping | 30–60 s | socket terminated, upstream closed | | Upstream handshake | 10 s | close `1011` | | Credit needed to open a session | 1 ¢ | `insufficient_funds` error event, `1008` | +| Credential re-check | every 60 s (kv.js cache, swept every 15 s) | `invalid_api_key` error event, `1008`; a failed lookup is retried, not fatal | +| New sessions while the node drains | refused | `server_restarting` error event, `1012` | | Session length | none here | OpenAI ends sessions at 60 min | The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep the two in step. @@ -234,15 +238,15 @@ The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep **The model is checked before the credential.** An unauthenticated caller can learn which model names are valid. The keyless puter.js API test relies on this order (`ai.suite.ts`, "assistant.connect surfaces relay errors…"). -**Metering depends on OpenAI's field order.** To avoid parsing every audio delta, the relay only parses frames whose first 64 bytes contain `"response.done"`. OpenAI currently serialises `type` first. If that changes, usage silently stops being recorded. Check for that before trusting billing numbers. +**Metering depends on OpenAI's field order.** To avoid parsing every audio delta, the relay only reads the `type` from each frame's first 96 bytes. OpenAI currently serialises `type` first. If that changes, usage silently stops being recorded. Check for that before trusting billing numbers. -**Billing happens after the fact.** No credit is held while a response runs. A response is charged when it finishes, and only then does the relay check for a positive balance, closing the session (`insufficient_funds`) if there is none. An account can therefore go negative by one response's worth. Responses cancelled before `response.done`, and anything OpenAI bills outside `response.done` (for example input transcription), are not metered. +**Billing happens after the fact.** No credit is held while a response runs. A response is charged when it finishes, and only then does the relay check for a positive balance, closing the session (`insufficient_funds`) if there is none. An account can therefore go negative by one response's worth. Responses cancelled before `response.done` are not metered. **It skips the shared AI gates.** Sessions don't count against `AI_RATE_LIMIT`/`AI_CONCURRENT` (`drivers/util/aiLimits.ts`), only against their own per-user cap. They also bypass the AI cost-factor hook (`withAiCostFactor`), so `ai.cost.factor.*` listeners don't apply. **It isn't paid-only.** Every other `/puterai/openai/v1/*` route requires a subscription and a full-access token. This one accepts session and app tokens on any plan. The docs say so. -**The credential is never re-checked.** Unlike `SocketService`, there is no periodic re-auth sweep. A revoked session or newly suspended account keeps an open session until it ends (60 min at most, OpenAI's cap). A team's 2FA requirement is not enforced either (`teamRequires2fa` isn't passed). +**The credential is re-checked every 60 s.** A successful auth is cached in the process kv.js singleton under a hash of the token for 60 s; a sweep every 15 s re-authenticates sessions whose entry expired (one lookup per token, shared by its sessions). A revoked token or suspended account is closed within about 75 s. A lookup that throws keeps the session and retries next sweep. A team's 2FA requirement is still not enforced (`teamRequires2fa` isn't passed). **Tokens can end up in logs.** The subprotocol header and `?auth_token=` query both carry the Puter token. Proxies or access logs that record them will capture it, so prefer the `Authorization` header from servers. @@ -256,7 +260,7 @@ The per-user cap is published in `src/docs/src/rate-limits-and-quotas.md`; keep **It lives in the service layer.** Only services receive the raw `http.Server` (`attachHttpServer`), so this edge component, which applies auth gates, sits in `services/` instead of `controllers/`. Move it if the relay outlives the experiment. -**Shutdown closes sessions immediately.** `onServerPrepareShutdown` closes every session with `1001`. They are not drained, so a deploy drops live conversations. +**Shutdown drains.** On `SIGTERM` (`onServerDrainStart`) new sessions are refused with `1012` and open ones keep running for the drain window (90 s with a `serverId`). At `onServerPrepareShutdown` each session gets up to 5 s to finish an in-flight response, then closes with `1012` (Service Restart) so the client reconnects elsewhere. Conversation state lives upstream and is lost on reconnect. ## Observing it diff --git a/src/backend/services/types.ts b/src/backend/services/types.ts index 98eb975b0..0aac60c21 100644 --- a/src/backend/services/types.ts +++ b/src/backend/services/types.ts @@ -62,7 +62,10 @@ export const PuterService = class PuterService implements WithLifecycle { public onServerStart() { return; } - public onServerPrepareShutdown() { + public onServerDrainStart(): Promise | void { + return; + } + public onServerPrepareShutdown(): Promise | void { return; } public onServerShutdown() { diff --git a/src/backend/types.ts b/src/backend/types.ts index a2e496b4c..b2613cb52 100644 --- a/src/backend/types.ts +++ b/src/backend/types.ts @@ -1308,6 +1308,12 @@ export interface WithLifecycle extends Object { onServerStart?: () => Promise | void; onServerShutdown?: () => Promise | void; onServerPrepareShutdown?: () => Promise | void; + /** + * When the drain starts, before `onServerPrepareShutdown`, which waits for + * every connection (upgraded sockets included) to close or the drain to + * end. + */ + onServerDrainStart?: () => Promise | void; } export interface WithCostsReporting extends WithLifecycle {