From 524f012739bf29210a7f2ef8ad3e20c86974ed2c Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sat, 22 Aug 2026 23:16:37 +0800 Subject: [PATCH] refactor(route): enforce global SFU admission --- .env.example | 14 ++- package.json | 2 +- scripts/peer-assisted-benchmark.ts | 4 +- src/server/app.ts | 7 +- src/server/config.ts | 50 ++++++-- src/server/hybrid-media-router.ts | 71 +++++++---- src/server/livekit-token.ts | 15 +-- src/server/media-route-controller.ts | 8 +- src/server/sfu-resource-admission.ts | 175 ++++++++++++++++++++++++++ src/shared/protocol.ts | 1 - tests/livekit-token.test.ts | 12 +- tests/media-route-controller.test.ts | 30 +++-- tests/peer-assisted-benchmark.test.ts | 20 +-- tests/route-transition.test.ts | 2 +- tests/server-config.test.ts | 66 +++++++--- tests/server-http.test.ts | 6 +- tests/server-signal.test.ts | 40 +++++- tests/sfu-client.test.ts | 6 +- tests/sfu-resource-admission.test.ts | 109 ++++++++++++++++ 19 files changed, 519 insertions(+), 119 deletions(-) create mode 100644 src/server/sfu-resource-admission.ts create mode 100644 tests/sfu-resource-admission.test.ts diff --git a/.env.example b/.env.example index 370e82a..d3c980e 100644 --- a/.env.example +++ b/.env.example @@ -23,14 +23,15 @@ PEER_ASSISTED_MEDIA=false # A peer child or Host SFU publication consumes one copy. Valid values: 1, 2, 3. ENDPOINT_MEDIA_COPY_CAPACITY=2 -# Optional automatic SFU fallback capacity. Set PEER_ASSISTED_MEDIA=true and -# supply all three values together; leaving them empty keeps LiveKit completely -# dormant. Production requires wss and an independent API secret of 32+ bytes. +# Optional automatic SFU fallback. Set PEER_ASSISTED_MEDIA=true and supply all +# five 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_KEY= LIVEKIT_API_SECRET= -# Effective only when the complete LiveKit tuple is configured. Valid range: 1-2. -MAX_SFU_ROOTS_PER_ROOM=2 +SFU_INGRESS_CAPACITY= +SFU_EGRESS_CAPACITY= # Optional one-shot TURN/UDP transport for one controller-selected edge after # peer and SFU recovery are exhausted. Set all three values together. @@ -51,5 +52,6 @@ STUN_URLS= # LIVEKIT_URL=wss://share.example.com # LIVEKIT_API_KEY=replace-with-a-livekit-api-key # LIVEKIT_API_SECRET=replace-with-at-least-32-independent-random-bytes -# MAX_SFU_ROOTS_PER_ROOM=2 +# SFU_INGRESS_CAPACITY=replace-with-measured-positive-integer +# SFU_EGRESS_CAPACITY=replace-with-measured-positive-integer # ENDPOINT_MEDIA_COPY_CAPACITY=2 diff --git a/package.json b/package.json index 5c57113..cb7e358 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-root-invariants": "vitest run 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 -t \"SFU root invariant gate\"", + "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: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/scripts/peer-assisted-benchmark.ts b/scripts/peer-assisted-benchmark.ts index 3eb6054..704a8bb 100644 --- a/scripts/peer-assisted-benchmark.ts +++ b/scripts/peer-assisted-benchmark.ts @@ -551,7 +551,7 @@ function inspectSfuPublication(pages: readonly PageObservation[]) { }; } - if (sfuViewers.length > 2 || publicationOwners.length !== 1) { + if (publicationOwners.length !== 1) { return { observed: true, rootCount: sfuViewers.length, coherent: false }; } const host = publicationOwners[0]!; @@ -872,7 +872,7 @@ export function buildRunChecks( passed: summary.sfuPublicationCoherent, actual: summary.sfuRootCount, expected: - "1-2 active SFU roots, one Host publication, and one active revision across all participants", + "admitted SFU subscriptions, one Host publication, and one active revision across all participants", } : { name: "no-orphan-sfu-publication", diff --git a/src/server/app.ts b/src/server/app.ts index 36c93f2..9c23418 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -12,6 +12,7 @@ import { SiteAccess } from "./access-session.js"; import { loadConfig, type ServerConfig } from "./config.js"; import { createIceConfig } from "./ice.js"; import type { SfuTokenIssuer } from "./livekit-token.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"; @@ -56,9 +57,11 @@ export async function createScreenerServer( apiKey: livekitFallback.apiKey, apiSecret: livekitFallback.apiSecret, maxViewersPerRoom: config.maxViewersPerRoom, - maxSfuRootsPerRoom: livekitFallback.maxSfuRootsPerRoom, }), - maxRoots: livekitFallback.maxSfuRootsPerRoom, + admission: new SfuResourceAdmission({ + ingressCapacity: livekitFallback.ingressCapacity, + egressCapacity: livekitFallback.egressCapacity, + }), } : undefined; const roomStore = diff --git a/src/server/config.ts b/src/server/config.ts index 4c7c056..48c341e 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -1,5 +1,4 @@ import { - CURRENT_SFU_ROOT_LIMIT, MAX_ICE_SERVER_URLS, MAX_VIEWERS_PER_ROOM_LIMIT, stunUrlSchema, @@ -20,7 +19,6 @@ const MAX_SELECTED_EDGE_TURN_SECRET_BYTES = 128; const MIN_SELECTED_EDGE_TURN_TTL_SECONDS = 60; const MAX_SELECTED_EDGE_TURN_TTL_SECONDS = 10 * 60; const DEFAULT_MAX_VIEWERS_PER_ROOM = 8; -const DEFAULT_MAX_SFU_ROOTS_PER_ROOM = CURRENT_SFU_ROOT_LIMIT; const VISIBLE_ASCII_PATTERN = /^[\x21-\x7e]+$/; const REMOVED_ENVIRONMENT_VARIABLES = [ "TURN_URLS", @@ -32,13 +30,15 @@ const REMOVED_ENVIRONMENT_VARIABLES = [ "PEER_ASSISTED_ROOM_IDS", "HOST_ADMISSION_PASSWORD", "MAX_PEER_RELAY_DOWNSTREAM_EDGES", + "MAX_SFU_ROOTS_PER_ROOM", ] as const; export interface LiveKitFallbackConfig { url: string; apiKey: string; apiSecret: string; - maxSfuRootsPerRoom: number; + ingressCapacity: number; + egressCapacity: number; } export interface SelectedEdgeTurnConfig { @@ -94,6 +94,20 @@ function parsePositiveInteger( return parsed; } +function parseRequiredPositiveInteger( + value: string | undefined, + name: string, +): number { + if (value === undefined || value === "") { + throw new Error(`${name} must be a positive integer`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new Error(`${name} must be a positive integer`); + } + return parsed; +} + function parseBoundedInteger( value: string | undefined, fallback: number, @@ -128,8 +142,18 @@ function parseLiveKitFallback( 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 ingressCapacity = environment.SFU_INGRESS_CAPACITY?.trim() || undefined; + const egressCapacity = environment.SFU_EGRESS_CAPACITY?.trim() || undefined; + const configuredCapacities = [ingressCapacity, egressCapacity].filter( + Boolean, + ).length; if (configuredValues === 0) { + if (configuredCapacities > 0) { + throw new Error( + "SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY require LiveKit fallback", + ); + } return undefined; } if (configuredValues !== 3) { @@ -137,6 +161,11 @@ function parseLiveKitFallback( "LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET must be configured together", ); } + if (configuredCapacities !== 2) { + throw new Error( + "SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY must be configured with LiveKit fallback", + ); + } if (Buffer.byteLength(apiSecret!) < MIN_LIVEKIT_API_SECRET_BYTES) { throw new Error("LIVEKIT_API_SECRET must contain at least 32 bytes"); } @@ -169,12 +198,13 @@ function parseLiveKitFallback( url: parsedUrl.origin, apiKey: apiKey!, apiSecret: apiSecret!, - maxSfuRootsPerRoom: parseBoundedInteger( - environment.MAX_SFU_ROOTS_PER_ROOM, - DEFAULT_MAX_SFU_ROOTS_PER_ROOM, - "MAX_SFU_ROOTS_PER_ROOM", - 1, - CURRENT_SFU_ROOT_LIMIT, + ingressCapacity: parseRequiredPositiveInteger( + ingressCapacity, + "SFU_INGRESS_CAPACITY", + ), + egressCapacity: parseRequiredPositiveInteger( + egressCapacity, + "SFU_EGRESS_CAPACITY", ), }; } @@ -281,6 +311,8 @@ export function loadConfig( ? `${name} is no longer supported; use SITE_ACCESS_PASSWORD` : name === "PEER_ASSISTED_ROOM_IDS" ? `${name} is no longer supported; peer-assisted media applies to every room when enabled` + : name === "MAX_SFU_ROOTS_PER_ROOM" + ? `${name} is no longer supported; use SFU_INGRESS_CAPACITY and SFU_EGRESS_CAPACITY` : `${name} is no longer supported; ordinary ICE accepts STUN_URLS only`, ); } diff --git a/src/server/hybrid-media-router.ts b/src/server/hybrid-media-router.ts index 717d6ba..2ad5085 100644 --- a/src/server/hybrid-media-router.ts +++ b/src/server/hybrid-media-router.ts @@ -1,7 +1,6 @@ import { randomBytes } from "node:crypto"; import { - CURRENT_SFU_ROOT_LIMIT, MAX_MEDIA_ROUTE_REVISION, type ClientMessage, type MediaAssignment, @@ -18,6 +17,10 @@ import { endpointMediaCopyCountFits, } from "../shared/media-copy-accounting.js"; import type { SfuTokenIssuer } from "./livekit-token.js"; +import type { + SfuResourceAdmission, + SfuResourceFence, +} from "./sfu-resource-admission.js"; import type { SelectedEdgeTurnConfig } from "./config.js"; import { MediaRouteController, @@ -55,6 +58,7 @@ interface PendingRoutePreparation { hostSfuIngressTurnAttempted: boolean; healthyReselection: HealthySfuReselectionProbe | null; peerQualityReselection: PeerQualityReselectionProbe | null; + sfuResourceFence: SfuResourceFence | null; timer: NodeJS.Timeout; } @@ -185,7 +189,7 @@ type SfuPrepareResult = "started" | "busy" | "unavailable"; export interface SfuFallbackOptions { url: string; tokenIssuer: SfuTokenIssuer; - maxRoots: number; + admission: SfuResourceAdmission; prepareTimeoutMs?: number; } @@ -293,15 +297,11 @@ export class HybridMediaRouter { return; } if ( - !Number.isSafeInteger(fallback.maxRoots) || - fallback.maxRoots < 1 || - fallback.maxRoots > - Math.min(CURRENT_SFU_ROOT_LIMIT, options.roomStore.maxViewersPerRoom) || - (fallback.prepareTimeoutMs !== undefined && - (!Number.isSafeInteger(fallback.prepareTimeoutMs) || - fallback.prepareTimeoutMs <= 0)) + fallback.prepareTimeoutMs !== undefined && + (!Number.isSafeInteger(fallback.prepareTimeoutMs) || + fallback.prepareTimeoutMs <= 0) ) { - throw new Error("SFU fallback limits are invalid"); + throw new Error("SFU fallback timeout is invalid"); } new URL(fallback.url); } @@ -335,6 +335,7 @@ export class HybridMediaRouter { this.viewerQualityEvidenceStates.clear(); this.heldQualityParentExclusionsByViewer.clear(); this.roomQualityMigrationCooldownUntilMs.clear(); + this.options.sfuFallback?.admission.releaseAll(); } connectParticipant(input: AuthenticatedRouteParticipant): HybridAuthenticationState { @@ -1038,6 +1039,7 @@ export class HybridMediaRouter { parentPeerId, }, peerQualityReselection: null, + sfuResourceFence: null, timer, }); for (const [sessionId, assignment] of [ @@ -2502,6 +2504,7 @@ export class HybridMediaRouter { newParentPeerId, newConnectionId: null, }, + sfuResourceFence: null, timer, }); this.startRoomQualityMigrationCooldown(roomId); @@ -2651,7 +2654,8 @@ export class HybridMediaRouter { ): SfuPrepareResult { const fallback = this.options.sfuFallback; const controller = this.mediaRouteControllers.get(roomId); - if (!fallback || !controller) { + const shareGeneration = this.options.getShareGeneration(roomId); + if (!fallback || !controller || !shareGeneration) { return "unavailable"; } if (controller.getPendingRoute()) { @@ -2721,16 +2725,11 @@ export class HybridMediaRouter { requestedAssignment.upstream.kind === "peer" ? findHostBranchRoot(active.assignments, failedPeerId, hostPeerId) : host.childPeerIds[0]; - if ( - !budgetRootPeerId || - budgetRootPeerId === hostPeerId || - (!rootPeerIds.includes(budgetRootPeerId) && - rootPeerIds.length >= fallback.maxRoots) - ) { + if (!budgetRootPeerId || budgetRootPeerId === hostPeerId) { this.sendError( sourceSessionId, "PEER_NOT_FOUND", - "SFU fallback root budget is exhausted", + "No bounded SFU fallback plan is available", ); return "unavailable"; } @@ -2749,7 +2748,6 @@ export class HybridMediaRouter { } if ( - rootPeerIds.length > fallback.maxRoots || !host || !endpointMediaCopyCountFits( countEndpointMediaCopies({ @@ -2762,7 +2760,7 @@ export class HybridMediaRouter { this.sendError( sourceSessionId, "PEER_NOT_FOUND", - "SFU fallback root budget is exhausted", + "Endpoint media copy capacity is exhausted", ); return "unavailable"; } @@ -2803,6 +2801,19 @@ export class HybridMediaRouter { } expectedSessionIds.set(peerId, peer.sessionId); } + const sfuResourceFence: SfuResourceFence = { + roomId, + shareGeneration, + publicationGeneration: generation, + }; + if (!fallback.admission.reserve(sfuResourceFence, rootPeerIds.length)) { + this.sendError( + sourceSessionId, + "PEER_NOT_FOUND", + "SFU server capacity is exhausted", + ); + return "unavailable"; + } let revision: number | undefined; try { revision = controller.prepare({ @@ -2812,6 +2823,7 @@ export class HybridMediaRouter { sfuRootPeerIds: rootPeerIds, }); } catch { + fallback.admission.release(sfuResourceFence); this.sendError( sourceSessionId, "SERVER_ERROR", @@ -2820,6 +2832,7 @@ export class HybridMediaRouter { return "unavailable"; } if (revision === undefined) { + fallback.admission.release(sfuResourceFence); return "unavailable"; } @@ -2843,6 +2856,7 @@ export class HybridMediaRouter { hostSfuIngressTurnAttempted: false, healthyReselection: null, peerQualityReselection: null, + sfuResourceFence, timer, }); void this.issueSfuPrepareGrants( @@ -2974,6 +2988,12 @@ 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"); + } clearTimeout(pending.timer); this.pendingRoutePreparations.delete(roomId); const active = controller.getActiveRoute(); @@ -3166,11 +3186,14 @@ export class HybridMediaRouter { ): void { const controller = this.mediaRouteControllers.get(roomId); const pending = this.pendingRoutePreparations.get(roomId); - if (!controller || !pending) { + if (!pending) { return; } clearTimeout(pending.timer); this.pendingRoutePreparations.delete(roomId); + if (pending.sfuResourceFence) { + this.options.sfuFallback?.admission.release(pending.sfuResourceFence); + } if (pending.healthyReselection) { this.options.deleteConnectionId(roomId, pending.healthyReselection.rootPeerId); } @@ -3179,7 +3202,7 @@ export class HybridMediaRouter { clearTimeout(ingress.timer); this.selectedSfuIngressAttempts.delete(roomId); } - if (controller.abort(pending.revision)) { + if (controller?.abort(pending.revision)) { this.broadcastActiveRoute( roomId, controller.getActiveRoute(), @@ -3498,6 +3521,9 @@ export class HybridMediaRouter { route: RoomMediaRoute, excludedPeerId?: string, ): void { + if (route.sfu.publicationGeneration === null) { + this.options.sfuFallback?.admission.releaseRoom(roomId); + } this.reconcileSelectedPeerEdgeLeases(roomId, route); for (const [key, attempt] of this.selectedSfuIngressAttempts) { if (attempt.roomId === roomId && attempt.revision !== route.revision) { @@ -3795,6 +3821,7 @@ export class HybridMediaRouter { this.sfuDisabledRoomIds.delete(roomId); this.clearRoomViewerQualityEvidenceStates(roomId); this.roomQualityMigrationCooldownUntilMs.delete(roomId); + this.options.sfuFallback?.admission.releaseRoom(roomId); } private clearSelectedEdgeTurnsForParticipant( diff --git a/src/server/livekit-token.ts b/src/server/livekit-token.ts index f6e030b..47247d5 100644 --- a/src/server/livekit-token.ts +++ b/src/server/livekit-token.ts @@ -5,7 +5,6 @@ import { } from "livekit-server-sdk"; import { - CURRENT_SFU_ROOT_LIMIT, MAX_VIEWERS_PER_ROOM_LIMIT, type Role, } from "../shared/protocol.js"; @@ -30,7 +29,6 @@ export interface LiveKitTokenIssuerOptions { apiKey: string; apiSecret: string; maxViewersPerRoom: number; - maxSfuRootsPerRoom: number; } export class LiveKitTokenIssuer implements SfuTokenIssuer { @@ -45,17 +43,10 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer { ) { throw new Error("LiveKit viewer limit is invalid"); } - if ( - !Number.isSafeInteger(options.maxSfuRootsPerRoom) || - options.maxSfuRootsPerRoom < 1 || - options.maxSfuRootsPerRoom > CURRENT_SFU_ROOT_LIMIT - ) { - throw new Error("LiveKit SFU root limit is invalid"); - } } async issueToken(request: SfuTokenRequest): Promise { - validateTokenRequest(request, this.options.maxSfuRootsPerRoom); + validateTokenRequest(request, this.options.maxViewersPerRoom); const isHost = request.role === "host"; const room = liveKitRoomName( @@ -92,7 +83,7 @@ export class LiveKitTokenIssuer implements SfuTokenIssuer { function validateTokenRequest( request: SfuTokenRequest, - maxSfuRootsPerRoom: number, + maxViewersPerRoom: number, ): void { if (!ROOM_ID_PATTERN.test(request.roomId)) { throw new Error("LiveKit room ID is invalid"); @@ -104,7 +95,7 @@ function validateTokenRequest( throw new Error("LiveKit participant identity is invalid"); } if ( - request.allowlistedRootPeerIds.length > maxSfuRootsPerRoom || + request.allowlistedRootPeerIds.length > maxViewersPerRoom || request.allowlistedRootPeerIds.some( (peerId) => !OPAQUE_ID_PATTERN.test(peerId), ) diff --git a/src/server/media-route-controller.ts b/src/server/media-route-controller.ts index 88ceecb..31b6b4b 100644 --- a/src/server/media-route-controller.ts +++ b/src/server/media-route-controller.ts @@ -1,5 +1,4 @@ import { - CURRENT_SFU_ROOT_LIMIT, MAX_MEDIA_ROUTE_REVISION, MAX_VIEWERS_PER_ROOM_LIMIT, participantRouteAssignmentSchema, @@ -282,11 +281,8 @@ function assertRouteInvariants( } const rootPeerIds = new Set(route.sfu.rootPeerIds); - if ( - rootPeerIds.size !== route.sfu.rootPeerIds.length || - rootPeerIds.size > CURRENT_SFU_ROOT_LIMIT - ) { - throw new Error("SFU root participants must be unique and bounded"); + if (rootPeerIds.size !== route.sfu.rootPeerIds.length) { + throw new Error("SFU root participants must be unique"); } if ( (route.sfu.publicationGeneration === null) !== (rootPeerIds.size === 0) diff --git a/src/server/sfu-resource-admission.ts b/src/server/sfu-resource-admission.ts new file mode 100644 index 0000000..f955f44 --- /dev/null +++ b/src/server/sfu-resource-admission.ts @@ -0,0 +1,175 @@ +export interface SfuResourceFence { + roomId: string; + shareGeneration: string; + publicationGeneration: string; +} + +export interface SfuResourceAdmissionOptions { + readonly ingressCapacity: number; + readonly egressCapacity: number; +} + +export interface SfuResourceUsage { + ingress: number; + egress: number; +} + +interface SfuResourceEntry { + fence: SfuResourceFence; + egress: number; +} + +interface RoomSfuResources { + committed?: SfuResourceEntry; + reserved?: SfuResourceEntry; +} + +export class SfuResourceAdmission { + private readonly rooms = new Map(); + private readonly ingressCapacity: number; + private readonly egressCapacity: number; + private ingressInUse = 0; + private egressInUse = 0; + + constructor(options: SfuResourceAdmissionOptions) { + assertPositiveSafeInteger(options.ingressCapacity, "SFU ingress capacity"); + assertPositiveSafeInteger(options.egressCapacity, "SFU egress capacity"); + this.ingressCapacity = options.ingressCapacity; + this.egressCapacity = options.egressCapacity; + } + + reserve(fence: SfuResourceFence, egress: number): boolean { + assertFence(fence); + assertPositiveSafeInteger(egress, "SFU egress reservation"); + const room = this.rooms.get(fence.roomId); + if (room?.reserved) { + return sameEntry(room.reserved, fence, egress); + } + if (room?.committed && sameFence(room.committed.fence, fence)) { + return room.committed.egress === egress; + } + if ( + this.ingressInUse >= this.ingressCapacity || + egress > this.egressCapacity - this.egressInUse + ) { + return false; + } + + const resources = room ?? {}; + resources.reserved = { fence: { ...fence }, egress }; + this.rooms.set(fence.roomId, resources); + this.ingressInUse += 1; + this.egressInUse += egress; + return true; + } + + commit(fence: SfuResourceFence): boolean { + assertFence(fence); + const room = this.rooms.get(fence.roomId); + if (room?.committed && sameFence(room.committed.fence, fence)) { + return true; + } + if (!room?.reserved || !sameFence(room.reserved.fence, fence)) { + return false; + } + + if (room.committed) { + this.subtract(room.committed); + } + room.committed = room.reserved; + room.reserved = undefined; + return true; + } + + release(fence: SfuResourceFence): boolean { + assertFence(fence); + const room = this.rooms.get(fence.roomId); + if (!room) { + return false; + } + + let released = false; + if (room.reserved && sameFence(room.reserved.fence, fence)) { + this.subtract(room.reserved); + room.reserved = undefined; + released = true; + } + if (room.committed && sameFence(room.committed.fence, fence)) { + this.subtract(room.committed); + room.committed = undefined; + released = true; + } + this.deleteEmptyRoom(fence.roomId, room); + return released; + } + + releaseRoom(roomId: string): boolean { + if (!roomId) { + throw new Error("SFU resource room ID is invalid"); + } + const room = this.rooms.get(roomId); + if (!room) { + return false; + } + if (room.reserved) { + this.subtract(room.reserved); + } + if (room.committed) { + this.subtract(room.committed); + } + this.rooms.delete(roomId); + return true; + } + + releaseAll(): void { + this.rooms.clear(); + this.ingressInUse = 0; + this.egressInUse = 0; + } + + usage(): SfuResourceUsage { + return { ingress: this.ingressInUse, egress: this.egressInUse }; + } + + private subtract(entry: SfuResourceEntry): void { + this.ingressInUse -= 1; + this.egressInUse -= entry.egress; + if (this.ingressInUse < 0 || this.egressInUse < 0) { + throw new Error("SFU resource accounting underflow"); + } + } + + private deleteEmptyRoom(roomId: string, room: RoomSfuResources): void { + if (!room.committed && !room.reserved) { + this.rooms.delete(roomId); + } + } +} + +function sameEntry( + entry: SfuResourceEntry, + fence: SfuResourceFence, + egress: number, +): boolean { + return entry.egress === egress && sameFence(entry.fence, fence); +} + +function sameFence(left: SfuResourceFence, right: SfuResourceFence): boolean { + return ( + left.roomId === right.roomId && + left.shareGeneration === right.shareGeneration && + left.publicationGeneration === right.publicationGeneration + ); +} + +function assertFence(fence: SfuResourceFence): void { + if (!fence.roomId || !fence.shareGeneration || !fence.publicationGeneration) { + throw new Error("SFU resource fence is invalid"); + } +} + +function assertPositiveSafeInteger(value: number, name: string): void { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive safe integer`); + } +} diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index a67e46c..4d52b26 100644 --- a/src/shared/protocol.ts +++ b/src/shared/protocol.ts @@ -205,7 +205,6 @@ export const DEFAULT_QUALITY_SETTINGS = { screenAudioQuality: "music", } as const satisfies QualitySettings; -export const CURRENT_SFU_ROOT_LIMIT = 2; export const relayDownstreamEdgesSchema = z .number() .int() diff --git a/tests/livekit-token.test.ts b/tests/livekit-token.test.ts index c02715a..8120fdd 100644 --- a/tests/livekit-token.test.ts +++ b/tests/livekit-token.test.ts @@ -10,20 +10,16 @@ const rootPeerId = "viewer_root_12345678"; const secondRootPeerId = "viewer_root_23456789"; const thirdRootPeerId = "viewer_root_34567890"; -function issuer( - maxViewersPerRoom = 8, - maxSfuRootsPerRoom = 2, -): LiveKitTokenIssuer { +function issuer(maxViewersPerRoom = 8): LiveKitTokenIssuer { return new LiveKitTokenIssuer({ apiKey, apiSecret, maxViewersPerRoom, - maxSfuRootsPerRoom, }); } describe("LiveKitTokenIssuer", () => { - it("SFU root invariant gate: allows exactly two roots and rejects a third", async () => { + it("allows every admitted root within the room Viewer bound", async () => { const tokenIssuer = issuer(); await expect( tokenIssuer.issueToken({ @@ -55,7 +51,7 @@ describe("LiveKitTokenIssuer", () => { thirdRootPeerId, ], }), - ).rejects.toThrow("root allowlist is invalid"); + ).resolves.toEqual(expect.any(String)); }); it("issues a generation-bound host token limited to screen sharing", async () => { @@ -123,7 +119,7 @@ describe("LiveKitTokenIssuer", () => { }); it("rejects duplicate, excessive, and malformed root allowlists", async () => { - const tokenIssuer = issuer(8, 1); + const tokenIssuer = issuer(1); const request = { roomId: "7", role: "host" as const, diff --git a/tests/media-route-controller.test.ts b/tests/media-route-controller.test.ts index 971b233..4e78008 100644 --- a/tests/media-route-controller.test.ts +++ b/tests/media-route-controller.test.ts @@ -73,7 +73,7 @@ function chainedAssignments(): Map { } describe("MediaRouteController", () => { - it("SFU root invariant gate: accepts two roots, rejects a third, and charges one Host publication edge", () => { + it("keeps SFU roots unique while charging one Host publication edge", () => { const assignments = new Map([ [ HOST, @@ -96,17 +96,23 @@ describe("MediaRouteController", () => { ]); expect(routes.hostActiveMediaEdges()).toBe(2); - const excessiveRoots = new Map(assignments); - excessiveRoots.set(SFU_ROOT_C, assignment({ kind: "sfu" })); - expect( - () => - new MediaRouteController({ - hostPeerId: HOST, - assignments: excessiveRoots, - sfuPublicationGeneration: GENERATION_A, - sfuRootPeerIds: [SFU_ROOT_A, SFU_ROOT_B, SFU_ROOT_C], - }), - ).toThrow("SFU root participants must be unique and bounded"); + const threeRootAssignments = new Map(assignments); + threeRootAssignments.set(SFU_ROOT_C, assignment({ kind: "sfu" })); + const threeRoots = new MediaRouteController({ + hostPeerId: HOST, + assignments: threeRootAssignments, + sfuPublicationGeneration: GENERATION_A, + sfuRootPeerIds: [SFU_ROOT_A, SFU_ROOT_B, SFU_ROOT_C], + }); + expect(threeRoots.getActiveRoute().sfu.rootPeerIds).toHaveLength(3); + expect(() => + new MediaRouteController({ + hostPeerId: HOST, + assignments, + sfuPublicationGeneration: GENERATION_A, + sfuRootPeerIds: [SFU_ROOT_A, SFU_ROOT_A], + }), + ).toThrow("SFU root participants must be unique"); const excessiveHostEdges = new Map(assignments); excessiveHostEdges.set( diff --git a/tests/peer-assisted-benchmark.test.ts b/tests/peer-assisted-benchmark.test.ts index 4807582..f0d6443 100644 --- a/tests/peer-assisted-benchmark.test.ts +++ b/tests/peer-assisted-benchmark.test.ts @@ -499,7 +499,7 @@ describe("peer topology loopback observations", () => { ); }); - it("SFU root invariant gate: accepts two roots and detects excess root or Host media edges", () => { + it("accepts admitted SFU subscriptions and detects excess Host media edges", () => { const generation = "publication_generation_12345678"; const host = page("host", "host", 2, 0); const direct = page("viewer", "viewer-1", 0, 1); @@ -554,24 +554,24 @@ describe("peer topology loopback observations", () => { thirdRoot.routeRevision = 7; thirdRoot.routeAssignment.upstream = { kind: "sfu" }; markActiveRouteReady(thirdRoot); - const excessiveRootInitial = [...structuredClone(initial), thirdRoot]; - const excessiveRootFinal = structuredClone(excessiveRootInitial); - for (const viewer of excessiveRootFinal.slice(1)) { + const threeRootInitial = [...structuredClone(initial), thirdRoot]; + const threeRootFinal = structuredClone(threeRootInitial); + for (const viewer of threeRootFinal.slice(1)) { viewer.connections.at(-1)!.receiveTotals!.framesTotal = 20; } - const excessiveRootSummary = summarizeSamples( + const threeRootSummary = summarizeSamples( [ - { atEpochMs: 2_000, elapsedMs: 0, pages: excessiveRootInitial }, - { atEpochMs: 4_000, elapsedMs: 2_000, pages: excessiveRootFinal }, + { atEpochMs: 2_000, elapsedMs: 0, pages: threeRootInitial }, + { atEpochMs: 4_000, elapsedMs: 2_000, pages: threeRootFinal }, ], 4, ); - expect(excessiveRootSummary.sfuRootCount).toBe(3); + expect(threeRootSummary.sfuRootCount).toBe(3); expect( - buildRunChecks(excessiveRootSummary, 4, "720p30").find( + buildRunChecks(threeRootSummary, 4, "720p30").find( (check) => check.name === "sfu-route-consistency", )?.passed, - ).toBe(false); + ).toBe(true); const excessiveHostInitial = structuredClone(initial); const excessiveHostFinal = structuredClone(final); diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index f875b9e..3e155f0 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -1203,7 +1203,7 @@ describe("HostSfuRoute", () => { }); describe("ViewerSfuRoute", () => { - it("SFU root invariant gate: retains an active subscriber when the root has zero peer children", async () => { + it("retains an active SFU subscriber when the root has zero peer children", async () => { const reconcileSfuChildren = vi.fn(); const { route, subscriber } = await activateViewerSfuRoute({ activatePeer: () => undefined, diff --git a/tests/server-config.test.ts b/tests/server-config.test.ts index f6d2b38..ffb8606 100644 --- a/tests/server-config.test.ts +++ b/tests/server-config.test.ts @@ -2,6 +2,11 @@ import { describe, expect, it } from "vitest"; import { loadConfig } from "../src/server/config.ts"; +const liveKitAdmission = { + SFU_INGRESS_CAPACITY: "4", + SFU_EGRESS_CAPACITY: "16", +} as const; + describe("server configuration", () => { it("allows local development without STUN and defaults the origin", () => { const config = loadConfig({ NODE_ENV: "development", PORT: "9123" }); @@ -23,6 +28,7 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }; const turn = { SELECTED_EDGE_TURN_URLS: "turn:turn.test:3478?transport=udp", @@ -54,6 +60,7 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, SELECTED_EDGE_TURN_URLS: url, SELECTED_EDGE_TURN_SHARED_SECRET: "t".repeat(32), SELECTED_EDGE_TURN_CREDENTIAL_TTL_SECONDS: "120", @@ -67,37 +74,40 @@ describe("server configuration", () => { LIVEKIT_URL: " ws://livekit.test:7880 ", LIVEKIT_API_KEY: " test-key ", LIVEKIT_API_SECRET: ` ${"s".repeat(32)} `, + SFU_INGRESS_CAPACITY: " 4 ", + SFU_EGRESS_CAPACITY: " 16 ", }); expect(config.livekitFallback).toEqual({ url: "ws://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - maxSfuRootsPerRoom: 2, + ingressCapacity: 4, + egressCapacity: 16, }); }); - it.each(["1", "2"])( - "accepts an SFU root limit of %s when fallback is configured", - (maxSfuRootsPerRoom) => { + it.each(["1", "9007199254740991"])( + "accepts an explicit positive safe SFU capacity of %s", + (capacity) => { const config = loadConfig({ PEER_ASSISTED_MEDIA: "true", LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), - MAX_SFU_ROOTS_PER_ROOM: maxSfuRootsPerRoom, + SFU_INGRESS_CAPACITY: capacity, + SFU_EGRESS_CAPACITY: capacity, }); - expect(config.livekitFallback?.maxSfuRootsPerRoom).toBe( - Number(maxSfuRootsPerRoom), - ); + expect(config.livekitFallback?.ingressCapacity).toBe(Number(capacity)); + expect(config.livekitFallback?.egressCapacity).toBe(Number(capacity)); }, ); - it("does not activate fallback from the root limit alone", () => { - expect( - loadConfig({ MAX_SFU_ROOTS_PER_ROOM: "2" }).livekitFallback, - ).toBeUndefined(); + it("rejects SFU capacity without the LiveKit credential tuple", () => { + expect(() => loadConfig({ SFU_INGRESS_CAPACITY: "2" })).toThrow( + "require LiveKit fallback", + ); }); it("requires peer-assisted media for LiveKit fallback", () => { @@ -107,6 +117,7 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }), ).toThrow("LiveKit fallback requires PEER_ASSISTED_MEDIA=true"); }); @@ -137,6 +148,7 @@ describe("server configuration", () => { LIVEKIT_URL: url, LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }), ).toThrow("LIVEKIT_URL"); }); @@ -151,30 +163,49 @@ describe("server configuration", () => { LIVEKIT_URL: "ws://livekit.test:7880", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }), ).toThrow("LIVEKIT_URL must use wss in production"); }); - it("rejects a short LiveKit API secret and invalid root limits", () => { + it("rejects a short LiveKit API secret and invalid SFU capacities", () => { const fallback = { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }; expect(() => loadConfig({ ...fallback, LIVEKIT_API_SECRET: "too-short" }), ).toThrow("LIVEKIT_API_SECRET must contain at least 32 bytes"); - for (const maxSfuRootsPerRoom of ["0", "3", "1.5"]) { + for (const capacity of ["0", "1.5", "9007199254740992"]) { expect(() => loadConfig({ ...fallback, - MAX_SFU_ROOTS_PER_ROOM: maxSfuRootsPerRoom, + SFU_EGRESS_CAPACITY: capacity, }), - ).toThrow("MAX_SFU_ROOTS_PER_ROOM"); + ).toThrow("SFU_EGRESS_CAPACITY"); } }); + it("requires both SFU capacities with LiveKit and rejects the removed root key", () => { + const fallback = { + LIVEKIT_URL: "wss://livekit.test", + LIVEKIT_API_KEY: "test-key", + LIVEKIT_API_SECRET: "s".repeat(32), + }; + expect(() => loadConfig(fallback)).toThrow( + "must be configured with LiveKit fallback", + ); + expect(() => + loadConfig({ ...fallback, SFU_INGRESS_CAPACITY: "2" }), + ).toThrow("must be configured with LiveKit fallback"); + expect(() => loadConfig({ MAX_SFU_ROOTS_PER_ROOM: "" })).toThrow( + "MAX_SFU_ROOTS_PER_ROOM is no longer supported", + ); + }); + it.each([ { SITE_ACCESS_PASSWORD: "x".repeat(32), @@ -182,12 +213,14 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "x".repeat(32), + ...liveKitAdmission, }, { PEER_ASSISTED_MEDIA: "true", LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "x".repeat(32), LIVEKIT_API_SECRET: "x".repeat(32), + ...liveKitAdmission, }, ])("rejects reused infrastructure secrets", (environment) => { expect(() => loadConfig(environment)).toThrow( @@ -311,6 +344,7 @@ describe("server configuration", () => { LIVEKIT_URL: "wss://livekit.test", LIVEKIT_API_KEY: "test-key", LIVEKIT_API_SECRET: "s".repeat(32), + ...liveKitAdmission, }); expect(config.livekitFallback?.url).toBe("wss://livekit.test"); diff --git a/tests/server-http.test.ts b/tests/server-http.test.ts index 6ca6aa5..3ed2506 100644 --- a/tests/server-http.test.ts +++ b/tests/server-http.test.ts @@ -386,7 +386,8 @@ describe("server HTTP listener and health", () => { url: "ws://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - maxSfuRootsPerRoom: 2, + ingressCapacity: 4, + egressCapacity: 16, }, }), { sfuTokenIssuer: { issueToken: async () => "unused-test-token" } }, @@ -418,7 +419,8 @@ describe("server HTTP listener and health", () => { url: "ws://livekit.test:7880", apiKey: "test-key", apiSecret: "s".repeat(32), - maxSfuRootsPerRoom: 2, + ingressCapacity: 4, + egressCapacity: 16, }, }), { diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 8159a57..7e7eb07 100644 --- a/tests/server-signal.test.ts +++ b/tests/server-signal.test.ts @@ -25,8 +25,11 @@ import { RoomStore, type CreatedRoom } from "../src/server/room-store.ts"; 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"; const allowedOrigin = "http://allowed.test"; +const TEST_SFU_INGRESS_CAPACITY = 10; +const TEST_SFU_EGRESS_CAPACITY = 16; const defaultQualitySettings = { resolution: "1080p", maxFramerate: 60, @@ -142,6 +145,7 @@ interface SignalHarness { roomStore: RoomStore; room: CreatedRoom; database?: RoomDatabase; + sfuAdmission?: SfuResourceAdmission; } function testConfig(): ServerConfig { @@ -640,7 +644,8 @@ async function startSfuHarness(options: { selectedEdgeTurn?: boolean; persistent?: boolean; prepareTimeoutMs?: number; - maxRoots?: number; + sfuIngressCapacity?: number; + sfuEgressCapacity?: number; endpointMediaCopyCapacity?: number; viewerDisconnectGraceMs?: number; stunUrls?: readonly string[]; @@ -660,6 +665,12 @@ async function startSfuHarness(options: { response.statusCode = 404; response.end(); }); + const sfuAdmission = new SfuResourceAdmission({ + ingressCapacity: + options.sfuIngressCapacity ?? TEST_SFU_INGRESS_CAPACITY, + egressCapacity: + options.sfuEgressCapacity ?? TEST_SFU_EGRESS_CAPACITY, + }); const signaling = new SignalingServer({ server: httpServer, roomStore, @@ -669,7 +680,7 @@ async function startSfuHarness(options: { sfuFallback: { url: "wss://sfu.example.test", tokenIssuer: options.tokenIssuer, - maxRoots: options.maxRoots ?? 2, + admission: sfuAdmission, prepareTimeoutMs: options.prepareTimeoutMs, }, ...(options.selectedEdgeTurn @@ -723,6 +734,7 @@ async function startSfuHarness(options: { webSocketUrl: `ws://127.0.0.1:${port}/signal`, roomStore, room, + sfuAdmission, }; } @@ -2286,6 +2298,7 @@ describe("WebSocket signaling", () => { const hostPrepare = await nextPreparedRoute(peers.host); const branchPrepare = await nextPreparedRoute(peers.secondRoot); const failedPrepare = await nextPreparedRoute(peers.failedViewer); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 2 }); await peers.host.inbox.next("sfu-config"); await peers.secondRoot.inbox.next("sfu-config"); await peers.failedViewer.inbox.next("sfu-config"); @@ -2368,6 +2381,7 @@ describe("WebSocket signaling", () => { childPeerIds: [], sfuPublicationGeneration: null, }); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); await expect(peers.host.inbox.next("signal", 30)).rejects.toThrow( "Timed out", ); @@ -6123,7 +6137,7 @@ describe("WebSocket signaling", () => { ); }); - it("SFU root invariant gate: retains a zero-child root across commit and reauthentication", async () => { + it("retains a zero-child SFU root across commit and reauthentication", async () => { const issued: Array< Parameters[0] > = []; @@ -6495,18 +6509,19 @@ describe("WebSocket signaling", () => { direct.host.inbox.next("selected-edge-turn"), ]); expect(viewerGrant.oldConnectionId).toBe(direct.oldConnectionId); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); }); it("commits one healthy SFU root probe only after the peer edge is proven", async () => { const harness = await startSfuHarness({ tokenIssuer: { async issueToken({ peerId }) { return `token-${peerId}`; } }, - maxRoots: 1, }); const active = await activateSingleViewerSfu( harness.webSocketUrl, harness.room, "healthy-commit", ); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); requestHealthyReselection(active.viewer, active.revision - 1); await expect(active.viewer.inbox.next("route-update", 40)).rejects.toThrow("Timed out"); @@ -6580,6 +6595,7 @@ describe("WebSocket signaling", () => { peerId: active.hostAuth.peerId, }); expect(hostActive.assignment.sfuPublicationGeneration).toBeNull(); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); active.viewer.socket.send(JSON.stringify({ type: "route-failed", @@ -8149,14 +8165,16 @@ describe("WebSocket signaling", () => { ); }); - it("fails a deep SFU fallback cleanly when its root budget is one", async () => { + it("fails a deep SFU fallback cleanly when deployment egress is exhausted", async () => { + let issuedTokens = 0; const harness = await startSfuHarness({ tokenIssuer: { async issueToken(request) { + issuedTokens += 1; return `token-${request.peerId}`; }, }, - maxRoots: 1, + sfuEgressCapacity: 1, }); const peers = await exhaustDeepViewerPeerRoutes( harness.webSocketUrl, @@ -8168,6 +8186,8 @@ describe("WebSocket signaling", () => { "PEER_NOT_FOUND", ); await expect(nextPreparedRoute(peers.host)).rejects.toThrow("Timed out"); + expect(issuedTokens).toBe(0); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); expect(peers.host.socket.readyState).toBe(WebSocket.OPEN); }); @@ -8355,6 +8375,7 @@ describe("WebSocket signaling", () => { ); expect(hostActive.assignment.sfuPublicationGeneration).toBeTruthy(); expect(issued).toHaveLength(3); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 2 }); await closeClient(peers.failedViewer); const retiredHost = await nextActiveRouteAfter( @@ -8372,6 +8393,7 @@ describe("WebSocket signaling", () => { }); await new Promise((resolve) => setTimeout(resolve, 20)); expect(issued).toHaveLength(3); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); }); it("separates host signaling reconnects from new sharing generations", async () => { @@ -8407,6 +8429,7 @@ describe("WebSocket signaling", () => { prepared.rootPrepare.revision, ); expect(active.assignment.upstream).toEqual({ kind: "sfu" }); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); const sameShareHost = await openClient(harness.webSocketUrl); const sameShareAuth = peerAssisted( @@ -8447,6 +8470,7 @@ describe("WebSocket signaling", () => { await prepared.failedViewer.inbox.next("sharing-stopped"); expect(nextShareAuth.routeRevision).toBeLessThanOrEqual(active.revision); expect(nextShareAuth.routeAssignment.sfuPublicationGeneration).toBeNull(); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); const staleStopClosed = new Promise((resolve) => nextShareHost.socket.once("close", (code) => resolve(code)), @@ -8657,6 +8681,7 @@ describe("WebSocket signaling", () => { harness.room, "timeout", ); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); const timedOutRollback = await nextActiveRouteAfter( timedOut.host, timedOut.hostPrepare.revision, @@ -8669,6 +8694,7 @@ describe("WebSocket signaling", () => { sfuPublicationGeneration: null, }, }); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); const replacementRoom = harness.roomStore.createRoom(); const replacing = await prepareFallbackForTwoViewers( @@ -8676,6 +8702,7 @@ describe("WebSocket signaling", () => { replacementRoom, "replacement", ); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 1, egress: 1 }); const replacementRoot = await openClient(harness.webSocketUrl); const replacementAuth = peerAssisted( await authenticate( @@ -8700,6 +8727,7 @@ describe("WebSocket signaling", () => { expect(replacementRollback.revision).toBe( replacing.hostPrepare.revision + 1, ); + expect(harness.sfuAdmission?.usage()).toEqual({ ingress: 0, egress: 0 }); replacing.rootViewer.socket.send( JSON.stringify({ diff --git a/tests/sfu-client.test.ts b/tests/sfu-client.test.ts index e87e68c..4909607 100644 --- a/tests/sfu-client.test.ts +++ b/tests/sfu-client.test.ts @@ -539,7 +539,7 @@ describe("SfuPublisher", () => { expect(updates.at(-1)).toBeNull(); }); - it("SFU root invariant gate: explicitly keeps Dynacast off while preparing", async () => { + it("explicitly keeps Dynacast off while preparing SFU fallback", async () => { const publisher = new SfuPublisher(); await expect(publisher.connect(connection)).resolves.toBe(true); @@ -598,7 +598,7 @@ describe("SfuPublisher", () => { expect(livekit.state.rooms[0]?.disconnect).toHaveBeenCalledWith(false); }); - it("SFU root invariant gate: publishes exactly the ordered q and h video encodings", async () => { + it("publishes exactly the ordered q and h video encodings", async () => { const publisher = new SfuPublisher(); const video = track("video", "video-1"); const audio = track("audio", "audio-1"); @@ -1229,7 +1229,7 @@ describe("SfuPublisher", () => { }); describe("SfuSubscriber", () => { - it("SFU root invariant gate: keeps the assigned screen subscription at a HIGH ceiling", async () => { + it("keeps the assigned screen subscription at a HIGH ceiling", async () => { const gate = deferred(); livekit.state.connectGate = gate.promise; const streams: Array = []; diff --git a/tests/sfu-resource-admission.test.ts b/tests/sfu-resource-admission.test.ts new file mode 100644 index 0000000..14c7c50 --- /dev/null +++ b/tests/sfu-resource-admission.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, it } from "vitest"; + +import { + SfuResourceAdmission, + type SfuResourceFence, +} from "../src/server/sfu-resource-admission.ts"; + +function fence( + roomId: string, + publicationGeneration: string, + shareGeneration = `share_${roomId}`, +): SfuResourceFence { + return { roomId, shareGeneration, publicationGeneration }; +} + +describe("SfuResourceAdmission", () => { + it("requires explicit positive safe capacities", () => { + for (const ingressCapacity of [0, 1.5, Number.MAX_SAFE_INTEGER + 1]) { + expect( + () => new SfuResourceAdmission({ ingressCapacity, egressCapacity: 1 }), + ).toThrow("positive safe integer"); + } + expect( + () => new SfuResourceAdmission({ ingressCapacity: 1, egressCapacity: 0 }), + ).toThrow("positive safe integer"); + }); + + it("enforces deployment-wide ingress and egress across rooms", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 2, + egressCapacity: 4, + }); + expect(admission.reserve(fence("1", "publication_a"), 2)).toBe(true); + expect(admission.reserve(fence("2", "publication_b"), 2)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 2, egress: 4 }); + + expect(admission.reserve(fence("3", "publication_c"), 1)).toBe(false); + expect(admission.usage()).toEqual({ ingress: 2, egress: 4 }); + }); + + it("charges old and candidate generations until exact commit", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 2, + egressCapacity: 5, + }); + 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.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.usage()).toEqual({ ingress: 1, egress: 2 }); + expect(admission.release(first)).toBe(false); + }); + + it("makes duplicate operations idempotent without weakening fences", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 2, + egressCapacity: 3, + }); + const pending = fence("1", "publication_pending"); + expect(admission.reserve(pending, 2)).toBe(true); + expect(admission.reserve(pending, 2)).toBe(true); + expect(admission.reserve(pending, 1)).toBe(false); + 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.usage()).toEqual({ ingress: 0, egress: 0 }); + }); + + it("releases a room's committed and reserved resources 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.reserve(fence("1", "publication_candidate"), 2)).toBe( + true, + ); + + expect(admission.releaseRoom("1")).toBe(true); + expect(admission.releaseRoom("1")).toBe(false); + expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); + }); + + it("releases every room when 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.reserve(fence("2", "publication_pending"), 2)).toBe( + true, + ); + + admission.releaseAll(); + admission.releaseAll(); + expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); + }); +});