diff --git a/docs/adr/0005-automatic-hybrid-media-routing.md b/docs/adr/0005-automatic-hybrid-media-routing.md index a71aef8..f4f1b69 100644 --- a/docs/adr/0005-automatic-hybrid-media-routing.md +++ b/docs/adr/0005-automatic-hybrid-media-routing.md @@ -89,12 +89,16 @@ root. A fresh healthy candidate sender proof is required before make-before-brea commit. Success moves one branch toward even fanout; failure consumes that one-shot intent. No timer, continuous rebalance or general load score is introduced. -A failed tuple is consumed for its exact candidate opportunity: tuple, relevant -endpoint sessions or publication generation, and endpoint-transition strength. -Only a new session or generation, or a strictly better transition, makes that -opportunity new. Candidate removal, worsening, reordering, and unrelated facts -do not retry it. An external SFU-resource wake may reopen SFU opportunities but -never Peer opportunities. There is no persistent parent blacklist. +A failed Peer tuple is consumed for its exact parent and endpoint sessions plus +endpoint-transition strength. SFU create, reuse, and replace are resource +actions for one logical SFU opportunity keyed by the child and Host sessions; +the publication generation remains a physical fence, not retry authority. +Controller-owned publication creation, replacement, or teardown cannot make the +same failed opportunity new. Only a relevant session, an external SFU-resource +wake, or a strictly better transition reopens it. Candidate removal, worsening, +reordering, and unrelated facts do not retry it. An external SFU-resource wake +may reopen SFU opportunities but never Peer opportunities. There is no +persistent parent blacklist. ### Prepare, Commit, And Rollback diff --git a/docs/product/routing-transport.md b/docs/product/routing-transport.md index dc58c07..9882685 100644 --- a/docs/product/routing-transport.md +++ b/docs/product/routing-transport.md @@ -60,7 +60,9 @@ ICE connected alone is not commit proof. Availability repair may instead use the explicit bounded-gap plan defined by ADR-0005 when no overlap slot exists. Failure or stale authority releases the candidate and preserves unaffected branches. An operation's purpose is immutable; availability work preempts a -background operation by replacing it rather than changing its meaning. +background operation by replacing it rather than changing its meaning. An +internally created or retired SFU publication does not by itself reopen an +exhausted media route. Direct peers receive the first bounded opportunity, with SFU used for availability when direct paths cannot provide media. Remaining direct paths may diff --git a/docs/status.md b/docs/status.md index dceca34..6696745 100644 --- a/docs/status.md +++ b/docs/status.md @@ -77,6 +77,13 @@ This is the current execution index. Git history owns completed timelines; pagehide no longer triggers LiveKit's automatic disconnect. Native Viewer controls still own playback, and manual reconnect stays on the current route. +## Source/Production Delta + +- Current source preserves failed SFU opportunities across controller-owned + publication generations, preventing an exhausted Viewer from repeatedly + bootstrapping and tearing down the same SFU path. Production release + `a48ea4c` predates this correction. + ## Current Milestone Measured route convergence, generation-owned Viewer presentation and diff --git a/docs/todo.md b/docs/todo.md index 95fda67..4d80763 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -7,7 +7,9 @@ topics are not implementation authority. ## Now -No implementation task is currently authorized. +No implementation task is currently authorized. The SFU opportunity lifecycle +correction is integrated and verified in current source but remains undeployed; +deployment requires explicit authority. ## Parked Product Work diff --git a/src/server/room-route-controller.ts b/src/server/room-route-controller.ts index d8f4e69..679fa6c 100644 --- a/src/server/room-route-controller.ts +++ b/src/server/room-route-controller.ts @@ -99,6 +99,7 @@ export interface ControllerOptions { const routeDebug = debuglog("screener-route"); const MAX_DIRECT_HEAD_START_MS = 5_000; +const SFU_OPPORTUNITY_PREFIX = "sfu\0"; interface Participant { peerId: string; @@ -2541,7 +2542,6 @@ export class RoomRouteController { participant.bootstrapFailureReported = undefined; } if (old?.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) { - this.clearSfuCandidateOpportunities(); this.hostPublication = null; } if ( @@ -2553,7 +2553,6 @@ export class RoomRouteController { demand.sfuFirstAtNextRoute = true; } this.sfuBootstrapIntent = undefined; - this.consumedSfuBootstrapOpportunities.clear(); } else if (this.usableRoute(operation.demandPeerId)) { this.clearSfuBootstrapForDemand(operation.demandPeerId); } @@ -2649,13 +2648,9 @@ export class RoomRouteController { this.participants.get(plan.tuple.parentPeerId)?.sessionId ?? "" }`; } - const publicationGeneration = - plan.tuple.publication === "create" - ? "" - : (this.hostPublication?.generation ?? ""); - return `${tupleKey(plan.tuple)}\0${childSessionId}\0${ + return `${SFU_OPPORTUNITY_PREFIX}${childSessionId}\0${ this.participants.get(this.options.hostPeerId)?.sessionId ?? "" - }\0${publicationGeneration}`; + }`; } private consumeCandidateOpportunity( @@ -2677,15 +2672,20 @@ export class RoomRouteController { child: Participant, edge: CommittedEdge, ): void { - const base = - edge.kind === "peer" - ? `peer:${edge.parentPeerId}\0${child.sessionId ?? ""}\0${ - edge.parentSessionId - }` - : `sfu:reuse\0${child.sessionId ?? ""}\0${ - this.hostPublication?.hostSessionId ?? "" - }\0${edge.publicationGeneration}`; - consumeOpportunity(child.consumedCandidateOpportunities, base, 0); + if (edge.kind === "peer") { + consumeOpportunity( + child.consumedCandidateOpportunities, + `peer:${edge.parentPeerId}\0${child.sessionId ?? ""}\0${ + edge.parentSessionId + }`, + 0, + ); + return; + } + this.consumeCandidateOpportunity(child.peerId, { + tuple: { kind: "sfu", publication: "reuse" }, + endpointTransition: { kind: "none" }, + }); } private consumeAvailabilityOperation( @@ -3257,10 +3257,18 @@ export class RoomRouteController { return Boolean( this.sfuBootstrapGloballyNeeded() && this.participants.get(demandPeerId)?.sessionId && - !this.usableRoute(demandPeerId), + !this.usableRoute(demandPeerId) && + this.sfuOpportunityAvailable(demandPeerId), ); } + private sfuOpportunityAvailable(demandPeerId: string): boolean { + return this.candidateOpportunityAvailable(demandPeerId, { + tuple: { kind: "sfu", publication: "reuse" }, + endpointTransition: { kind: "none" }, + }); + } + private sfuBootstrapGloballyNeeded(): boolean { return Boolean( this.options.sfuEnabled && @@ -3598,7 +3606,6 @@ export class RoomRouteController { if (!this.hasSfuSubscribers() && this.hostPublication) { this.clearSfuQuality(); if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); - this.clearSfuCandidateOpportunities(); this.hostPublication = null; } this.revision = this.allocateRevision(); @@ -4108,11 +4115,14 @@ export class RoomRouteController { for (const key of child.consumedCandidateOpportunities.keys()) { if ( key.startsWith(parentPrefix) || - (role === "host" && key.startsWith("sfu:")) + (role === "host" && key.startsWith(SFU_OPPORTUNITY_PREFIX)) ) { child.consumedCandidateOpportunities.delete(key); } } + if (role === "host") { + child.bootstrapFailureReported = undefined; + } } if (role === "host") { this.consumedSfuBootstrapOpportunities.clear(); @@ -4129,7 +4139,7 @@ export class RoomRouteController { private clearSfuCandidateOpportunities(): void { for (const participant of this.participants.values()) { for (const key of participant.consumedCandidateOpportunities.keys()) { - if (key.startsWith("sfu:")) { + if (key.startsWith(SFU_OPPORTUNITY_PREFIX)) { participant.consumedCandidateOpportunities.delete(key); } } @@ -4307,7 +4317,6 @@ export class RoomRouteController { if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource); this.hostPublication = null; } - this.clearSfuCandidateOpportunities(); return released; } diff --git a/tests/hybrid-media-router.test.ts b/tests/hybrid-media-router.test.ts index 3ee7409..9a799e5 100644 --- a/tests/hybrid-media-router.test.ts +++ b/tests/hybrid-media-router.test.ts @@ -761,6 +761,185 @@ describe("HybridMediaRouter v9 runtime", () => { } }); + it("does not recreate SFU after an exhausted reuse and carrier teardown", async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const { store, sent, admission, roomControl, router } = harness( + 2, + true, + 300, + ); + const sfuConfigCount = () => + [...sent.values()] + .flat() + .filter((message) => message.type === "sfu-config").length; + const ownPrepare = (participant: AuthenticatedRouteParticipant) => + sent + .get(participant.sessionId) + ?.findLast( + ( + message, + ): message is Extract< + ServerMessage, + { type: "route-update"; phase: "prepare" } + > => + message.type === "route-update" && + message.phase === "prepare" && + message.candidate.childPeerId === participant.peerId, + ); + try { + const room = await store.createRoom(); + const host = connectHost(store, room); + complete(router, host); + + const firstRoot = connectViewer(store, room, "reuse-loop-first-root"); + complete(router, firstRoot); + await vi.waitFor(() => + expect(preparedFor(sent, firstRoot.sessionId)?.candidate.transport).toBe( + "direct", + ), + ); + const firstDirect = preparedFor(sent, firstRoot.sessionId)!; + router.handleRouteReady(firstRoot, { + type: "route-ready", + revision: firstDirect.revision, + phase: "prepare", + }); + router.setViewerRelayCapacity(firstRoot, 1); + + const secondRoot = connectViewer(store, room, "reuse-loop-second-root"); + complete(router, secondRoot); + await vi.waitFor(() => + expect(preparedFor(sent, secondRoot.sessionId)?.candidate.transport).toBe( + "direct", + ), + ); + const secondDirect = preparedFor(sent, secondRoot.sessionId)!; + router.handleRouteReady(secondRoot, { + type: "route-ready", + revision: secondDirect.revision, + phase: "prepare", + }); + + const departedParent = connectViewer( + store, + room, + "reuse-loop-departed-parent", + ); + complete(router, departedParent); + await vi.waitFor(() => + expect( + preparedFor(sent, departedParent.sessionId)?.assignment.upstream, + ).toEqual({ kind: "peer", peerId: firstRoot.peerId }), + ); + const parentDirect = preparedFor(sent, departedParent.sessionId)!; + router.handleRouteReady(departedParent, { + type: "route-ready", + revision: parentDirect.revision, + phase: "prepare", + }); + router.setViewerRelayCapacity(departedParent, 1); + + const demand = connectViewer(store, room, "reuse-loop-demand"); + complete(router, demand); + await vi.waitFor(() => + expect(preparedFor(sent, demand.sessionId)?.assignment.upstream).toEqual( + { kind: "peer", peerId: departedParent.peerId }, + ), + ); + const demandDirect = preparedFor(sent, demand.sessionId)!; + router.handleRouteReady(demand, { + type: "route-ready", + revision: demandDirect.revision, + phase: "prepare", + }); + + const departureAt = Date.now(); + store.disconnectParticipant( + room.roomId, + departedParent.peerId, + departedParent.sessionId, + ); + router.disconnectParticipant( + room.roomId, + departedParent.peerId, + departedParent.sessionId, + ); + router.removeViewer(room.roomId, departedParent.peerId); + router.setViewerRelayCapacity(secondRoot, 1); + + await vi.waitFor(() => { + const recovery = ownPrepare(demand); + expect(recovery?.revision).toBeGreaterThan(demandDirect.revision); + expect(recovery?.candidate).toMatchObject({ + childPeerId: demand.peerId, + transport: "direct", + }); + }); + await vi.advanceTimersByTimeAsync( + Math.max(0, departureAt + 150 - Date.now()), + ); + + const carrier = await vi.waitFor(() => { + const value = [firstRoot, secondRoot].find( + (viewer) => + ownPrepare(viewer)?.candidate.transport === "sfu", + ); + expect(value).toBeDefined(); + return value!; + }); + const bootstrap = ownPrepare(carrier)!; + router.handleRouteReady(carrier, { + type: "route-ready", + revision: bootstrap.revision, + phase: "prepare", + }); + + await vi.waitFor(() => + expect(preparedFor(sent, demand.sessionId)?.candidate.transport).toBe( + "sfu", + ), + ); + expect(roomControl?.created).toHaveLength(1); + const configCountAfterBootstrap = sfuConfigCount(); + expect(configCountAfterBootstrap).toBeGreaterThanOrEqual(3); + + await vi.advanceTimersByTimeAsync( + Math.max(0, departureAt + 300 - Date.now()), + ); + await vi.waitFor(() => + expect(ownPrepare(carrier)?.candidate.transport).toBe( + "direct", + ), + ); + expect(router.routeDiagnosticSnapshot(room.roomId).operation).toMatchObject({ + reason: "direct-convergence", + }); + const carrierDirect = ownPrepare(carrier)!; + router.handleRouteReady(carrier, { + type: "route-ready", + revision: carrierDirect.revision, + phase: "prepare", + }); + + await vi.waitFor(() => + expect( + router.resolveActiveViewerMediaEdge(room.roomId, carrier.peerId) + ?.upstream, + ).toEqual({ kind: "peer", peerId: host.peerId }), + ); + await vi.waitFor(() => expect(roomControl?.deleted).toHaveLength(1)); + await vi.advanceTimersByTimeAsync(300); + + expect(roomControl?.created).toHaveLength(1); + expect(sfuConfigCount()).toBe(configCountAfterBootstrap); + expect(admission?.usage()).toEqual({ ingress: 0, egress: 0 }); + } finally { + await router.close(); + vi.useRealTimers(); + } + }); + it("reports a late final SFU bootstrap ready only to the waiting demand", async () => { vi.useFakeTimers(); vi.setSystemTime(0); diff --git a/tests/room-route-controller.test.ts b/tests/room-route-controller.test.ts index da67899..9ebc56d 100644 --- a/tests/room-route-controller.test.ts +++ b/tests/room-route-controller.test.ts @@ -130,6 +130,130 @@ function peerEdge( }; } +function retireLastSfuPublicationAfterReuseDeadline(): { + routes: RoomRouteController; + nowMs: number; +} { + const routes = controller(2, { sfuEnabled: true }); + for (const [peerId, connectionId] of [ + [B, "b_from_host"], + [C, "c_from_host"], + ] as const) { + addViewer(routes, peerId, 0); + routes.hydrateEdge(peerId, peerEdge(HOST, connectionId)); + } + addViewer(routes, A, 0, 0); + + const bootstrap = routes.reconcile(0).operation!; + expect(bootstrap).toMatchObject({ + demandPeerId: A, + reason: "sfu-bootstrap", + candidates: [ + { + tuple: { kind: "sfu", publication: "create" }, + }, + ], + }); + const carrierPeerId = bootstrap.childPeerId; + const carrier = beginCandidate(routes, { + nowMs: 1, + connectionId: "carrier_sfu", + publicationGeneration: "publication_generation", + publicationConnectionId: "publication_connection", + reservation: { + kind: "sfu-create", + edge: "carrier_subscription", + publication: "publication_resource", + overlap: "publication_overlap", + }, + }).operation!; + expect( + routes.candidateReady( + { + childPeerId: carrier.childPeerId, + childSessionId: carrier.childSessionId, + revision: carrier.current!.revision, + connectionId: "carrier_sfu", + }, + 2, + ).accepted, + ).toBe(true); + + const demand = routes.reconcile(3).operation!; + expect(demand).toMatchObject({ + childPeerId: A, + demandPeerId: A, + candidates: [ + { + tuple: { kind: "sfu", publication: "reuse" }, + }, + ], + }); + beginCandidate(routes, { + nowMs: 4, + connectionId: "demand_sfu_reuse", + reservation: { kind: "sfu-reuse", edge: "demand_subscription" }, + }); + expect(routes.operationExpired(demand.deadlineAtMs).failedPeerIds).toEqual([ + A, + ]); + + const convergence = routes.reconcile(demand.deadlineAtMs + 1).operation!; + expect(convergence).toMatchObject({ + childPeerId: carrierPeerId, + reason: "direct-convergence", + candidates: [ + { + tuple: { kind: "peer", parentPeerId: HOST, transport: "direct" }, + }, + ], + }); + const direct = beginCandidate(routes, { + nowMs: demand.deadlineAtMs + 2, + connectionId: "carrier_direct", + reservation: { kind: "direct", overlap: "carrier_direct_overlap" }, + }).operation!; + const committed = routes.candidateReady( + { + childPeerId: direct.childPeerId, + childSessionId: direct.childSessionId, + revision: direct.current!.revision, + connectionId: "carrier_direct", + }, + demand.deadlineAtMs + 3, + ); + expect(committed.accepted).toBe(true); + expect(committed.released).toEqual( + expect.arrayContaining(["carrier_subscription", "publication_resource"]), + ); + expect(routes.snapshot().hostPublication).toBeNull(); + return { routes, nowMs: demand.deadlineAtMs + 4 }; +} + +function invalidateActiveSfuEdge(): RoomRouteController { + const routes = controller(1, { sfuEnabled: true }); + addViewer(routes, A, 0); + routes.hydrateHostPublication("active_publication", "publication_resource"); + routes.hydrateEdge(A, { + kind: "sfu", + publicationGeneration: "active_publication", + transport: "sfu", + connectionId: "active_sfu_edge", + usable: true, + physicalActive: true, + resource: "active_subscription", + }); + expect( + routes.invalidateEdge({ + childPeerId: A, + childSessionId: `${A}_session`, + routeRevision: routes.snapshot().revision, + connectionId: "active_sfu_edge", + }), + ).toBe(true); + return routes; +} + function observeSenderState( routes: RoomRouteController, input: { @@ -3374,7 +3498,7 @@ describe("RoomRouteController", () => { expect(routes.snapshot().hostPublication).toBeNull(); }); - it("does not reopen a consumed none transition when it worsens to overlap", () => { + it("does not reopen consumed none transitions after capacity reduction", () => { const routes = controller(2, { sfuEnabled: true }); addViewer(routes, A, 0); routes.hydrateHostPublication("publication", "publication_resource"); @@ -3403,20 +3527,7 @@ describe("RoomRouteController", () => { expect(routes.operationExpired(operation.deadlineAtMs).failedPeerIds).toEqual([ A, ]); - const retry = routes.reconcile(operation.deadlineAtMs + 1).operation!; - expect(retry.candidates).toEqual([ - { - tuple: { kind: "sfu", publication: "replace" }, - endpointTransition: { kind: "none", producerPeerId: HOST }, - }, - ]); - expect( - retry.candidates.some( - (candidate) => - candidate.tuple.kind === "peer" && - candidate.tuple.parentPeerId === HOST, - ), - ).toBe(false); + expect(routes.reconcile(operation.deadlineAtMs + 1).operation).toBeUndefined(); }); it("does not unblock an exhausted demand for an unrelated join", () => { @@ -3598,6 +3709,144 @@ describe("RoomRouteController", () => { }); }); + it("does not bootstrap a new SFU generation after the last reuse opportunity expires", () => { + const { routes, nowMs } = retireLastSfuPublicationAfterReuseDeadline(); + + const reconciled = routes.reconcile(nowMs); + expect(reconciled.operation).toBeUndefined(); + expect(reconciled.failedPeerIds).toEqual([]); + }); + + it.each(["child-session", "host-session", "external-resource"] as const)( + "reopens the exhausted SFU opportunity for a new %s authority", + (authority) => { + const { routes, nowMs } = retireLastSfuPublicationAfterReuseDeadline(); + if (authority === "child-session") { + routes.upsertParticipant({ + peerId: A, + role: "viewer", + sessionId: "replacement_child_session", + effectiveDownstreamCapacity: 0, + }); + } else if (authority === "host-session") { + routes.upsertParticipant({ + peerId: HOST, + role: "host", + sessionId: "replacement_host_session", + effectiveDownstreamCapacity: 2, + }); + } else { + routes.touchExternalFacts(); + } + + expect(routes.reconcile(nowMs).operation).toMatchObject({ + demandPeerId: A, + reason: "sfu-bootstrap", + candidates: [ + { + tuple: { kind: "sfu", publication: "create" }, + }, + ], + }); + }, + ); + + it("consumes the canonical SFU opportunity when an active edge fails", () => { + const routes = invalidateActiveSfuEdge(); + + const operation = routes.reconcile(0).operation; + expect( + operation?.candidates.some((candidate) => candidate.tuple.kind === "sfu"), + ).toBe(false); + }); + + it.each(["child-session", "host-session", "external-resource"] as const)( + "reopens an active-edge SFU opportunity for a new %s authority", + (authority) => { + const routes = invalidateActiveSfuEdge(); + if (authority === "child-session") { + routes.upsertParticipant({ + peerId: A, + role: "viewer", + sessionId: "active_edge_child_replacement", + effectiveDownstreamCapacity: 0, + }); + } else if (authority === "host-session") { + routes.upsertParticipant({ + peerId: HOST, + role: "host", + sessionId: "active_edge_host_replacement", + effectiveDownstreamCapacity: 1, + }); + } else { + routes.touchExternalFacts(); + } + + expect( + routes + .reconcile(0) + .operation?.candidates.some( + (candidate) => + candidate.tuple.kind === "sfu" && + candidate.tuple.publication === "reuse", + ), + ).toBe(true); + }, + ); + + it("does not turn a failed same-rank SFU replace into a create retry", () => { + const routes = controller(2, { sfuEnabled: true }); + routes.hydrateHostPublication( + "failed_publication", + "failed_publication_resource", + ); + expect( + routes.invalidateHostPublication({ + hostSessionId: "host_session", + routeRevision: routes.snapshot().revision, + generation: "failed_publication", + connectionId: "publication:failed_publication", + }), + ).toBe(true); + addViewer(routes, A, 0); + const operation = routes.reconcile(0).operation!; + expect(operation.candidates[0]?.tuple).toMatchObject({ kind: "peer" }); + skipCandidate(routes, 1); + expect(routes.snapshot().operation?.candidates[1]).toMatchObject({ + tuple: { kind: "sfu", publication: "replace" }, + endpointTransition: { kind: "none", producerPeerId: HOST }, + }); + const replace = beginCandidate(routes, { + nowMs: 2, + connectionId: "failed_replace", + publicationGeneration: "replacement_publication", + publicationConnectionId: "replacement_ingress", + reservation: { + kind: "sfu-create", + edge: "replacement_subscription", + publication: "replacement_publication_resource", + }, + }).operation!; + expect( + routes.candidateFailed( + { + childPeerId: A, + childSessionId: `${A}_session`, + revision: replace.current!.revision, + connectionId: "failed_replace", + }, + 3, + ).failedPeerIds, + ).toEqual([A]); + + addViewer(routes, B, 0); + expect(routes.confirmDeparture(B)).toBe(true); + const reconciled = routes.reconcile(4); + expect(routes.snapshot().hostPublication).toBeNull(); + expect(reconciled.operation).toBeUndefined(); + expect(reconciled.failedPeerIds).toEqual([]); + }); + it("reopens only SFU opportunities for a new external fact", () => { const routes = controller(1, { sfuEnabled: true }); addViewer(routes, A, 0, 0);