AAA
89a03e6470
## Problem and outcome / 问题与结果 This maintenance phase fixes failures around native capture startup, room reconnection and media handoffs. Capture mutations stay cancellable, relayed-content failures remain request-scoped, and failed media connections return to the existing route controller. Windows Auto codec selection measures delivered cadence instead of timer-sensitive per-frame work estimates. The phase preserves P2P-first routing, one committed graph and one route operation. It adds no signaling replay mechanism, changes no wire/schema version, and includes no interaction prototype or 4K feature work. ## Release notes ### 中文 Piik v1.6.8 是一次稳定性修复，重点改善 Windows 原生采集、断线恢复和分享过程中的控制响应。 #### 修复与改善 - **更准确地选择编码器。** Windows 的自动模式按实际出帧表现评估 H.264，并正确区分启动预热和持续性能不足，减少计时误差导致的不必要 VP8 回退。硬件编码事件改为异步等待。 - **改善断线与线路切换。** 房间信令短暂断开后保留观众成员身份 20 秒；修复线路切换、SFU 订阅恢复及声音重协商卡住的问题。失效连接及时交回现有路由控制器处理，原生 SFU 发送端不再把短暂断连直接判为终止。 - **保持 App 控制响应。** 开始分享和更换来源时仍可取消操作；合法消息突发不再直接关闭控制连接。中继信令的单条内容错误不会连带结束整场原生分享。 - **修正设备列表与配置保存。** 来源或麦克风数量达到上限时保留可用列表；配置替换失败时保留原有设置。 - **改善紧凑布局和辅助功能。** 调整手机横屏下的画面与播控布局、画质预设标签和长邀请链接显示，修正部分提示与辅助功能语义。 - **加强站点访问保护。** 站点登录 Cookie 使用独立随机签名密钥，应用自身阻止被其他页面嵌套。发布说明也会在合并前检查。 #### 升级说明 通信协议和持久化格式保持不变。设置了站点密码时，App 或 Server 重启后需要重新输入站点密码；持久化房间和邀请授权不受这项 Cookie 调整影响。Piik 操作页面不再支持嵌入 iframe。 ### English Piik v1.6.8 is a stability update focused on Windows native capture, connection recovery and responsive sharing controls. #### Fixes and improvements - **More accurate codec selection.** Windows Auto mode evaluates H.264 using delivered frames and distinguishes startup warmup from sustained performance limits, reducing unnecessary VP8 fallback caused by timing noise. Hardware encoder events now use asynchronous waiting. - **Improved reconnection and route handoffs.** Viewer membership survives brief signaling outages for 20 seconds. Fixes cover route activation, SFU subscription recovery and stalled audio renegotiation. Failed connections return to the existing route controller, and the native SFU publisher no longer treats a transient disconnect as immediate termination. - **Responsive App controls.** Starting a share and replacing its source remain cancellable. Valid message bursts no longer close the control connection, and an invalid relayed payload cannot terminate the entire native sharing session. - **Device lists and saved settings.** Source and microphone enumeration retain a usable list at their limits. Failed configuration replacement preserves the previous settings. - **Compact layouts and accessibility.** Improved picture and playback-control layout on landscape phones, quality preset labels and long invitation links, with corrected tooltip and accessibility semantics. - **Stronger site-access protection.** Login cookies use an independent random signing key, and the application prevents embedding by other pages. Release notes are also checked before merge. #### Upgrade notes Wire protocols and persistent formats are unchanged. If a site password is configured, restarting the App or Server requires entering it again; this cookie change does not revoke persisted rooms or invitation grants. Piik operation pages can no longer be embedded in an iframe. ## Verification / 验证 - `npm run check`: passed, 76 test files / 1,171 tests, TypeScript and production Web build. - `npm run check:go-core`: passed; full Go/native checks also passed during integration of the capture changes. macOS cgo builds remain native-runner checks. - Windows native Host gates: Auto, H.264 and VP8 passed during the phase; final control changes rechecked with Auto, including live/paused quality, source replacement, failure retirement and subsequent sharing. - Embedded SFU browser VP8 recovery: passed with production Host/Viewer route owners. - Embedded SFU native H.264, two rooms: passed, including recovered decoded video/audio, independent room retirement and resource cleanup. - Documentation, repository hygiene and diff checks: passed. Final follow-up changes only align operations documentation with the PR note check. - Go race checks could not run locally because no cgo/GCC toolchain is available; the existing Linux CI race job remains required. AMD/Intel hardware, broader device combinations, public-NAT field reports and long endurance are not established by these local checks. ## Impact / 影响 Compatible patch release after v1.6.7. The cookie and frame-protection effects are disclosed in the public upgrade notes. No automatic retry budget, room-code admission policy, room lifetime or media topology is replaced. Reports without matched diagnostics and unmeasured native performance costs remain in the TODO ledger.
1813 lines
63 KiB
TypeScript
1813 lines
63 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import type {
|
|
ClientMessage,
|
|
ParticipantRouteAssignment,
|
|
PreparedRouteCandidate,
|
|
} from "../src/shared/protocol.ts";
|
|
import { HostSfuRoute } from "../src/client/media/host-sfu-route.ts";
|
|
import { QUALITY_PROFILES } from "../src/client/media/quality.ts";
|
|
import {
|
|
exactPeerSignalOwner,
|
|
MediaRouteTransition,
|
|
reportActivePeerRouteFailure,
|
|
} from "../src/client/media/route-transition.ts";
|
|
import { ViewerSfuRoute } from "../src/client/media/viewer-sfu-route.ts";
|
|
import { t } from "../src/client/ui/copy.ts";
|
|
import { resolveMediaFailure } from "../src/client/ui/media-failure.ts";
|
|
import {
|
|
EMPTY_METRICS,
|
|
type ConnectionMetrics,
|
|
} from "../src/client/types.ts";
|
|
|
|
const peerAssignment = (
|
|
parentPeerId: string,
|
|
childPeerIds: string[] = [],
|
|
): ParticipantRouteAssignment => ({
|
|
upstream: { kind: "peer", peerId: parentPeerId },
|
|
childPeerIds,
|
|
sfuPublicationGeneration: null,
|
|
});
|
|
|
|
const viewerSfuAssignment = (
|
|
childPeerIds: string[] = [],
|
|
publicationGeneration = "publication_generation_12345678",
|
|
): ParticipantRouteAssignment => ({
|
|
upstream: { kind: "sfu" },
|
|
childPeerIds,
|
|
sfuPublicationGeneration: publicationGeneration,
|
|
});
|
|
|
|
const emptyViewerAssignment = (): ParticipantRouteAssignment => ({
|
|
upstream: { kind: "none" },
|
|
childPeerIds: [],
|
|
sfuPublicationGeneration: null,
|
|
});
|
|
|
|
const hostAssignment = (
|
|
publicationGeneration: string | null,
|
|
childPeerIds: string[] = [],
|
|
): ParticipantRouteAssignment => ({
|
|
upstream: { kind: "none" },
|
|
childPeerIds,
|
|
sfuPublicationGeneration: publicationGeneration,
|
|
});
|
|
|
|
function sfuConfig(revision: number, publicationGeneration = "publication_generation_12345678") {
|
|
return {
|
|
type: "sfu-config" as const,
|
|
revision,
|
|
publicationGeneration,
|
|
connectionId: `sfu_connection_${revision}_12345678`,
|
|
};
|
|
}
|
|
|
|
function createFakePublisher(log: string[], label: string) {
|
|
return {
|
|
updateConfig: vi.fn(),
|
|
acceptSignal: vi.fn(async () => undefined),
|
|
connect: vi.fn(async (_config: unknown) => {
|
|
log.push(`${label}:connect`);
|
|
return true;
|
|
}),
|
|
activate: vi.fn(async () => {
|
|
log.push(`${label}:activate`);
|
|
return true;
|
|
}),
|
|
deactivate: vi.fn(async () => {
|
|
log.push(`${label}:deactivate`);
|
|
return true;
|
|
}),
|
|
replaceStream: vi.fn(async () => true),
|
|
updateProfile: vi.fn(async () => true),
|
|
setPaused: vi.fn(),
|
|
disconnect: vi.fn(async () => {
|
|
log.push(`${label}:disconnect`);
|
|
}),
|
|
};
|
|
}
|
|
|
|
type SubscriberEvents = Parameters<
|
|
NonNullable<ConstructorParameters<typeof ViewerSfuRoute>[1]["createSubscriber"]>
|
|
>[0];
|
|
|
|
const candidate = (
|
|
revision: number,
|
|
childPeerId = "viewer_12345678",
|
|
transport: "direct" | "sfu" = "direct",
|
|
connectionId = `candidate_${revision}_12345678`,
|
|
qualityProbe = false,
|
|
): PreparedRouteCandidate => ({
|
|
childPeerId,
|
|
connectionId,
|
|
transport,
|
|
qualityProbe,
|
|
});
|
|
|
|
function createFakeSubscriber(events: SubscriberEvents, log: string[], label: string) {
|
|
return {
|
|
resync: vi.fn(),
|
|
updateConfig: vi.fn(),
|
|
acceptSignal: vi.fn(async () => undefined),
|
|
reconnect: vi.fn(() => true),
|
|
events,
|
|
connect: vi.fn(async () => {
|
|
log.push(`${label}:connect`);
|
|
return true;
|
|
}),
|
|
activate: vi.fn(() => {
|
|
log.push(`${label}:activate`);
|
|
return true;
|
|
}),
|
|
deactivate: vi.fn(() => {
|
|
log.push(`${label}:deactivate`);
|
|
return true;
|
|
}),
|
|
armDecodedFrameProof: vi.fn((requireProgress = false) => {
|
|
log.push(`${label}:proof:${requireProgress ? "progress" : "fresh"}`);
|
|
}),
|
|
stopDecodedFrameProof: vi.fn(() => {
|
|
log.push(`${label}:proof:stop`);
|
|
}),
|
|
disconnect: vi.fn(async () => {
|
|
log.push(`${label}:disconnect`);
|
|
}),
|
|
};
|
|
}
|
|
|
|
async function prepareSfu(
|
|
route: ViewerSfuRoute,
|
|
revision: number,
|
|
assignment = viewerSfuAssignment(),
|
|
config = sfuConfig(revision, assignment.sfuPublicationGeneration ?? undefined),
|
|
) {
|
|
route.accept({ revision, phase: "prepare", assignment,
|
|
candidate: candidate(revision, "viewer_12345678", "sfu", config.connectionId) });
|
|
await route.acceptConfig(config);
|
|
route.accept({ revision, phase: "active", assignment });
|
|
}
|
|
|
|
function receiveMetrics(
|
|
timestampMs: number,
|
|
overrides: Partial<ConnectionMetrics> = {},
|
|
): ConnectionMetrics {
|
|
return {
|
|
...EMPTY_METRICS,
|
|
sampleTimestampMs: timestampMs,
|
|
sampleWindowMs: 2_000,
|
|
frameWidth: 1920,
|
|
frameHeight: 1080,
|
|
framesPerSecond: 60,
|
|
bitrateKbps: 5_000,
|
|
intervalFramesDecoded: 120,
|
|
intervalFreezeCount: 0,
|
|
intervalFreezeDurationMs: 0,
|
|
intervalPauseCount: 0,
|
|
intervalPauseDurationMs: 0,
|
|
...overrides,
|
|
};
|
|
}
|
|
|
|
describe("minimal route transition contracts", () => {
|
|
it("reports active failure when a prepared Peer disappears before commit delivery", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const onSfuStream = vi.fn();
|
|
let subscriber!: ReturnType<typeof createFakeSubscriber>;
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => false,
|
|
onSfuStream,
|
|
send: (message) => { messages.push(message); return true; },
|
|
createSubscriber: (events) => {
|
|
subscriber = createFakeSubscriber(events, [], "current");
|
|
return subscriber;
|
|
},
|
|
});
|
|
await prepareSfu(route, 1, viewerSfuAssignment(), sfuConfig(1));
|
|
subscriber.events.onStream({} as MediaStream);
|
|
subscriber.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(onSfuStream).toHaveBeenCalledOnce());
|
|
messages.length = 0;
|
|
|
|
const assignment = peerAssignment("host_12345678");
|
|
route.accept({ revision: 2, phase: "prepare", assignment, candidate: candidate(2) });
|
|
// First-frame readiness reached the server, then the local candidate failed.
|
|
// Its prepare failure can arrive after the server committed this revision.
|
|
route.accept({ revision: 2, phase: "active", assignment });
|
|
await vi.waitFor(() => expect(messages).toContainEqual({
|
|
type: "route-failed", revision: 2, phase: "active", connectionId: candidate(2).connectionId,
|
|
}));
|
|
expect(messages).not.toContainEqual({ type: "route-ready", revision: 2, phase: "active" });
|
|
expect(subscriber.disconnect).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it.each([false, true])("ignores a superseded Peer promotion resolving to %s", async (promoted) => {
|
|
const messages: ClientMessage[] = [];
|
|
const onSfuStream = vi.fn();
|
|
let finishPromotion!: (value: boolean) => void;
|
|
const promotion = new Promise<boolean>((resolve) => { finishPromotion = resolve; });
|
|
const activatePeer = vi.fn((_assignment: ParticipantRouteAssignment, revision?: number) =>
|
|
revision === 2 ? promotion : true,
|
|
);
|
|
let subscriber!: ReturnType<typeof createFakeSubscriber>;
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer,
|
|
onSfuStream,
|
|
send: (message) => { messages.push(message); return true; },
|
|
createSubscriber: (events) => {
|
|
subscriber = createFakeSubscriber(events, [], "current");
|
|
return subscriber;
|
|
},
|
|
});
|
|
await prepareSfu(route, 1, viewerSfuAssignment(), sfuConfig(1));
|
|
subscriber.events.onStream({} as MediaStream);
|
|
subscriber.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(onSfuStream).toHaveBeenCalledOnce());
|
|
messages.length = 0;
|
|
|
|
const first = peerAssignment("first_parent");
|
|
route.accept({ revision: 2, phase: "prepare", assignment: first, candidate: candidate(2) });
|
|
route.accept({ revision: 2, phase: "active", assignment: first });
|
|
await vi.waitFor(() => expect(activatePeer).toHaveBeenCalledWith(first, 2));
|
|
const next = peerAssignment("next_parent");
|
|
route.accept({ revision: 3, phase: "prepare", assignment: next, candidate: candidate(3) });
|
|
finishPromotion(promoted);
|
|
await promotion;
|
|
await Promise.resolve();
|
|
expect(messages).toEqual([]);
|
|
expect(subscriber.disconnect).not.toHaveBeenCalled();
|
|
|
|
route.accept({ revision: 3, phase: "active", assignment: next });
|
|
await vi.waitFor(() => expect(activatePeer).toHaveBeenCalledWith(next, 3));
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("rebinds Viewer route ownership before a post-restart prepare", async () => {
|
|
const prepared: Array<{ parentPeerId: string | null; revision?: number }> = [];
|
|
const route = new ViewerSfuRoute("viewer_old_12345678", {
|
|
activatePeer: () => true,
|
|
preparePeer: (assignment, revision) =>
|
|
prepared.push({
|
|
parentPeerId:
|
|
assignment?.upstream.kind === "peer"
|
|
? assignment.upstream.peerId
|
|
: null,
|
|
...(revision === undefined ? {} : { revision }),
|
|
}),
|
|
resetMedia: () => undefined,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
send: () => true,
|
|
});
|
|
route.accept({
|
|
revision: 9,
|
|
phase: "active",
|
|
assignment: peerAssignment("old_parent_12345678"),
|
|
});
|
|
|
|
const resync = route.resyncAuthoritative(
|
|
{
|
|
revision: 0,
|
|
phase: "active",
|
|
assignment: emptyViewerAssignment(),
|
|
},
|
|
"viewer_new_12345678",
|
|
);
|
|
expect(
|
|
route.accept({
|
|
revision: 1,
|
|
phase: "prepare",
|
|
assignment: peerAssignment("new_parent_12345678"),
|
|
candidate: candidate(
|
|
1,
|
|
"viewer_new_12345678",
|
|
"direct",
|
|
"post_restart_connection_12345678",
|
|
),
|
|
}),
|
|
).toBe("accepted");
|
|
await resync;
|
|
|
|
expect(prepared.at(-1)).toEqual({
|
|
parentPeerId: "new_parent_12345678",
|
|
revision: 1,
|
|
});
|
|
});
|
|
|
|
it("keeps old media through prepare and commits only exact active authority", () => {
|
|
const route = new MediaRouteTransition();
|
|
const oldAssignment = peerAssignment("old-parent");
|
|
const nextAssignment = peerAssignment("next-parent");
|
|
|
|
route.accept({ revision: 1, phase: "active", assignment: oldAssignment });
|
|
const oldToken = route.token()!;
|
|
expect(route.markMediaActive(oldToken)).toBe(true);
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: nextAssignment,
|
|
candidate: candidate(2),
|
|
});
|
|
expect(route.getMediaAssignment()).toEqual(oldAssignment);
|
|
expect(route.markMediaActive(route.token()!)).toBe(false);
|
|
|
|
route.accept({ revision: 2, phase: "active", assignment: nextAssignment });
|
|
expect(route.markMediaActive(route.token()!)).toBe(true);
|
|
expect(route.getMediaAssignment()).toEqual(nextAssignment);
|
|
expect(
|
|
route.accept({ revision: 1, phase: "active", assignment: oldAssignment }),
|
|
).toBe("stale");
|
|
});
|
|
|
|
it("treats child assignment order as presentation-only", () => {
|
|
const route = new MediaRouteTransition();
|
|
route.accept({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: peerAssignment("parent", ["first", "second"]),
|
|
});
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: peerAssignment("parent", ["second", "first"]),
|
|
candidate: candidate(2, "second"),
|
|
});
|
|
|
|
expect(
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: peerAssignment("parent", ["first", "second"]),
|
|
}),
|
|
).toBe("accepted");
|
|
});
|
|
|
|
|
|
it("keeps duplicate peer prepare idempotent and releases it before takeover", async () => {
|
|
const prepared: Array<{ parentPeerId: string | null; revision?: number }> = [];
|
|
const activatePeer = vi.fn(() => true);
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer,
|
|
preparePeer: (assignment, revision) =>
|
|
prepared.push({
|
|
parentPeerId:
|
|
assignment?.upstream.kind === "peer"
|
|
? assignment.upstream.peerId
|
|
: null,
|
|
...(revision === undefined ? {} : { revision }),
|
|
}),
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
send: () => true,
|
|
});
|
|
const oldAssignment = peerAssignment("old-parent");
|
|
const nextAssignment = peerAssignment("next-parent");
|
|
|
|
route.accept({ revision: 1, phase: "active", assignment: oldAssignment });
|
|
await vi.waitFor(() => expect(activatePeer).toHaveBeenCalled());
|
|
prepared.length = 0;
|
|
|
|
expect(
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: nextAssignment,
|
|
candidate: candidate(2),
|
|
}),
|
|
).toBe("accepted");
|
|
expect(prepared).toEqual([
|
|
{ parentPeerId: "next-parent", revision: 2 },
|
|
]);
|
|
|
|
expect(
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: nextAssignment,
|
|
candidate: candidate(2),
|
|
}),
|
|
).toBe("duplicate");
|
|
expect(prepared).toEqual([
|
|
{ parentPeerId: "next-parent", revision: 2 },
|
|
]);
|
|
|
|
route.accept({
|
|
revision: 3,
|
|
phase: "prepare",
|
|
assignment: peerAssignment("third-parent"),
|
|
candidate: candidate(3),
|
|
});
|
|
expect(prepared.slice(-2)).toEqual([
|
|
{ parentPeerId: null },
|
|
{ parentPeerId: "third-parent", revision: 3 },
|
|
]);
|
|
|
|
route.accept({
|
|
revision: 4,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment(),
|
|
candidate: candidate(4, "viewer_12345678", "sfu"),
|
|
});
|
|
expect(prepared.at(-1)).toEqual({ parentPeerId: null });
|
|
});
|
|
|
|
it("routes same-parent overlap signals only by exact connection identity", () => {
|
|
expect(exactPeerSignalOwner("pending", "pending", "active")).toBe(
|
|
"pending",
|
|
);
|
|
expect(exactPeerSignalOwner("active", "pending", "active")).toBe(
|
|
"active",
|
|
);
|
|
expect(exactPeerSignalOwner("unknown", "pending", "active")).toBeNull();
|
|
});
|
|
|
|
it("reports only the exact active peer failure", () => {
|
|
const route = new MediaRouteTransition();
|
|
const sent: ClientMessage[] = [];
|
|
route.accept({
|
|
revision: 4,
|
|
phase: "active",
|
|
assignment: peerAssignment("parent"),
|
|
});
|
|
|
|
expect(
|
|
reportActivePeerRouteFailure(route, "other", "connection", (message) => {
|
|
sent.push(message);
|
|
return true;
|
|
}),
|
|
).toBe(true);
|
|
expect(sent).toEqual([]);
|
|
|
|
reportActivePeerRouteFailure(route, "parent", "connection", (message) => {
|
|
sent.push(message);
|
|
return true;
|
|
});
|
|
expect(sent).toEqual([
|
|
{
|
|
type: "route-failed",
|
|
revision: 4,
|
|
phase: "active",
|
|
connectionId: "connection",
|
|
},
|
|
]);
|
|
});
|
|
|
|
it("reports committed Peer failure during another preparation using active authority", () => {
|
|
const route = new MediaRouteTransition();
|
|
const send = vi.fn(() => false);
|
|
route.accept({ revision: 7, phase: "active", assignment: peerAssignment("parent") });
|
|
route.accept({
|
|
revision: 8, phase: "prepare",
|
|
assignment: peerAssignment("parent", ["child"]), candidate: candidate(8, "child"),
|
|
});
|
|
|
|
expect(reportActivePeerRouteFailure(route, "parent", "active-connection", send)).toBe(false);
|
|
expect(send).toHaveBeenLastCalledWith({
|
|
type: "route-failed", revision: 7, phase: "active", connectionId: "active-connection",
|
|
});
|
|
send.mockReturnValue(true);
|
|
expect(reportActivePeerRouteFailure(route, "parent", "active-connection", send)).toBe(true);
|
|
|
|
route.accept({ revision: 9, phase: "active", assignment: peerAssignment("replacement") });
|
|
send.mockClear();
|
|
expect(reportActivePeerRouteFailure(route, "parent", "active-connection", send)).toBe(true);
|
|
route.reset();
|
|
expect(reportActivePeerRouteFailure(route, "replacement", "new-connection", send)).toBe(true);
|
|
expect(send).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("reports the committed SFU publisher when a Peer child is preparing", async () => {
|
|
const send = vi.fn(() => true);
|
|
let fail!: () => void;
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "h264",
|
|
reconcileChildren: () => undefined,
|
|
send,
|
|
createPublisher: (onDisconnected) => {
|
|
fail = onDisconnected;
|
|
return createFakePublisher([], "active");
|
|
},
|
|
});
|
|
const config = sfuConfig(7);
|
|
await route.acceptAndWait({ revision: 7, phase: "active", assignment: hostAssignment(config.publicationGeneration) });
|
|
await route.acceptConfig(config);
|
|
send.mockClear();
|
|
route.accept({
|
|
revision: 8, phase: "prepare",
|
|
assignment: hostAssignment(config.publicationGeneration, ["child"]), candidate: candidate(8, "child"),
|
|
});
|
|
fail();
|
|
fail();
|
|
expect(send.mock.calls).toEqual([[{
|
|
type: "route-failed", revision: 7, phase: "active", connectionId: config.connectionId,
|
|
}]]);
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("reports the committed SFU subscription when a downstream Peer is preparing", async () => {
|
|
const send = vi.fn(() => true);
|
|
const stream = vi.fn();
|
|
let events!: SubscriberEvents;
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
onSfuStream: stream,
|
|
send,
|
|
createSubscriber: (callbacks) => {
|
|
events = callbacks;
|
|
return createFakeSubscriber(callbacks, [], "active");
|
|
},
|
|
});
|
|
const config = sfuConfig(7);
|
|
await prepareSfu(route, 7, viewerSfuAssignment(), config);
|
|
events.onStream({} as MediaStream);
|
|
events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(stream).toHaveBeenCalledOnce());
|
|
send.mockClear();
|
|
route.accept({
|
|
revision: 8, phase: "prepare",
|
|
assignment: viewerSfuAssignment(["child"]), candidate: candidate(8, "child"),
|
|
});
|
|
events.onDisconnected();
|
|
events.onDisconnected();
|
|
expect(send.mock.calls).toEqual([[{
|
|
type: "route-failed", revision: 7, phase: "active", connectionId: config.connectionId,
|
|
}]]);
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("publishes SFU media during prepare and promotes before retiring old media", async () => {
|
|
const log: string[] = [];
|
|
const messages: ClientMessage[] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "h264",
|
|
reconcileChildren: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
log,
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-a"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-a"));
|
|
expect(publishers[0]?.activate).toHaveBeenCalledOnce();
|
|
expect(publishers[0]?.activate).toHaveBeenCalledWith(
|
|
expect.anything(),
|
|
QUALITY_PROFILES["720p30"],
|
|
"h264",
|
|
);
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-b"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2, "publication-b"));
|
|
expect(publishers[1]?.activate).toHaveBeenCalledOnce();
|
|
expect(publishers[0]?.deactivate).not.toHaveBeenCalled();
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
|
|
await route.acceptAndWait({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-b"),
|
|
});
|
|
expect(publishers[0]?.deactivate).toHaveBeenCalledOnce();
|
|
expect(publishers[0]?.disconnect).toHaveBeenCalledOnce();
|
|
expect(publishers[1]?.disconnect).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("renders an SFU connection failure in each language without placeholders", async () => {
|
|
const publisher = createFakePublisher([], "publisher");
|
|
publisher.connect.mockResolvedValue(false);
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: () => true,
|
|
createPublisher: () => ({
|
|
...publisher,
|
|
getFailureStage: () => "connect",
|
|
}),
|
|
});
|
|
route.accept({
|
|
revision: 1,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication"),
|
|
candidate: candidate(1, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication"));
|
|
const warning = route.getQualityWarning();
|
|
|
|
for (const lang of ["en", "zh"] as const) {
|
|
const text = resolveMediaFailure(warning, {
|
|
lang,
|
|
t: (key, vars) => t(lang, key, vars),
|
|
});
|
|
expect(text).toBe(t(lang, "host.warn.sfuRecover", {
|
|
params: t(lang, "host.warn.sfuStage.connect"),
|
|
}));
|
|
expect(text).not.toMatch(/\{[^}]+\}/);
|
|
}
|
|
});
|
|
|
|
it("changes Host children only after active authority commits", async () => {
|
|
const reconciledChildren: string[][] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: (children) => reconciledChildren.push(children),
|
|
send: () => true,
|
|
createPublisher: () => createFakePublisher([], "publisher"),
|
|
});
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication", ["committed-child"]),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication"));
|
|
reconciledChildren.length = 0;
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment(null, ["candidate-child"]),
|
|
candidate: candidate(2, "candidate-child"),
|
|
});
|
|
expect(reconciledChildren).toEqual([]);
|
|
|
|
await route.acceptAndWait({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: hostAssignment(null, ["candidate-child"]),
|
|
});
|
|
expect(reconciledChildren).toEqual([["candidate-child"]]);
|
|
});
|
|
|
|
it("replaces a Host physical connection within one publication and fences stale signaling", async () => {
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: () => true,
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher([], "publisher");
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
await route.acceptAndWait({ revision: 1, phase: "active", assignment: hostAssignment("publication") });
|
|
const first = sfuConfig(1, "publication");
|
|
await route.acceptConfig(first);
|
|
const replacement = { ...first, connectionId: "replacement_connection" };
|
|
await route.acceptConfig(replacement);
|
|
expect(publishers).toHaveLength(2);
|
|
expect(publishers[0]!.disconnect).toHaveBeenCalledOnce();
|
|
expect(publishers[1]!.disconnect).not.toHaveBeenCalled();
|
|
await route.acceptSignal({ ...first, type: "sfu-signal", kind: "candidate", candidate: { candidate: "old" } });
|
|
expect(publishers[0]!.acceptSignal).not.toHaveBeenCalled();
|
|
await route.acceptSignal({ ...replacement, type: "sfu-signal", kind: "candidate", candidate: { candidate: "current" } });
|
|
expect(publishers[1]!.acceptSignal).toHaveBeenCalledOnce();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("reports and retires an exact active SFU source replacement failure", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher([], "publisher-active");
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-active"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-active"));
|
|
publishers[0]!.replaceStream.mockResolvedValue(false);
|
|
|
|
await expect(route.replaceStream({} as MediaStream)).resolves.toBe(false);
|
|
expect(publishers[0]!.disconnect).toHaveBeenCalledOnce();
|
|
expect(messages).toContainEqual({ type: "refresh-sfu", revision: 1 });
|
|
});
|
|
|
|
it("retires a failed pending source without invalidating healthy active SFU", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-active"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-active"));
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-pending"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2, "publication-pending"));
|
|
publishers[1]!.replaceStream.mockResolvedValue(false);
|
|
|
|
await expect(route.replaceStream({} as MediaStream)).resolves.toBe(true);
|
|
expect(publishers[0]!.disconnect).not.toHaveBeenCalled();
|
|
expect(publishers[1]!.disconnect).toHaveBeenCalledOnce();
|
|
expect(messages).toContainEqual({
|
|
type: "route-failed",
|
|
revision: 2,
|
|
phase: "prepare",
|
|
connectionId: null,
|
|
});
|
|
});
|
|
|
|
it("reports success when a pending SFU promotion supersedes an old failure", async () => {
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: () => true,
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-old"));
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-new"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2, "publication-new"));
|
|
|
|
let resolveOldReplacement!: (replaced: boolean) => void;
|
|
publishers[0]!.replaceStream.mockImplementation(
|
|
() =>
|
|
new Promise<boolean>((resolve) => {
|
|
resolveOldReplacement = resolve;
|
|
}),
|
|
);
|
|
const oldReplacementCalls = publishers[0]!.replaceStream.mock.calls.length;
|
|
const replacing = route.replaceStream({} as MediaStream);
|
|
await vi.waitFor(() =>
|
|
expect(publishers[0]!.replaceStream).toHaveBeenCalledTimes(
|
|
oldReplacementCalls + 1,
|
|
),
|
|
);
|
|
|
|
await route.acceptAndWait({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-new"),
|
|
});
|
|
resolveOldReplacement(false);
|
|
|
|
await expect(replacing).resolves.toBe(true);
|
|
expect(publishers[1]!.disconnect).not.toHaveBeenCalled();
|
|
});
|
|
|
|
|
|
|
|
it("reconciles paused Host active truth by exact SFU publication", async () => {
|
|
const reconciledChildren: string[][] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: (children) => reconciledChildren.push(children),
|
|
send: () => true,
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-old"));
|
|
route.setPaused(true);
|
|
expect(publishers[0]?.setPaused).toHaveBeenCalledWith(true);
|
|
|
|
await route.acceptAndWait({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old", ["retained-child"]),
|
|
});
|
|
expect(publishers[0]?.disconnect).not.toHaveBeenCalled();
|
|
expect(reconciledChildren.at(-1)).toEqual(["retained-child"]);
|
|
expect(
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old", ["retained-child"]),
|
|
}),
|
|
).toBe("duplicate");
|
|
|
|
await route.acceptAndWait({
|
|
revision: 3,
|
|
phase: "active",
|
|
assignment: hostAssignment(null, ["retained-child"]),
|
|
});
|
|
expect(publishers[0]?.deactivate).toHaveBeenCalledOnce();
|
|
expect(publishers[0]?.disconnect).toHaveBeenCalledOnce();
|
|
expect(reconciledChildren.at(-1)).toEqual(["retained-child"]);
|
|
});
|
|
|
|
it("rebuilds a failed Host SFU publisher while sharing is paused", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const disconnects: Array<() => void> = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createPublisher: (onDisconnected) => {
|
|
disconnects.push(onDisconnected);
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-paused"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-paused"));
|
|
route.setPaused(true);
|
|
disconnects[0]!();
|
|
expect(messages).toContainEqual({ type: "refresh-sfu", revision: 1 });
|
|
|
|
await route.acceptConfig({ ...sfuConfig(1, "publication-paused"), connectionId: "recovery_connection" });
|
|
expect(publishers).toHaveLength(2);
|
|
expect(publishers[1]!.setPaused).toHaveBeenCalledWith(true);
|
|
|
|
route.setPaused(false);
|
|
expect(publishers[1]!.setPaused).toHaveBeenLastCalledWith(false);
|
|
});
|
|
|
|
it("does not activate a fresh SFU publisher after its exact route becomes stale", async () => {
|
|
let releaseFreshPublisher!: () => void;
|
|
const freshPublisherGate = new Promise<void>((resolve) => {
|
|
releaseFreshPublisher = resolve;
|
|
});
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: () => true,
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
if (publishers.length === 1) {
|
|
publisher.activate.mockImplementation(async () => {
|
|
await freshPublisherGate;
|
|
return true;
|
|
});
|
|
}
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-old"));
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-fresh"),
|
|
candidate: candidate(
|
|
2,
|
|
"viewer_12345678",
|
|
"sfu",
|
|
"candidate_2_12345678",
|
|
),
|
|
});
|
|
const freshPublication = route.acceptConfig(sfuConfig(2, "publication-fresh"));
|
|
await vi.waitFor(() =>
|
|
expect(publishers[1]?.activate).toHaveBeenCalledOnce(),
|
|
);
|
|
|
|
const staleRoute = route.acceptAndWait({
|
|
revision: 3,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
releaseFreshPublisher();
|
|
await Promise.all([freshPublication, staleRoute]);
|
|
|
|
expect(publishers[0]?.disconnect).not.toHaveBeenCalled();
|
|
expect(publishers[1]?.disconnect).toHaveBeenCalled();
|
|
});
|
|
|
|
it.each(["disconnect", "resync"])("cancels pending Host preparation before joining its queue during %s", async (reason) => {
|
|
let cancel!: (result: boolean) => void;
|
|
const activating = new Promise<boolean>(resolve => { cancel = resolve; });
|
|
const publisher = createFakePublisher([], "pending");
|
|
publisher.activate.mockImplementation(() => activating);
|
|
publisher.disconnect.mockImplementation(async () => { cancel(false); });
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: () => true,
|
|
createPublisher: () => publisher,
|
|
});
|
|
await route.acceptAndWait({
|
|
revision: 2, phase: "active", assignment: hostAssignment("pending"),
|
|
});
|
|
const preparing = route.acceptConfig(sfuConfig(2, "pending"));
|
|
await vi.waitFor(() => expect(publisher.activate).toHaveBeenCalledOnce());
|
|
const retiring = reason === "disconnect" ? route.disconnect() : route.resyncAuthoritative({
|
|
revision: 1, phase: "active", assignment: hostAssignment(null),
|
|
});
|
|
expect(publisher.disconnect).toHaveBeenCalled();
|
|
await Promise.all([preparing, retiring]);
|
|
expect(publisher.setPaused).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("ignores a retired publisher failure after same-generation resync", async () => {
|
|
let releaseOldConnect!: () => void;
|
|
const oldConnectGate = new Promise<void>((resolve) => {
|
|
releaseOldConnect = resolve;
|
|
});
|
|
const messages: ClientMessage[] = [];
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"],
|
|
getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher(
|
|
[],
|
|
`publisher-${publishers.length + 1}`,
|
|
);
|
|
if (publishers.length === 1) {
|
|
publisher.connect.mockImplementation(async () => {
|
|
await oldConnectGate;
|
|
throw new Error("retired publisher");
|
|
});
|
|
}
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
|
|
await route.acceptAndWait({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(1, "publication-old"));
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-new"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
const stale = route.acceptConfig(sfuConfig(2, "publication-new"));
|
|
await vi.waitFor(() => expect(publishers[1]?.connect).toHaveBeenCalledOnce());
|
|
|
|
await route.resyncAuthoritative({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: hostAssignment("publication-old"),
|
|
});
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: hostAssignment("publication-new"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
const current = route.acceptConfig(sfuConfig(2, "publication-new"));
|
|
await vi.waitFor(() => expect(publishers).toHaveLength(3));
|
|
releaseOldConnect();
|
|
await Promise.all([stale, current]);
|
|
|
|
expect(messages.filter((message) => message.type === "route-failed")).toEqual(
|
|
[],
|
|
);
|
|
expect(publishers[2]?.activate).toHaveBeenCalledOnce();
|
|
});
|
|
|
|
it("waits for pending retirement before activating a same-key replacement", async () => {
|
|
let finishClose!: () => void;
|
|
const closing = new Promise<void>(resolve => { finishClose = resolve; });
|
|
const publishers: ReturnType<typeof createFakePublisher>[] = [];
|
|
const route = new HostSfuRoute({
|
|
getStream: () => ({}) as MediaStream,
|
|
getProfile: () => QUALITY_PROFILES["720p30"], getVideoCodec: () => "vp8",
|
|
reconcileChildren: () => {}, send: () => true,
|
|
createPublisher: () => {
|
|
const publisher = createFakePublisher([], `publisher-${publishers.length}`);
|
|
if (!publishers.length) publisher.disconnect.mockImplementation(() => closing);
|
|
publishers.push(publisher);
|
|
return publisher;
|
|
},
|
|
});
|
|
route.accept({ revision: 1, phase: "prepare", assignment: hostAssignment("publication"),
|
|
candidate: candidate(1, "viewer_12345678", "sfu") });
|
|
await route.acceptConfig(sfuConfig(1, "publication"));
|
|
route.setPaused(true);
|
|
route.setPaused(false);
|
|
const replacement = route.acceptConfig(sfuConfig(1, "publication"));
|
|
await vi.waitFor(() => expect(publishers[1]?.connect).toHaveBeenCalledOnce());
|
|
expect(publishers[1]?.activate).not.toHaveBeenCalled();
|
|
finishClose();
|
|
await replacement;
|
|
expect(publishers[1]?.activate).toHaveBeenCalledOnce();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("ignores a retired subscriber failure after same-generation resync", async () => {
|
|
let releaseOldConnect!: () => void;
|
|
const oldConnectGate = new Promise<void>((resolve) => {
|
|
releaseOldConnect = resolve;
|
|
});
|
|
const messages: ClientMessage[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
[],
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
if (subscribers.length === 1) {
|
|
subscriber.connect.mockImplementation(async () => {
|
|
await oldConnectGate;
|
|
throw new Error("retired subscriber");
|
|
});
|
|
}
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
await prepareSfu(route, 1, viewerSfuAssignment([], "publication-old"), sfuConfig(1, "publication-old"));
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment([], "publication-new"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
const stale = route.acceptConfig(sfuConfig(2, "publication-new"));
|
|
await vi.waitFor(() => expect(subscribers[1]?.connect).toHaveBeenCalledOnce());
|
|
|
|
await route.resyncAuthoritative(
|
|
{
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: viewerSfuAssignment([], "publication-old"),
|
|
},
|
|
"viewer_12345678",
|
|
);
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment([], "publication-new"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
const current = route.acceptConfig(sfuConfig(2, "publication-new"));
|
|
await vi.waitFor(() => expect(subscribers).toHaveLength(3));
|
|
releaseOldConnect();
|
|
await Promise.all([stale, current]);
|
|
|
|
expect(messages.filter((message) => message.type === "route-failed")).toEqual(
|
|
[],
|
|
);
|
|
expect(subscribers[2]?.activate).toHaveBeenCalledOnce();
|
|
});
|
|
|
|
|
|
it("uses the subscriber first decoded frame as the only SFU prepare ready", async () => {
|
|
const log: string[] = [];
|
|
const messages: ClientMessage[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
log,
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
await prepareSfu(route, 1, viewerSfuAssignment(), sfuConfig(1));
|
|
const firstStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(firstStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([firstStream]));
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment([], "publication-generation-2"),
|
|
candidate: candidate(2, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2, "publication-generation-2"));
|
|
const pendingStream = {} as MediaStream;
|
|
subscribers[1]?.events.onStream(pendingStream);
|
|
subscribers[1]?.events.onStats?.({
|
|
intervalFramesDecoded: 2,
|
|
} as ConnectionMetrics);
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
|
|
subscribers[1]?.events.onFirstDecodedFrame();
|
|
subscribers[1]?.events.onFirstDecodedFrame();
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([
|
|
{ type: "route-ready", revision: 2, phase: "prepare" },
|
|
]);
|
|
expect(streams).toEqual([firstStream]);
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: viewerSfuAssignment([], "publication-generation-2"),
|
|
});
|
|
await vi.waitFor(() => expect(streams).toEqual([firstStream, pendingStream]));
|
|
expect(subscribers[0]?.deactivate).toHaveBeenCalledOnce();
|
|
expect(subscribers[0]?.disconnect).toHaveBeenCalledOnce();
|
|
});
|
|
|
|
it("keeps a quality SFU candidate pending until three non-regressing windows", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
let currentMetrics = receiveMetrics(1_000);
|
|
let qualityProbeEligible = true;
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
currentPeerMetrics: () => currentMetrics,
|
|
qualityProbeEligible: () => qualityProbeEligible,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
[],
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
route.accept({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: peerAssignment("parent_12345678"),
|
|
});
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment(),
|
|
candidate: candidate(
|
|
2,
|
|
"viewer_12345678",
|
|
"sfu",
|
|
"candidate_2_12345678",
|
|
true,
|
|
),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2));
|
|
const subscriber = subscribers[0]!;
|
|
subscriber.events.onStream({} as MediaStream);
|
|
subscriber.events.onFirstDecodedFrame();
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
|
|
for (const timestampMs of [1_100, 3_100]) {
|
|
currentMetrics = receiveMetrics(timestampMs - 100);
|
|
subscriber.events.onStats(receiveMetrics(timestampMs));
|
|
}
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
|
|
qualityProbeEligible = false;
|
|
for (const timestampMs of [5_100, 7_100, 9_100]) {
|
|
currentMetrics = receiveMetrics(timestampMs - 100);
|
|
subscriber.events.onStats(receiveMetrics(timestampMs));
|
|
}
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
|
|
qualityProbeEligible = true;
|
|
for (const timestampMs of [11_100, 13_100, 15_100]) {
|
|
currentMetrics = receiveMetrics(timestampMs - 100);
|
|
subscriber.events.onStats(receiveMetrics(timestampMs));
|
|
}
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([
|
|
{ type: "route-ready", revision: 2, phase: "prepare" },
|
|
]);
|
|
});
|
|
|
|
it("rejects a quality SFU candidate after three comparable regressions", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
let currentMetrics = receiveMetrics(1_000);
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
currentPeerMetrics: () => currentMetrics,
|
|
qualityProbeEligible: () => true,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "candidate");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
route.accept({
|
|
revision: 1,
|
|
phase: "active",
|
|
assignment: peerAssignment("parent_12345678"),
|
|
});
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment(),
|
|
candidate: candidate(
|
|
2,
|
|
"viewer_12345678",
|
|
"sfu",
|
|
"candidate_2_12345678",
|
|
true,
|
|
),
|
|
});
|
|
await route.acceptConfig(sfuConfig(2));
|
|
const subscriber = subscribers[0]!;
|
|
subscriber.events.onStream({} as MediaStream);
|
|
subscriber.events.onFirstDecodedFrame();
|
|
|
|
for (const timestampMs of [1_100, 3_100, 5_100]) {
|
|
currentMetrics = receiveMetrics(timestampMs - 100);
|
|
subscriber.events.onStats(
|
|
receiveMetrics(timestampMs, { framesPerSecond: 30 }),
|
|
);
|
|
}
|
|
|
|
expect(messages.at(-1)).toEqual({
|
|
type: "route-failed",
|
|
revision: 2,
|
|
phase: "prepare",
|
|
connectionId: null,
|
|
});
|
|
expect(subscriber.disconnect).toHaveBeenCalledOnce();
|
|
expect(
|
|
messages.filter((message) => message.type === "route-ready"),
|
|
).toEqual([]);
|
|
});
|
|
|
|
it("reconnects only the active SFU subscriber on the current route", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const resetMedia = vi.fn();
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
resetMedia,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
[],
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
await prepareSfu(route, 7, viewerSfuAssignment(), sfuConfig(7));
|
|
const firstStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(firstStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([firstStream]));
|
|
|
|
expect(route.reconnectActive()).toBe(true);
|
|
expect(subscribers[0]?.reconnect).toHaveBeenCalledOnce();
|
|
expect(resetMedia).not.toHaveBeenCalled();
|
|
expect(subscribers[0]?.deactivate).not.toHaveBeenCalled();
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
|
|
await route.acceptConfig(sfuConfig(7));
|
|
expect(subscribers).toHaveLength(1);
|
|
expect(streams).toEqual([firstStream]);
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("keeps an active SFU subscriber across same-publication resync", async () => {
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const resetMedia = vi.fn();
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
resetMedia,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: () => true,
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "active");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
const assignment = viewerSfuAssignment();
|
|
await prepareSfu(route, 7, assignment, sfuConfig(7));
|
|
const initialStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(initialStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([initialStream]));
|
|
|
|
await expect(
|
|
route.resyncAuthoritative(
|
|
{ revision: 8, phase: "active", assignment },
|
|
"viewer_12345678",
|
|
),
|
|
).resolves.toBe("accepted");
|
|
expect(subscribers).toHaveLength(1);
|
|
expect(subscribers[0]?.deactivate).not.toHaveBeenCalled();
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
expect(resetMedia).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("retains the physical SFU subscriber when a room revision advances during reconnect", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "active");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
const assignment = viewerSfuAssignment();
|
|
await prepareSfu(route, 7, assignment, sfuConfig(7));
|
|
const initialStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(initialStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([initialStream]));
|
|
messages.length = 0;
|
|
expect(route.reconnectActive()).toBe(true);
|
|
|
|
route.accept({ revision: 8, phase: "active", assignment });
|
|
await vi.waitFor(() => expect(subscribers[0]?.updateConfig).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({ revision: 8 }),
|
|
));
|
|
expect(messages.filter((message) => message.type === "refresh-sfu")).toEqual([]);
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("keeps SFU reconnect on the physical subscriber through an unrelated prepare", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "active");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
const assignment = viewerSfuAssignment(["relay-child"]);
|
|
await prepareSfu(route, 7, assignment, sfuConfig(7));
|
|
subscribers[0]!.events.onStream({} as MediaStream);
|
|
subscribers[0]!.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toHaveLength(1));
|
|
expect(route.reconnectActive()).toBe(true);
|
|
|
|
route.accept({
|
|
revision: 8,
|
|
phase: "prepare",
|
|
assignment,
|
|
candidate: candidate(8, "relay-child"),
|
|
});
|
|
route.accept({ revision: 8, phase: "active", assignment });
|
|
|
|
await vi.waitFor(() => expect(subscribers[0]?.updateConfig).toHaveBeenLastCalledWith(
|
|
expect.objectContaining({ revision: 8 }),
|
|
));
|
|
expect(subscribers[0]?.reconnect).toHaveBeenCalledOnce();
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
expect(messages.filter((message) => message.type === "refresh-sfu").every((message) => message.revision === 7)).toBe(true);
|
|
});
|
|
|
|
it("reports a failed subscription again after offline reauthentication without creating an empty peer", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const streams: MediaStream[] = [];
|
|
let acceptsFailure = false;
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return message.type !== "route-failed" || acceptsFailure;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "active");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
await prepareSfu(route, 7, viewerSfuAssignment(), sfuConfig(7));
|
|
subscribers[0]!.events.onStream({} as MediaStream);
|
|
subscribers[0]!.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toHaveLength(1));
|
|
subscribers[0]!.events.onDisconnected();
|
|
acceptsFailure = true;
|
|
|
|
await route.resyncAuthoritative({ revision: 7, phase: "active", assignment: viewerSfuAssignment() }, "viewer_12345678");
|
|
await route.acceptConfig(sfuConfig(7));
|
|
expect(subscribers).toHaveLength(1);
|
|
expect(messages.filter(message => message.type === "route-failed")).toContainEqual({
|
|
type: "route-failed", revision: 7, phase: "active", connectionId: sfuConfig(7).connectionId,
|
|
});
|
|
messages.length = 0;
|
|
|
|
route.accept({
|
|
revision: 8,
|
|
phase: "active",
|
|
assignment: viewerSfuAssignment(),
|
|
});
|
|
await vi.waitFor(() =>
|
|
expect(
|
|
messages.filter((message) => message.type === "route-failed"),
|
|
).toEqual([
|
|
{ type: "route-failed", revision: 8, phase: "active", connectionId: sfuConfig(7).connectionId },
|
|
]),
|
|
);
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("does not resync a retained subscriber after a newer preparation supersedes authentication", async () => {
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true, send: () => true,
|
|
onSfuStream: stream => streams.push(stream),
|
|
createSubscriber: events => {
|
|
const subscriber = createFakeSubscriber(events, [], "retained");
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
await prepareSfu(route, 7, viewerSfuAssignment(), sfuConfig(7));
|
|
subscribers[0]!.events.onStream({} as MediaStream);
|
|
subscribers[0]!.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toHaveLength(1));
|
|
const resync = route.resyncAuthoritative({ revision: 7, phase: "active",
|
|
assignment: viewerSfuAssignment() }, "viewer_12345678");
|
|
route.accept({ revision: 8, phase: "prepare", assignment: peerAssignment("parent_12345678"),
|
|
candidate: candidate(8) });
|
|
await resync;
|
|
expect(subscribers[0]!.resync).not.toHaveBeenCalled();
|
|
expect(subscribers[0]!.disconnect).not.toHaveBeenCalled();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("recovers a failed subscription through a new server-owned preparation", async () => {
|
|
let releaseRecovery!: (connected: boolean) => void;
|
|
const recoveryConnected = new Promise<boolean>((resolve) => {
|
|
releaseRecovery = resolve;
|
|
});
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: () => true,
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(events, [], "subscriber");
|
|
if (subscribers.length === 1) {
|
|
subscriber.connect.mockImplementation(() => recoveryConnected);
|
|
}
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
const assignment = viewerSfuAssignment();
|
|
await prepareSfu(route, 7, assignment, sfuConfig(7));
|
|
const initialStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(initialStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([initialStream]));
|
|
|
|
subscribers[0]!.events.onDisconnected();
|
|
route.accept({ revision: 8, phase: "prepare", assignment,
|
|
candidate: candidate(8, "viewer_12345678", "sfu", "recovery_connection") });
|
|
const recovery = route.acceptConfig({
|
|
...sfuConfig(8),
|
|
connectionId: "recovery_connection",
|
|
});
|
|
await vi.waitFor(() => expect(subscribers).toHaveLength(2));
|
|
expect(subscribers).toHaveLength(2);
|
|
expect(subscribers[1]?.disconnect).not.toHaveBeenCalled();
|
|
|
|
releaseRecovery(true);
|
|
await recovery;
|
|
const recoveredStream = {} as MediaStream;
|
|
subscribers[1]?.events.onStream(recoveredStream);
|
|
subscribers[1]?.events.onFirstDecodedFrame();
|
|
route.accept({ revision: 8, phase: "active", assignment });
|
|
await vi.waitFor(() =>
|
|
expect(streams).toEqual([initialStream, recoveredStream]),
|
|
);
|
|
expect(subscribers[0]?.disconnect).toHaveBeenCalledOnce();
|
|
await route.disconnect();
|
|
});
|
|
|
|
it("discards a paused pending subscriber and ignores its later evidence", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
let subscriber!: ReturnType<typeof createFakeSubscriber>;
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: () => undefined,
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
subscriber = createFakeSubscriber(events, [], "pending");
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
route.accept({
|
|
revision: 3,
|
|
phase: "prepare",
|
|
assignment: viewerSfuAssignment(),
|
|
candidate: candidate(3, "viewer_12345678", "sfu"),
|
|
});
|
|
await route.acceptConfig(sfuConfig(3));
|
|
route.setPaused(true);
|
|
expect(subscriber.disconnect).toHaveBeenCalledOnce();
|
|
|
|
subscriber.events.onStream({} as MediaStream);
|
|
subscriber.events.onFirstDecodedFrame();
|
|
expect(messages.filter((message) => message.type === "route-ready")).toEqual([]);
|
|
});
|
|
|
|
it("reports paused subscription failure and waits for the server to prepare its replacement", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const streams: MediaStream[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: () => true,
|
|
reconcileSfuChildren: () => undefined,
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
[],
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
await prepareSfu(route, 7, viewerSfuAssignment(), sfuConfig(7));
|
|
subscribers[0]!.events.onStream({} as MediaStream);
|
|
subscribers[0]!.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toHaveLength(1));
|
|
|
|
route.setPaused(true);
|
|
subscribers[0]!.events.onDisconnected();
|
|
expect(messages).toContainEqual({ type: "route-failed", revision: 7, phase: "active", connectionId: sfuConfig(7).connectionId });
|
|
|
|
await route.acceptConfig({ ...sfuConfig(7), connectionId: "recovery_connection" });
|
|
expect(subscribers).toHaveLength(1);
|
|
|
|
route.setPaused(false);
|
|
await prepareSfu(route, 8);
|
|
expect(subscribers[1]!.armDecodedFrameProof).toHaveBeenCalledOnce();
|
|
await route.disconnect();
|
|
});
|
|
|
|
|
|
it("retires only an unreferenced paused Viewer SFU slot", async () => {
|
|
const messages: ClientMessage[] = [];
|
|
const activatedPeers: ParticipantRouteAssignment[] = [];
|
|
const activatedChildren: string[][] = [];
|
|
const streams: MediaStream[] = [];
|
|
const subscribers: ReturnType<typeof createFakeSubscriber>[] = [];
|
|
const route = new ViewerSfuRoute("viewer_12345678", {
|
|
activatePeer: (assignment) => {
|
|
activatedPeers.push(assignment);
|
|
return true;
|
|
},
|
|
activateChildren: (children) => activatedChildren.push([...children]),
|
|
onSfuStream: (stream) => streams.push(stream),
|
|
send: (message) => {
|
|
messages.push(message);
|
|
return true;
|
|
},
|
|
createSubscriber: (events) => {
|
|
const subscriber = createFakeSubscriber(
|
|
events,
|
|
[],
|
|
`subscriber-${subscribers.length + 1}`,
|
|
);
|
|
subscribers.push(subscriber);
|
|
return subscriber;
|
|
},
|
|
});
|
|
|
|
await prepareSfu(route, 1, viewerSfuAssignment([], "publication-old"), sfuConfig(1, "publication-old"));
|
|
const oldStream = {} as MediaStream;
|
|
subscribers[0]?.events.onStream(oldStream);
|
|
subscribers[0]?.events.onFirstDecodedFrame();
|
|
await vi.waitFor(() => expect(streams).toEqual([oldStream]));
|
|
route.setPaused(true);
|
|
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: viewerSfuAssignment(
|
|
["retained-child"],
|
|
"publication-old",
|
|
),
|
|
});
|
|
await vi.waitFor(() =>
|
|
expect(activatedChildren.at(-1)).toEqual(["retained-child"]),
|
|
);
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
expect(
|
|
route.accept({
|
|
revision: 2,
|
|
phase: "active",
|
|
assignment: viewerSfuAssignment(
|
|
["retained-child"],
|
|
"publication-old",
|
|
),
|
|
}),
|
|
).toBe("duplicate");
|
|
expect(subscribers[0]?.disconnect).not.toHaveBeenCalled();
|
|
|
|
route.accept({
|
|
revision: 3,
|
|
phase: "active",
|
|
assignment: peerAssignment("retained-parent", ["retained-child"]),
|
|
});
|
|
await vi.waitFor(() =>
|
|
expect(subscribers[0]?.disconnect).toHaveBeenCalledOnce(),
|
|
);
|
|
expect(activatedPeers.at(-1)).toEqual(
|
|
peerAssignment("retained-parent", ["retained-child"]),
|
|
);
|
|
});
|
|
});
|