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,