From 78e0d4194a998420b6d5077693a8dbfdf28d1389 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Thu, 17 Sep 2026 22:56:06 +0800 Subject: [PATCH] fix(routing): recover failed connections across route transitions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem and outcome / 问题与结果 An active media connection could fail while a replacement was being prepared, but its failure report was suppressed or attributed to the pending route. Separately, a room revision advanced by another Viewer could invalidate a delayed failure report for a connection that was still current, leaving playback stalled after the browser had already reported the failure. Failure reporting now follows committed connection ownership. P2P and SFU reports carry the existing physical connection identity; the server accepts a delayed active failure only while that connection and its endpoint session still match. Preparation reports retain their exact operation checks, and stale connections, stale sessions and future revisions remain rejected. This adds no wire fields, recovery timers, dependencies or routing policy. ## Release notes ### 中文 这是一次连接恢复修复，改善分享过程中断线或画面停滞后的自动恢复。 - 修复自动切换线路期间，当前连接失效后可能未触发正确恢复的问题，覆盖 P2P 和服务器兜底线路。 - 修复其他观众加入等房间变化恰好发生时，延迟到达的故障反馈可能被忽略，导致画面停滞后无法自动恢复的问题。 - App 公网邀请或本地模式请房主更新 App 后重新分享，观众刷新页面。使用已部署站点时，请管理员更新 Server，房主和观众刷新页面。 ### English This patch improves automatic recovery when a connection fails or playback stalls during sharing. - Fixed failures on the current connection being missed or attributed to the wrong route while Piik prepares a replacement. This covers P2P and server fallback connections. - Fixed delayed failure reports being ignored after another Viewer joins or the room changes, which could leave playback stalled without automatic recovery. - For App public invitations or Local mode, the Host should update the App and start sharing again; Viewers should reload the page. For a deployed site, update the Server, then refresh the page for the Host and all Viewers. ## Verification / 验证 - `npm run check`: typecheck, 71 test files / 1,003 tests, and the Web production build passed. - `npm run check:go-core`: Go tests, formatting, vet and supported core target builds passed. - Browser fault injection passed for delayed reports across room changes, Host and Viewer signaling reconnects, pending-candidate reauthentication, and source replacement during preparation. The previously reproduced stall recovered automatically with one failure report; process, profile and port cleanup passed. - The embedded SFU H.264 simulcast gate passed with decoded video and audio, a live 720p to 480p profile change, and publication retirement. - Focused regressions cover Host/Viewer and P2P/SFU report ownership, retired connections, wrong identities, stale sessions and future revisions. Documentation and repository hygiene checks passed. These results do not establish the original reporter's specific cause. Physical Native capture, representative public networks and long-duration recovery remain outside this local acceptance. ## Impact / 影响 Patch release: **v1.5.3**. No wire, configuration or storage format changes; existing published message handlers remain compatible. The routing contract owns the updated failure-report rule, and TODO retains the outstanding reporter/device evidence boundaries. --- docs/standards/routing-transport.md | 6 + docs/todo.md | 11 +- internal/server/signal/router.go | 6 +- internal/server/signal/router_failure_test.go | 132 ++++++++++++++++++ internal/server/signal/router_test.go | 6 + src/client/media/host-sfu-route.ts | 20 ++- src/client/media/route-transition.ts | 11 +- src/client/media/viewer-sfu-route.ts | 24 +++- src/client/pages/ViewerPage.tsx | 24 ++-- tests/route-transition.test.ts | 92 +++++++++++- 10 files changed, 298 insertions(+), 34 deletions(-) create mode 100644 internal/server/signal/router_failure_test.go diff --git a/docs/standards/routing-transport.md b/docs/standards/routing-transport.md index f330a22..ef83bba 100644 --- a/docs/standards/routing-transport.md +++ b/docs/standards/routing-transport.md @@ -127,6 +127,12 @@ perform quality selection or choose another route. Current-edge P2P signaling remains valid during candidate overlap. Accepting a replacement from that parent cancels conflicting optional preparation so the replacement can complete; a retired candidate cannot take its place afterward. +Failure reports belong to the failed physical connection, not another +preparation's phase. A committed connection's identified failure remains valid +across unrelated room revisions only while its endpoint session and connection +identity still match; future revisions and retired connections are rejected. +Preparation reports retain their exact operation fence. Reports without a +connection identity retain their exact room-revision fence. ## SFU Fallback diff --git a/docs/todo.md b/docs/todo.md index 239c521..ab3af35 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -7,10 +7,8 @@ history. A parked idea is not implementation authority. ## Now -No additional implementation is scheduled. Routine release and postflight -follow the [deployment runbook](./deployment.md); public demo and private service -deployments remain separate. Physical limits are owned by -[verification status](./verification-status.md). +No implementation work is currently scheduled. Reporter/device follow-ups and +evidence-dependent work remain below. ## Awaiting Device Or Reporter Evidence @@ -89,9 +87,8 @@ not establish better connection success or speed; this note adds no retry policy resource owners, fewer shared writers and a smaller change surface under [engineering review](./standards/engineering.md#ablation-and-review); file size alone does not justify a split. The completed audits do not close this - candidate. Reopen C=3 structural-intent retention, SFU failure during unrelated - prepare and multi-child evidence ownership only with current-contract - reproductions. + candidate. Reopen C=3 structural-intent retention and multi-child evidence + ownership only with current-contract reproductions. When related behavior changes, compare the Browser/Native recovery budget; preserve Native bridge versus network failure distinctions when sharing code. No new revision namespace, failure-state mirror or topology queue by default. diff --git a/internal/server/signal/router.go b/internal/server/signal/router.go index 0391f9e..60c6780 100644 --- a/internal/server/signal/router.go +++ b/internal/server/signal/router.go @@ -860,7 +860,11 @@ func (r *router) handleRouteFailed(participant authenticatedRouteParticipant, me return } - if snapshot.Revision != revision { + // A room revision can advance for another Viewer while this report is in + // flight. Only an exact still-owned connection may outlive that revision; + // identity-free reports retain their snapshot fence. Endpoint/session and + // connection checks below still reject retired resources. + if revision > snapshot.Revision || (revision != snapshot.Revision && connectionID == "") { return } if participant.role == protocol.RoleHost { diff --git a/internal/server/signal/router_failure_test.go b/internal/server/signal/router_failure_test.go new file mode 100644 index 0000000..cc092ab --- /dev/null +++ b/internal/server/signal/router_failure_test.go @@ -0,0 +1,132 @@ +package signal + +import ( + "testing" + + "github.com/TNTcraftHIM/Piik/internal/server/protocol" +) + +func TestRouterFencesDelayedActiveFailureByPhysicalConnection(t *testing.T) { + for _, transport := range []string{"peer", "sfu"} { + for _, role := range []string{"host", "viewer"} { + for _, report := range []string{"exact", "no-identity", "other-connection", "old-session", "future"} { + t.Run(transport+"/"+role+"/"+report, func(t *testing.T) { + h := newRouterHarness(t, routerHarnessOptions{capacity: 2, withSfu: transport == "sfu", fakeTimers: true}) + created := h.createRoom() + host := h.connectHost(created, nil, "", "") + viewer := h.connectViewer(created, "delayed-failure") + h.locked(func() { h.doComplete(host); h.doComplete(viewer) }) + prepared := h.waitPreparedTransport(viewer.sessionID, "direct") + if transport == "sfu" { + h.routeFailed(viewer, routeFailedMessage(int64(prepared.Revision), "prepare", prepared.Candidate.ConnectionID)) + prepared = h.waitPreparedTransport(viewer.sessionID, "sfu") + } + h.routeReady(viewer, int64(prepared.Revision)) + var message protocol.RouteFailedMessage + h.locked(func() { + edge := h.doMustEdge(created.RoomID, viewer.peerID) + id := edge.connectionID + if transport == "sfu" && role == "host" { + rm, _ := h.router.rooms.Get(created.RoomID) + id = rm.controller.Snapshot().HostPublication.ConnectionID + } + message = routeFailedMessage(edge.revision, "active", id) + }) + + other := h.connectViewer(created, "unrelated-join") + h.complete(other) + candidate := h.waitPreparedTransport(other.sessionID, "direct") + h.routeReady(other, int64(candidate.Revision)) + h.locked(func() { + rm, _ := h.router.rooms.Get(created.RoomID) + before := rm.controller.Snapshot() + if before.Revision <= int64(message.Revision) { + t.Fatal("another Viewer must advance the room revision") + } + participant := viewer + if role == "host" { + participant = host + } + switch report { + case "no-identity": + message.ConnectionID = nil + case "other-connection": + id := "retired_connection_12345678" + message.ConnectionID = &id + case "old-session": + participant.sessionID = "retired_session_12345678" + case "future": + message.Revision = protocol.Int(before.Revision + 1) + } + h.router.handleRouteFailed(participant, message) + after := rm.controller.Snapshot() + edge, ok := after.UpstreamByViewer.Get(viewer.peerID) + usable := ok && edge.Usable + if transport == "sfu" && role == "host" { + usable = after.HostPublication != nil && after.HostPublication.Usable + } + if usable != (report != "exact") { + t.Fatalf("active usable=%v after %s report", usable, report) + } + unaffected, ok := after.UpstreamByViewer.Get(other.peerID) + if !ok || !unaffected.Usable { + t.Fatal("unrelated direct Viewer must remain usable") + } + }) + }) + } + } + } +} + +func TestRouterActiveFailureDuringPeerPreparationKeepsItsSubject(t *testing.T) { + for _, role := range []string{"host", "viewer"} { + t.Run(role, func(t *testing.T) { + capacity := 2 + if role == "viewer" { + capacity = 1 + } + h := newRouterHarness(t, routerHarnessOptions{capacity: capacity, withSfu: true, fakeTimers: true}) + created := h.createRoom() + host := h.connectHost(created, nil, "", "") + viewer := h.connectViewer(created, "sfu-ingress") + h.locked(func() { h.doComplete(host); h.doComplete(viewer) }) + direct := h.waitPreparedTransport(viewer.sessionID, "direct") + h.routeFailed(viewer, routeFailedMessage(int64(direct.Revision), "prepare", direct.Candidate.ConnectionID)) + sfu := h.waitPreparedTransport(viewer.sessionID, "sfu") + h.routeReady(viewer, int64(sfu.Revision)) + if role == "viewer" { + h.locked(func() { h.doRelay(viewer, 1) }) + } + child := h.connectViewer(created, "preparing-child") + h.complete(child) + candidate := h.waitPreparedTransport(child.sessionID, "direct") + h.locked(func() { + rm, _ := h.router.rooms.Get(created.RoomID) + before := rm.controller.Snapshot() + participant := host + id := before.HostPublication.ConnectionID + if role == "viewer" { + participant = viewer + edge, _ := before.UpstreamByViewer.Get(viewer.peerID) + id = edge.ConnectionID + } + if before.Operation.Current.Tuple.ParentPeerID != participant.peerID { + t.Fatal("preparation must belong to the reporting parent") + } + h.router.handleRouteFailed(participant, routeFailedMessage(before.Revision, "active", id)) + after := rm.controller.Snapshot() + if role == "host" { + if after.HostPublication != nil && after.HostPublication.Usable { + t.Fatal("failed publication remains usable") + } + if after.Operation == nil || after.Operation.Current == nil || after.Operation.Current.ConnectionID != candidate.Candidate.ConnectionID { + t.Fatal("independent direct candidate was retired") + } + } else if edge, ok := after.UpstreamByViewer.Get(viewer.peerID); ok && edge.Usable { + t.Fatal("failed subscriber remains usable") + } + }) + }) + } +} diff --git a/internal/server/signal/router_test.go b/internal/server/signal/router_test.go index ab87e44..2fa405a 100644 --- a/internal/server/signal/router_test.go +++ b/internal/server/signal/router_test.go @@ -1150,6 +1150,12 @@ func TestRouterAcceptsActiveFailureForDirectConnectionAdoptedDuringRebuild(t *te if adopted.revision != active.revision || adopted.connectionID != rebuiltConnectionID { t.Fatalf("adopted edge = %+v", adopted) } + h.doFailed(viewer, routeFailedMessage(active.revision, "active", active.connectionID)) + rm, _ := h.router.rooms.Get(created.RoomID) + edge, ok := rm.controller.Snapshot().UpstreamByViewer.Get(viewer.peerID) + if !ok || !edge.Usable || edge.ConnectionID != rebuiltConnectionID { + t.Fatal("retired-connection failure affected its replacement") + } h.doFailed(viewer, routeFailedMessage(active.revision, "active", rebuiltConnectionID)) }) h.waitFailed(viewer.sessionID) diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index 4493d0a..b9e1d76 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -659,7 +659,16 @@ export class HostSfuRoute { return; } if (phase === "prepare") { - if (matchesPlannedRoute || wasActive) { + const activeRevision = this.route.getActiveRevision(); + if (wasActive) { + if ( + activeRevision !== null && + this.route.getActiveAssignment()?.sfuPublicationGeneration === + slot.publicationGeneration + ) { + this.routeFailed(activeRevision, "active", slot.connectionId); + } + } else if (matchesPlannedRoute) { this.routeFailed(revision, "prepare", null); } return; @@ -667,7 +676,7 @@ export class HostSfuRoute { if (!matchesPlannedRoute) { return; } - this.requestRecovery(revision); + this.requestRecovery(revision, slot.connectionId); } private failPublisherSlot(slot: HostPublisherSlot): void { @@ -678,7 +687,10 @@ export class HostSfuRoute { this.handleFailure(slot); } - private requestRecovery(revision: number): void { + private requestRecovery( + revision: number, + connectionId: string | null = null, + ): void { if (!this.recovery || this.recovery.revision !== revision) { this.recovery = { revision, refreshed: false }; } @@ -688,7 +700,7 @@ export class HostSfuRoute { return; } } - this.routeFailed(revision, "active", null); + this.routeFailed(revision, "active", connectionId); } private routeFailed( diff --git a/src/client/media/route-transition.ts b/src/client/media/route-transition.ts index 744b1ee..1303c66 100644 --- a/src/client/media/route-transition.ts +++ b/src/client/media/route-transition.ts @@ -42,6 +42,8 @@ export class MediaRouteTransition { private phase: MediaRoutePhase | null = null; private plannedAssignment: ParticipantRouteAssignment | null = null; private activeAssignment: ParticipantRouteAssignment | null = null; + // Preparation advances the operation without replacing committed authority. + private activeRevision: number | null = null; private mediaAssignment: ParticipantRouteAssignment | null = null; private preparedCandidate: PreparedRouteCandidate | null = null; private generation = 0; @@ -73,6 +75,7 @@ export class MediaRouteTransition { update.phase === "prepare" ? cloneCandidate(update.candidate) : null; if (update.phase === "active") { this.activeAssignment = cloneAssignment(update.assignment); + this.activeRevision = update.revision; } this.generation += 1; return "accepted"; @@ -123,6 +126,10 @@ export class MediaRouteTransition { return this.activeAssignment ? cloneAssignment(this.activeAssignment) : null; } + getActiveRevision(): number | null { + return this.activeRevision; + } + getMediaAssignment(): ParticipantRouteAssignment | null { return this.mediaAssignment ? cloneAssignment(this.mediaAssignment) : null; } @@ -138,6 +145,7 @@ export class MediaRouteTransition { this.phase = null; this.plannedAssignment = null; this.activeAssignment = null; + this.activeRevision = null; this.mediaAssignment = null; this.preparedCandidate = null; this.generation += 1; @@ -169,9 +177,8 @@ export function reportActivePeerRouteFailure( send: (message: ClientMessage) => boolean, ): boolean { const assignment = route.getActiveAssignment(); - const revision = route.getRevision(); + const revision = route.getActiveRevision(); if ( - route.getPhase() !== "active" || revision === null || assignment?.upstream.kind !== "peer" || assignment.upstream.peerId !== parentPeerId diff --git a/src/client/media/viewer-sfu-route.ts b/src/client/media/viewer-sfu-route.ts index 81139cc..8341ab7 100644 --- a/src/client/media/viewer-sfu-route.ts +++ b/src/client/media/viewer-sfu-route.ts @@ -86,6 +86,7 @@ interface ViewerSfuRouteEvents { framesDecodedDelta: number | null, revision: number, mediaIdentity: string, + connectionId: string, ) => void; onSfuState?: ( state: "connected" | "reconnecting", @@ -474,6 +475,7 @@ export class ViewerSfuRoute { framesDecodedDelta, slot.revision, slot.mediaIdentity, + slot.connectionId, ); } }, @@ -949,7 +951,17 @@ export class ViewerSfuRoute { return; } if (phase === "prepare") { - if (matchesPlannedRoute || wasActive) { + const activeRevision = this.route.getActiveRevision(); + const activeAssignment = this.route.getActiveAssignment(); + if (wasActive) { + if ( + activeRevision !== null && + activeAssignment?.upstream.kind === "sfu" && + activeAssignment.sfuPublicationGeneration === slot.publicationGeneration + ) { + this.routeFailed(activeRevision, "active", slot.connectionId); + } + } else if (matchesPlannedRoute) { this.routeFailed(revision, "prepare"); } return; @@ -961,7 +973,7 @@ export class ViewerSfuRoute { this.recovery?.revision === revision && this.recovery.refreshed ) { - this.routeFailed(revision, "active"); + this.routeFailed(revision, "active", slot.connectionId); return; } this.requestRecovery(revision); @@ -1007,12 +1019,16 @@ export class ViewerSfuRoute { : null; } - private routeFailed(revision: number, phase: MediaRoutePhase): boolean { + private routeFailed( + revision: number, + phase: MediaRoutePhase, + connectionId: string | null = null, + ): boolean { return this.events.send({ type: "route-failed", revision, phase, - connectionId: null, + connectionId, }); } diff --git a/src/client/pages/ViewerPage.tsx b/src/client/pages/ViewerPage.tsx index 1ff7617..55637de 100644 --- a/src/client/pages/ViewerPage.tsx +++ b/src/client/pages/ViewerPage.tsx @@ -924,7 +924,7 @@ export function ViewerPage({ identity: string, framesDecodedDelta: number | null, authorityRevision: number, - connectionId?: string, + connectionId: string, ): void { if ( currentHostPaused || @@ -936,20 +936,12 @@ export function ViewerPage({ decodedFrameStall.allowReportRetry(); return; } - let sent = false; - if (route === "peer" && connectionId) { - sent = signal.send({ - type: "route-failed", - revision: authorityRevision, - phase: "active", - connectionId, - }); - } else if (route === "sfu") { - sent = signal.send({ - type: "route-media-unavailable", - revision: authorityRevision, - }); - } + const sent = signal.send({ + type: "route-failed", + revision: authorityRevision, + phase: "active", + connectionId, + }); if (!sent) decodedFrameStall.allowReportRetry(); } @@ -1127,6 +1119,7 @@ export function ViewerPage({ framesDecodedDelta, revision, mediaIdentity, + connectionId, ) => { if (active && viewerSfuRoute === route) { observeActiveDecodedFrames( @@ -1134,6 +1127,7 @@ export function ViewerPage({ mediaIdentity, framesDecodedDelta, revision, + connectionId, ); } }, diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index 81e0e3f..dae8b43 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -367,6 +367,92 @@ describe("minimal route transition contracts", () => { ]); }); + 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); + route.accept({ revision: 7, phase: "active", assignment: viewerSfuAssignment() }); + await route.acceptConfig(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[] = []; @@ -970,6 +1056,7 @@ describe("minimal route transition contracts", () => { framesDecodedDelta: number | null; revision: number; mediaIdentity: string; + connectionId: string; }> = []; const subscribers: ReturnType[] = []; const route = new ViewerSfuRoute("viewer_12345678", { @@ -980,7 +1067,8 @@ describe("minimal route transition contracts", () => { framesDecodedDelta, revision, mediaIdentity, - ) => decodedSamples.push({ framesDecodedDelta, revision, mediaIdentity }), + connectionId, + ) => decodedSamples.push({ framesDecodedDelta, revision, mediaIdentity, connectionId }), send: (message) => { messages.push(message); return true; @@ -1012,6 +1100,7 @@ describe("minimal route transition contracts", () => { framesDecodedDelta: 3, revision: 1, mediaIdentity: expect.stringContaining("publication_generation_12345678:"), + connectionId: sfuConfig(1).connectionId, }, ]); @@ -1050,6 +1139,7 @@ describe("minimal route transition contracts", () => { expect(decodedSamples[1]).toMatchObject({ framesDecodedDelta: null, revision: 2, + connectionId: sfuConfig(2).connectionId, }); expect(decodedSamples[1]!.mediaIdentity).not.toBe( decodedSamples[0]!.mediaIdentity,