fix(routing): recover failed connections across route transitions

## 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.
2026-09-17 22:56:06 +08:00
1 parent 083c9273a7
commit 78e0d4194a
@@ -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 {
@@ -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")
+				}
+			})
+		})
+	}
+}
@@ -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)