okooooooo/Piik
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: finish audited state ownership cleanup (#357)

Browse Source
This commit is contained in:
AAA authored and GitHub committed 2026-08-31 12:27:13 +08:00
1 parent bd2e9b67de
commit ae2ab6f78e
7 files changed
+322 -25
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/client/lib/signaling.ts
+1 -1
View File
@@ -288,8 +288,8 @@ export class SignalingClient {
this.clearAuthenticationTimer();
if (this.terminalMessage) {
const terminalMessage = this.terminalMessage;
this.terminalMessage = null;
if (this.send(terminalMessage)) {
this.terminalMessage = null;
this.stop();
return;
}
src/client/media/viewer-quality-evidence.ts
+13
View File
@@ -460,6 +460,19 @@ export function freshViewerQualityEvidence(
return presentation?.fresh ? presentation.evidence : null;
}
export function retainPresentViewerQualityEvidence(
presentations: Map<string, ViewerQualityEvidencePresentation>,
presentPeerIds: ReadonlySet<string>,
): Map<string, ViewerQualityEvidencePresentation> {
let retained: Map<string, ViewerQualityEvidencePresentation> | null = null;
for (const peerId of presentations.keys()) {
if (presentPeerIds.has(peerId)) continue;
retained ??= new Map(presentations);
retained.delete(peerId);
}
return retained ?? presentations;
}
export function classifyHostViewerQualityEvidence(
evidence: ViewerQualityEvidence,
hostPeerId: string | null,
src/client/pages/HostPage.tsx
+64 -21
View File
@@ -134,6 +134,7 @@ import {
qualityEvidenceUpstreamMatches,
reconcileViewerQualityEvidencePresentation,
refreshViewerQualityEvidencePresentation,
retainPresentViewerQualityEvidence,
type ViewerQualityEvidencePresentation,
} from "../media/viewer-quality-evidence";
import type {
@@ -436,6 +437,9 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
>(() => new Map());
const [noticeValue, setNoticeValue] = useState<NoticeValue | null>(null);
const [noticeComic, setNoticeComic] = useState<ComicKind | null>(null);
const [hostSfuQualityWarning, setHostSfuQualityWarning] = useState<
string | null
>(null);
function setNotice(value: string | null, comic: ComicKind | null = null): void {
setNoticeValue(value ? { kind: "text", text: value } : null);
setNoticeComic(comic);
@@ -729,6 +733,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
function clearHostSfuRoute(): void {
const route = hostSfuRouteRef.current;
hostSfuRouteRef.current = null;
setHostSfuQualityWarning(null);
sfuStandbyPrewarmerRef.current?.setUrl(null);
void route?.disconnect();
}
@@ -742,19 +747,19 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
sfuStandbyPrewarmerRef.current.setUrl(url);
}
function showHostSfuQualityWarning(
function syncHostSfuQualityWarning(
route: HostSfuRoute,
generation: number,
): void {
): string | null {
if (
isCurrentGeneration(generation) &&
hostSfuRouteRef.current === route
!isCurrentGeneration(generation) ||
hostSfuRouteRef.current !== route
) {
return null;
}
const warning = route.getQualityWarning();
if (warning) {
setNotice(warning, "route-failed");
}
}
setHostSfuQualityWarning(warning);
return warning;
}
function disposeResources(notifyServer: boolean): void {
@@ -765,6 +770,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
codecProbeAbortRef.current = null;
videoCodecRef.current = VP8_ONLY_VIDEO_CODEC;
setResolvedVideoCodec(null);
setHostSfuQualityWarning(null);
const signal = signalRef.current;
if (signal) {
if (notifyServer) {
@@ -1032,6 +1038,30 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
commitViewerQualityEvidence(peerId, null);
}
function retainViewerQualityEvidenceForPresence(
entries: readonly ParticipantPresenceEntry[],
): void {
const presentPeerIds = new Set(
entries
.filter((entry) => entry.role === "viewer")
.map((entry) => entry.peerId),
);
const current = viewerQualityEvidenceRef.current;
const retained = retainPresentViewerQualityEvidence(
current,
presentPeerIds,
);
if (retained === current) return;
for (const peerId of current.keys()) {
if (retained.has(peerId)) continue;
const timer = viewerQualityEvidenceTimersRef.current.get(peerId);
if (timer !== undefined) window.clearTimeout(timer);
viewerQualityEvidenceTimersRef.current.delete(peerId);
}
viewerQualityEvidenceRef.current = retained;
setViewerQualityEvidence(retained);
}
function acceptViewerQualityEvidence(evidence: ViewerQualityEvidence): void {
const peer = peersRef.current.get(evidence.viewerPeerId);
const directSnapshot = peer?.getSnapshot() ?? null;
@@ -1211,14 +1241,13 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
) {
const failed = results.filter((updated) => !updated).length;
const sfuWarning =
hostSfuRouteRef.current === activeSfuRoute
? (activeSfuRoute?.getQualityWarning() ?? null)
activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute
? syncHostSfuQualityWarning(activeSfuRoute, generation)
: null;
const connectionWarning =
failed > 0 || !sfuUpdated
? say("host.notice.partialApply")
: null;
const warning = sfuWarning ?? connectionWarning;
const successNotice =
audioChanged && !videoChanged
? say("host.notice.audioSet", { label: say(AUDIO_QUALITY_CAPTIONS[resolveScreenAudioQuality(nextProfile.screenAudioQuality)]) })
@@ -1227,7 +1256,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
: say("host.notice.qualitySet", {
label: qualitySettingsLabel(nextProfile),
});
setNotice(warning || successNotice);
setNotice(connectionWarning ?? (sfuWarning ? null : successNotice));
}
} catch (error) {
if (
@@ -1676,7 +1705,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
if (reauthenticated && hostSfuRouteRef.current === route) {
await route.updateProfile(currentQualitySettings);
}
showHostSfuQualityWarning(route, generation);
syncHostSfuQualityWarning(route, generation);
});
return;
}
@@ -1687,6 +1716,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
return;
}
if (message.type === "viewer-presence") {
retainViewerQualityEvidenceForPresence(message.viewers);
setParticipantPresence(message.viewers);
return;
}
@@ -1732,7 +1762,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
) {
activeRouteRevisionRef.current = message.revision;
}
showHostSfuQualityWarning(route, generation);
syncHostSfuQualityWarning(route, generation);
}
return;
}
@@ -1745,7 +1775,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
const route = ensureHostSfuRoute(generation);
void route
.acceptConfig(message)
.then(() => showHostSfuQualityWarning(route, generation));
.then(() => syncHostSfuQualityWarning(route, generation));
}
return;
}
@@ -2183,7 +2213,7 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
const sfuWarning =
activeSfuRoute && hostSfuRouteRef.current === activeSfuRoute
? activeSfuRoute.getQualityWarning()
? syncHostSfuQualityWarning(activeSfuRoute, generation)
: null;
previousStream.getTracks().forEach((track) => track.stop());
if (retiringStreamRef.current === previousStream) {
@@ -2216,13 +2246,18 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
isCurrentGeneration(generation) &&
sourceSwitchRef.current === token
) {
setNotice(
sourceSwitchNotice({
const sourceNotice =
sfuReplaced && sfuWarning && failedPeerIds.length === 0
? null
: sourceSwitchNotice({
failedPeerCount: failedPeerIds.length,
sfuReplaced,
sfuWarning,
}),
sfuWarning || !sfuReplaced || failedPeerIds.length > 0
sfuWarning: null,
});
setNotice(
sourceNotice,
sourceNotice &&
(sfuWarning || !sfuReplaced || failedPeerIds.length > 0)
? "warning"
: null,
);
@@ -2897,6 +2932,14 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
}
/>
) : null}
{hostSfuQualityWarning &&
hostSfuQualityWarning !== noticeText ? (
<Pill
icon="alert"
label={hostSfuQualityWarning}
comic="route-failed"
/>
) : null}
{noticeText && (vis || noticeText !== phaseLine) ? (
<Pill
icon={noticeComic ? "alert" : "check"}
src/server/signaling.ts
+33 -3
View File
@@ -112,6 +112,10 @@ export class SignalingServer {
private readonly socketsBySessionId = new Map<string, WebSocket>();
private readonly viewerGraceTimers = new Map<string, NodeJS.Timeout>();
private readonly connectionIdsByViewer = new Map<string, string>();
private readonly ordinaryReadyConnectionIdsByViewer = new Map<
string,
string
>();
private readonly viewerQualityEvidenceGates = new Map<
string,
ViewerQualityEvidenceGate
@@ -271,6 +275,7 @@ export class SignalingServer {
this.shareGenerationsByRoom.clear();
this.pausedShareGenerationsByRoom.clear();
this.ordinaryActiveHostChildrenByRoom.clear();
this.ordinaryReadyConnectionIdsByViewer.clear();
await closed;
clearTimeout(forceCloseTimer);
if (routeCloseError) {
@@ -1462,6 +1467,18 @@ export class SignalingServer {
fromPeerId: source.peerId,
payload: message.payload,
});
const connectionKey = viewerConnectionKey(source.roomId, source.peerId);
if (
description?.type === "answer" &&
this.connectionIdsByViewer.get(connectionKey) ===
message.payload.connectionId
) {
this.ordinaryReadyConnectionIdsByViewer.set(
connectionKey,
message.payload.connectionId,
);
this.sendViewerPresence(source.roomId);
}
}
private routePeerAssistedSignal(
@@ -1934,13 +1951,19 @@ export class SignalingServer {
viewer.peerId,
)
: undefined;
const mediaReady =
activeEdge !== undefined &&
const connectionKey = viewerConnectionKey(roomId, viewer.peerId);
const mediaReady = this.isHybridMediaEnabled()
? activeEdge !== undefined &&
upstream.kind !== "none" &&
(activeEdge.upstream.kind === "sfu"
? upstream.kind === "sfu"
: upstream.kind === "peer" &&
activeEdge.upstream.peerId === upstream.peerId);
activeEdge.upstream.peerId === upstream.peerId)
: upstream.kind === "peer" &&
this.connectionIdsByViewer.has(connectionKey) &&
this.ordinaryReadyConnectionIdsByViewer.get(
connectionKey,
) === this.connectionIdsByViewer.get(connectionKey);
viewers.push({
role: "viewer",
peerId: viewer.peerId,
@@ -2081,6 +2104,11 @@ export class SignalingServer {
this.connectionIdsByViewer.delete(key);
}
}
for (const key of this.ordinaryReadyConnectionIdsByViewer.keys()) {
if (key.startsWith(prefix)) {
this.ordinaryReadyConnectionIdsByViewer.delete(key);
}
}
for (const key of this.viewerQualityEvidenceGates.keys()) {
if (key.startsWith(prefix)) {
this.viewerQualityEvidenceGates.delete(key);
@@ -2100,6 +2128,7 @@ export class SignalingServer {
): void {
const key = viewerConnectionKey(roomId, viewerPeerId);
if (this.connectionIdsByViewer.get(key) !== connectionId) {
this.ordinaryReadyConnectionIdsByViewer.delete(key);
this.viewerQualityEvidenceGates.delete(key);
this.viewerQualityEvidenceAttemptAtMs.delete(key);
}
@@ -2112,6 +2141,7 @@ export class SignalingServer {
): void {
const key = viewerConnectionKey(roomId, viewerPeerId);
this.connectionIdsByViewer.delete(key);
this.ordinaryReadyConnectionIdsByViewer.delete(key);
this.viewerQualityEvidenceGates.delete(key);
this.viewerQualityEvidenceAttemptAtMs.delete(key);
}
tests/client-state.test.ts
+86
View File
@@ -814,6 +814,92 @@ describe("client signaling recovery policy", () => {
signal.stop();
});
it("retains a terminal command until an authenticated socket accepts it", () => {
vi.useFakeTimers();
const sockets: FakeWebSocket[] = [];
class FakeWebSocket extends EventTarget {
static readonly OPEN = 1;
static readonly CLOSING = 2;
readyState = FakeWebSocket.OPEN;
readonly send = vi.fn();
readonly close = vi.fn(() => {
this.readyState = FakeWebSocket.CLOSING;
});
constructor(readonly url: string) {
super();
sockets.push(this);
}
}
vi.stubGlobal("WebSocket", FakeWebSocket);
vi.stubGlobal("window", {
location: new URL("https://share.test/r/1234"),
setTimeout: globalThis.setTimeout,
clearTimeout: globalThis.clearTimeout,
});
const signal = new SignalingClient(
{
roomId: "1234",
role: "viewer",
clientId: "viewer-client",
},
{
onMessage: () => undefined,
onStatus: () => undefined,
onTerminated: () => undefined,
onAccessRequired: () => undefined,
},
);
const receiveAuthenticated = (socket: FakeWebSocket) => {
const event = new Event("message");
Object.defineProperty(event, "data", {
value: JSON.stringify({
type: "authenticated",
protocol: SIGNALING_PROTOCOL,
role: "viewer",
peerId: "viewer_12345678",
roomExpiresAt: null,
maxViewers: 8,
endpointMediaCopyCapacity: 2,
hostOnline: true,
connectionId: null,
viewerPeerIds: [],
iceConfig: { iceServers: [] },
codeEntryPolicy: "open",
viewerAuthorizationGeneration: "viewer_generation_12345678",
}),
});
socket.dispatchEvent(event);
};
signal.start();
signal.sendThenStop({ type: "abandon-room" });
sockets[0]!.dispatchEvent(new Event("open"));
sockets[0]!.readyState = FakeWebSocket.CLOSING;
receiveAuthenticated(sockets[0]!);
expect(sockets[0]!.send).toHaveBeenCalledOnce();
const closed = new Event("close");
Object.defineProperties(closed, {
code: { value: 1006 },
reason: { value: "network interrupted" },
});
sockets[0]!.dispatchEvent(closed);
vi.advanceTimersByTime(750);
sockets[1]!.dispatchEvent(new Event("open"));
receiveAuthenticated(sockets[1]!);
expect(
sockets[1]!.send.mock.calls.map(([value]) =>
JSON.parse(String(value)),
),
).toEqual([
expect.objectContaining({ type: "authenticate" }),
{ type: "abandon-room" },
]);
expect(sockets[1]!.close).toHaveBeenCalledWith(1000, "client closed");
});
it("classifies a close-only authentication failure without claiming replacement", () => {
const sockets: FakeWebSocket[] = [];
class FakeWebSocket extends EventTarget {
tests/server-signal.test.ts
+94
View File
@@ -2216,6 +2216,100 @@ describe("WebSocket signaling", () => {
});
});
it("marks an ordinary Host edge ready after its exact SDP handshake", async () => {
const harness = await startHarness({ peerAssistedMedia: false });
const host = await openClient(harness.webSocketUrl);
const hostAuth = await authenticate(
host,
harness.room,
"host",
"ordinary-ready-host",
1,
undefined,
{ viewerPresence: true },
);
await host.inbox.next("viewer-presence");
const viewer = await openClient(harness.webSocketUrl);
const viewerAuth = await authenticate(
viewer,
harness.room,
"viewer",
"ordinary-ready-viewer",
1,
undefined,
{ viewerPresence: true },
);
expect((await host.inbox.next("peer-joined")).peerId).toBe(
viewerAuth.peerId,
);
const assigned = await nextViewerPresenceMatching(
host,
(message) =>
viewerPresenceEntries(message).some(
(entry) => entry.peerId === viewerAuth.peerId,
),
);
expect(viewerPresenceEntries(assigned)[0]).not.toHaveProperty("mediaReady");
const connectionId = "ordinary_ready_connection_12345678";
host.socket.send(
JSON.stringify({
type: "signal",
targetPeerId: viewerAuth.peerId,
payload: {
kind: "description",
connectionId,
description: { type: "offer", sdp: "v=0\r\n" },
},
}),
);
expect(await viewer.inbox.next("signal")).toMatchObject({
fromPeerId: hostAuth.peerId,
payload: { connectionId, description: { type: "offer" } },
});
host.socket.send(
JSON.stringify({
type: "set-display-name",
displayName: "等待应答",
}),
);
const offered = await nextViewerPresenceMatching(
host,
(message) =>
viewerPresenceEntries(message).some(
(entry) => entry.peerId === viewerAuth.peerId,
),
);
expect(viewerPresenceEntries(offered)[0]).not.toHaveProperty("mediaReady");
viewer.socket.send(
JSON.stringify({
type: "signal",
payload: {
kind: "description",
connectionId,
description: { type: "answer", sdp: "v=0\r\n" },
},
}),
);
expect(await host.inbox.next("signal")).toMatchObject({
fromPeerId: viewerAuth.peerId,
payload: { connectionId, description: { type: "answer" } },
});
const ready = await nextViewerPresenceMatching(
host,
(message) =>
viewerPresenceEntries(message).some(
(entry) =>
entry.peerId === viewerAuth.peerId && entry.mediaReady === true,
),
);
expect(viewerPresenceEntries(ready)[0]).toMatchObject({
peerId: viewerAuth.peerId,
upstream: { kind: "peer", peerId: hostAuth.peerId },
mediaReady: true,
});
});
it("binds new-share quality and preserves active-share quality across reconnect", async () => {
const harness = await startHarness({ peerAssistedMedia: true });
const initialSettings: QualitySettings = {
tests/viewer-quality-evidence.test.ts
+31
View File
@@ -16,6 +16,7 @@ import {
qualityEvidenceWindowFromMetrics,
reconcileViewerQualityEvidencePresentation,
refreshViewerQualityEvidencePresentation,
retainPresentViewerQualityEvidence,
type ViewerQualityEvidence,
ViewerQualityEvidenceReporter,
} from "../src/client/media/viewer-quality-evidence.ts";
@@ -809,4 +810,34 @@ describe("viewer quality evidence", () => {
).toBeNull();
});
it("retains evidence only for Viewers in the authoritative presence", () => {
const first = presentViewerQualityEvidence(
null,
serverEvidence({ viewerPeerId: "viewer_first_12345678" }),
0,
);
const second = presentViewerQualityEvidence(
null,
serverEvidence({ viewerPeerId: "viewer_second_12345678" }),
0,
);
const presentations = new Map([
[first.evidence.viewerPeerId, first],
[second.evidence.viewerPeerId, second],
]);
expect(
retainPresentViewerQualityEvidence(
presentations,
new Set(presentations.keys()),
),
).toBe(presentations);
const retained = retainPresentViewerQualityEvidence(
presentations,
new Set([second.evidence.viewerPeerId]),
);
expect([...retained.keys()]).toEqual([second.evidence.viewerPeerId]);
expect(presentations.size).toBe(2);
});
});