From 14ee256232840a5f410b7e4eab03dd67de360772 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sun, 23 Aug 2026 00:41:22 +0800 Subject: [PATCH] refactor(route): reconcile SFU room lifecycle --- .env.example | 4 +- deploy/livekit/livekit.yaml.example | 4 +- package.json | 2 +- src/server/app.ts | 167 ++++++++++---- src/server/config.ts | 46 +++- src/server/hybrid-media-router.ts | 317 +++++++++++++++++++++++++-- src/server/livekit-token.ts | 21 +- src/server/sfu-resource-admission.ts | 82 ++++--- src/server/sfu-room-control.ts | 193 ++++++++++++++++ src/server/signaling.ts | 34 ++- tests/fake-sfu-room-control.ts | 63 ++++++ tests/livekit-token.test.ts | 20 +- tests/server-config.test.ts | 56 ++++- tests/server-http.test.ts | 220 ++++++++++++++++++- tests/server-signal.test.ts | 273 +++++++++++++++++++++++ tests/sfu-resource-admission.test.ts | 45 ++-- tests/sfu-room-control.test.ts | 189 ++++++++++++++++ 17 files changed, 1602 insertions(+), 134 deletions(-) create mode 100644 src/server/sfu-room-control.ts create mode 100644 tests/fake-sfu-room-control.ts create mode 100644 tests/sfu-room-control.test.ts diff --git a/.env.example b/.env.example index d3c980e..db88058 100644 --- a/.env.example +++ b/.env.example @@ -24,10 +24,11 @@ PEER_ASSISTED_MEDIA=false ENDPOINT_MEDIA_COPY_CAPACITY=2 # Optional automatic SFU fallback. Set PEER_ASSISTED_MEDIA=true and supply all -# five values together; leaving them empty keeps LiveKit completely dormant. +# six values together; leaving them empty keeps LiveKit completely dormant. # Capacities have no defaults and must come from this deployment's measurements. # Production requires wss and an independent API secret of 32+ bytes. LIVEKIT_URL= +LIVEKIT_API_URL= LIVEKIT_API_KEY= LIVEKIT_API_SECRET= SFU_INGRESS_CAPACITY= @@ -50,6 +51,7 @@ STUN_URLS= # Automatic SFU fallback (requires PEER_ASSISTED_MEDIA=true and does not # replace STUN above): # LIVEKIT_URL=wss://share.example.com +# LIVEKIT_API_URL=http://127.0.0.1:7880 # LIVEKIT_API_KEY=replace-with-a-livekit-api-key # LIVEKIT_API_SECRET=replace-with-at-least-32-independent-random-bytes # SFU_INGRESS_CAPACITY=replace-with-measured-positive-integer diff --git a/deploy/livekit/livekit.yaml.example b/deploy/livekit/livekit.yaml.example index 15b44bd..3928838 100644 --- a/deploy/livekit/livekit.yaml.example +++ b/deploy/livekit/livekit.yaml.example @@ -19,7 +19,9 @@ rtc: use_external_ip: true room: - auto_create: true + # Screener creates exact-generation rooms through the private RoomService API. + # Keep this false so a stale participant token cannot recreate a drained room. + auto_create: false # Screener enforces its configured admission limit before issuing a token. # This server-side ceiling covers the repository's maximum 16 viewers + host. max_participants: 17 diff --git a/package.json b/package.json index cb7e358..63f174d 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ "build:server": "tsc -p tsconfig.server.json", "start": "node dist/server/server/index.js", "gate:peer-topology-loopback": "tsx scripts/peer-assisted-benchmark.ts", - "gate:sfu-admission": "vitest run tests/sfu-resource-admission.test.ts tests/media-route-controller.test.ts tests/livekit-token.test.ts tests/sfu-client.test.ts tests/route-transition.test.ts tests/peer-assisted-benchmark.test.ts tests/server-signal.test.ts", + "gate:sfu-admission": "vitest run tests/sfu-resource-admission.test.ts tests/sfu-room-control.test.ts tests/media-route-controller.test.ts tests/livekit-token.test.ts tests/sfu-client.test.ts tests/route-transition.test.ts tests/peer-assisted-benchmark.test.ts tests/server-config.test.ts tests/server-http.test.ts tests/server-signal.test.ts", "gate:access-privacy": "npm run build:client && vitest run tests/server-signal.test.ts && tsx scripts/access-privacy-gate.ts", "gate:turn-udp-allocation": "tsx scripts/turn-allocation-gate.ts", "probe:native-one-viewer": "tsx scripts/native-one-viewer-gate.ts", diff --git a/src/server/app.ts b/src/server/app.ts index 9c23418..597ed7b 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -11,11 +11,13 @@ import { import { SiteAccess } from "./access-session.js"; import { loadConfig, type ServerConfig } from "./config.js"; import { createIceConfig } from "./ice.js"; +import type { SfuFallbackOptions } from "./hybrid-media-router.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; +import type { SfuRoomControl } from "./sfu-room-control.js"; import { SfuResourceAdmission } from "./sfu-resource-admission.js"; import { RoomDatabase } from "./room-database.js"; import { RoomStore, RoomStoreError } from "./room-store.js"; -import { SignalingServer } from "./signaling.js"; +import { SignalingServer, type SignalingOptions } from "./signaling.js"; export interface CreateServerOptions { config?: ServerConfig; @@ -31,6 +33,7 @@ export interface CreateServerOptions { maxUnauthenticatedSignalConnections?: number; siteAccessTtlSeconds?: number; sfuTokenIssuer?: SfuTokenIssuer; + sfuRoomControl?: SfuRoomControl; } export interface ScreenerServer { @@ -48,22 +51,33 @@ export async function createScreenerServer( const config = options.config ?? loadConfig(); const now = options.now ?? Date.now; const livekitFallback = config.livekitFallback; - const sfuFallback = livekitFallback - ? { - url: livekitFallback.url, - tokenIssuer: - options.sfuTokenIssuer ?? - new (await import("./livekit-token.js")).LiveKitTokenIssuer({ - apiKey: livekitFallback.apiKey, - apiSecret: livekitFallback.apiSecret, - maxViewersPerRoom: config.maxViewersPerRoom, - }), - admission: new SfuResourceAdmission({ - ingressCapacity: livekitFallback.ingressCapacity, - egressCapacity: livekitFallback.egressCapacity, + let sfuFallback: SfuFallbackOptions | undefined; + let sfuRoomControl: SfuRoomControl | undefined; + if (livekitFallback) { + sfuRoomControl = + options.sfuRoomControl ?? + new (await import("./sfu-room-control.js")).LiveKitSfuRoomControl({ + apiUrl: livekitFallback.apiUrl, + apiKey: livekitFallback.apiKey, + apiSecret: livekitFallback.apiSecret, + maxViewersPerRoom: config.maxViewersPerRoom, + }); + sfuFallback = { + url: livekitFallback.url, + tokenIssuer: + options.sfuTokenIssuer ?? + new (await import("./livekit-token.js")).LiveKitTokenIssuer({ + apiKey: livekitFallback.apiKey, + apiSecret: livekitFallback.apiSecret, + maxViewersPerRoom: config.maxViewersPerRoom, }), - } - : undefined; + admission: new SfuResourceAdmission({ + ingressCapacity: livekitFallback.ingressCapacity, + egressCapacity: livekitFallback.egressCapacity, + }), + roomControl: sfuRoomControl, + }; + } const roomStore = options.roomStore ?? new RoomStore({ @@ -89,7 +103,14 @@ export async function createScreenerServer( let frontendHandler: FrontendHandler | undefined; let vite: ViteDevServer | undefined; + let acceptingTraffic = false; const httpServer = createServer((request, response) => { + if (!acceptingTraffic) { + response.setHeader("Cache-Control", "no-store"); + response.setHeader("Retry-After", "1"); + sendJson(response, 503, { error: "Service starting" }); + return; + } void handleRequest( request, response, @@ -111,7 +132,7 @@ export async function createScreenerServer( }); }); - const signaling = new SignalingServer({ + const signalingOptions: SignalingOptions = { server: httpServer, roomStore, peerAssistedMedia: config.peerAssistedMedia, @@ -133,7 +154,10 @@ export async function createScreenerServer( options.maxUnauthenticatedSignalConnections, passThroughUnknownUpgrades: config.nodeEnv === "development" && options.serveFrontend !== false, - }); + }; + let signaling: SignalingServer | undefined; + let startupOperation: Promise | undefined; + let closing = false; if (options.serveFrontend !== false) { if (config.nodeEnv === "development") { @@ -176,42 +200,109 @@ export async function createScreenerServer( return { httpServer, roomStore, - listen(port = config.port, host = config.listenHost) { - return new Promise((resolve, reject) => { - const onError = (error: Error) => { - httpServer.off("listening", onListening); - reject(error); - }; - const onListening = () => { - httpServer.off("error", onError); - const address = httpServer.address(); - if (!address || typeof address === "string") { - reject(new Error("Server did not bind to a TCP port")); - return; + async listen(port = config.port, host = config.listenHost) { + if (startupOperation) { + throw new Error("Screener server startup was already requested"); + } + if (closing) { + throw new Error("Screener server is closing"); + } + startupOperation = (async () => { + const boundPort = await bindHttpServer(httpServer, port, host); + try { + await sfuRoomControl?.initialize(); + if (!closing) { + signaling = new SignalingServer(signalingOptions); + acceptingTraffic = true; } - resolve(address.port); - }; - httpServer.once("error", onError); - httpServer.once("listening", onListening); - httpServer.listen(port, host); - }); + return boundPort; + } catch (error) { + try { + await closeHttpServer(httpServer); + } catch (closeError) { + throw new AggregateError( + [error, closeError], + "Screener startup reconciliation failed", + ); + } + throw error; + } + })(); + return await startupOperation; }, async close() { + acceptingTraffic = false; + closing = true; + const errors: unknown[] = []; + try { + await startupOperation; + } catch { + // The listen caller owns the startup error; shutdown still closes resources. + } + try { + await signaling?.close(); + } catch (error) { + errors.push(error); + } try { - await signaling.close(); if (httpServer.listening) { await new Promise((resolve, reject) => { httpServer.close((error) => (error ? reject(error) : resolve())); }); } + } catch (error) { + errors.push(error); + } + try { await vite?.close(); - } finally { + } catch (error) { + errors.push(error); + } + try { roomStore.close(); + } catch (error) { + errors.push(error); + } + if (errors.length > 0) { + throw new AggregateError(errors, "Screener server shutdown failed"); } }, }; } +function bindHttpServer( + httpServer: ReturnType, + port: number, + host: string, +): Promise { + return new Promise((resolve, reject) => { + const onError = (error: Error) => { + httpServer.off("listening", onListening); + reject(error); + }; + const onListening = () => { + httpServer.off("error", onError); + const address = httpServer.address(); + if (!address || typeof address === "string") { + reject(new Error("Server did not bind to a TCP port")); + return; + } + resolve(address.port); + }; + httpServer.once("error", onError); + httpServer.once("listening", onListening); + httpServer.listen(port, host); + }); +} + +function closeHttpServer( + httpServer: ReturnType, +): Promise { + return new Promise((resolve, reject) => { + httpServer.close((error) => (error ? reject(error) : resolve())); + }); +} + async function handleRequest( request: IncomingMessage, response: ServerResponse, diff --git a/src/server/config.ts b/src/server/config.ts index 48c341e..774978e 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -35,6 +35,7 @@ const REMOVED_ENVIRONMENT_VARIABLES = [ export interface LiveKitFallbackConfig { url: string; + apiUrl: string; apiKey: string; apiSecret: string; ingressCapacity: number; @@ -139,9 +140,10 @@ function parseLiveKitFallback( nodeEnv: RuntimeEnvironment, ): LiveKitFallbackConfig | undefined { const url = environment.LIVEKIT_URL?.trim() || undefined; + const apiUrl = environment.LIVEKIT_API_URL?.trim() || undefined; const apiKey = environment.LIVEKIT_API_KEY?.trim() || undefined; const apiSecret = environment.LIVEKIT_API_SECRET?.trim() || undefined; - const configuredValues = [url, apiKey, apiSecret].filter(Boolean).length; + const configuredValues = [url, apiUrl, apiKey, apiSecret].filter(Boolean).length; const ingressCapacity = environment.SFU_INGRESS_CAPACITY?.trim() || undefined; const egressCapacity = environment.SFU_EGRESS_CAPACITY?.trim() || undefined; const configuredCapacities = [ingressCapacity, egressCapacity].filter( @@ -156,9 +158,9 @@ function parseLiveKitFallback( } return undefined; } - if (configuredValues !== 3) { + if (configuredValues !== 4) { throw new Error( - "LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", + "LIVEKIT_URL, LIVEKIT_API_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", ); } if (configuredCapacities !== 2) { @@ -194,8 +196,37 @@ function parseLiveKitFallback( throw new Error("LIVEKIT_URL must use wss in production"); } + let parsedApiUrl: URL; + try { + parsedApiUrl = new URL(apiUrl!); + } catch { + throw new Error("LIVEKIT_API_URL must be a valid http or https origin"); + } + if (parsedApiUrl.protocol !== "http:" && parsedApiUrl.protocol !== "https:") { + throw new Error("LIVEKIT_API_URL must use http or https"); + } + if ( + parsedApiUrl.username || + parsedApiUrl.password || + parsedApiUrl.pathname !== "/" || + parsedApiUrl.search || + parsedApiUrl.hash + ) { + throw new Error( + "LIVEKIT_API_URL must be an origin without credentials, path, query, or fragment", + ); + } + if ( + nodeEnv === "production" && + parsedApiUrl.protocol !== "https:" && + !isLoopbackHostname(parsedApiUrl.hostname) + ) { + throw new Error("LIVEKIT_API_URL must use https or loopback in production"); + } + return { url: parsedUrl.origin, + apiUrl: parsedApiUrl.origin, apiKey: apiKey!, apiSecret: apiSecret!, ingressCapacity: parseRequiredPositiveInteger( @@ -209,6 +240,15 @@ function parseLiveKitFallback( }; } +function isLoopbackHostname(hostname: string): boolean { + return ( + hostname === "localhost" || + hostname === "127.0.0.1" || + hostname === "[::1]" || + hostname === "::1" + ); +} + function parseUrlList(value: string | undefined, name: string): string[] { if (!value?.trim()) { return []; diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 2ad5085..fa2699e 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -21,6 +21,10 @@ import type { SfuResourceAdmission, SfuResourceFence, } from "./sfu-resource-admission.js"; +import { + managedSfuRoomName, + type SfuRoomControl, +} from "./sfu-room-control.js"; import type { SelectedEdgeTurnConfig } from "./config.js"; import { MediaRouteController, @@ -37,6 +41,8 @@ import { issueSelectedEdgeTurnCredential } from "./selected-edge-turn.js"; type ErrorCode = Extract["code"]; const DEFAULT_SFU_PREPARE_TIMEOUT_MS = 20_000; +const DEFAULT_SFU_DRAIN_RETRY_MS = 1_000; +const DEFAULT_HOST_OFFLINE_CHECK_MS = 5_000; const SELECTED_EDGE_TURN_ATTEMPT_TIMEOUT_MS = 30_000; const VIEWER_QUALITY_BAD_WINDOWS_TO_REASSIGN = 3; const VIEWER_QUALITY_EVIDENCE_GAP_MS = 5_000; @@ -62,6 +68,18 @@ interface PendingRoutePreparation { timer: NodeJS.Timeout; } +interface SfuDrainTask { + fence: SfuResourceFence; + operation?: Promise; + retryTimer?: NodeJS.Timeout; +} + +interface HostOfflineCheck { + hostPeerId: string; + fence: SfuResourceFence; + timer: NodeJS.Timeout; +} + interface ViewerRouteIntent { failedParentPeerId: string | null; failedParentSessionId?: string; @@ -190,7 +208,10 @@ export interface SfuFallbackOptions { url: string; tokenIssuer: SfuTokenIssuer; admission: SfuResourceAdmission; + roomControl: SfuRoomControl; prepareTimeoutMs?: number; + drainRetryMs?: number; + hostOfflineCheckMs?: number; } export interface HybridMediaRouterOptions { @@ -283,6 +304,9 @@ export class HybridMediaRouter { string, DeferredHealthySfuReselection >(); + private readonly sfuDrainTasks = new Map(); + private readonly hostOfflineChecks = new Map(); + private closing = false; constructor(private readonly options: HybridMediaRouterOptions) { assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); this.peerRelayTopology = new PeerRelayTopology( @@ -303,10 +327,22 @@ export class HybridMediaRouter { ) { throw new Error("SFU fallback timeout is invalid"); } + for (const [value, name] of [ + [fallback.drainRetryMs, "SFU drain retry"], + [fallback.hostOfflineCheckMs, "SFU Host-offline check"], + ] as const) { + if ( + value !== undefined && + (!Number.isSafeInteger(value) || value <= 0) + ) { + throw new Error(`${name} interval is invalid`); + } + } new URL(fallback.url); } - close(): void { + async close(): Promise { + this.closing = true; for (const pending of this.pendingRoutePreparations.values()) { clearTimeout(pending.timer); } @@ -325,6 +361,10 @@ export class HybridMediaRouter { } } this.deferredHealthySfuReselections.clear(); + for (const check of this.hostOfflineChecks.values()) { + clearTimeout(check.timer); + } + this.hostOfflineChecks.clear(); this.activeSelectedSfuIngresses.clear(); this.mediaRouteControllers.clear(); this.failedParentPeerIdsByViewer.clear(); @@ -335,7 +375,37 @@ export class HybridMediaRouter { this.viewerQualityEvidenceStates.clear(); this.heldQualityParentExclusionsByViewer.clear(); this.roomQualityMigrationCooldownUntilMs.clear(); - this.options.sfuFallback?.admission.releaseAll(); + const fallback = this.options.sfuFallback; + if (!fallback) { + return; + } + for (const fence of fallback.admission.beginDrainAll()) { + this.scheduleSfuDrain(fence); + } + const results = await Promise.allSettled( + [...this.sfuDrainTasks].map(async ([key, task]) => { + if (task.retryTimer) { + clearTimeout(task.retryTimer); + task.retryTimer = undefined; + } + if (task.operation) { + await task.operation; + } + if (this.sfuDrainTasks.get(key) === task) { + await this.runSfuDrain(key, task); + } + }), + ); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === "rejected") + .map((result) => result.reason); + const usage = fallback.admission.usage(); + if (failures.length > 0 || usage.ingress !== 0 || usage.egress !== 0) { + throw new AggregateError( + failures, + "LiveKit rooms were not fully drained during shutdown", + ); + } } connectParticipant(input: AuthenticatedRouteParticipant): HybridAuthenticationState { @@ -348,6 +418,9 @@ export class HybridMediaRouter { this.clearRoomViewerQualityEvidenceStates(input.roomId); this.clearHeldQualityParentExclusion(input.roomId, input.peerId); this.clearSfuRefreshesForPeer(input.roomId, input.peerId); + if (input.role === "host") { + this.cancelHostOfflineCheck(input.roomId); + } if (input.role === "viewer") { this.clearDeferredHealthySfuReselectionForParticipant( input.roomId, @@ -1507,6 +1580,9 @@ export class HybridMediaRouter { this.clearSelectedEdgeTurnsForParticipant(roomId, peerId); this.clearViewerQualityStateForParticipant(roomId, peerId); this.clearDeferredHealthySfuReselectionForParticipant(roomId, peerId); + if (this.peerRelayTopology.getHostPeerId(roomId) === peerId) { + this.scheduleHostOfflineCheck(roomId, peerId); + } const pending = this.pendingRoutePreparations.get(roomId); if (!pending || this.pendingSessionsAreCurrent(roomId, pending)) { return; @@ -2823,7 +2899,7 @@ export class HybridMediaRouter { sfuRootPeerIds: rootPeerIds, }); } catch { - fallback.admission.release(sfuResourceFence); + this.beginSfuDrain(sfuResourceFence); this.sendError( sourceSessionId, "SERVER_ERROR", @@ -2832,7 +2908,7 @@ export class HybridMediaRouter { return "unavailable"; } if (revision === undefined) { - fallback.admission.release(sfuResourceFence); + this.beginSfuDrain(sfuResourceFence); return "unavailable"; } @@ -2863,7 +2939,7 @@ export class HybridMediaRouter { roomId, controller, revision, - generation, + sfuResourceFence, rootPeerIds, expectedSessionIds, ); @@ -2874,7 +2950,7 @@ export class HybridMediaRouter { roomId: string, controller: MediaRouteController, revision: number, - generation: string, + sfuResourceFence: SfuResourceFence, rootPeerIds: readonly string[], expectedSessionIds: ReadonlyMap, ): Promise { @@ -2883,6 +2959,24 @@ export class HybridMediaRouter { return; } try { + await fallback.roomControl.createRoom(sfuResourceFence); + let pending = this.pendingRoutePreparations.get(roomId); + let pendingRoute = controller.getPendingRoute(); + if ( + this.mediaRouteControllers.get(roomId) !== controller || + pending?.revision !== revision || + pendingRoute?.route.revision !== revision + ) { + this.beginSfuDrain(sfuResourceFence); + return; + } + if (!this.pendingSessionsAreCurrent(roomId, pending)) { + this.abortPendingRoute(roomId); + return; + } + if (this.abortStalePendingQualityIntent(roomId, pending)) { + return; + } const grants = await Promise.all( [...expectedSessionIds].map(async ([peerId, sessionId]) => ({ peerId, @@ -2894,18 +2988,20 @@ export class HybridMediaRouter { ? "host" : "viewer", peerId, - publicationGeneration: generation, + shareGeneration: sfuResourceFence.shareGeneration, + publicationGeneration: sfuResourceFence.publicationGeneration, allowlistedRootPeerIds: rootPeerIds, }), })), ); - const pending = this.pendingRoutePreparations.get(roomId); - const pendingRoute = controller.getPendingRoute(); + pending = this.pendingRoutePreparations.get(roomId); + pendingRoute = controller.getPendingRoute(); if ( this.mediaRouteControllers.get(roomId) !== controller || pending?.revision !== revision || pendingRoute?.route.revision !== revision ) { + this.beginSfuDrain(sfuResourceFence); return; } if (!this.pendingSessionsAreCurrent(roomId, pending)) { @@ -2943,6 +3039,8 @@ export class HybridMediaRouter { pending?.revision === revision ) { this.abortPendingRoute(roomId); + } else { + this.beginSfuDrain(sfuResourceFence); } } } @@ -2988,11 +3086,15 @@ export class HybridMediaRouter { if (!controller.commit(revision)) { return; } - if ( - pending.sfuResourceFence && - !this.options.sfuFallback?.admission.commit(pending.sfuResourceFence) - ) { - throw new Error("Committed SFU route has no resource reservation"); + let supersededSfuFences: readonly SfuResourceFence[] = []; + if (pending.sfuResourceFence) { + const result = this.options.sfuFallback?.admission.commit( + pending.sfuResourceFence, + ); + if (!result) { + throw new Error("Committed SFU route has no resource reservation"); + } + supersededSfuFences = result; } clearTimeout(pending.timer); this.pendingRoutePreparations.delete(roomId); @@ -3020,6 +3122,9 @@ export class HybridMediaRouter { this.clearChangedRouteConnectionIds(roomId, before, active); } this.broadcastActiveRoute(roomId, active); + for (const fence of supersededSfuFences) { + this.scheduleSfuDrain(fence); + } this.drainViewerRouteIntents(roomId); } @@ -3192,7 +3297,7 @@ export class HybridMediaRouter { clearTimeout(pending.timer); this.pendingRoutePreparations.delete(roomId); if (pending.sfuResourceFence) { - this.options.sfuFallback?.admission.release(pending.sfuResourceFence); + this.beginSfuDrain(pending.sfuResourceFence); } if (pending.healthyReselection) { this.options.deleteConnectionId(roomId, pending.healthyReselection.rootPeerId); @@ -3325,12 +3430,14 @@ export class HybridMediaRouter { const fallback = this.options.sfuFallback; const controller = this.mediaRouteControllers.get(roomId); const active = controller?.getActiveRoute(); + const shareGeneration = this.options.getShareGeneration(roomId); if ( !fallback || !controller || !active || active.revision !== revision || !active.sfu.publicationGeneration || + !shareGeneration || (peerId !== this.peerRelayTopology.getHostPeerId(roomId) && !active.sfu.rootPeerIds.includes(peerId)) ) { @@ -3383,6 +3490,7 @@ export class HybridMediaRouter { ? "host" : "viewer", peerId, + shareGeneration, publicationGeneration: active.sfu.publicationGeneration, allowlistedRootPeerIds: active.sfu.rootPeerIds, }); @@ -3392,7 +3500,8 @@ export class HybridMediaRouter { this.connectedPeer(roomId, peerId)?.sessionId !== sessionId || current.revision !== revision || current.sfu.publicationGeneration !== - active.sfu.publicationGeneration + active.sfu.publicationGeneration || + this.options.getShareGeneration(roomId) !== shareGeneration ) { return; } @@ -3410,7 +3519,8 @@ export class HybridMediaRouter { this.connectedPeer(roomId, peerId)?.sessionId === sessionId && current.revision === revision && current.sfu.publicationGeneration === - active.sfu.publicationGeneration + active.sfu.publicationGeneration && + this.options.getShareGeneration(roomId) === shareGeneration ) { if ( active.sfu.rootPeerIds.includes(peerId) && @@ -3522,7 +3632,7 @@ export class HybridMediaRouter { excludedPeerId?: string, ): void { if (route.sfu.publicationGeneration === null) { - this.options.sfuFallback?.admission.releaseRoom(roomId); + this.beginSfuRoomDrain(roomId); } this.reconcileSelectedPeerEdgeLeases(roomId, route); for (const [key, attempt] of this.selectedSfuIngressAttempts) { @@ -3796,7 +3906,168 @@ export class HybridMediaRouter { } } + private beginSfuDrain(fence: SfuResourceFence): boolean { + const fallback = this.options.sfuFallback; + if (!fallback || !fallback.admission.beginDrain(fence)) { + return false; + } + this.scheduleSfuDrain(fence); + return true; + } + + private beginSfuRoomDrain(roomId: string): void { + const fallback = this.options.sfuFallback; + if (!fallback) { + return; + } + for (const fence of fallback.admission.beginDrainRoom(roomId)) { + this.scheduleSfuDrain(fence); + } + } + + private scheduleSfuDrain(fence: SfuResourceFence): void { + const fallback = this.options.sfuFallback; + if (!fallback) { + return; + } + const key = managedSfuRoomName(fence); + let task = this.sfuDrainTasks.get(key); + if (!task) { + task = { fence: { ...fence } }; + this.sfuDrainTasks.set(key, task); + } + if (!task.operation && !task.retryTimer) { + void this.runSfuDrain(key, task).catch(() => undefined); + } + } + + private runSfuDrain(key: string, task: SfuDrainTask): Promise { + if (task.operation) { + return task.operation; + } + const fallback = this.options.sfuFallback; + if (!fallback) { + return Promise.resolve(); + } + let tracked!: Promise; + tracked = (async () => { + try { + await fallback.roomControl.deleteRoom(task.fence); + if (this.sfuDrainTasks.get(key) !== task) { + return; + } + if (!fallback.admission.completeDrain(task.fence)) { + throw new Error("LiveKit drain has no matching resource generation"); + } + if (task.retryTimer) { + clearTimeout(task.retryTimer); + } + this.sfuDrainTasks.delete(key); + } catch (error) { + if (this.sfuDrainTasks.get(key) !== task) { + return; + } + if (this.closing) { + throw error; + } + console.error("SFU room drain failed; capacity remains reserved"); + if (!task.retryTimer) { + task.retryTimer = setTimeout(() => { + task.retryTimer = undefined; + void this.runSfuDrain(key, task).catch(() => undefined); + }, fallback.drainRetryMs ?? DEFAULT_SFU_DRAIN_RETRY_MS); + task.retryTimer.unref(); + } + } + })().finally(() => { + if (task.operation === tracked) { + task.operation = undefined; + } + }); + task.operation = tracked; + return tracked; + } + + private activeSfuFence(roomId: string): SfuResourceFence | null { + const publicationGeneration = this.mediaRouteControllers + .get(roomId) + ?.getActiveRoute().sfu.publicationGeneration; + const shareGeneration = this.options.getShareGeneration(roomId); + return publicationGeneration && shareGeneration + ? { roomId, shareGeneration, publicationGeneration } + : null; + } + + private scheduleHostOfflineCheck(roomId: string, hostPeerId: string): void { + this.cancelHostOfflineCheck(roomId); + const fallback = this.options.sfuFallback; + const fence = this.activeSfuFence(roomId); + if (!fallback || !fence || this.closing) { + return; + } + const timer = setTimeout(() => { + void this.checkHostOffline(roomId).catch(() => undefined); + }, fallback.hostOfflineCheckMs ?? DEFAULT_HOST_OFFLINE_CHECK_MS); + timer.unref(); + this.hostOfflineChecks.set(roomId, { + hostPeerId, + fence, + timer, + }); + } + + private async checkHostOffline(roomId: string): Promise { + const check = this.hostOfflineChecks.get(roomId); + const fallback = this.options.sfuFallback; + if (!check || !fallback || this.closing) { + return; + } + if (this.options.roomStore.getConnectedHost(roomId)) { + this.cancelHostOfflineCheck(roomId); + return; + } + + let hostExists: boolean; + try { + hostExists = await fallback.roomControl.hostParticipantExists(check.fence); + } catch { + if (this.hostOfflineChecks.get(roomId) === check) { + console.error("SFU Host-offline check failed; capacity remains reserved"); + this.hostOfflineChecks.delete(roomId); + this.scheduleHostOfflineCheck(roomId, check.hostPeerId); + } + return; + } + if (this.hostOfflineChecks.get(roomId) !== check) { + return; + } + if (this.options.roomStore.getConnectedHost(roomId)) { + this.cancelHostOfflineCheck(roomId); + return; + } + const currentFence = this.activeSfuFence(roomId); + if (!currentFence || !sameSfuFence(currentFence, check.fence)) { + this.cancelHostOfflineCheck(roomId); + return; + } + this.hostOfflineChecks.delete(roomId); + if (hostExists) { + this.scheduleHostOfflineCheck(roomId, check.hostPeerId); + return; + } + this.clearRoomMediaRouteState(roomId); + } + + private cancelHostOfflineCheck(roomId: string): void { + const check = this.hostOfflineChecks.get(roomId); + if (check) { + clearTimeout(check.timer); + this.hostOfflineChecks.delete(roomId); + } + } + private clearRoomMediaRouteState(roomId: string): void { + this.cancelHostOfflineCheck(roomId); const pending = this.pendingRoutePreparations.get(roomId); if (pending) { clearTimeout(pending.timer); @@ -3821,7 +4092,7 @@ export class HybridMediaRouter { this.sfuDisabledRoomIds.delete(roomId); this.clearRoomViewerQualityEvidenceStates(roomId); this.roomQualityMigrationCooldownUntilMs.delete(roomId); - this.options.sfuFallback?.admission.releaseRoom(roomId); + this.beginSfuRoomDrain(roomId); } private clearSelectedEdgeTurnsForParticipant( @@ -4049,6 +4320,14 @@ function viewerConnectionKey(roomId: string, peerId: string): string { return `${roomId}:${peerId}`; } +function sameSfuFence(left: SfuResourceFence, right: SfuResourceFence): boolean { + return ( + left.roomId === right.roomId && + left.shareGeneration === right.shareGeneration && + left.publicationGeneration === right.publicationGeneration + ); +} + function sameQualityRouteIntentGuard( left: QualityRouteIntentGuard, right: QualityRouteIntentGuard, diff --git a/src/server/livekit-token.ts b/src/server/livekit-token.ts index 47247d5..8ae525a 100644 --- a/src/server/livekit-token.ts +++ b/src/server/livekit-token.ts @@ -1,6 +1,5 @@ import { AccessToken, - RoomConfiguration, TrackSource, } from "livekit-server-sdk"; @@ -8,6 +7,7 @@ import { MAX_VIEWERS_PER_ROOM_LIMIT, type Role, } from "../shared/protocol.js"; +import { managedSfuRoomName } from "./sfu-room-control.js"; const LIVEKIT_TOKEN_TTL_SECONDS = 5 * 60; const ROOM_ID_PATTERN = /^[1-9]\d{0,11}$/; @@ -17,6 +17,7 @@ export interface SfuTokenRequest { roomId: string; role: Role; peerId: string; + shareGeneration: string; publicationGeneration: string; allowlistedRootPeerIds: readonly string[]; } @@ -49,10 +50,7 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer { validateTokenRequest(request, this.options.maxViewersPerRoom); const isHost = request.role === "host"; - const room = liveKitRoomName( - request.roomId, - request.publicationGeneration, - ); + const room = managedSfuRoomName(request); const token = new AccessToken( this.options.apiKey, this.options.apiSecret, @@ -72,11 +70,6 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer { canPublishData: false, canUpdateOwnMetadata: false, }); - token.roomConfig = new RoomConfiguration({ - name: room, - maxParticipants: this.options.maxViewersPerRoom + 1, - }); - return await token.toJwt(); } } @@ -90,6 +83,7 @@ function validateTokenRequest( } if ( !OPAQUE_ID_PATTERN.test(request.peerId) || + !OPAQUE_ID_PATTERN.test(request.shareGeneration) || !OPAQUE_ID_PATTERN.test(request.publicationGeneration) ) { throw new Error("LiveKit participant identity is invalid"); @@ -111,10 +105,3 @@ function validateTokenRequest( throw new Error("LiveKit viewer is not an allowlisted SFU root"); } } - -function liveKitRoomName( - roomId: string, - publicationGeneration: string, -): string { - return `screener-${roomId}-${publicationGeneration}`; -} diff --git a/src/server/sfu-resource-admission.ts b/src/server/sfu-resource-admission.ts index f955f44..e365aaa 100644 --- a/src/server/sfu-resource-admission.ts +++ b/src/server/sfu-resource-admission.ts @@ -22,6 +22,7 @@ interface SfuResourceEntry { interface RoomSfuResources { committed?: SfuResourceEntry; reserved?: SfuResourceEntry; + draining: Map; } export class SfuResourceAdmission { @@ -48,6 +49,9 @@ export class SfuResourceAdmission { if (room?.committed && sameFence(room.committed.fence, fence)) { return room.committed.egress === egress; } + if (room?.draining.has(fenceKey(fence))) { + return false; + } if ( this.ingressInUse >= this.ingressCapacity || egress > this.egressCapacity - this.egressInUse @@ -55,7 +59,7 @@ export class SfuResourceAdmission { return false; } - const resources = room ?? {}; + const resources: RoomSfuResources = room ?? { draining: new Map() }; resources.reserved = { fence: { ...fence }, egress }; this.rooms.set(fence.roomId, resources); this.ingressInUse += 1; @@ -63,68 +67,88 @@ export class SfuResourceAdmission { return true; } - commit(fence: SfuResourceFence): boolean { + commit(fence: SfuResourceFence): readonly SfuResourceFence[] | null { assertFence(fence); const room = this.rooms.get(fence.roomId); if (room?.committed && sameFence(room.committed.fence, fence)) { - return true; + return []; } if (!room?.reserved || !sameFence(room.reserved.fence, fence)) { - return false; + return null; } + const draining: SfuResourceFence[] = []; if (room.committed) { - this.subtract(room.committed); + this.addDraining(room, room.committed); + draining.push({ ...room.committed.fence }); } room.committed = room.reserved; room.reserved = undefined; - return true; + return draining; } - release(fence: SfuResourceFence): boolean { + beginDrain(fence: SfuResourceFence): boolean { assertFence(fence); const room = this.rooms.get(fence.roomId); if (!room) { return false; } - let released = false; + if (room.draining.has(fenceKey(fence))) { + return true; + } + let entry: SfuResourceEntry | undefined; if (room.reserved && sameFence(room.reserved.fence, fence)) { - this.subtract(room.reserved); + entry = room.reserved; room.reserved = undefined; - released = true; } if (room.committed && sameFence(room.committed.fence, fence)) { - this.subtract(room.committed); + entry = room.committed; room.committed = undefined; - released = true; } - this.deleteEmptyRoom(fence.roomId, room); - return released; + if (!entry) { + return false; + } + this.addDraining(room, entry); + return true; } - releaseRoom(roomId: string): boolean { + completeDrain(fence: SfuResourceFence): boolean { + assertFence(fence); + const room = this.rooms.get(fence.roomId); + const entry = room?.draining.get(fenceKey(fence)); + if (!room || !entry) { + return false; + } + room.draining.delete(fenceKey(fence)); + this.subtract(entry); + this.deleteEmptyRoom(fence.roomId, room); + return true; + } + + beginDrainRoom(roomId: string): readonly SfuResourceFence[] { if (!roomId) { throw new Error("SFU resource room ID is invalid"); } const room = this.rooms.get(roomId); if (!room) { - return false; + return []; } if (room.reserved) { - this.subtract(room.reserved); + this.addDraining(room, room.reserved); + room.reserved = undefined; } if (room.committed) { - this.subtract(room.committed); + this.addDraining(room, room.committed); + room.committed = undefined; } - this.rooms.delete(roomId); - return true; + return [...room.draining.values()].map((entry) => ({ ...entry.fence })); } - releaseAll(): void { - this.rooms.clear(); - this.ingressInUse = 0; - this.egressInUse = 0; + beginDrainAll(): readonly SfuResourceFence[] { + return [...this.rooms.keys()].flatMap((roomId) => + this.beginDrainRoom(roomId), + ); } usage(): SfuResourceUsage { @@ -139,13 +163,21 @@ export class SfuResourceAdmission { } } + private addDraining(room: RoomSfuResources, entry: SfuResourceEntry): void { + room.draining.set(fenceKey(entry.fence), entry); + } + private deleteEmptyRoom(roomId: string, room: RoomSfuResources): void { - if (!room.committed && !room.reserved) { + if (!room.committed && !room.reserved && room.draining.size === 0) { this.rooms.delete(roomId); } } } +function fenceKey(fence: SfuResourceFence): string { + return `${fence.shareGeneration}\u0000${fence.publicationGeneration}`; +} + function sameEntry( entry: SfuResourceEntry, fence: SfuResourceFence, diff --git a/src/server/sfu-room-control.ts b/src/server/sfu-room-control.ts new file mode 100644 index 0000000..e3281da --- /dev/null +++ b/src/server/sfu-room-control.ts @@ -0,0 +1,193 @@ +import { RoomServiceClient, ServerError } from "livekit-server-sdk"; + +import { MAX_VIEWERS_PER_ROOM_LIMIT } from "../shared/protocol.js"; +import type { SfuResourceFence } from "./sfu-resource-admission.js"; + +const MANAGED_ROOM_PREFIX = "screener-v1."; +const ROOM_ID_PATTERN = /^[1-9]\d{0,11}$/; +const OPAQUE_ID_PATTERN = /^[A-Za-z0-9_-]{8,128}$/; +const LEGACY_MANAGED_ROOM_PATTERN = /^screener-[1-9]\d{0,11}-[A-Za-z0-9_-]{8,128}$/; +const ROOM_SERVICE_TIMEOUT_SECONDS = 5; + +interface LiveKitRoomRecord { + name: string; +} + +interface LiveKitParticipantRecord { + identity: string; +} + +export interface LiveKitRoomService { + createRoom(options: { + name: string; + maxParticipants: number; + }): Promise; + listRooms(names?: string[]): Promise; + deleteRoom(room: string): Promise; + listParticipants(room: string): Promise; +} + +export interface SfuRoomControl { + initialize(): Promise; + createRoom(fence: SfuResourceFence): Promise; + deleteRoom(fence: SfuResourceFence): Promise; + hostParticipantExists(fence: SfuResourceFence): Promise; +} + +export interface LiveKitSfuRoomControlOptions { + apiUrl: string; + apiKey: string; + apiSecret: string; + maxViewersPerRoom: number; + roomService?: LiveKitRoomService; +} + +export class LiveKitSfuRoomControl implements SfuRoomControl { + private readonly roomService: LiveKitRoomService; + private readonly maxParticipants: number; + private readonly roomOperations = new Map>(); + + constructor(options: LiveKitSfuRoomControlOptions) { + if (!options.apiUrl || !options.apiKey || Buffer.byteLength(options.apiSecret) < 32) { + throw new Error("LiveKit RoomService configuration is invalid"); + } + if ( + !Number.isSafeInteger(options.maxViewersPerRoom) || + options.maxViewersPerRoom < 1 || + options.maxViewersPerRoom > MAX_VIEWERS_PER_ROOM_LIMIT + ) { + throw new Error("LiveKit viewer limit is invalid"); + } + this.maxParticipants = options.maxViewersPerRoom + 1; + this.roomService = + options.roomService ?? + new RoomServiceClient( + options.apiUrl, + options.apiKey, + options.apiSecret, + { + requestTimeout: ROOM_SERVICE_TIMEOUT_SECONDS, + failover: false, + }, + ); + } + + async initialize(): Promise { + const rooms = await this.roomService.listRooms(); + if (rooms.some((room) => !isManagedSfuRoomName(room.name))) { + throw new Error("Dedicated LiveKit instance contains a foreign room"); + } + for (const room of rooms) { + await this.deleteRoomByName(room.name); + } + if ((await this.roomService.listRooms()).length !== 0) { + throw new Error("Dedicated LiveKit instance is not empty after startup drain"); + } + } + + createRoom(fence: SfuResourceFence): Promise { + const roomName = managedSfuRoomName(fence); + return this.serialize(roomName, async () => { + if ((await this.roomService.listRooms([roomName])).length !== 0) { + throw new Error("Managed LiveKit room already exists"); + } + await this.roomService.createRoom({ + name: roomName, + maxParticipants: this.maxParticipants, + }); + const rooms = await this.roomService.listRooms([roomName]); + if (rooms.length !== 1 || rooms[0]?.name !== roomName) { + throw new Error("Managed LiveKit room creation was not confirmed"); + } + }); + } + + deleteRoom(fence: SfuResourceFence): Promise { + const roomName = managedSfuRoomName(fence); + return this.serialize(roomName, async () => { + await this.deleteRoomByName(roomName); + if ((await this.roomService.listRooms([roomName])).length !== 0) { + throw new Error("Managed LiveKit room deletion was not confirmed"); + } + }); + } + + hostParticipantExists(fence: SfuResourceFence): Promise { + const roomName = managedSfuRoomName(fence); + return this.serialize(roomName, async () => { + try { + return (await this.roomService.listParticipants(roomName)).some( + (participant) => participant.identity === "host", + ); + } catch (error) { + if (isNotFound(error)) { + return false; + } + throw error; + } + }); + } + + private async deleteRoomByName(roomName: string): Promise { + try { + await this.roomService.deleteRoom(roomName); + } catch (error) { + if (!isNotFound(error)) { + throw error; + } + } + } + + private serialize( + roomName: string, + operation: () => Promise, + ): Promise { + const previous = this.roomOperations.get(roomName) ?? Promise.resolve(); + const result = previous.catch(() => undefined).then(operation); + const tail = result.then( + () => undefined, + () => undefined, + ); + this.roomOperations.set(roomName, tail); + void tail.then(() => { + if (this.roomOperations.get(roomName) === tail) { + this.roomOperations.delete(roomName); + } + }); + return result; + } +} + +export function managedSfuRoomName(fence: SfuResourceFence): string { + if ( + !ROOM_ID_PATTERN.test(fence.roomId) || + !OPAQUE_ID_PATTERN.test(fence.shareGeneration) || + !OPAQUE_ID_PATTERN.test(fence.publicationGeneration) + ) { + throw new Error("Managed LiveKit room fence is invalid"); + } + return `${MANAGED_ROOM_PREFIX}${fence.roomId}.${fence.shareGeneration}.${fence.publicationGeneration}`; +} + +export function isManagedSfuRoomName(roomName: string): boolean { + if (LEGACY_MANAGED_ROOM_PATTERN.test(roomName)) { + return true; + } + if (!roomName.startsWith(MANAGED_ROOM_PREFIX)) { + return false; + } + const parts = roomName.slice(MANAGED_ROOM_PREFIX.length).split("."); + return ( + parts.length === 3 && + ROOM_ID_PATTERN.test(parts[0] ?? "") && + OPAQUE_ID_PATTERN.test(parts[1] ?? "") && + OPAQUE_ID_PATTERN.test(parts[2] ?? "") + ); +} + +function isNotFound(error: unknown): boolean { + return ( + error instanceof ServerError && + (error.status === 404 || error.code === "not_found") + ); +} diff --git a/src/server/signaling.ts b/src/server/signaling.ts index 77c85c3..f8d27e0 100644 --- a/src/server/signaling.ts +++ b/src/server/signaling.ts @@ -216,6 +216,10 @@ export class SignalingServer { } return; } + if (this.closing) { + rejectUpgrade(socket, 503, "Service Unavailable"); + return; + } if (!this.isAllowedOrigin(request.headers.origin)) { rejectUpgrade(socket, 403, "Forbidden"); return; @@ -250,26 +254,33 @@ export class SignalingServer { async close(): Promise { this.closing = true; + this.options.server.off("upgrade", this.upgradeHandler); clearInterval(this.heartbeatTimer); clearInterval(this.cleanupTimer); for (const timer of this.viewerGraceTimers.values()) { clearTimeout(timer); } this.viewerGraceTimers.clear(); - this.hybridMediaRouter?.close(); + let routeCloseError: unknown; + try { + await this.hybridMediaRouter?.close(); + } catch (error) { + routeCloseError = error; + } this.viewerQualityEvidenceGates.clear(); this.viewerMediaReadyByRoom.clear(); this.shareGenerationsByRoom.clear(); this.pausedShareGenerationsByRoom.clear(); this.ordinaryActiveHostChildrenByRoom.clear(); - this.options.server.off("upgrade", this.upgradeHandler); - for (const socket of this.webSocketServer.clients) { socket.terminate(); } await new Promise((resolve) => { this.webSocketServer.close(() => resolve()); }); + if (routeCloseError) { + throw routeCloseError; + } } private accept(socket: WebSocket, siteAccessAuthenticated: boolean): void { @@ -313,7 +324,7 @@ export class SignalingServer { private handleMessage(socket: WebSocket, encoded: string): void { const state = this.socketStates.get(socket); - if (!state) { + if (!state || this.closing) { return; } if (state.revoked) { @@ -420,6 +431,7 @@ export class SignalingServer { sessionId: state.sessionId, }, () => + !this.closing && this.socketStates.get(socket) === state && !state.revoked && !state.authenticated && @@ -462,6 +474,20 @@ export class SignalingServer { return; } + if ( + this.closing || + this.socketStates.get(socket) !== state || + state.revoked || + socket.readyState !== WebSocket.OPEN + ) { + this.options.roomStore.disconnectParticipant( + participant.roomId, + participant.peerId, + state.sessionId, + ); + return; + } + clearTimeout(state.authenticationTimer); this.unauthenticatedConnections -= 1; let shareGeneration: string | null = null; diff --git a/tests/fake-sfu-room-control.ts b/tests/fake-sfu-room-control.ts new file mode 100644 index 0000000..d89d031 --- /dev/null +++ b/tests/fake-sfu-room-control.ts @@ -0,0 +1,63 @@ +import type { SfuResourceFence } from "../src/server/sfu-resource-admission.ts"; +import { + managedSfuRoomName, + type SfuRoomControl, +} from "../src/server/sfu-room-control.ts"; + +export class FakeSfuRoomControl implements SfuRoomControl { + readonly rooms = new Map>(); + readonly created: SfuResourceFence[] = []; + readonly deleted: SfuResourceFence[] = []; + readonly startupDeletedRoomNames: string[] = []; + initializeCalls = 0; + initializeBarrier?: Promise; + initializeError?: Error; + createBarrier?: Promise; + deleteBarrier?: Promise; + failDelete = false; + failHostCheck = false; + + async initialize(): Promise { + this.initializeCalls += 1; + await this.initializeBarrier; + if (this.initializeError) { + throw this.initializeError; + } + this.startupDeletedRoomNames.push(...this.rooms.keys()); + this.rooms.clear(); + } + + async createRoom(fence: SfuResourceFence): Promise { + await this.createBarrier; + const roomName = managedSfuRoomName(fence); + if (this.rooms.has(roomName)) { + throw new Error("room already exists"); + } + this.rooms.set(roomName, new Set()); + this.created.push({ ...fence }); + } + + async deleteRoom(fence: SfuResourceFence): Promise { + await this.deleteBarrier; + if (this.failDelete) { + throw new Error("room deletion failed"); + } + this.rooms.delete(managedSfuRoomName(fence)); + this.deleted.push({ ...fence }); + } + + async hostParticipantExists(fence: SfuResourceFence): Promise { + if (this.failHostCheck) { + throw new Error("host check failed"); + } + return this.rooms.get(managedSfuRoomName(fence))?.has("host") ?? false; + } + + seedRoom(fence: SfuResourceFence, identities: readonly string[] = []): void { + this.rooms.set(managedSfuRoomName(fence), new Set(identities)); + } + + canJoin(roomName: string): boolean { + return this.rooms.has(roomName); + } +} diff --git a/tests/livekit-token.test.ts b/tests/livekit-token.test.ts index 8120fdd..230db06 100644 --- a/tests/livekit-token.test.ts +++ b/tests/livekit-token.test.ts @@ -6,6 +6,7 @@ import { LiveKitTokenIssuer } from "../src/server/livekit-token.ts"; const apiKey = "test-api-key"; const apiSecret = "s".repeat(32); const publicationGeneration = "publication_12345678"; +const shareGeneration = "share_generation_12345678"; const rootPeerId = "viewer_root_12345678"; const secondRootPeerId = "viewer_root_23456789"; const thirdRootPeerId = "viewer_root_34567890"; @@ -26,6 +27,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "42", role: "host", peerId: "host_peer_12345678", + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [rootPeerId, secondRootPeerId], }), @@ -35,6 +37,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "42", role: "viewer", peerId: secondRootPeerId, + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [rootPeerId, secondRootPeerId], }), @@ -44,6 +47,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "42", role: "host", peerId: "host_peer_12345678", + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [ rootPeerId, @@ -60,6 +64,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "42", role: "host", peerId: "host_peer_12345678", + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [rootPeerId], }), @@ -69,17 +74,14 @@ describe("LiveKitTokenIssuer", () => { expect(claims.exp! - claims.nbf!).toBe(5 * 60); expect(claims.video).toMatchObject({ roomJoin: true, - room: `screener-42-${publicationGeneration}`, + room: `screener-v1.42.${shareGeneration}.${publicationGeneration}`, canPublish: true, canSubscribe: false, canPublishData: false, canUpdateOwnMetadata: false, canPublishSources: ["screen_share", "screen_share_audio"], }); - expect(claims.roomConfig).toMatchObject({ - name: `screener-42-${publicationGeneration}`, - maxParticipants: 9, - }); + expect(claims.roomConfig).toBeUndefined(); }); it("issues a subscribe-only token to a current fallback root", async () => { @@ -88,6 +90,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "7", role: "viewer", peerId: rootPeerId, + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [rootPeerId], }), @@ -96,14 +99,14 @@ describe("LiveKitTokenIssuer", () => { expect(claims.sub).toBe(`viewer:${rootPeerId}`); expect(claims.video).toMatchObject({ roomJoin: true, - room: `screener-7-${publicationGeneration}`, + room: `screener-v1.7.${shareGeneration}.${publicationGeneration}`, canPublish: false, canSubscribe: true, canPublishData: false, canUpdateOwnMetadata: false, }); expect(claims.video?.canPublishSources).toBeUndefined(); - expect(claims.roomConfig?.maxParticipants).toBe(4); + expect(claims.roomConfig).toBeUndefined(); }); it("does not issue viewer tokens outside the current root allowlist", async () => { @@ -112,6 +115,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "7", role: "viewer", peerId: "viewer_other_12345678", + shareGeneration, publicationGeneration, allowlistedRootPeerIds: [rootPeerId], }), @@ -124,6 +128,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "7", role: "host" as const, peerId: "host_peer_12345678", + shareGeneration, publicationGeneration, }; @@ -154,6 +159,7 @@ describe("LiveKitTokenIssuer", () => { roomId: "7", role: "host" as const, peerId: "host_peer_12345678", + shareGeneration, allowlistedRootPeerIds: [rootPeerId], }; const first = await verifier.verify( diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index ffb8606..056d082 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from "vitest"; import { loadConfig } from "../src/server/config.ts"; const liveKitAdmission = { + LIVEKIT_API_URL: "https://livekit-api.test", SFU_INGRESS_CAPACITY: "4", SFU_EGRESS_CAPACITY: "16", } as const; @@ -72,6 +73,7 @@ describe("server configuration", () => { const config = loadConfig({ PEER_ASSISTED_MEDIA: "true", LIVEKIT_URL: " ws://livekit.test:7880 ", + LIVEKIT_API_URL: " http://livekit.test:7880 ", LIVEKIT_API_KEY: " test-key ", LIVEKIT_API_SECRET: ` ${"s".repeat(32)} `, SFU_INGRESS_CAPACITY: " 4 ", @@ -80,6 +82,7 @@ describe("server configuration", () => { expect(config.livekitFallback).toEqual({ url: "ws://livekit.test:7880", + apiUrl: "http://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), ingressCapacity: 4, @@ -93,6 +96,7 @@ describe("server configuration", () => { const config = loadConfig({ PEER_ASSISTED_MEDIA: "true", LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_URL: "https://livekit-api.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), SFU_INGRESS_CAPACITY: capacity, @@ -124,15 +128,17 @@ describe("server configuration", () => { it.each([ { LIVEKIT_URL: "wss://livekit.test" }, + { LIVEKIT_API_URL: "https://livekit-api.test" }, { LIVEKIT_API_KEY: "test-key" }, { LIVEKIT_API_SECRET: "s".repeat(32) }, { LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_URL: "https://livekit-api.test", LIVEKIT_API_KEY: "test-key", }, ])("rejects a partial LiveKit credential tuple", (partial) => { expect(() => loadConfig(partial)).toThrow( - "LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", + "LIVEKIT_URL, LIVEKIT_API_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", ); }); @@ -168,6 +174,53 @@ describe("server configuration", () => { ).toThrow("LIVEKIT_URL must use wss in production"); }); + it.each([ + "ws://livekit-api.test", + "http://user:pass@livekit-api.test", + "http://livekit-api.test/rtc", + "http://livekit-api.test?token=value", + "http://livekit-api.test#fragment", + ])("rejects an invalid LiveKit control origin: %s", (apiUrl) => { + expect(() => + loadConfig({ + PEER_ASSISTED_MEDIA: "true", + LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_URL: apiUrl, + LIVEKIT_API_KEY: "test-key", + LIVEKIT_API_SECRET: "s".repeat(32), + SFU_INGRESS_CAPACITY: "4", + SFU_EGRESS_CAPACITY: "16", + }), + ).toThrow("LIVEKIT_API_URL"); + }); + + it("allows only HTTPS or loopback LiveKit control in production", () => { + const production = { + NODE_ENV: "production", + PUBLIC_BASE_URL: "https://share.test", + SITE_ACCESS_PASSWORD: "host-password-12", + STUN_URLS: "stun:stun.test:3478", + PEER_ASSISTED_MEDIA: "true", + LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_KEY: "test-key", + LIVEKIT_API_SECRET: "s".repeat(32), + SFU_INGRESS_CAPACITY: "4", + SFU_EGRESS_CAPACITY: "16", + }; + expect(() => + loadConfig({ + ...production, + LIVEKIT_API_URL: "http://livekit-api.test:7880", + }), + ).toThrow("must use https or loopback"); + expect( + loadConfig({ + ...production, + LIVEKIT_API_URL: "http://127.0.0.1:7880", + }).livekitFallback?.apiUrl, + ).toBe("http://127.0.0.1:7880"); + }); + it("rejects a short LiveKit API secret and invalid SFU capacities", () => { const fallback = { LIVEKIT_URL: "wss://livekit.test", @@ -192,6 +245,7 @@ describe("server configuration", () => { it("requires both SFU capacities with LiveKit and rejects the removed root key", () => { const fallback = { LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_URL: "https://livekit-api.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), }; diff --git a/tests/server-http.test.ts b/tests/server-http.test.ts index 3ed2506..49f965b 100644 --- a/tests/server-http.test.ts +++ b/tests/server-http.test.ts @@ -2,7 +2,7 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { createRoomResponseSchema } from "../src/shared/protocol.ts"; import { @@ -11,6 +11,7 @@ import { type ScreenerServer, } from "../src/server/app.ts"; import type { ServerConfig } from "../src/server/config.ts"; +import { FakeSfuRoomControl } from "./fake-sfu-room-control.ts"; const allowedOrigin = "http://allowed.test"; const siteAccessPassword = "instance-access-password"; @@ -53,11 +54,15 @@ async function start( config = testConfig(), options: Pick< CreateServerOptions, - "now" | "siteAccessTtlSeconds" | "sfuTokenIssuer" + "now" | "siteAccessTtlSeconds" | "sfuTokenIssuer" | "sfuRoomControl" > = {}, ): Promise { + const sfuRoomControl = + options.sfuRoomControl ?? + (config.livekitFallback ? new FakeSfuRoomControl() : undefined); runningServer = await createScreenerServer({ ...options, + ...(sfuRoomControl ? { sfuRoomControl } : {}), config, serveFrontend: false, }); @@ -384,6 +389,7 @@ describe("server HTTP listener and health", () => { stunUrls: ["stun:stun.example.test:3478"], livekitFallback: { url: "ws://livekit.test:7880", + apiUrl: "http://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), ingressCapacity: 4, @@ -417,6 +423,7 @@ describe("server HTTP listener and health", () => { peerAssistedMedia: true, livekitFallback: { url: "ws://livekit.test:7880", + apiUrl: "http://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), ingressCapacity: 4, @@ -434,6 +441,215 @@ describe("server HTTP listener and health", () => { expect(response.status).toBe(200); }); + it("clears a stale managed LiveKit room before a restarted server serves traffic", async () => { + const roomControl = new FakeSfuRoomControl(); + const staleFence = { + roomId: "42", + shareGeneration: "stale_share_generation", + publicationGeneration: "stale_publication_generation", + }; + roomControl.seedRoom(staleFence, ["host", "viewer:stale"]); + const baseUrl = await start( + testConfig({ + peerAssistedMedia: true, + livekitFallback: { + url: "ws://livekit.test:7880", + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + ingressCapacity: 4, + egressCapacity: 16, + }, + }), + { + sfuTokenIssuer: { + issueToken: async () => "unused-test-token", + }, + sfuRoomControl: roomControl, + }, + ); + + expect(roomControl.initializeCalls).toBe(1); + expect(roomControl.startupDeletedRoomNames).toHaveLength(1); + expect(roomControl.rooms.size).toBe(0); + expect((await fetch(`${baseUrl}/healthz`)).status).toBe(200); + }); + + it("binds the application listener before LiveKit reconciliation", async () => { + let releaseInitialization!: () => void; + const roomControl = new FakeSfuRoomControl(); + roomControl.initializeBarrier = new Promise((resolve) => { + releaseInitialization = resolve; + }); + runningServer = await createScreenerServer({ + config: testConfig({ + peerAssistedMedia: true, + livekitFallback: { + url: "ws://livekit.test:7880", + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + ingressCapacity: 4, + egressCapacity: 16, + }, + }), + serveFrontend: false, + sfuRoomControl: roomControl, + sfuTokenIssuer: { + issueToken: async () => "unused-test-token", + }, + }); + + const listen = runningServer.listen(0, "127.0.0.1"); + try { + await vi.waitFor(() => { + expect(roomControl.initializeCalls).toBe(1); + expect(runningServer?.httpServer.address()).not.toBeNull(); + }); + const address = runningServer.httpServer.address(); + if (!address || typeof address === "string") { + throw new Error("Expected a TCP server address"); + } + expect(runningServer.httpServer.listenerCount("upgrade")).toBe(0); + const response = await fetch(`http://127.0.0.1:${address.port}/healthz`); + expect(response.status).toBe(503); + expect(response.headers.get("retry-after")).toBe("1"); + } finally { + releaseInitialization(); + } + + await expect(listen).resolves.toBeGreaterThan(0); + expect(runningServer.httpServer.listenerCount("upgrade")).toBe(1); + }); + + it("makes no LiveKit call when another process owns the listener", async () => { + const owner = await createScreenerServer({ + config: testConfig(), + serveFrontend: false, + }); + const ownerPort = await owner.listen(0, "127.0.0.1"); + const roomControl = new FakeSfuRoomControl(); + const contender = await createScreenerServer({ + config: testConfig({ + peerAssistedMedia: true, + livekitFallback: { + url: "ws://livekit.test:7880", + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + ingressCapacity: 4, + egressCapacity: 16, + }, + }), + serveFrontend: false, + sfuRoomControl: roomControl, + sfuTokenIssuer: { + issueToken: async () => "unused-test-token", + }, + }); + + try { + await expect(contender.listen(ownerPort, "127.0.0.1")).rejects.toMatchObject({ + code: "EADDRINUSE", + }); + expect(roomControl.initializeCalls).toBe(0); + } finally { + await contender.close(); + await owner.close(); + } + }); + + it("holds listener ownership until an in-flight startup reconciliation settles", async () => { + let releaseInitialization!: () => void; + const roomControl = new FakeSfuRoomControl(); + roomControl.initializeBarrier = new Promise((resolve) => { + releaseInitialization = resolve; + }); + runningServer = await createScreenerServer({ + config: testConfig({ + peerAssistedMedia: true, + livekitFallback: { + url: "ws://livekit.test:7880", + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + ingressCapacity: 4, + egressCapacity: 16, + }, + }), + serveFrontend: false, + sfuRoomControl: roomControl, + sfuTokenIssuer: { + issueToken: async () => "unused-test-token", + }, + }); + const listen = runningServer.listen(0, "127.0.0.1"); + await vi.waitFor(() => { + expect(roomControl.initializeCalls).toBe(1); + expect(runningServer?.httpServer.address()).not.toBeNull(); + }); + const address = runningServer.httpServer.address(); + if (!address || typeof address === "string") { + throw new Error("Expected a TCP server address"); + } + + let closed = false; + const closing = runningServer.close().then(() => { + closed = true; + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(closed).toBe(false); + expect(runningServer.httpServer.listening).toBe(true); + + const contender = await createScreenerServer({ + config: testConfig(), + serveFrontend: false, + }); + try { + await expect( + contender.listen(address.port, "127.0.0.1"), + ).rejects.toMatchObject({ code: "EADDRINUSE" }); + } finally { + await contender.close(); + } + + releaseInitialization(); + await expect(listen).resolves.toBe(address.port); + await closing; + runningServer = undefined; + expect(closed).toBe(true); + }); + + it("closes the listener when LiveKit startup reconciliation fails", async () => { + const roomControl = new FakeSfuRoomControl(); + roomControl.initializeError = new Error("startup reconciliation failed"); + runningServer = await createScreenerServer({ + config: testConfig({ + peerAssistedMedia: true, + livekitFallback: { + url: "ws://livekit.test:7880", + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + ingressCapacity: 4, + egressCapacity: 16, + }, + }), + serveFrontend: false, + sfuRoomControl: roomControl, + sfuTokenIssuer: { + issueToken: async () => "unused-test-token", + }, + }); + + await expect(runningServer.listen(0, "127.0.0.1")).rejects.toThrow( + "startup reconciliation failed", + ); + expect(roomControl.initializeCalls).toBe(1); + expect(runningServer.httpServer.listening).toBe(false); + expect(runningServer.httpServer.listenerCount("upgrade")).toBe(0); + }); + it("uses the configured listen host by default", async () => { runningServer = await createScreenerServer({ config: testConfig(), diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 7e7eb07..0606735 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -26,6 +26,8 @@ import { SignalingServer } from "../src/server/signaling.ts"; import type { SfuTokenIssuer } from "../src/server/livekit-token.ts"; import { HybridMediaRouter } from "../src/server/hybrid-media-router.ts"; import { SfuResourceAdmission } from "../src/server/sfu-resource-admission.ts"; +import { managedSfuRoomName } from "../src/server/sfu-room-control.ts"; +import { FakeSfuRoomControl } from "./fake-sfu-room-control.ts"; const allowedOrigin = "http://allowed.test"; const TEST_SFU_INGRESS_CAPACITY = 10; @@ -146,6 +148,7 @@ interface SignalHarness { room: CreatedRoom; database?: RoomDatabase; sfuAdmission?: SfuResourceAdmission; + sfuRoomControl?: FakeSfuRoomControl; } function testConfig(): ServerConfig { @@ -650,6 +653,9 @@ async function startSfuHarness(options: { viewerDisconnectGraceMs?: number; stunUrls?: readonly string[]; now?: () => number; + roomControl?: FakeSfuRoomControl; + hostOfflineCheckMs?: number; + drainRetryMs?: number; }): Promise { const roomStore = new RoomStore({ ttlMs: 14_400_000, @@ -671,6 +677,7 @@ async function startSfuHarness(options: { egressCapacity: options.sfuEgressCapacity ?? TEST_SFU_EGRESS_CAPACITY, }); + const sfuRoomControl = options.roomControl ?? new FakeSfuRoomControl(); const signaling = new SignalingServer({ server: httpServer, roomStore, @@ -681,7 +688,10 @@ async function startSfuHarness(options: { url: "wss://sfu.example.test", tokenIssuer: options.tokenIssuer, admission: sfuAdmission, + roomControl: sfuRoomControl, prepareTimeoutMs: options.prepareTimeoutMs, + hostOfflineCheckMs: options.hostOfflineCheckMs, + drainRetryMs: options.drainRetryMs, }, ...(options.selectedEdgeTurn ? { @@ -735,6 +745,7 @@ async function startSfuHarness(options: { roomStore, room, sfuAdmission, + sfuRoomControl, }; } @@ -1515,6 +1526,14 @@ async function closeClient(client: TestClient): Promise { await closed; } +function deferredVoid(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + describe("WebSocket signaling", () => { it("answers only opted-in current sockets and rate-limits each socket", async () => { let now = Date.now(); @@ -10055,4 +10074,258 @@ describe("WebSocket signaling", () => { ); expect(await rejectedUpgradeStatus(harness.webSocketUrl)).toBe(503); }); + + it("keeps a draining generation charged and blocks a second room until deletion proof", async () => { + let issuedTokens = 0; + const roomControl = new FakeSfuRoomControl(); + const harness = await startSfuHarness({ + roomControl, + sfuIngressCapacity: 1, + sfuEgressCapacity: 1, + tokenIssuer: { + async issueToken({ peerId }) { + issuedTokens += 1; + return `token-${peerId}`; + }, + }, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "draining-capacity", + ); + const resourceFence = roomControl.created[0]!; + const staleTokenRoom = managedSfuRoomName(resourceFence); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + + const deletion = deferredVoid(); + roomControl.deleteBarrier = deletion.promise; + active.host.socket.send(JSON.stringify({ + type: "stop-sharing", + shareGeneration: "draining-capacity-share", + })); + await active.viewer.inbox.next("sharing-stopped"); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + expect(roomControl.canJoin(staleTokenRoom)).toBe(true); + + const secondRoom = harness.roomStore.createRoom(); + const second = await failSingleViewerDirect( + harness.webSocketUrl, + secondRoom, + "draining-second-room", + ); + expect((await second.viewer.inbox.next("error")).message).toBe( + "SFU server capacity is exhausted", + ); + expect(roomControl.created).toHaveLength(1); + expect(issuedTokens).toBe(2); + + deletion.resolve(); + await vi.waitFor(() => + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }), + ); + expect(roomControl.canJoin(staleTokenRoom)).toBe(false); + }); + + it("keeps two active rooms isolated under one deployment-wide owner", async () => { + const roomControl = new FakeSfuRoomControl(); + const harness = await startSfuHarness({ + roomControl, + sfuIngressCapacity: 2, + sfuEgressCapacity: 2, + tokenIssuer: { + async issueToken({ peerId }) { + return `token-${peerId}`; + }, + }, + }); + const first = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "two-room-first", + ); + const secondRoom = harness.roomStore.createRoom(); + const second = await activateSingleViewerSfu( + harness.webSocketUrl, + secondRoom, + "two-room-second", + ); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 2, egress: 2 }); + expect(new Set(roomControl.created.map(({ roomId }) => roomId))).toEqual( + new Set([harness.room.roomId, secondRoom.roomId]), + ); + + first.host.socket.send(JSON.stringify({ + type: "stop-sharing", + shareGeneration: "two-room-first-share", + })); + await first.viewer.inbox.next("sharing-stopped"); + await vi.waitFor(() => + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }), + ); + expect( + roomControl.rooms.has(managedSfuRoomName(roomControl.created[1]!)), + ).toBe(true); + + second.viewer.socket.send(JSON.stringify({ + type: "refresh-sfu", + revision: second.revision, + })); + expect(await second.viewer.inbox.next("sfu-config")).toMatchObject({ + revision: second.revision, + }); + }); + + it("keeps a signaling-offline Host charged while LiveKit is live and reclaims it after departure", async () => { + const roomControl = new FakeSfuRoomControl(); + const error = vi.spyOn(console, "error").mockImplementation(() => undefined); + const harness = await startSfuHarness({ + roomControl, + hostOfflineCheckMs: 20, + tokenIssuer: { + async issueToken({ peerId }) { + return `token-${peerId}`; + }, + }, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "host-offline", + ); + const resourceFence = roomControl.created[0]!; + roomControl.seedRoom(resourceFence, ["host"]); + roomControl.failHostCheck = true; + + await closeClient(active.host); + await new Promise((resolve) => setTimeout(resolve, 55)); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + expect(roomControl.rooms.has(managedSfuRoomName(resourceFence))).toBe(true); + expect(error).toHaveBeenCalledWith( + "SFU Host-offline check failed; capacity remains reserved", + ); + + roomControl.failHostCheck = false; + await new Promise((resolve) => setTimeout(resolve, 55)); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + roomControl.rooms.get(managedSfuRoomName(resourceFence))?.delete("host"); + await vi.waitFor( + () => + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }), + { timeout: 300 }, + ); + expect(roomControl.rooms.has(managedSfuRoomName(resourceFence))).toBe(false); + error.mockRestore(); + }); + + it("keeps failed room deletion charged until a retry proves absence", async () => { + const roomControl = new FakeSfuRoomControl(); + const error = vi.spyOn(console, "error").mockImplementation(() => undefined); + const harness = await startSfuHarness({ + roomControl, + drainRetryMs: 20, + tokenIssuer: { + async issueToken({ peerId }) { + return `token-${peerId}`; + }, + }, + }); + const active = await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "delete-retry", + ); + const resourceFence = roomControl.created[0]!; + roomControl.failDelete = true; + + active.host.socket.send(JSON.stringify({ + type: "stop-sharing", + shareGeneration: "delete-retry-share", + })); + await active.viewer.inbox.next("sharing-stopped"); + await new Promise((resolve) => setTimeout(resolve, 55)); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + expect(roomControl.rooms.has(managedSfuRoomName(resourceFence))).toBe(true); + expect(error).toHaveBeenCalledWith( + "SFU room drain failed; capacity remains reserved", + ); + + roomControl.failDelete = false; + await vi.waitFor( + () => + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }), + { timeout: 300 }, + ); + expect(roomControl.rooms.has(managedSfuRoomName(resourceFence))).toBe(false); + error.mockRestore(); + }); + + it("waits for managed LiveKit rooms to drain before graceful shutdown completes", async () => { + const roomControl = new FakeSfuRoomControl(); + const harness = await startSfuHarness({ + roomControl, + tokenIssuer: { + async issueToken({ peerId }) { + return `token-${peerId}`; + }, + }, + }); + await activateSingleViewerSfu( + harness.webSocketUrl, + harness.room, + "graceful-drain", + ); + const secondRoom = harness.roomStore.createRoom(); + const secondHost = await openClient(harness.webSocketUrl); + await authenticate( + secondHost, + secondRoom, + "host", + "graceful-second-host", + 1, + "graceful-second-share", + ); + const secondViewer = await openClient(harness.webSocketUrl); + const secondViewerAuth = peerAssisted(await authenticate( + secondViewer, + secondRoom, + "viewer", + "graceful-second-viewer", + )); + secondHost.socket.send(JSON.stringify({ + type: "signal", + targetPeerId: secondViewerAuth.peerId, + payload: { + kind: "description", + connectionId: "graceful-second-direct", + description: { type: "offer", sdp: "v=0\r\n" }, + }, + })); + await secondViewer.inbox.next("signal"); + const deletion = deferredVoid(); + roomControl.deleteBarrier = deletion.promise; + + let closed = false; + const closing = runningServer!.close().then(() => { + closed = true; + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(closed).toBe(false); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); + secondViewer.socket.send(JSON.stringify({ + type: "route-failed", + revision: secondViewerAuth.routeRevision, + phase: "active", + connectionId: "graceful-second-direct", + })); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(roomControl.created).toHaveLength(1); + + deletion.resolve(); + await closing; + runningServer = undefined; + expect(closed).toBe(true); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); + expect(roomControl.created).toHaveLength(1); + }); }); diff --git a/tests/sfu-resource-admission.test.ts b/tests/sfu-resource-admission.test.ts index 14c7c50..c122a19 100644 --- a/tests/sfu-resource-admission.test.ts +++ b/tests/sfu-resource-admission.test.ts @@ -38,7 +38,7 @@ describe("SfuResourceAdmission", () => { expect(admission.usage()).toEqual({ ingress: 2, egress: 4 }); }); - it("charges old and candidate generations until exact commit", () => { + it("charges old and candidate generations until exact drain proof", () => { const admission = new SfuResourceAdmission({ ingressCapacity: 2, egressCapacity: 5, @@ -46,14 +46,16 @@ describe("SfuResourceAdmission", () => { const first = fence("1", "publication_first"); const second = fence("1", "publication_second"); expect(admission.reserve(first, 1)).toBe(true); - expect(admission.commit(first)).toBe(true); + expect(admission.commit(first)).toEqual([]); expect(admission.reserve(second, 2)).toBe(true); expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); - expect(admission.commit(fence("1", "publication_stale"))).toBe(false); - expect(admission.commit(second)).toBe(true); + expect(admission.commit(fence("1", "publication_stale"))).toBeNull(); + expect(admission.commit(second)).toEqual([first]); + expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); + expect(admission.completeDrain(first)).toBe(true); expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); - expect(admission.release(first)).toBe(false); + expect(admission.completeDrain(first)).toBe(false); }); it("makes duplicate operations idempotent without weakening fences", () => { @@ -68,42 +70,55 @@ describe("SfuResourceAdmission", () => { expect(admission.reserve(fence("1", "publication_other"), 1)).toBe(false); expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); - expect(admission.release(pending)).toBe(true); - expect(admission.release(pending)).toBe(false); + expect(admission.beginDrain(pending)).toBe(true); + expect(admission.beginDrain(pending)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); + expect(admission.completeDrain(pending)).toBe(true); + expect(admission.completeDrain(pending)).toBe(false); expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); }); - it("releases a room's committed and reserved resources together", () => { + it("moves a room's committed and reserved resources into draining together", () => { const admission = new SfuResourceAdmission({ ingressCapacity: 2, egressCapacity: 4, }); const active = fence("1", "publication_active"); expect(admission.reserve(active, 1)).toBe(true); - expect(admission.commit(active)).toBe(true); + expect(admission.commit(active)).toEqual([]); expect(admission.reserve(fence("1", "publication_candidate"), 2)).toBe( true, ); - expect(admission.releaseRoom("1")).toBe(true); - expect(admission.releaseRoom("1")).toBe(false); + const draining = admission.beginDrainRoom("1"); + expect(draining).toHaveLength(2); + expect(admission.beginDrainRoom("1")).toEqual(draining); + expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); + for (const resourceFence of draining) { + expect(admission.completeDrain(resourceFence)).toBe(true); + } expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); }); - it("releases every room when its process owner closes", () => { + it("moves every room to draining before its process owner closes", () => { const admission = new SfuResourceAdmission({ ingressCapacity: 2, egressCapacity: 3, }); const active = fence("1", "publication_active"); expect(admission.reserve(active, 1)).toBe(true); - expect(admission.commit(active)).toBe(true); + expect(admission.commit(active)).toEqual([]); expect(admission.reserve(fence("2", "publication_pending"), 2)).toBe( true, ); - admission.releaseAll(); - admission.releaseAll(); + const draining = admission.beginDrainAll(); + expect(draining).toHaveLength(2); + expect(admission.beginDrainAll()).toEqual(draining); + expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); + for (const resourceFence of draining) { + expect(admission.completeDrain(resourceFence)).toBe(true); + } expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); }); }); diff --git a/tests/sfu-room-control.test.ts b/tests/sfu-room-control.test.ts new file mode 100644 index 0000000..15d4b88 --- /dev/null +++ b/tests/sfu-room-control.test.ts @@ -0,0 +1,189 @@ +import { readFileSync } from "node:fs"; + +import { ServerError } from "livekit-server-sdk"; +import { describe, expect, it, vi } from "vitest"; + +import type { SfuResourceFence } from "../src/server/sfu-resource-admission.ts"; +import { + LiveKitSfuRoomControl, + isManagedSfuRoomName, + managedSfuRoomName, + type LiveKitRoomService, +} from "../src/server/sfu-room-control.ts"; + +class FakeRoomService implements LiveKitRoomService { + readonly rooms = new Map>(); + readonly operations: string[] = []; + readonly createOptions: Array<{ name: string; maxParticipants: number }> = []; + createBarrier?: Promise; + retainDeletedRoom = false; + + async createRoom(options: { name: string; maxParticipants: number }) { + this.operations.push(`create:${options.name}`); + this.createOptions.push(options); + await this.createBarrier; + if (this.rooms.has(options.name)) { + throw new Error("room already exists"); + } + this.rooms.set(options.name, new Set()); + return { name: options.name }; + } + + async listRooms(names?: string[]) { + const selected = names?.length + ? [...this.rooms.keys()].filter((name) => names.includes(name)) + : [...this.rooms.keys()]; + return selected.map((name) => ({ name })); + } + + async deleteRoom(room: string): Promise { + this.operations.push(`delete:${room}`); + if (!this.rooms.has(room)) { + throw new ServerError("Not Found", "missing", 404, "not_found"); + } + if (!this.retainDeletedRoom) { + this.rooms.delete(room); + } + } + + async listParticipants(room: string) { + const participants = this.rooms.get(room); + if (!participants) { + throw new ServerError("Not Found", "missing", 404, "not_found"); + } + return [...participants].map((identity) => ({ identity })); + } + + join(room: string, identity: string): boolean { + const participants = this.rooms.get(room); + if (!participants) { + return false; + } + participants.add(identity); + return true; + } +} + +function fence( + roomId = "42", + shareGeneration = "share_generation_12345678", + publicationGeneration = "publication_generation_12345678", +): SfuResourceFence { + return { roomId, shareGeneration, publicationGeneration }; +} + +function control(roomService: FakeRoomService): LiveKitSfuRoomControl { + return new LiveKitSfuRoomControl({ + apiUrl: "http://127.0.0.1:7880", + apiKey: "test-key", + apiSecret: "s".repeat(32), + maxViewersPerRoom: 8, + roomService, + }); +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve }; +} + +describe("LiveKitSfuRoomControl", () => { + it("drains current and legacy managed rooms before startup", async () => { + const roomService = new FakeRoomService(); + const current = managedSfuRoomName(fence()); + const legacy = "screener-7-publication_12345678"; + roomService.rooms.set(current, new Set(["host"])); + roomService.rooms.set(legacy, new Set(["host"])); + + await expect(control(roomService).initialize()).resolves.toBeUndefined(); + expect(roomService.rooms.size).toBe(0); + expect(roomService.operations).toEqual([ + `delete:${current}`, + `delete:${legacy}`, + ]); + }); + + it("rejects a shared instance and an unconfirmed startup drain", async () => { + const foreign = new FakeRoomService(); + foreign.rooms.set("another-application", new Set()); + await expect(control(foreign).initialize()).rejects.toThrow("foreign room"); + expect(foreign.operations).toEqual([]); + + const retained = new FakeRoomService(); + retained.retainDeletedRoom = true; + retained.rooms.set(managedSfuRoomName(fence()), new Set()); + await expect(control(retained).initialize()).rejects.toThrow( + "not empty after startup drain", + ); + }); + + it("serializes create and delete and leaves a stale token with no room", async () => { + const roomService = new FakeRoomService(); + const roomControl = control(roomService); + const resourceFence = fence(); + const roomName = managedSfuRoomName(resourceFence); + const gate = deferred(); + roomService.createBarrier = gate.promise; + + const creating = roomControl.createRoom(resourceFence); + await vi.waitFor(() => + expect(roomService.operations).toEqual([`create:${roomName}`]), + ); + const deleting = roomControl.deleteRoom(resourceFence); + await Promise.resolve(); + expect(roomService.operations).toEqual([`create:${roomName}`]); + + gate.resolve(); + await Promise.all([creating, deleting]); + expect(roomService.operations).toEqual([ + `create:${roomName}`, + `delete:${roomName}`, + ]); + expect(roomService.createOptions[0]?.maxParticipants).toBe(9); + expect(roomService.join(roomName, "host")).toBe(false); + }); + + it("rejects an off-ledger exact room before issuing another generation", async () => { + const roomService = new FakeRoomService(); + const resourceFence = fence(); + roomService.rooms.set(managedSfuRoomName(resourceFence), new Set(["host"])); + + await expect(control(roomService).createRoom(resourceFence)).rejects.toThrow( + "already exists", + ); + expect(roomService.operations).toEqual([]); + }); + + it("checks the exact Host participant and treats an absent room as offline", async () => { + const roomService = new FakeRoomService(); + const roomControl = control(roomService); + const resourceFence = fence(); + await roomControl.createRoom(resourceFence); + const roomName = managedSfuRoomName(resourceFence); + + expect(roomService.join(roomName, "viewer:one")).toBe(true); + await expect( + roomControl.hostParticipantExists(resourceFence), + ).resolves.toBe(false); + expect(roomService.join(roomName, "host")).toBe(true); + await expect( + roomControl.hostParticipantExists(resourceFence), + ).resolves.toBe(true); + await roomControl.deleteRoom(resourceFence); + await expect( + roomControl.hostParticipantExists(resourceFence), + ).resolves.toBe(false); + }); + + it("keeps the tracked deployment unable to auto-create rooms", () => { + const configuration = readFileSync( + new URL("../deploy/livekit/livekit.yaml.example", import.meta.url), + "utf8", + ); + expect(configuration).toMatch(/room:\s+[\s\S]*auto_create: false/); + expect(isManagedSfuRoomName("screener-v1.42.bad.segment.extra")).toBe(false); + }); +});