refactor(route): reconcile SFU room lifecycle

This commit is contained in:
AAA committed 2026-08-23 00:52:53 +08:00
1 parent a30f9c2382
commit 14ee256232
17 files changed
+1602 -134

No files matched your search

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