diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts new file mode 100644 index 0000000..25f422f --- /dev/null +++ b/src/server/room-route-controller.ts @@ -0,0 +1,1166 @@ +import { MAX_MEDIA_ROUTE_REVISION } from "../shared/protocol.js"; +import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js"; + +export type CandidateTuple = + | { kind: "peer"; parentPeerId: string; transport: "direct" | "selected-turn" } + | { kind: "sfu"; publication: "reuse" | "create" | "replace"; ingress: "existing" | "direct" | "selected-turn" }; + +export type CandidateReservation = + | { kind: "direct"; overlap?: Resource } + | { kind: "selected-turn"; edge: Resource; overlap?: Resource } + | { kind: "sfu-reuse"; edge: Resource; overlap?: Resource } + | { kind: "sfu-create"; edge: Resource; publication: Resource; overlap?: Resource }; + +export type EndpointRetirement = + | { + kind: "edge"; + childPeerId: string; + childSessionId: string; + parentPeerId: string; + parentSessionId: string; + transport: "direct" | "selected-turn"; + connectionId: string; + } + | { + kind: "publication"; + hostSessionId: string; + generation: string; + connectionId: string; + }; + +export type EndpointTransition = + | { kind: "none"; producerPeerId?: string } + | { kind: "overlap"; producerPeerId: string } + | { kind: "bounded-gap"; producerPeerId: string; retire: EndpointRetirement }; + +export interface CandidatePlan { + tuple: CandidateTuple; + endpointTransition: EndpointTransition; +} + +export type CommittedEdge = + | { kind: "peer"; childSessionId: string; parentPeerId: string; parentSessionId: string; transport: "direct"; connectionId: string; usable: boolean; physicalActive: boolean } + | { kind: "peer"; childSessionId: string; parentPeerId: string; parentSessionId: string; transport: "selected-turn"; connectionId: string; usable: boolean; physicalActive: boolean; resource: Resource } + | { kind: "sfu"; childSessionId: string; publicationGeneration: string; transport: "sfu"; connectionId: string; usable: boolean; physicalActive: boolean; resource: Resource }; + +export type CommittedEdgeSeed = + | Omit, { transport: "direct" }>, "childSessionId" | "parentSessionId"> + | Omit, { transport: "selected-turn" }>, "childSessionId" | "parentSessionId"> + | Omit, { transport: "sfu" }>, "childSessionId">; + +export interface ParticipantInput { + peerId: string; + role: "host" | "viewer"; + sessionId: string; + effectiveDownstreamCapacity: number; +} + +export interface EdgeGuard { + childPeerId: string; + childSessionId: string; + routeRevision: number; + connectionId: string; + parentSessionId?: string; +} + +export interface CandidateGuard { + childPeerId: string; + childSessionId: string; + revision: number; + connectionId: string; +} + +export interface CandidateCursorGuard { + childPeerId: string; + childSessionId: string; + baseRevision: number; + factVersion: number; + cursor: number; + plan: CandidatePlan; +} + +export interface ControllerOptions { + hostPeerId: string; + endpointMediaCopyCapacity: number; + operationTimeoutMs: number; + selectedTurnEnabled?: boolean; + hostSfuSelectedTurnEnabled?: boolean; + sfuEnabled?: boolean; +} + +interface Participant { + peerId: string; + role: "host" | "viewer"; + sessionId: string | null; + departureConfirmed: boolean; + joinOrder: number; + effectiveDownstreamCapacity: number; + blockedAtFactVersion?: number; +} + +interface Attempt { + tuple: CandidateTuple; + revision: number; + connectionId: string; + childSessionId: string; + parentSessionId?: string; + publicationGeneration?: string; + publicationConnectionId?: string; + reservation: CandidateReservation; +} + +interface ChildOperation { + childPeerId: string; + childSessionId: string; + baseRevision: number; + candidates: CandidatePlan[]; + cursor: number; + deadlineAtMs: number; + builtAtFactVersion: number; + current?: Attempt; +} + +export interface OperationSnapshot { + childPeerId: string; + childSessionId: string; + baseRevision: number; + factVersion: number; + candidates: readonly CandidatePlan[]; + cursor: number; + deadlineAtMs: number; + current?: { tuple: CandidateTuple; revision: number; connectionId: string }; +} + +export interface RouteSnapshot { + revision: number; + paused: boolean; + factVersion: number; + upstreamByViewer: ReadonlyMap>; + hostPublication: HostPublication | null; + operation?: OperationSnapshot; +} + +export interface ReconcileResult { + operation?: OperationSnapshot; + removedPeerIds: readonly string[]; + released: readonly Resource[]; +} + +export interface SettleResult { + accepted: boolean; + exhausted?: boolean; + activeRevision: number; + released: readonly Resource[]; +} + +export interface BeginResult { + accepted: boolean; + operation?: OperationSnapshot; + exhausted?: boolean; + released: readonly Resource[]; +} + +interface HostPublication { + generation: string; + hostSessionId: string; + connectionId: string; + ingress: "direct" | "selected-turn"; + usable: boolean; + physicalActive: boolean; + resource: Resource; +} + +export class RoomRouteController { + private readonly participants = new Map(); + private readonly upstreamByViewer = new Map>(); + private hostPublication: HostPublication | null = null; + private operation?: ChildOperation; + private revision = 0; + private factVersion = 0; + private nextJoinOrder = 0; + private paused = false; + + constructor(private readonly options: ControllerOptions) { + assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity); + if (!Number.isSafeInteger(options.operationTimeoutMs) || options.operationTimeoutMs <= 0) { + throw new Error("Route operation timeout must be a positive integer"); + } + } + + snapshot(): RouteSnapshot { + return { + revision: this.revision, + paused: this.paused, + factVersion: this.factVersion, + upstreamByViewer: new Map([...this.upstreamByViewer].map(([id, edge]) => [id, { ...edge }])), + hostPublication: this.hostPublication ? { ...this.hostPublication } : null, + operation: this.operationSnapshot(), + }; + } + + upsertParticipant(input: ParticipantInput): readonly Resource[] { + if ((input.peerId === this.options.hostPeerId) !== (input.role === "host")) { + throw new Error("Route Host identity is inconsistent"); + } + const capacity = this.effectiveCapacity(input.effectiveDownstreamCapacity); + const current = this.participants.get(input.peerId); + if (current && current.role !== input.role) throw new Error("Route role cannot change"); + if (current) { + let changed = false; + const released: Resource[] = []; + const previousSessionId = current.sessionId; + if (previousSessionId !== input.sessionId) { + current.sessionId = input.sessionId; + released.push(...this.rebindCommittedSession(input.peerId, input.sessionId)); + changed = true; + } + if (current.departureConfirmed) { + current.departureConfirmed = false; + changed = true; + } + if (current.effectiveDownstreamCapacity !== capacity) { + current.effectiveDownstreamCapacity = capacity; + changed = true; + } + if (changed && current.blockedAtFactVersion !== undefined) { + current.blockedAtFactVersion = undefined; + } + if (changed) this.touchFacts(); + return released; + } else { + this.participants.set(input.peerId, { + ...input, + effectiveDownstreamCapacity: capacity, + departureConfirmed: false, + joinOrder: this.nextJoinOrder++, + }); + } + this.touchFacts(); + return []; + } + + disconnectSession(peerId: string, sessionId: string): boolean { + const participant = this.participants.get(peerId); + if (!participant || participant.sessionId !== sessionId) return false; + participant.sessionId = null; + this.touchFacts(); + return true; + } + + confirmDeparture(peerId: string): boolean { + const participant = this.participants.get(peerId); + if (!participant || participant.role === "host") return false; + if (participant.departureConfirmed && participant.sessionId === null && + participant.effectiveDownstreamCapacity === 0) return true; + participant.sessionId = null; + participant.departureConfirmed = true; + participant.effectiveDownstreamCapacity = 0; + this.touchFacts(); + return true; + } + + setEffectiveCapacity(peerId: string, sessionId: string, value: number): boolean { + const participant = this.participants.get(peerId); + if (!participant || participant.sessionId !== sessionId) return false; + const capacity = this.effectiveCapacity(value); + const changed = participant.effectiveDownstreamCapacity !== capacity; + if (!changed) return true; + participant.effectiveDownstreamCapacity = capacity; + participant.blockedAtFactVersion = undefined; + this.touchFacts(); + return true; + } + + touchExternalFacts(): void { + this.touchFacts(); + } + + hydrateEdge( + childPeerId: string, + edge: CommittedEdgeSeed, + ): void { + if (this.operation) throw new Error("Cannot hydrate while an operation is active"); + this.assertViewer(childPeerId); + const childSessionId = this.participants.get(childPeerId)!.sessionId; + if (!childSessionId) throw new Error("Route child has no current session"); + this.upstreamByViewer.set(childPeerId, edge.kind === "peer" ? { + ...edge, + childSessionId, + parentSessionId: + this.participants.get(edge.parentPeerId)?.sessionId ?? "", + } : { ...edge, childSessionId }); + this.assertGraph(); + } + + hydrateHostPublication( + generation: string, + resource: Resource, + connectionId = `publication:${generation}`, + ingress: "direct" | "selected-turn" = "direct", + ): void { + if (this.operation) throw new Error("Cannot hydrate while an operation is active"); + const hostSessionId = this.participants.get(this.options.hostPeerId)?.sessionId; + if (!hostSessionId) throw new Error("Route Host has no current session"); + this.hostPublication = { + generation, + hostSessionId, + connectionId, + ingress, + usable: true, + physicalActive: true, + resource, + }; + this.assertGraph(); + } + + invalidateEdge(guard: EdgeGuard): boolean { + const child = this.participants.get(guard.childPeerId); + const edge = this.upstreamByViewer.get(guard.childPeerId); + if (!child || child.sessionId !== guard.childSessionId || this.revision !== guard.routeRevision || + !edge || edge.connectionId !== guard.connectionId || + edge.childSessionId !== guard.childSessionId || + (edge.kind === "peer" && edge.parentSessionId !== guard.parentSessionId)) { + return false; + } + if (edge.usable) { + edge.usable = false; + child.blockedAtFactVersion = undefined; + this.touchFacts(); + } + return true; + } + + invalidateHostPublication(guard: { + hostSessionId: string; + routeRevision: number; + generation: string; + connectionId: string; + }): boolean { + const publication = this.hostPublication; + if (!publication || this.revision !== guard.routeRevision || + publication.hostSessionId !== guard.hostSessionId || + publication.generation !== guard.generation || + publication.connectionId !== guard.connectionId) return false; + if (publication.usable) { + publication.usable = false; + this.touchFacts(); + } + return true; + } + + retireCommittedTransport(guard: EdgeGuard): readonly Resource[] { + const child = this.participants.get(guard.childPeerId); + const edge = this.upstreamByViewer.get(guard.childPeerId); + if (this.operation?.current || !child || child.sessionId !== guard.childSessionId || this.revision !== guard.routeRevision || + !edge || edge.connectionId !== guard.connectionId || !edge.physicalActive || + edge.childSessionId !== guard.childSessionId || + (edge.kind === "peer" && edge.parentSessionId !== guard.parentSessionId)) { + return []; + } + edge.physicalActive = false; + edge.usable = false; + child.blockedAtFactVersion = undefined; + this.revision = this.nextRevision(); + if (this.operation) this.operation.baseRevision = this.revision; + this.touchFacts(); + return edge.transport === "direct" ? [] : [edge.resource]; + } + + retireHostPublication(guard: { + hostSessionId: string; + routeRevision: number; + generation: string; + connectionId: string; + }): readonly Resource[] { + const publication = this.hostPublication; + if (this.operation?.current || !publication || !publication.physicalActive || + this.revision !== guard.routeRevision || + publication.hostSessionId !== guard.hostSessionId || + publication.generation !== guard.generation || + publication.connectionId !== guard.connectionId) return []; + publication.physicalActive = false; + publication.usable = false; + this.revision = this.nextRevision(); + if (this.operation) this.operation.baseRevision = this.revision; + this.touchFacts(); + return [publication.resource]; + } + + setPaused(paused: boolean): readonly Resource[] { + if (this.paused === paused) return []; + this.paused = paused; + this.touchFacts(); + return paused ? this.abortOperation() : []; + } + + reconcile(nowMs: number): ReconcileResult { + const released: Resource[] = []; + const validation = this.validateOrAdvance(nowMs); + released.push(...validation.released); + if (this.paused || this.operation) { + return { operation: this.operationSnapshot(), removedPeerIds: [], released }; + } + const removedPeerIds = this.pruneDepartedLeaves(released); + + for (let remaining = this.participants.size; remaining > 0; remaining -= 1) { + const bootstrap = this.bootstrapForBlockedDemand(); + const childPeerId = bootstrap ?? this.selectNextChild(); + if (!childPeerId) return { removedPeerIds, released }; + const child = this.participants.get(childPeerId)!; + const candidates = this.buildCandidates(childPeerId, bootstrap !== undefined); + if (candidates.length === 0) { + child.blockedAtFactVersion = this.factVersion; + continue; + } + this.operation = { + childPeerId, + childSessionId: child.sessionId!, + baseRevision: this.revision, + candidates, + cursor: 0, + deadlineAtMs: nowMs + this.options.operationTimeoutMs, + builtAtFactVersion: this.factVersion, + }; + return { operation: this.operationSnapshot(), removedPeerIds, released }; + } + return { removedPeerIds, released }; + } + + beginCurrentCandidate(input: { + guard: CandidateCursorGuard; + nowMs: number; + connectionId: string; + reservation: CandidateReservation; + publicationGeneration?: string; + publicationConnectionId?: string; + }): BeginResult { + const validation = this.validateOrAdvance(input.nowMs); + const operation = this.operation; + if (!operation || operation.current || operation.baseRevision !== this.revision || + !this.cursorGuardMatches(input.guard, operation)) { + return { + accepted: false, + released: [...validation.released, ...reservationResources(input.reservation)], + exhausted: validation.exhausted, + }; + } + const plan = operation.candidates[operation.cursor]; + if (!plan || !this.candidateValid(operation.childPeerId, plan)) { + return { + accepted: false, + released: [...validation.released, ...reservationResources(input.reservation)], + exhausted: validation.exhausted, + }; + } + if (plan.endpointTransition.kind === "bounded-gap") { + return { + accepted: false, + operation: this.operationSnapshot(), + released: [...validation.released, ...reservationResources(input.reservation)], + }; + } + const tuple = plan.tuple; + this.assertReservation(tuple, input.reservation, plan.endpointTransition.kind === "overlap"); + const publicationGeneration = tuple.kind === "sfu" + ? tuple.publication === "reuse" ? this.hostPublication?.generation : input.publicationGeneration + : undefined; + if (tuple.kind === "sfu" && !publicationGeneration) throw new Error("SFU candidate needs a publication generation"); + if (tuple.kind === "sfu" && tuple.publication !== "reuse" && !input.publicationConnectionId) { + throw new Error("SFU publication candidate needs an ingress connection identity"); + } + operation.current = { + tuple, + revision: this.nextRevision(), + connectionId: input.connectionId, + childSessionId: operation.childSessionId, + parentSessionId: tuple.kind === "peer" ? this.participants.get(tuple.parentPeerId)?.sessionId ?? undefined : undefined, + publicationGeneration, + publicationConnectionId: input.publicationConnectionId, + reservation: input.reservation, + }; + return { accepted: true, operation: this.operationSnapshot(), released: validation.released }; + } + + skipCurrentCandidate(guard: CandidateCursorGuard, nowMs: number): BeginResult { + const validation = this.validateOrAdvance(nowMs); + const operation = this.operation; + if (!operation || operation.current || !this.cursorGuardMatches(guard, operation)) { + return { accepted: false, released: validation.released, exhausted: validation.exhausted }; + } + operation.cursor += 1; + const advanced = this.validateOrAdvance(nowMs); + return { + operation: this.operationSnapshot(), + accepted: true, + released: [...validation.released, ...advanced.released], + exhausted: advanced.exhausted, + }; + } + + retireCurrentCandidateProducer( + guard: CandidateCursorGuard, + nowMs: number, + ): BeginResult { + const validation = this.validateOrAdvance(nowMs); + const operation = this.operation; + if (!operation || operation.current || !this.cursorGuardMatches(guard, operation)) { + return { accepted: false, released: validation.released, exhausted: validation.exhausted }; + } + const plan = operation.candidates[operation.cursor]; + if (!plan || plan.endpointTransition.kind !== "bounded-gap") { + return { accepted: false, operation: this.operationSnapshot(), released: validation.released }; + } + const released = [...validation.released]; + const retirement = plan.endpointTransition.retire; + if (retirement.kind === "edge") { + const edge = this.upstreamByViewer.get(retirement.childPeerId); + if (!edge || edge.kind !== "peer" || !edge.physicalActive || + edge.childSessionId !== retirement.childSessionId || + edge.parentPeerId !== retirement.parentPeerId || + edge.parentSessionId !== retirement.parentSessionId || + edge.transport !== retirement.transport || + edge.connectionId !== retirement.connectionId) { + return { accepted: false, operation: this.operationSnapshot(), released }; + } + if (edge.transport === "selected-turn") released.push(edge.resource); + this.upstreamByViewer.delete(retirement.childPeerId); + } else { + const publication = this.hostPublication; + if (!publication || !publication.physicalActive || + publication.hostSessionId !== retirement.hostSessionId || + publication.generation !== retirement.generation || + publication.connectionId !== retirement.connectionId) { + return { accepted: false, operation: this.operationSnapshot(), released }; + } + released.push(...this.removePublicationGeneration(retirement.generation)); + } + this.advanceActiveRevision(operation); + const nextTuple = retirement.kind === "publication" && plan.tuple.kind === "sfu" && + plan.tuple.publication === "replace" + ? { ...plan.tuple, publication: "create" as const } + : plan.tuple; + const nextPlan = this.planCandidate(operation.childPeerId, nextTuple); + if (!nextPlan) { + operation.cursor += 1; + const advanced = this.validateOrAdvance(nowMs); + return { + accepted: true, + operation: this.operationSnapshot(), + exhausted: advanced.exhausted, + released: [...released, ...advanced.released], + }; + } + operation.candidates[operation.cursor] = nextPlan; + return { accepted: true, operation: this.operationSnapshot(), released }; + } + + candidateReady(guard: CandidateGuard, nowMs: number): SettleResult { + const validation = this.validateOrAdvance(nowMs); + const operation = this.operation; + const attempt = operation?.current; + if (!operation || !attempt || !this.guardMatches(guard, operation, attempt)) { + return { accepted: false, exhausted: validation.exhausted, activeRevision: this.revision, released: validation.released }; + } + const released = [...validation.released, ...this.commitAttempt(operation, attempt)]; + return { accepted: true, activeRevision: this.revision, released }; + } + + candidateFailed(guard: CandidateGuard, nowMs: number): SettleResult { + const validation = this.validateOrAdvance(nowMs, guard); + if (validation.consumedGuard) { + return { accepted: true, exhausted: validation.exhausted, activeRevision: this.revision, released: validation.released }; + } + return { accepted: false, exhausted: validation.exhausted, activeRevision: this.revision, released: validation.released }; + } + + operationExpired(nowMs: number): SettleResult { + const validation = this.validateOrAdvance(nowMs); + return { accepted: validation.expired, exhausted: validation.exhausted, activeRevision: this.revision, released: validation.released }; + } + + private validateOrAdvance(nowMs: number, failedGuard?: CandidateGuard): { + released: Resource[]; exhausted?: boolean; expired: boolean; consumedGuard: boolean; + } { + const released: Resource[] = []; + const result = { released, exhausted: undefined as boolean | undefined, expired: false, consumedGuard: false }; + let operation = this.operation; + if (!operation) return result; + if (nowMs >= operation.deadlineAtMs) { + if (operation.current) released.push(...reservationResources(operation.current.reservation)); + if (operation.current) this.advanceActiveRevision(operation); + const factsChanged = operation.builtAtFactVersion !== this.factVersion; + this.blockAndClear(operation, !factsChanged); + return { ...result, exhausted: !factsChanged, expired: true }; + } + if (this.participants.get(operation.childPeerId)?.sessionId !== operation.childSessionId) { + if (operation.current) released.push(...reservationResources(operation.current.reservation)); + if (operation.current) this.advanceActiveRevision(operation); + this.operation = undefined; + return result; + } + + while ((operation = this.operation)) { + const attempt = operation.current; + if (attempt) { + const guardFailed = failedGuard && this.guardMatches(failedGuard, operation, attempt); + const plan = operation.candidates[operation.cursor]; + if (!guardFailed && plan && this.candidateValid(operation.childPeerId, plan, attempt)) return result; + released.push(...reservationResources(attempt.reservation)); + this.advanceActiveRevision(operation); + operation.current = undefined; + operation.cursor += 1; + result.consumedGuard = Boolean(guardFailed); + } + while (operation.cursor < operation.candidates.length && + !this.candidateValid(operation.childPeerId, operation.candidates[operation.cursor]!)) { + operation.cursor += 1; + } + if (operation.cursor < operation.candidates.length) return result; + const factsChanged = operation.builtAtFactVersion !== this.factVersion; + this.blockAndClear(operation, !factsChanged); + result.exhausted = !factsChanged; + } + return result; + } + + private commitAttempt(operation: ChildOperation, attempt: Attempt): Resource[] { + const released: Resource[] = []; + const old = this.upstreamByViewer.get(operation.childPeerId); + if ("overlap" in attempt.reservation && attempt.reservation.overlap !== undefined) released.push(attempt.reservation.overlap); + + if (attempt.tuple.kind === "sfu" && attempt.tuple.publication === "replace") { + released.push(...this.removePublicationGeneration(this.hostPublication!.generation)); + } else if (old && old.transport !== "direct" && old.physicalActive) { + released.push(old.resource); + } + + if (attempt.tuple.kind === "peer") { + const parentSessionId = this.participants.get(attempt.tuple.parentPeerId)?.sessionId; + if (!parentSessionId) throw new Error("Candidate parent session is unavailable"); + this.upstreamByViewer.set(operation.childPeerId, attempt.tuple.transport === "direct" ? { + kind: "peer", childSessionId: operation.childSessionId, + parentPeerId: attempt.tuple.parentPeerId, parentSessionId, transport: "direct", + connectionId: attempt.connectionId, usable: true, physicalActive: true, + } : { + kind: "peer", childSessionId: operation.childSessionId, + parentPeerId: attempt.tuple.parentPeerId, parentSessionId, transport: "selected-turn", + connectionId: attempt.connectionId, usable: true, physicalActive: true, + resource: (attempt.reservation as Extract, { kind: "selected-turn" }>).edge, + }); + } else { + const generation = attempt.publicationGeneration!; + if (attempt.tuple.publication !== "reuse") { + const reservation = attempt.reservation as Extract, { kind: "sfu-create" }>; + const hostSessionId = this.participants.get(this.options.hostPeerId)?.sessionId; + if (!hostSessionId || !attempt.publicationConnectionId) { + throw new Error("SFU publication identity is unavailable"); + } + this.hostPublication = { + generation, + hostSessionId, + connectionId: attempt.publicationConnectionId, + ingress: attempt.tuple.ingress === "selected-turn" ? "selected-turn" : "direct", + usable: true, + physicalActive: true, + resource: reservation.publication, + }; + } + const edge = (attempt.reservation as Extract, { kind: "sfu-reuse" | "sfu-create" }>).edge; + this.upstreamByViewer.set(operation.childPeerId, { + kind: "sfu", childSessionId: operation.childSessionId, + publicationGeneration: generation, transport: "sfu", + connectionId: attempt.connectionId, usable: true, physicalActive: true, resource: edge, + }); + } + this.revision = attempt.revision; + this.operation = undefined; + this.touchFacts(); + const participant = this.participants.get(operation.childPeerId); + if (participant) participant.blockedAtFactVersion = undefined; + if (old?.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) { + if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); + this.hostPublication = null; + } + this.assertGraph(); + return released; + } + + private buildCandidates(childPeerId: string, sfuOnly: boolean): CandidatePlan[] { + const failed = this.upstreamByViewer.get(childPeerId); + const failedKey = failed?.usable === false ? edgeTupleKey(failed) : undefined; + const descendants = this.descendantsOf(childPeerId); + const parents = [...this.participants.values()] + .filter((parent) => parent.sessionId && !parent.departureConfirmed && parent.peerId !== childPeerId && + !descendants.has(parent.peerId) && this.sourceUsable(parent.peerId) && this.targetFits(parent.peerId, childPeerId)) + .sort((left, right) => this.depth(left.peerId) - this.depth(right.peerId) || + this.remaining(right.peerId, childPeerId) - this.remaining(left.peerId, childPeerId) || + compareParticipant(left, right)); + const candidates: CandidateTuple[] = []; + if (!sfuOnly) { + for (const parent of parents) candidates.push({ kind: "peer", parentPeerId: parent.peerId, transport: "direct" }); + if (this.options.selectedTurnEnabled) { + for (const parent of parents) if (parent.role !== "host") { + candidates.push({ kind: "peer", parentPeerId: parent.peerId, transport: "selected-turn" }); + } + } + } + if (this.options.sfuEnabled) { + if (this.hostPublication?.usable && this.hostPublication.physicalActive) { + candidates.push({ kind: "sfu", publication: "reuse", ingress: "existing" }); + } else if (this.hostPublication || this.publicationFits(childPeerId, false)) { + const publication = this.hostPublication ? "replace" : "create"; + candidates.push({ kind: "sfu", publication, ingress: "direct" }); + if (this.options.hostSfuSelectedTurnEnabled) { + candidates.push({ kind: "sfu", publication, ingress: "selected-turn" }); + } + } + } + return candidates.filter((candidate, index, all) => tupleKey(candidate) !== failedKey && + all.findIndex((other) => tupleKey(other) === tupleKey(candidate)) === index) + .map((candidate) => this.planCandidate(childPeerId, candidate)) + .filter((candidate): candidate is CandidatePlan => candidate !== null && + this.candidateValid(childPeerId, candidate)); + } + + private selectNextChild(): string | undefined { + const viewers = this.availableViewers(); + const departedParent = viewers.find(({ peerId }) => { + const edge = this.upstreamByViewer.get(peerId); + return edge?.kind === "peer" && this.participants.get(edge.parentPeerId)?.departureConfirmed; + }); + if (departedParent) return departedParent.peerId; + const overflow = this.overflowChildren(); + if (overflow.length > 0) return overflow[0]; + const unusable = viewers.find(({ peerId }) => this.upstreamByViewer.get(peerId)?.usable === false); + const staleSfu = viewers.find(({ peerId }) => { + const edge = this.upstreamByViewer.get(peerId); + return edge?.kind === "sfu" && (!this.hostPublication?.usable || + this.hostPublication.generation !== edge.publicationGeneration); + }); + return unusable?.peerId ?? staleSfu?.peerId ?? viewers.find(({ peerId }) => !this.upstreamByViewer.has(peerId))?.peerId; + } + + private bootstrapForBlockedDemand(): string | undefined { + if (!this.options.sfuEnabled || this.hostPublication || this.hostHasPublicationSlot()) return undefined; + const demand = this.availableViewers(true).find((viewer) => viewer.blockedAtFactVersion === this.factVersion); + if (!demand) return undefined; + return this.childrenOf(this.options.hostPeerId) + .filter((id) => { + const participant = this.participants.get(id); + const edge = this.upstreamByViewer.get(id); + return participant?.sessionId && participant.blockedAtFactVersion !== this.factVersion && + edge?.kind === "peer" && edge.transport === "direct" && edge.usable; + }) + .sort((left, right) => compareParticipant(this.participants.get(right)!, this.participants.get(left)!))[0]; + } + + private availableViewers(includeBlocked = false): Participant[] { + return [...this.participants.values()].filter((participant) => participant.role === "viewer" && + participant.sessionId && !participant.departureConfirmed && + (includeBlocked || participant.blockedAtFactVersion !== this.factVersion)) + .sort(compareParticipant); + } + + private overflowChildren(): string[] { + const result: string[] = []; + for (const parent of this.participants.values()) { + const capacity = Math.max(0, parent.effectiveDownstreamCapacity - (parent.role === "host" && this.hostPublication ? 1 : 0)); + const children = this.childrenOf(parent.peerId).filter((id) => { + const child = this.participants.get(id); + return child?.sessionId && !child.departureConfirmed && child.blockedAtFactVersion !== this.factVersion; + }).sort((left, right) => compareParticipant(this.participants.get(left)!, this.participants.get(right)!)); + result.push(...children.slice(capacity).reverse()); + } + return result; + } + + private candidateValid(childPeerId: string, plan: CandidatePlan, attempt?: Attempt): boolean { + const tuple = plan.tuple; + const child = this.participants.get(childPeerId); + if (!child?.sessionId || child.departureConfirmed || (attempt && attempt.childSessionId !== child.sessionId)) return false; + if (tuple.kind === "peer") { + const parent = this.participants.get(tuple.parentPeerId); + if (!parent?.sessionId || parent.departureConfirmed || (attempt && parent.sessionId !== attempt.parentSessionId) || + this.descendantsOf(childPeerId).has(parent.peerId) || !this.sourceUsable(parent.peerId) || + !this.targetFits(parent.peerId, childPeerId)) return false; + } else { + if (!this.options.sfuEnabled) return false; + if (tuple.publication === "reuse") { + if (!this.hostPublication?.usable || !this.hostPublication.physicalActive || + (attempt && attempt.publicationGeneration !== this.hostPublication.generation)) return false; + } else if (tuple.publication === "replace") { + if (!this.hostPublication || this.hostPublication.usable || !this.publicationFits(childPeerId, true)) return false; + } else if (this.hostPublication || !this.publicationFits(childPeerId, false)) { + return false; + } + } + const currentPlan = this.planCandidate(childPeerId, tuple); + return Boolean(currentPlan && endpointTransitionEquals(currentPlan.endpointTransition, plan.endpointTransition)); + } + + private planCandidate(childPeerId: string, tuple: CandidateTuple): CandidatePlan | null { + if (tuple.kind === "sfu" && tuple.publication === "reuse") { + return { tuple, endpointTransition: { kind: "none" } }; + } + const producerPeerId = tuple.kind === "peer" ? tuple.parentPeerId : this.options.hostPeerId; + const producer = this.participants.get(producerPeerId); + if (!producer?.sessionId || producer.departureConfirmed) return null; + const copies = this.physicalCopies(producerPeerId); + if (copies + 1 <= producer.effectiveDownstreamCapacity) { + return { tuple, endpointTransition: { kind: "none", producerPeerId } }; + } + if (copies + 1 <= Math.min(this.options.endpointMediaCopyCapacity + 1, 3)) { + return { tuple, endpointTransition: { kind: "overlap", producerPeerId } }; + } + const retire = this.retirementFor(childPeerId, producerPeerId, tuple); + if (!retire || copies > producer.effectiveDownstreamCapacity) return null; + return { tuple, endpointTransition: { kind: "bounded-gap", producerPeerId, retire } }; + } + + private retirementFor( + childPeerId: string, + producerPeerId: string, + tuple: CandidateTuple, + ): EndpointRetirement | null { + const old = this.upstreamByViewer.get(childPeerId); + if (old?.kind === "peer" && old.parentPeerId === producerPeerId && old.physicalActive) { + return { + kind: "edge", + childPeerId, + childSessionId: old.childSessionId, + parentPeerId: old.parentPeerId, + parentSessionId: old.parentSessionId, + transport: old.transport, + connectionId: old.connectionId, + }; + } + if (producerPeerId !== this.options.hostPeerId || !this.hostPublication?.physicalActive) return null; + const replacesPublication = tuple.kind === "sfu" && tuple.publication === "replace"; + const releasesLastPublication = tuple.kind === "peer" && old?.kind === "sfu" && + this.sfuSubscriberCount() === 1; + if (!replacesPublication && !releasesLastPublication) return null; + return { + kind: "publication", + hostSessionId: this.hostPublication.hostSessionId, + generation: this.hostPublication.generation, + connectionId: this.hostPublication.connectionId, + }; + } + + private physicalCopies(peerId: string): number { + let copies = [...this.upstreamByViewer.values()].filter((edge) => + edge.kind === "peer" && edge.parentPeerId === peerId && edge.physicalActive).length; + if (peerId === this.options.hostPeerId && this.hostPublication?.physicalActive) copies += 1; + return copies; + } + + private assertReservation(tuple: CandidateTuple, reservation: CandidateReservation, overlap: boolean): void { + const expected = tuple.kind === "peer" ? tuple.transport === "direct" ? "direct" : "selected-turn" : + tuple.publication === "reuse" ? "sfu-reuse" : "sfu-create"; + if (reservation.kind !== expected) throw new Error("Candidate reservation kind does not match tuple"); + if (overlap && !("overlap" in reservation && reservation.overlap !== undefined)) { + throw new Error("Candidate requires an endpoint overlap reservation"); + } + } + + private pruneDepartedLeaves(released: Resource[]): string[] { + const removed: string[] = []; + let changed = true; + while (changed) { + changed = false; + for (const participant of this.participants.values()) { + if (participant.role !== "viewer" || !participant.departureConfirmed || this.childrenOf(participant.peerId).length > 0) continue; + const edge = this.upstreamByViewer.get(participant.peerId); + if (edge && edge.transport !== "direct" && edge.physicalActive) released.push(edge.resource); + this.upstreamByViewer.delete(participant.peerId); + this.participants.delete(participant.peerId); + removed.push(participant.peerId); + changed = true; + } + } + if (removed.length > 0) { + if (!this.hasSfuSubscribers() && this.hostPublication) { + if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); + this.hostPublication = null; + } + this.revision = this.nextRevision(); + this.touchFacts(); + } + return removed; + } + + private abortOperation(): Resource[] { + if (!this.operation) return []; + const released = this.operation.current ? reservationResources(this.operation.current.reservation) : []; + if (this.operation.current) this.advanceActiveRevision(this.operation); + this.operation = undefined; + return released; + } + + private advanceActiveRevision(operation: ChildOperation): void { + this.revision = this.nextRevision(); + operation.baseRevision = this.revision; + } + + private blockAndClear(operation: ChildOperation, block = true): void { + const child = this.participants.get(operation.childPeerId); + if (block && child) child.blockedAtFactVersion = this.factVersion; + this.operation = undefined; + } + + private operationSnapshot(): OperationSnapshot | undefined { + const operation = this.operation; + if (!operation) return undefined; + return { childPeerId: operation.childPeerId, childSessionId: operation.childSessionId, + baseRevision: operation.baseRevision, factVersion: operation.builtAtFactVersion, + candidates: operation.candidates.map(cloneCandidatePlan), + cursor: operation.cursor, deadlineAtMs: operation.deadlineAtMs, + current: operation.current ? { tuple: { ...operation.current.tuple }, revision: operation.current.revision, + connectionId: operation.current.connectionId } : undefined }; + } + + private guardMatches(guard: CandidateGuard, operation: ChildOperation, attempt: Attempt): boolean { + return guard.childPeerId === operation.childPeerId && guard.childSessionId === operation.childSessionId && + guard.revision === attempt.revision && guard.connectionId === attempt.connectionId; + } + + private cursorGuardMatches(guard: CandidateCursorGuard, operation: ChildOperation): boolean { + const plan = operation.candidates[operation.cursor]; + return Boolean(plan && guard.childPeerId === operation.childPeerId && + guard.childSessionId === operation.childSessionId && guard.baseRevision === operation.baseRevision && + guard.factVersion === operation.builtAtFactVersion && guard.cursor === operation.cursor && + candidatePlanEquals(guard.plan, plan)); + } + + private childrenOf(parentPeerId: string): string[] { + return [...this.upstreamByViewer].filter(([, edge]) => edge.kind === "peer" && edge.parentPeerId === parentPeerId).map(([id]) => id); + } + + private descendantsOf(peerId: string): Set { + const descendants = new Set(); + const queue = [peerId]; + while (queue.length) for (const child of this.childrenOf(queue.pop()!)) if (!descendants.has(child)) { + descendants.add(child); queue.push(child); + } + return descendants; + } + + private sourceUsable(peerId: string): boolean { + if (peerId === this.options.hostPeerId) return true; + const seen = new Set(); + while (!seen.has(peerId)) { + seen.add(peerId); + const edge = this.upstreamByViewer.get(peerId); + if (!edge?.usable || !edge.physicalActive) return false; + if (edge.kind === "sfu") return Boolean(this.hostPublication?.usable && + this.hostPublication.physicalActive && this.hostPublication.generation === edge.publicationGeneration); + if (edge.parentPeerId === this.options.hostPeerId) return true; + peerId = edge.parentPeerId; + } + return false; + } + + private targetFits(parentPeerId: string, childPeerId: string): boolean { + const parent = this.participants.get(parentPeerId)!; + const edge = this.upstreamByViewer.get(childPeerId); + const already = edge?.kind === "peer" && edge.parentPeerId === parentPeerId; + const releasesLastPublication = parent.role === "host" && edge?.kind === "sfu" && + this.sfuSubscriberCount() === 1; + return this.usedSlots(parentPeerId) - (releasesLastPublication ? 1 : 0) + + (already ? 0 : 1) <= parent.effectiveDownstreamCapacity; + } + + private publicationFits(childPeerId: string, replacing: boolean): boolean { + const host = this.participants.get(this.options.hostPeerId); + const edge = this.upstreamByViewer.get(childPeerId); + const releasesHost = edge?.kind === "peer" && edge.parentPeerId === this.options.hostPeerId; + const nextSlots = this.usedSlots(this.options.hostPeerId) - (releasesHost ? 1 : 0) + + (replacing ? 0 : 1); + return Boolean(host?.sessionId && !host.departureConfirmed && nextSlots <= host.effectiveDownstreamCapacity); + } + + private hostHasPublicationSlot(): boolean { + const host = this.participants.get(this.options.hostPeerId); + return Boolean(host && this.usedSlots(this.options.hostPeerId) + 1 <= host.effectiveDownstreamCapacity); + } + + private usedSlots(peerId: string): number { + const participant = this.participants.get(peerId); + return this.childrenOf(peerId).length + (participant?.role === "host" && this.hostPublication ? 1 : 0); + } + + private remaining(parentPeerId: string, childPeerId: string): number { + const parent = this.participants.get(parentPeerId)!; + const edge = this.upstreamByViewer.get(childPeerId); + const already = edge?.kind === "peer" && edge.parentPeerId === parentPeerId; + return parent.effectiveDownstreamCapacity - this.usedSlots(parentPeerId) - (already ? 0 : 1); + } + + private depth(peerId: string): number { + if (peerId === this.options.hostPeerId) return 1; + let depth = 1; + const seen = new Set(); + while (!seen.has(peerId)) { + seen.add(peerId); + const edge = this.upstreamByViewer.get(peerId); + if (!edge || edge.kind === "sfu") return depth + 1; + depth += 1; + if (edge.parentPeerId === this.options.hostPeerId) return depth; + peerId = edge.parentPeerId; + } + return Number.MAX_SAFE_INTEGER; + } + + private hasSfuSubscribers(): boolean { + return [...this.upstreamByViewer.values()].some((edge) => edge.kind === "sfu"); + } + + private sfuSubscriberCount(): number { + return [...this.upstreamByViewer.values()].filter((edge) => edge.kind === "sfu").length; + } + + private rebindCommittedSession(peerId: string, sessionId: string): Resource[] { + const released: Resource[] = []; + const ownEdge = this.upstreamByViewer.get(peerId); + if (ownEdge?.kind === "peer" && ownEdge.transport === "selected-turn") { + if (ownEdge.physicalActive) released.push(ownEdge.resource); + ownEdge.physicalActive = false; + ownEdge.usable = false; + } else if (ownEdge) { + ownEdge.childSessionId = sessionId; + } + for (const edge of this.upstreamByViewer.values()) { + if (edge.kind !== "peer" || edge.parentPeerId !== peerId) continue; + if (edge.transport === "selected-turn") { + if (edge.physicalActive) released.push(edge.resource); + edge.physicalActive = false; + edge.usable = false; + } else { + edge.parentSessionId = sessionId; + } + } + if (peerId === this.options.hostPeerId && this.hostPublication) { + if (this.hostPublication.ingress === "selected-turn") { + if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); + this.hostPublication.physicalActive = false; + this.hostPublication.usable = false; + } else { + this.hostPublication.hostSessionId = sessionId; + } + } + return released; + } + + private removePublicationGeneration(generation: string): Resource[] { + const released: Resource[] = []; + for (const [viewerPeerId, edge] of this.upstreamByViewer) { + if (edge.kind !== "sfu" || edge.publicationGeneration !== generation) continue; + if (edge.physicalActive) released.push(edge.resource); + this.upstreamByViewer.delete(viewerPeerId); + } + if (this.hostPublication?.generation === generation) { + if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); + this.hostPublication = null; + } + return released; + } + + private assertGraph(): void { + const host = this.participants.get(this.options.hostPeerId); + if (!host || host.role !== "host") throw new Error("Route Host is missing"); + for (const [child, edge] of this.upstreamByViewer) { + this.assertViewer(child); + if (edge.kind === "peer" && !this.participants.has(edge.parentPeerId)) throw new Error("Route parent is missing"); + if (edge.kind === "sfu" && this.hostPublication?.generation !== edge.publicationGeneration) throw new Error("SFU publication is stale"); + const seen = new Set(); + let current = child; + while (current !== this.options.hostPeerId) { + if (seen.has(current)) throw new Error("Peer route contains a cycle"); + seen.add(current); + const currentEdge = this.upstreamByViewer.get(current); + if (!currentEdge) throw new Error("Route is not source-reachable"); + if (currentEdge.kind === "sfu") break; + current = currentEdge.parentPeerId; + } + } + for (const participant of this.participants.values()) if (this.usedSlots(participant.peerId) > this.options.endpointMediaCopyCapacity) { + throw new Error("Committed route exceeds endpoint capacity"); + } + } + + private assertViewer(peerId: string): void { + if (this.participants.get(peerId)?.role !== "viewer") throw new Error("Route child must be a Viewer"); + } + + private effectiveCapacity(value: number): number { + if (!Number.isSafeInteger(value) || value < 0) throw new Error("Effective capacity is invalid"); + return Math.min(value, this.options.endpointMediaCopyCapacity); + } + + private touchFacts(): void { this.factVersion += 1; } + + private nextRevision(): number { + if (this.revision >= MAX_MEDIA_ROUTE_REVISION) throw new Error("Media route revision space exhausted"); + return this.revision + 1; + } +} + +function compareParticipant(left: Participant, right: Participant): number { + return left.joinOrder - right.joinOrder || left.peerId.localeCompare(right.peerId); +} + +function tupleKey(tuple: CandidateTuple): string { + return tuple.kind === "peer" ? `peer:${tuple.parentPeerId}:${tuple.transport}` : `sfu:${tuple.publication}:${tuple.ingress}`; +} + +function edgeTupleKey(edge: CommittedEdge): string { + return edge.kind === "peer" ? `peer:${edge.parentPeerId}:${edge.transport}` : "sfu:reuse:existing"; +} + +function cloneCandidatePlan(plan: CandidatePlan): CandidatePlan { + return { + tuple: { ...plan.tuple }, + endpointTransition: plan.endpointTransition.kind === "bounded-gap" + ? { ...plan.endpointTransition, retire: { ...plan.endpointTransition.retire } } + : { ...plan.endpointTransition }, + }; +} + +function candidatePlanEquals(left: CandidatePlan, right: CandidatePlan): boolean { + return tupleKey(left.tuple) === tupleKey(right.tuple) && + endpointTransitionEquals(left.endpointTransition, right.endpointTransition); +} + +function endpointTransitionEquals(left: EndpointTransition, right: EndpointTransition): boolean { + if (left.kind !== right.kind) return false; + if (left.kind === "none" && right.kind === "none") { + return left.producerPeerId === right.producerPeerId; + } + if (left.kind === "overlap" && right.kind === "overlap") { + return left.producerPeerId === right.producerPeerId; + } + if (left.kind !== "bounded-gap" || right.kind !== "bounded-gap" || + left.producerPeerId !== right.producerPeerId || left.retire.kind !== right.retire.kind) return false; + if (left.retire.kind === "publication" && right.retire.kind === "publication") { + return left.retire.hostSessionId === right.retire.hostSessionId && + left.retire.generation === right.retire.generation && + left.retire.connectionId === right.retire.connectionId; + } + if (left.retire.kind === "edge" && right.retire.kind === "edge") { + return left.retire.childPeerId === right.retire.childPeerId && + left.retire.childSessionId === right.retire.childSessionId && + left.retire.parentPeerId === right.retire.parentPeerId && + left.retire.parentSessionId === right.retire.parentSessionId && + left.retire.transport === right.retire.transport && + left.retire.connectionId === right.retire.connectionId; + } + return false; +} + +function reservationResources(reservation: CandidateReservation): Resource[] { + const resources: Resource[] = []; + if ("edge" in reservation) resources.push(reservation.edge); + if ("publication" in reservation) resources.push(reservation.publication); + if ("overlap" in reservation && reservation.overlap !== undefined) resources.push(reservation.overlap); + return resources; +} diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts new file mode 100644 index 0000000..074846f --- /dev/null +++ b/tests/room-route-controller.test.ts @@ -0,0 +1,905 @@ +import { describe, expect, it } from "vitest"; + +import { + RoomRouteController, + type CandidateGuard, + type CandidateCursorGuard, + type CandidateReservation, + type CommittedEdgeSeed, + type OperationSnapshot, +} from "../src/server/room-route-controller.ts"; + +const HOST = "host_12345678"; +const A = "viewer_a_12345678"; +const B = "viewer_b_12345678"; +const C = "viewer_c_12345678"; +const D = "viewer_d_12345678"; +const E = "viewer_e_12345678"; +const F = "viewer_f_12345678"; + +function controller( + capacity: 1 | 2 | 3 = 2, + options: { selectedTurnEnabled?: boolean; sfuEnabled?: boolean } = {}, +) { + const routes = new RoomRouteController({ + hostPeerId: HOST, + endpointMediaCopyCapacity: capacity, + operationTimeoutMs: 10_000, + ...options, + }); + routes.upsertParticipant({ + peerId: HOST, + role: "host", + sessionId: "host_session", + effectiveDownstreamCapacity: capacity, + }); + return routes; +} + +function addViewer( + routes: RoomRouteController, + peerId: string, + capacity: 0 | 1 | 2 | 3, +) { + routes.upsertParticipant({ + peerId, + role: "viewer", + sessionId: `${peerId}_session`, + effectiveDownstreamCapacity: capacity, + }); +} + +function cursorGuard(operation: OperationSnapshot): CandidateCursorGuard { + return { + childPeerId: operation.childPeerId, + childSessionId: operation.childSessionId, + baseRevision: operation.baseRevision, + factVersion: operation.factVersion, + cursor: operation.cursor, + plan: operation.candidates[operation.cursor]!, + }; +} + +function beginCandidate( + routes: RoomRouteController, + input: { + nowMs: number; + connectionId: string; + reservation: CandidateReservation; + publicationGeneration?: string; + publicationConnectionId?: string; + }, +) { + const operation = routes.snapshot().operation!; + return routes.beginCurrentCandidate({ + guard: cursorGuard(operation), + ...input, + }); +} + +function skipCandidate(routes: RoomRouteController, nowMs: number) { + const operation = routes.snapshot().operation!; + return routes.skipCurrentCandidate(cursorGuard(operation), nowMs); +} + +function commitCurrent( + routes: RoomRouteController, + nowMs: number, + connectionId: string, +) { + const begin = beginCandidate(routes, { + nowMs, + connectionId, + reservation: { kind: "direct" }, + }); + const prepared = begin.operation!; + const current = prepared.current!; + const guard: CandidateGuard = { + childPeerId: prepared.childPeerId, + childSessionId: prepared.childSessionId, + revision: current.revision, + connectionId, + }; + expect(routes.candidateReady(guard, nowMs + 1).accepted).toBe(true); + return guard; +} + +function peerEdge( + parentPeerId: string, + connectionId: string, + usable = true, +): CommittedEdgeSeed { + return { + kind: "peer", + parentPeerId, + transport: "direct", + connectionId, + usable, + physicalActive: true, + }; +} + +describe("RoomRouteController", () => { + it.each([1, 2, 3] as const)( + "builds a deterministic acyclic route without a depth cap at C=%i", + (capacity) => { + const routes = controller(capacity); + for (const peerId of [A, B, C, D]) { + addViewer(routes, peerId, capacity); + const result = routes.reconcile(0); + expect(result.operation?.childPeerId).toBe(peerId); + commitCurrent(routes, 1, `${peerId}_connection`); + } + + const snapshot = routes.snapshot(); + expect(snapshot.upstreamByViewer).toHaveLength(4); + for (const [childPeerId, edge] of snapshot.upstreamByViewer) { + expect(edge.kind).toBe("peer"); + expect(edge.kind === "peer" && edge.parentPeerId).not.toBe(childPeerId); + } + if (capacity === 1) { + expect(snapshot.upstreamByViewer.get(D)).toMatchObject({ + kind: "peer", + parentPeerId: C, + }); + } + }, + ); + + it("keeps the old edge until exact candidate first-frame ready", () => { + const routes = controller(2); + addViewer(routes, A, 2); + addViewer(routes, B, 2); + routes.reconcile(0); + commitCurrent(routes, 1, "old_connection"); + routes.reconcile(2); + commitCurrent(routes, 3, "b_connection"); + const revision = routes.snapshot().revision; + expect( + routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + parentSessionId: "host_session", + routeRevision: revision, + connectionId: "old_connection", + }), + ).toBe(true); + + routes.reconcile(100); + const prepared = beginCandidate(routes, { + nowMs: 101, + connectionId: "new_connection", + reservation: { kind: "direct" }, + }).operation!; + expect(routes.snapshot().upstreamByViewer.get(A)?.connectionId).toBe( + "old_connection", + ); + expect( + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: prepared.current!.revision - 1, + connectionId: "new_connection", + }, 102).accepted, + ).toBe(false); + expect( + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: prepared.current!.revision, + connectionId: "new_connection", + }, 102), + ).toMatchObject({ accepted: true }); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + connectionId: "new_connection", + parentPeerId: B, + usable: true, + }); + }); + + it("advances one candidate cursor without resetting the operation deadline", () => { + const routes = controller(2, { + selectedTurnEnabled: true, + sfuEnabled: true, + }); + addViewer(routes, A, 2); + const operation = routes.reconcile(1_000).operation!; + const deadline = operation.deadlineAtMs; + const first = beginCandidate(routes, { + nowMs: 1_001, + connectionId: "candidate_1", + reservation: { kind: "direct" }, + }).operation!; + const failed = routes.candidateFailed({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: first.current!.revision, + connectionId: "candidate_1", + }, 1_001); + expect(failed.accepted).toBe(true); + expect(failed.exhausted).toBeUndefined(); + expect(failed.released).toEqual([]); + expect(routes.snapshot().operation).toMatchObject({ + cursor: 1, + deadlineAtMs: deadline, + }); + + const second = beginCandidate(routes, { + nowMs: 1_002, + connectionId: "candidate_2", + publicationGeneration: "publication_2", + publicationConnectionId: "publication_connection_2", + reservation: { + kind: "sfu-create", + edge: "candidate_2_edge", + publication: "candidate_2_publication", + }, + }).operation!; + const lateReady = routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: second.current!.revision, + connectionId: "candidate_2", + }, deadline); + expect(lateReady).toMatchObject({ accepted: false, exhausted: true }); + expect(lateReady.released).toEqual([ + "candidate_2_edge", + "candidate_2_publication", + ]); + expect(routes.snapshot().operation).toBeUndefined(); + expect( + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: second.current!.revision, + connectionId: "candidate_2", + }, deadline + 1).accepted, + ).toBe(false); + expect(routes.reconcile(deadline + 1).operation).toBeUndefined(); + routes.touchExternalFacts(); + expect(routes.reconcile(deadline + 2).operation?.childPeerId).toBe(A); + + const overlap = controller(2, { selectedTurnEnabled: true }); + addViewer(overlap, A, 1); + addViewer(overlap, B, 0); + overlap.hydrateEdge(A, peerEdge(HOST, "a_parent")); + overlap.hydrateEdge(B, peerEdge(A, "b_direct")); + expect(overlap.invalidateEdge({ + childPeerId: B, + childSessionId: `${B}_session`, + parentSessionId: `${A}_session`, + routeRevision: 0, + connectionId: "b_direct", + })).toBe(true); + overlap.reconcile(0); + const hostAttempt = beginCandidate(overlap, { + nowMs: 1, + connectionId: "b_host", + reservation: { kind: "direct" }, + }).operation!; + overlap.candidateFailed({ + childPeerId: B, + childSessionId: `${B}_session`, + revision: hostAttempt.current!.revision, + connectionId: "b_host", + }, 2); + expect(() => beginCandidate(overlap, { + nowMs: 3, + connectionId: "b_turn", + reservation: { kind: "selected-turn", edge: "turn_edge" }, + })).toThrow("endpoint overlap reservation"); + beginCandidate(overlap, { + nowMs: 3, + connectionId: "b_turn", + reservation: { + kind: "selected-turn", + edge: "turn_edge", + overlap: "overlap_slot", + }, + }); + overlap.setEffectiveCapacity(A, `${A}_session`, 0); + expect(overlap.reconcile(4).released).toEqual([ + "turn_edge", + "overlap_slot", + ]); + }); + + it("retains another exact failed edge while one child operation is busy", () => { + const routes = controller(3); + addViewer(routes, A, 1); + addViewer(routes, B, 1); + addViewer(routes, C, 1); + routes.hydrateEdge(A, peerEdge(HOST, "a_old")); + routes.hydrateEdge(B, peerEdge(HOST, "b_old")); + routes.hydrateEdge(C, peerEdge(HOST, "c_old")); + const revision = routes.snapshot().revision; + expect( + routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + parentSessionId: "host_session", + routeRevision: revision, + connectionId: "a_old", + }), + ).toBe(true); + routes.reconcile(0); + const current = beginCandidate(routes, { + nowMs: 1, + connectionId: "a_new", + reservation: { kind: "direct" }, + }).operation!; + expect( + routes.invalidateEdge({ + childPeerId: B, + childSessionId: `${B}_session`, + parentSessionId: "host_session", + routeRevision: revision, + connectionId: "b_old", + }), + ).toBe(true); + expect( + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: current.current!.revision, + connectionId: "a_new", + }, 2).accepted, + ).toBe(false); + const replacement = beginCandidate(routes, { + nowMs: 2, + connectionId: "a_new_2", + reservation: { kind: "direct" }, + }).operation!; + expect( + routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: replacement.current!.revision, + connectionId: "a_new_2", + }, 3).accepted, + ).toBe(true); + expect(routes.reconcile(3).operation?.childPeerId).toBe(B); + }); + + it("keeps grace media, then reparents a confirmed relay departure", () => { + const routes = controller(2); + addViewer(routes, A, 2); + addViewer(routes, B, 1); + addViewer(routes, C, 0); + routes.hydrateEdge(A, peerEdge(HOST, "a_connection")); + routes.hydrateEdge(B, peerEdge(A, "b_connection")); + routes.hydrateEdge(C, peerEdge(B, "c_connection")); + + expect(routes.disconnectSession(A, `${A}_session`)).toBe(true); + expect(routes.reconcile(0).operation).toBeUndefined(); + expect(routes.confirmDeparture(A)).toBe(true); + expect(routes.reconcile(1).operation?.childPeerId).toBe(B); + commitCurrent(routes, 2, "b_reparented"); + const beforePrune = routes.snapshot().revision; + const after = routes.reconcile(3); + expect(after.removedPeerIds).toContain(A); + expect(routes.snapshot().revision).toBeGreaterThan(beforePrune); + expect(routes.snapshot().upstreamByViewer.get(C)).toMatchObject({ + parentPeerId: B, + connectionId: "c_connection", + }); + }); + + it("moves newest overflow children and supports effective capacity zero", () => { + const routes = controller(3); + addViewer(routes, A, 3); + routes.hydrateEdge(A, peerEdge(HOST, "a_connection")); + for (const peerId of [B, C, D]) { + addViewer(routes, peerId, 0); + routes.hydrateEdge(peerId, peerEdge(A, `${peerId}_connection`)); + } + expect(routes.setEffectiveCapacity(A, `${A}_session`, 1)).toBe(true); + expect(routes.reconcile(0).operation?.childPeerId).toBe(D); + expect(routes.setPaused(true)).toEqual([]); + expect(routes.setEffectiveCapacity(A, `${A}_session`, 0)).toBe(true); + expect(routes.setPaused(false)).toEqual([]); + expect(routes.reconcile(1).operation?.childPeerId).toBe(D); + + const blocked = controller(1); + addViewer(blocked, A, 0); + addViewer(blocked, B, 0); + blocked.hydrateEdge(A, peerEdge(HOST, "blocked_a")); + blocked.hydrateEdge(B, peerEdge(A, "blocked_b")); + expect(blocked.reconcile(0).operation).toBeUndefined(); + expect(blocked.reconcile(1).operation).toBeUndefined(); + }); + + it("uses an SFU-fed endpoint as an ordinary parent and aborts pause resources", () => { + const routes = controller(1, { sfuEnabled: true }); + addViewer(routes, A, 1); + addViewer(routes, B, 0); + routes.hydrateHostPublication("publication_1", "publication_resource"); + routes.hydrateEdge(A, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: "a_sfu", + usable: true, + physicalActive: true, + resource: "a_subscription", + }); + const operation = routes.reconcile(0).operation!; + expect(operation.candidates[0]?.tuple).toEqual({ + kind: "peer", + parentPeerId: A, + transport: "direct", + }); + beginCandidate(routes, { + nowMs: 1, + connectionId: "b_candidate", + reservation: { kind: "direct" }, + }); + expect(routes.setPaused(true)).toEqual([]); + expect(routes.snapshot().operation).toBeUndefined(); + expect(routes.snapshot().upstreamByViewer.get(A)?.connectionId).toBe("a_sfu"); + routes.setPaused(false); + expect(routes.invalidateHostPublication({ + hostSessionId: "host_session", + routeRevision: routes.snapshot().revision, + generation: "publication_1", + connectionId: "publication:publication_1", + })).toBe(true); + const repair = routes.reconcile(2).operation!; + expect(repair.childPeerId).toBe(A); + expect(repair.candidates[0]?.tuple).toEqual({ + kind: "peer", + parentPeerId: HOST, + transport: "direct", + }); + skipCandidate(routes, 3); + expect(routes.snapshot().operation?.candidates[1]?.tuple).toEqual({ + kind: "sfu", + publication: "replace", + ingress: "direct", + }); + expect(() => beginCandidate(routes, { + nowMs: 3, + connectionId: "a_sfu_2", + publicationGeneration: "publication_2", + publicationConnectionId: "publication_connection_2", + reservation: { + kind: "sfu-create", + edge: "a_subscription_2", + publication: "publication_resource_2", + }, + })).toThrow("endpoint overlap reservation"); + const publicationRepair = beginCandidate(routes, { + nowMs: 3, + connectionId: "a_sfu_2", + publicationGeneration: "publication_2", + publicationConnectionId: "publication_connection_2", + reservation: { + kind: "sfu-create", + edge: "a_subscription_2", + publication: "publication_resource_2", + overlap: "publication_overlap", + }, + }).operation!; + const repaired = routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: publicationRepair.current!.revision, + connectionId: "a_sfu_2", + }, 4); + expect(repaired.accepted).toBe(true); + expect(repaired.released).toEqual([ + "publication_overlap", + "a_subscription", + "publication_resource", + ]); + expect(routes.snapshot().hostPublication).toMatchObject({ + generation: "publication_2", + usable: true, + physicalActive: true, + }); + + const bootstrap = controller(1, { sfuEnabled: true }); + addViewer(bootstrap, A, 0); + addViewer(bootstrap, B, 0); + bootstrap.hydrateEdge(A, peerEdge(HOST, "bootstrap_a")); + const bootstrapOperation = bootstrap.reconcile(0).operation!; + expect(bootstrapOperation.childPeerId).toBe(A); + expect(bootstrapOperation.candidates[0]?.tuple).toEqual({ + kind: "sfu", + publication: "create", + ingress: "direct", + }); + expect(() => beginCandidate(bootstrap, { + nowMs: 2, + connectionId: "bootstrap_sfu", + publicationGeneration: "bootstrap_publication", + publicationConnectionId: "bootstrap_ingress_connection", + reservation: { + kind: "sfu-create", + edge: "bootstrap_subscription", + publication: "bootstrap_ingress", + }, + })).toThrow("endpoint overlap reservation"); + const preparedBootstrap = beginCandidate(bootstrap, { + nowMs: 2, + connectionId: "bootstrap_sfu", + publicationGeneration: "bootstrap_publication", + publicationConnectionId: "bootstrap_ingress_connection", + reservation: { + kind: "sfu-create", + edge: "bootstrap_subscription", + publication: "bootstrap_ingress", + overlap: "bootstrap_overlap", + }, + }).operation!; + expect(bootstrap.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: preparedBootstrap.current!.revision, + connectionId: "bootstrap_sfu", + }, 3).accepted).toBe(true); + expect(bootstrap.reconcile(4).operation?.childPeerId).toBe(B); + }); + + it("binds reserve and skip results to the exact candidate cursor", () => { + const reserve = controller(2); + addViewer(reserve, A, 2); + reserve.hydrateEdge(A, peerEdge(HOST, "a_active")); + addViewer(reserve, B, 0); + const first = reserve.reconcile(0).operation!; + expect(first.candidates[0]?.tuple).toMatchObject({ + kind: "peer", + parentPeerId: HOST, + }); + reserve.setEffectiveCapacity(HOST, "host_session", 0); + const staleReserve = reserve.beginCurrentCandidate({ + guard: cursorGuard(first), + nowMs: 1, + connectionId: "reserved_for_old_cursor", + reservation: { kind: "direct", overlap: "stale_reservation" }, + }); + expect(staleReserve).toMatchObject({ accepted: false }); + expect(staleReserve.released).toEqual(["stale_reservation"]); + expect(reserve.snapshot().operation).toMatchObject({ cursor: 1 }); + expect(reserve.snapshot().operation?.current).toBeUndefined(); + + const skip = controller(2); + addViewer(skip, A, 2); + skip.hydrateEdge(A, peerEdge(HOST, "a_active")); + addViewer(skip, B, 0); + const staleCursor = skip.reconcile(0).operation!; + skip.setEffectiveCapacity(HOST, "host_session", 0); + expect(skip.skipCurrentCandidate(cursorGuard(staleCursor), 1).accepted).toBe(false); + expect(skip.snapshot().operation).toMatchObject({ cursor: 1 }); + expect(skip.snapshot().operation?.candidates[1]?.tuple).toMatchObject({ + kind: "peer", + parentPeerId: A, + }); + }); + + it("replaces one publication and makes every old subscriber waiting", () => { + const routes = controller(1, { sfuEnabled: true }); + addViewer(routes, A, 0); + addViewer(routes, B, 0); + routes.hydrateHostPublication("publication_1", "publication_resource_1"); + for (const [peerId, resource] of [[A, "subscription_a"], [B, "subscription_b"]] as const) { + routes.hydrateEdge(peerId, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: `${peerId}_sfu_1`, + usable: true, + physicalActive: true, + resource, + }); + } + expect(routes.invalidateHostPublication({ + hostSessionId: "host_session", + routeRevision: 0, + generation: "publication_1", + connectionId: "publication:publication_1", + })).toBe(true); + const operation = routes.reconcile(0).operation!; + expect(operation.candidates[0]).toMatchObject({ + tuple: { kind: "sfu", publication: "replace", ingress: "direct" }, + endpointTransition: { kind: "overlap", producerPeerId: HOST }, + }); + const prepared = beginCandidate(routes, { + nowMs: 1, + connectionId: "a_sfu_2", + publicationGeneration: "publication_2", + publicationConnectionId: "publication_connection_2", + reservation: { + kind: "sfu-create", + edge: "subscription_a_2", + publication: "publication_resource_2", + overlap: "publication_overlap", + }, + }).operation!; + const settled = routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: prepared.current!.revision, + connectionId: "a_sfu_2", + }, 2); + expect(settled.accepted).toBe(true); + expect(settled.released).toEqual([ + "publication_overlap", + "subscription_a", + "subscription_b", + "publication_resource_1", + ]); + expect(routes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + publicationGeneration: "publication_2", + }); + expect(routes.snapshot().upstreamByViewer.has(B)).toBe(false); + expect(routes.reconcile(3).operation?.childPeerId).toBe(B); + }); + + it("releases session-bound transports while rebinding direct media", () => { + const peerRoutes = controller(2); + addViewer(peerRoutes, A, 1); + addViewer(peerRoutes, B, 0); + peerRoutes.hydrateEdge(A, { + kind: "peer", + parentPeerId: HOST, + transport: "selected-turn", + connectionId: "a_selected", + usable: true, + physicalActive: true, + resource: "a_turn_allocation", + }); + peerRoutes.hydrateEdge(B, peerEdge(HOST, "b_direct")); + expect(peerRoutes.upsertParticipant({ + peerId: A, + role: "viewer", + sessionId: "a_replacement", + effectiveDownstreamCapacity: 1, + })).toEqual(["a_turn_allocation"]); + expect(peerRoutes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + childSessionId: `${A}_session`, + usable: false, + physicalActive: false, + }); + expect(peerRoutes.upsertParticipant({ + peerId: B, + role: "viewer", + sessionId: "b_replacement", + effectiveDownstreamCapacity: 0, + })).toEqual([]); + expect(peerRoutes.snapshot().upstreamByViewer.get(B)).toMatchObject({ + childSessionId: "b_replacement", + usable: true, + physicalActive: true, + }); + + const sfuRoutes = controller(1, { sfuEnabled: true }); + addViewer(sfuRoutes, A, 0); + sfuRoutes.hydrateHostPublication( + "selected_publication", + "selected_ingress_resource", + "selected_ingress_connection", + "selected-turn", + ); + sfuRoutes.hydrateEdge(A, { + kind: "sfu", + publicationGeneration: "selected_publication", + transport: "sfu", + connectionId: "subscription_connection", + usable: true, + physicalActive: true, + resource: "subscription_resource", + }); + expect(sfuRoutes.upsertParticipant({ + peerId: HOST, + role: "host", + sessionId: "host_replacement", + effectiveDownstreamCapacity: 1, + })).toEqual(["selected_ingress_resource"]); + expect(sfuRoutes.snapshot().hostPublication).toMatchObject({ + hostSessionId: "host_session", + ingress: "selected-turn", + usable: false, + physicalActive: false, + }); + expect(sfuRoutes.upsertParticipant({ + peerId: A, + role: "viewer", + sessionId: "a_replacement", + effectiveDownstreamCapacity: 0, + })).toEqual([]); + expect(sfuRoutes.snapshot().upstreamByViewer.get(A)).toMatchObject({ + childSessionId: "a_replacement", + physicalActive: true, + }); + }); + + it("does not reopen an exhausted bootstrap without a new fact", () => { + const routes = controller(1, { sfuEnabled: true }); + addViewer(routes, A, 0); + routes.hydrateEdge(A, peerEdge(HOST, "failed_direct")); + expect(routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + parentSessionId: "host_session", + routeRevision: 0, + connectionId: "failed_direct", + })).toBe(true); + const operation = routes.reconcile(0).operation!; + const prepared = beginCandidate(routes, { + nowMs: 1, + connectionId: "failed_sfu", + publicationGeneration: "failed_publication", + publicationConnectionId: "failed_publication_connection", + reservation: { + kind: "sfu-create", + edge: "failed_subscription_resource", + publication: "failed_publication_resource", + overlap: "failed_overlap", + }, + }).operation!; + expect(routes.candidateFailed({ + childPeerId: operation.childPeerId, + childSessionId: operation.childSessionId, + revision: prepared.current!.revision, + connectionId: "failed_sfu", + }, 2)).toMatchObject({ accepted: true, exhausted: true }); + const factVersion = routes.snapshot().factVersion; + expect(routes.reconcile(3).operation).toBeUndefined(); + expect(routes.setEffectiveCapacity(A, `${A}_session`, 0)).toBe(true); + expect(routes.snapshot().factVersion).toBe(factVersion); + expect(routes.reconcile(4).operation).toBeUndefined(); + expect(routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + parentSessionId: "host_session", + routeRevision: routes.snapshot().revision, + connectionId: "failed_direct", + })).toBe(true); + expect(routes.snapshot().factVersion).toBe(factVersion); + expect(routes.reconcile(5).operation).toBeUndefined(); + }); + + it.each([1, 2] as const)( + "requires one overlap slot for an SFU-to-Host transition at C=%i", + (capacity) => { + const routes = controller(capacity, { sfuEnabled: true }); + addViewer(routes, A, 0); + if (capacity === 2) { + addViewer(routes, E, 0); + routes.hydrateEdge(E, peerEdge(HOST, "e_direct")); + } + routes.hydrateHostPublication("publication_1", "publication_resource_1"); + routes.hydrateEdge(A, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: "a_sfu", + usable: true, + physicalActive: true, + resource: "a_subscription", + }); + routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: 0, + connectionId: "a_sfu", + }); + const operation = routes.reconcile(0).operation!; + expect(operation.candidates[0]).toMatchObject({ + tuple: { kind: "peer", parentPeerId: HOST, transport: "direct" }, + endpointTransition: { kind: "overlap", producerPeerId: HOST }, + }); + expect(() => beginCandidate(routes, { + nowMs: 1, + connectionId: "a_host_direct", + reservation: { kind: "direct" }, + })).toThrow("endpoint overlap reservation"); + }, + ); + + it("uses one bounded gap instead of a fourth Host copy at C=3", () => { + const routes = controller(3, { sfuEnabled: true }); + addViewer(routes, A, 0); + for (const peerId of [E, F]) { + addViewer(routes, peerId, 0); + routes.hydrateEdge(peerId, peerEdge(HOST, `${peerId}_direct`)); + } + routes.hydrateHostPublication("publication_1", "publication_resource_1"); + routes.hydrateEdge(A, { + kind: "sfu", + publicationGeneration: "publication_1", + transport: "sfu", + connectionId: "a_sfu", + usable: true, + physicalActive: true, + resource: "a_subscription", + }); + routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: 0, + connectionId: "a_sfu", + }); + const operation = routes.reconcile(10).operation!; + const deadline = operation.deadlineAtMs; + expect(operation.candidates[0]).toMatchObject({ + tuple: { kind: "peer", parentPeerId: HOST, transport: "direct" }, + endpointTransition: { + kind: "bounded-gap", + producerPeerId: HOST, + retire: { kind: "publication", generation: "publication_1" }, + }, + }); + const gap = routes.retireCurrentCandidateProducer(cursorGuard(operation), 11); + expect(gap).toMatchObject({ accepted: true }); + expect(gap.released).toEqual(["a_subscription", "publication_resource_1"]); + expect(gap.operation).toMatchObject({ + deadlineAtMs: deadline, + candidates: [{ endpointTransition: { kind: "none", producerPeerId: HOST } }], + }); + const prepared = beginCandidate(routes, { + nowMs: 12, + connectionId: "a_host_direct", + reservation: { kind: "direct" }, + }).operation!; + expect(routes.candidateReady({ + childPeerId: A, + childSessionId: `${A}_session`, + revision: prepared.current!.revision, + connectionId: "a_host_direct", + }, 13).accepted).toBe(true); + expect(routes.snapshot().hostPublication).toBeNull(); + }); + + it("retires one exact peer edge for a C=3 bounded gap", () => { + const routes = controller(3, { selectedTurnEnabled: true }); + addViewer(routes, A, 3); + routes.hydrateEdge(A, peerEdge(HOST, "a_direct")); + for (const peerId of [B, C, D]) { + addViewer(routes, peerId, 0); + routes.hydrateEdge(peerId, peerEdge(A, `${peerId}_from_a`)); + } + routes.invalidateEdge({ + childPeerId: B, + childSessionId: `${B}_session`, + parentSessionId: `${A}_session`, + routeRevision: 0, + connectionId: `${B}_from_a`, + }); + const first = routes.reconcile(0).operation!; + expect(first.candidates[0]?.tuple).toMatchObject({ + kind: "peer", + parentPeerId: HOST, + transport: "direct", + }); + expect(skipCandidate(routes, 1).accepted).toBe(true); + const bounded = routes.snapshot().operation!; + expect(bounded.candidates[bounded.cursor]).toMatchObject({ + tuple: { kind: "peer", parentPeerId: A, transport: "selected-turn" }, + endpointTransition: { + kind: "bounded-gap", + producerPeerId: A, + retire: { kind: "edge", childPeerId: B, connectionId: `${B}_from_a` }, + }, + }); + const gap = routes.retireCurrentCandidateProducer(cursorGuard(bounded), 2); + expect(gap).toMatchObject({ accepted: true, released: [] }); + const prepared = beginCandidate(routes, { + nowMs: 3, + connectionId: "b_selected_turn", + reservation: { kind: "selected-turn", edge: "b_turn_resource" }, + }).operation!; + expect(routes.candidateReady({ + childPeerId: B, + childSessionId: `${B}_session`, + revision: prepared.current!.revision, + connectionId: "b_selected_turn", + }, 4).accepted).toBe(true); + expect(routes.snapshot().upstreamByViewer.get(B)).toMatchObject({ + parentPeerId: A, + transport: "selected-turn", + resource: "b_turn_resource", + }); + }); +});