fix(route): stop SFU opportunity generation churn (#317)

This commit is contained in:
AAA authored and GitHub committed 2026-08-28 21:31:02 +08:00
1 parent ce7de88034
commit 27421f458c
7 files changed
+497 -45

No files matched your search

@@ -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
+3 -1
View File
@@ -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
+7
View File
@@ -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
+3 -1
View File
@@ -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
+31 -22
View File
@@ -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<Resource = unknown> {
participant.bootstrapFailureReported = undefined;
}
if (old?.kind === "sfu" && !this.hasSfuSubscribers() && this.hostPublication) {
this.clearSfuCandidateOpportunities();
this.hostPublication = null;
}
if (
@@ -2553,7 +2553,6 @@ export class RoomRouteController<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
child: Participant,
edge: CommittedEdge<Resource>,
): 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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
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<Resource = unknown> {
if (this.hostPublication.physicalActive) released.push(this.hostPublication.resource);
this.hostPublication = null;
}
this.clearSfuCandidateOpportunities();
return released;
}
+179
View File
@@ -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);
+264 -15
View File
@@ -130,6 +130,130 @@ function peerEdge(
};
}
function retireLastSfuPublicationAfterReuseDeadline(): {
routes: RoomRouteController<string>;
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<string> {
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<string>,
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);