refactor(route): enforce global SFU admission

This commit is contained in:
AAA committed 2026-08-23 00:50:11 +08:00
1 parent 56ae363838
commit 524f012739
19 files changed
+519 -119

No files matched your search

+8 -6
View File
@@ -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
+1 -1
View File
@@ -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",
+2 -2
View File
@@ -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",
+5 -2
View File
@@ -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 =
+41 -9
View File
@@ -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`,
);
}
+49 -22
View File
@@ -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(
+3 -12
View File
@@ -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<string> {
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),
)
+2 -6
View File
@@ -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)
+175
View File
@@ -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<string, RoomSfuResources>();
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`);
}
}
-1
View File
@@ -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()
+4 -8
View File
@@ -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,
+18 -12
View File
@@ -73,7 +73,7 @@ function chainedAssignments(): Map<string, ParticipantRouteAssignment> {
}
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<string, ParticipantRouteAssignment>([
[
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(
+10 -10
View File
@@ -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);
+1 -1
View File
@@ -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,
+50 -16
View File
@@ -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");
+4 -2
View File
@@ -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,
},
}),
{
+34 -6
View File
@@ -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<SfuTokenIssuer["issueToken"]>[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<number>((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({
+3 -3
View File
@@ -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<MediaStream | null> = [];
+109
View File
@@ -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 });
});
});