From ff50e0609e77fd41e828cda016dd5977acee2762 Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sun, 20 Sep 2026 09:45:55 +0800 Subject: [PATCH] fix(media): preserve playback through native connection recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem and outcome / 问题与结果 Compatible Native receiver renegotiation rebuilt the encoded source and its playback/relay consumers. Replacement backends could also lose committed-route recovery ownership, while queued events and unfinished SFU cleanup could reach a replacement with the same logical identity. Retain compatible receivers, keep recovery on the committed route, and validate physical ownership through final control delivery. SFU activation waits for the existing retirement barrier. This phase also consumes renewed same-page invitations through the existing grant owner and avoids closing uninitialized Windows capture objects after preview failure. Camera, microphone, interaction and control redesign remain outside this patch. ## Release notes ### 中文 这是一次以连接稳定性为重点的小修复更新。 - 修复 App 接收端重新协商时不必要地中断播放和中继的问题，并保留正确的连接恢复行为。 - 修复旧连接的迟到事件及关闭时序干扰替代连接的问题。 - 修复重新打开同页邀请链接后仍沿用旧邀请的问题。 - 修复 Windows 选源预览失败时的一处异常退出。 更新 App 并刷新网页即可使用这些修复，无需修改配置。 ### English This maintenance update focuses on connection stability. - Preserve App playback, relay connections and recovery behavior during compatible renegotiation. - Prevent delayed events and cleanup from retired connections from disrupting their replacements. - Use the new invitation when another invite link opens in the same page. - Fix a Windows crash during failed source-preview cleanup. Update Piik App and refresh the page to use these fixes. No configuration changes are required. ## Verification / 验证 - TypeScript type checking, 71 Vitest files / 1,015 tests and Web production build. - Go core tests, vet and supported cross-target builds; receiver tests exercise H.264/VP8/Opus, ICE restart, fresh gathering and retained downstream media. - Windows native compile/probe and deterministic capture checks, including failed-preview cleanup. - Documentation and repository hygiene checks; independent review of the extracted patch and old/new Web/App capability negotiation. - Local full-suite runs reached the unchanged Git fixture's 5-second timeout; that suite passed alone. The complete suite was rerun with two workers and a 15-second test deadline. ## Impact / 影响 Planned version: v1.5.6. Existing public wire versions and configuration are retained. Receiver reuse is capability-negotiated: old Apps receive no new request field, and old pages receive no new answer field. Room authority, routing policy and retry budgets are unchanged. The established-viewing interruption report still lacks paired diagnostics; these reproduced defects do not establish that reporter's exact cause. Physical device/network endurance remains outside local deterministic acceptance. Git/PRs retain this repair history; the TODO ledger retains outstanding reports and separate feature candidates. --- docs/standards/engineering.md | 3 + docs/standards/routing-transport.md | 4 + docs/todo.md | 35 ++++-- internal/app/app.go | 4 +- internal/app/loopback/server.go | 18 ++- internal/app/loopback/server_test.go | 15 +++ internal/app/mediaedge/receiver.go | 110 +++++++++++++++--- internal/app/mediaedge/receiver_test.go | 95 ++++++++++++++- internal/app/nativecontrol/session.go | 25 ++-- internal/app/nativecontrol/session_test.go | 39 +++++++ internal/app/nativecontrol/wire.go | 18 +-- internal/app/nativehost/publication.go | 21 +++- internal/app/nativehost/publication_test.go | 34 +++++- internal/app/nativehost/session.go | 2 + internal/app/nativeviewer/session.go | 45 +++++-- internal/app/nativeviewer/session_test.go | 60 ++++++++++ native/capture/windows/build.ps1 | 5 +- native/capture/windows/capture_target.cpp | 4 +- .../capture/windows/capture_target.test.cpp | 30 +++++ src/client/App.tsx | 19 ++- src/client/lib/session.ts | 12 ++ src/client/media/host-sfu-route.ts | 3 +- src/client/native/client.ts | 3 + src/client/native/native-sfu-publisher.ts | 55 +++++---- src/client/native/native-viewer-peer.ts | 92 ++++++++++----- src/client/native/wire.ts | 2 + tests/client-state.test.ts | 42 +++++++ tests/native-client.test.ts | 39 ++++++- tests/native-sfu-publisher.test.ts | 43 ++++++- tests/native-viewer-peer.test.ts | 109 ++++++++++++++++- tests/route-transition.test.ts | 29 +++++ 31 files changed, 893 insertions(+), 122 deletions(-) create mode 100644 native/capture/windows/capture_target.test.cpp diff --git a/docs/standards/engineering.md b/docs/standards/engineering.md index 2000d0d..029fb0f 100644 --- a/docs/standards/engineering.md +++ b/docs/standards/engineering.md @@ -142,6 +142,9 @@ above rather than searching the tree. retirement. After an await/callback, and inside cleanup, validate the original operation/resource identity before changing current state. Cancellation, failure, missing observation and success must retain distinct meanings. + Queued events retain physical-owner validity until the final serialized write; + reused logical IDs do not identify a replacement instance. New subscriptions + follow the old owner's retirement, including repeated cleanup callers. Later cancellation must not erase a failure already observed before cleanup. A caller deadline does not cancel a dependency that ignores it. Keep late work bounded and serialize its cleanup; verify the dependency's actual lifetime. diff --git a/docs/standards/routing-transport.md b/docs/standards/routing-transport.md index ef83bba..175ea94 100644 --- a/docs/standards/routing-transport.md +++ b/docs/standards/routing-transport.md @@ -110,6 +110,10 @@ connection-attempt budget after rollback. The waiting display uses the actual server-issued attempt ordinal; it never counts time as an attempted connection. Committed Browser P2P recovery first restarts ICE on the same connection, then rebuilds that connection if necessary, within the existing two-request budget. +Capable Native receivers also renegotiate that connection while retaining the +encoded source, local playback and downstream edges when its codec/audio shape +is unchanged. Pion owns ICE restart; each actual gathering owns fresh supplemental +candidates. A changed media shape replaces the source and its dependent edges. It does not require NAT prediction or reopen the route candidate. A prepared candidate instead follows its current route operation's failure path, without an independent restart loop. diff --git a/docs/todo.md b/docs/todo.md index ab3af35..1a34e0a 100644 --- a/docs/todo.md +++ b/docs/todo.md @@ -1,17 +1,35 @@ # Current TODO Ledger -Last reviewed: 2026-09-17 +Last reviewed: 2026-09-20 Only **Now** is executable. Product modules own behavior; Git/PRs own completed history. A parked idea is not implementation authority. ## Now -No implementation work is currently scheduled. Reporter/device follow-ups and -evidence-dependent work remain below. +- [ ] **Review camera and Host microphone in the separate capture candidate.** + Keep device selection in the existing source/settings surfaces. Finish control + grouping, dark-theme feedback and microphone volume presentation; verify native + mixing, device replacement and mobile camera behavior before acceptance. + These features are excluded from the bug-fix release. +- [ ] **Refine the isolated room-interaction prototype.** Review chat, optional + danmaku and participant-targeted reactions for usability, placement and intended + cross-view synchronization. Preserve room authorization and media-route owners. + This experiment does not authorize integration or publication. ## Awaiting Device Or Reporter Evidence +- [ ] **Interruption during established viewing.** A Viewer reportedly returns + to P2P connecting after watching for a while. Native receiver renegotiation, + retired event delivery and SFU replacement have locally reproduced defects + and regression checks, but paired diagnostics and device/network details are + still needed to establish this reporter's cause. Include an upstream relay's + report when present. The existing five-second Viewer membership grace during + signaling loss is unchanged. This is distinct from the first-frame report below. +- [ ] **Windows 11 capture border remains visible.** Identify the App/Browser + capture path, Windows build and capture-border permission result. A local + Windows 11 request succeeded; the reported failure remains unconfirmed. Check + system consent and other active captures before changing the WGC path. - [ ] **Brief connection followed by repeated Viewer loss on v1.5.0.** One Viewer reportedly drops just after connection details appear, while other Viewers work; App public-link mode is suspected. Check signaling, first-frame @@ -98,21 +116,16 @@ not establish better connection success or speed; this note adds no retry policy 6. **Platform output.** Reopen for a registered receiver acting as an ordinary Viewer only after the [platform-output gate](./research/platform-output.md) passes. -7. **Full UI themes.** After the base version, research themes that can change - layout, composition and motion, including a restrained graphic/cinematic - direction with original Piik assets. Assess extension boundaries and cost - before scheduling any theme/plugin API. Design experiments remain in Git - history, outside the main source tree. -8. **Additional languages.** Review community catalogs and their rendered UI +7. **Additional languages.** Review community catalogs and their rendered UI following the [translation guide](./guide/translating.md). Add website, documentation or App console translations as contributed; verify text direction and layout when a language requires it. Check contributed language names and rendered menu navigation when registering a new catalog. -9. **Windows code signing.** Revisit after enrollment in a trusted signing +8. **Windows code signing.** Revisit after enrollment in a trusted signing service. Sign Piik's executables before archive checksums are computed; signing improves publisher identity but does not guarantee that antivirus cloud scanning stops. Service selection and enrollment remain pending. -10. **Gitee download-source warning.** Paused by the owner. Keep GitHub primary +9. **Gitee download-source warning.** Paused by the owner. Keep GitHub primary and retain Gitee; do not add a self-hosted mirror. Chrome still blocks the Gitee attachment when Referer is removed. Reopen for new evidence or a provider review; the warning remains unresolved. diff --git a/internal/app/app.go b/internal/app/app.go index bd16239..54a0402 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -613,13 +613,15 @@ func discoverNativeMedia(ctx context.Context, configuredPath string) nativeRunti } capabilities, err := nativecapture.Discover(ctx, path) if err != nil { - return nativeRuntime{captureProcess: path, discoveryErr: err} + return nativeRuntime{captureProcess: path, discoveryErr: err, + capabilities: loopback.NativeMediaCapabilities{ReceiverReuse: true}} } summary := capabilities.Summary() return nativeRuntime{ captureProcess: path, capture: capabilities, capabilities: loopback.NativeMediaCapabilities{ + ReceiverReuse: true, Video: summary.Video, CaptureBorderControl: summary.CaptureBorderControl, ProcessAudio: summary.ProcessAudio, diff --git a/internal/app/loopback/server.go b/internal/app/loopback/server.go index 0e42bd2..b695907 100644 --- a/internal/app/loopback/server.go +++ b/internal/app/loopback/server.go @@ -46,6 +46,13 @@ type ControlSession interface { Close() error } +// ControlEvent retains a physical owner's validity through asynchronous queues. +// Current must only inspect ownership; it runs under the WebSocket write lock. +type ControlEvent struct { + Value any + Current func() bool +} + type Endpoint struct { URL string `json:"url"` Host string `json:"host"` @@ -62,6 +69,7 @@ type Health struct { } type NativeMediaCapabilities struct { + ReceiverReuse bool `json:"receiverReuse,omitempty"` Video bool `json:"video"` CaptureBorderControl bool `json:"captureBorderControl,omitempty"` ProcessAudio bool `json:"processAudio"` @@ -318,12 +326,18 @@ func (server *Server) handleControl(response http.ResponseWriter, request *http. } var writeMutex sync.Mutex write := func(ctx context.Context, value any) error { + writeMutex.Lock() + defer writeMutex.Unlock() + if event, ok := value.(ControlEvent); ok { + if event.Current != nil && !event.Current() { + return nil + } + value = event.Value + } encoded, encodeErr := encodeMessage(value) if encodeErr != nil { return encodeErr } - writeMutex.Lock() - defer writeMutex.Unlock() return connection.Write(ctx, websocket.MessageText, encoded) } if err = write(server.ctx, controlMessage{ diff --git a/internal/app/loopback/server_test.go b/internal/app/loopback/server_test.go index af814ac..e7787ca 100644 --- a/internal/app/loopback/server_test.go +++ b/internal/app/loopback/server_test.go @@ -245,6 +245,21 @@ func TestControlSessionSharesOneBoundedSocketForResponsesAndEvents(t *testing.T) writeControl(t, connection, requestJSON("request_hello", "hello")) var ready controlMessage readControl(t, connection, &ready) + // The old owner's event was queued while valid, but retired before delivery. + current := true + retired := ControlEvent{Value: controlMessage{Version: ProtocolVersion, Type: "retired-event"}, + Current: func() bool { return current }} + current = false + extension.events <- retired + extension.events <- ControlEvent{ + Value: controlMessage{Version: ProtocolVersion, Type: "current-event"}, + Current: func() bool { return true }, + } + var currentEvent controlMessage + readControl(t, connection, ¤tEvent) + if currentEvent.Type != "current-event" { + t.Fatalf("retired event escaped the delivery fence: %+v", currentEvent) + } extension.events <- controlMessage{ Version: ProtocolVersion, ID: "event_123456", Type: "extension-event", } diff --git a/internal/app/mediaedge/receiver.go b/internal/app/mediaedge/receiver.go index 671b522..6ef91f4 100644 --- a/internal/app/mediaedge/receiver.go +++ b/internal/app/mediaedge/receiver.go @@ -13,8 +13,9 @@ import ( ) type ReceiverEvents struct { - LocalCandidate func(*webrtc.ICECandidateInit) - ConnectionState func(webrtc.PeerConnectionState, *SelectedPair) + // The final consumer must check current again after any event queues. + LocalCandidate func(candidate *webrtc.ICECandidateInit, current func() bool) + ConnectionState func(state webrtc.PeerConnectionState, pair *SelectedPair, current func() bool) } type ReceiverOptions struct { @@ -29,11 +30,16 @@ type ReceiverOptions struct { // ordinary bounded Sources. Consumers can attach local playback and downstream // P2P edges without adding a decoder or encoder. type Receiver struct { + engine *Engine connection *webrtc.PeerConnection source *Source audioSource *AudioSource events ReceiverEvents localCandidates *localCandidateGathering + iceServers []webrtc.ICEServer + prepareMapping func() int + offerMu sync.Mutex + candidateMu sync.Mutex mu sync.Mutex closed bool @@ -49,7 +55,7 @@ func (engine *Engine) NewReceiver(options ReceiverOptions) (*Receiver, webrtc.Se return nil, webrtc.SessionDescription{}, err } var prepareMapping func() int - if engine.portMapping != nil && len(options.ICEServers) > 0 { + if engine.portMapping != nil { prepareMapping = engine.portMapping.Prepare } connection, err := engine.api.NewPeerConnection(webrtc.Configuration{}) @@ -66,8 +72,11 @@ func (engine *Engine) NewReceiver(options ReceiverOptions) (*Receiver, webrtc.Se return nil, webrtc.SessionDescription{}, err } receiver := &Receiver{ - connection: connection, - events: options.Events, + engine: engine, + connection: connection, + events: options.Events, + iceServers: options.ICEServers, + prepareMapping: prepareMapping, } layers := 1 if _, supported := relayBackend(videoCodec, options.Relay); supported { @@ -89,14 +98,18 @@ func (engine *Engine) NewReceiver(options ReceiverOptions) (*Receiver, webrtc.Se return nil, webrtc.SessionDescription{}, err } } - receiver.localCandidates = newLocalCandidateGathering( - engine, - options.ICEServers, - prepareMapping, - options.Events.LocalCandidate, - ) - connection.OnICECandidate(receiver.localCandidates.addPion) + // Pion announces gathering before emitting candidates, including when a + // remote ICE restart starts inside SetRemoteDescription. Ordinary SDP + // renegotiation leaves this collector and its completion fence alone. + connection.OnICEGatheringStateChange(func(state webrtc.ICEGatheringState) { + if state == webrtc.ICEGatheringStateGathering { + receiver.beginGathering() + } + }) connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { + if !receiver.active() { + return + } debug := slog.Default().Enabled(engine.ctx, slog.LevelDebug) if debug { slog.Debug("piik-client", "event", "media-connection-state", "direction", "inbound", "state", state.String()) @@ -112,7 +125,7 @@ func (engine *Engine) NewReceiver(options ReceiverOptions) (*Receiver, webrtc.Se } } if receiver.events.ConnectionState != nil { - receiver.events.ConnectionState(state, selected) + receiver.events.ConnectionState(state, selected, receiver.active) } }) connection.OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { @@ -132,6 +145,76 @@ func (receiver *Receiver) AudioSource() *AudioSource { return receiver.audioSour func (receiver *Receiver) HasAudio() bool { return receiver.audioSource != nil } func (receiver *Receiver) Codec() string { return receiver.source.Codec() } +func (receiver *Receiver) active() bool { + receiver.mu.Lock() + defer receiver.mu.Unlock() + return !receiver.closed +} + +// Renegotiate keeps the encoded source and its consumers when the offer still +// describes the same media shape. A changed codec/audio shape needs replacement. +func (receiver *Receiver) Renegotiate(offer webrtc.SessionDescription, servers []webrtc.ICEServer) (webrtc.SessionDescription, bool, error) { + receiver.offerMu.Lock() + defer receiver.offerMu.Unlock() + if offer.Type != webrtc.SDPTypeOffer || offer.SDP == "" { + return webrtc.SessionDescription{}, false, errors.New("native media receiver input is invalid") + } + hasAudio, err := offerSendsCodec(offer.SDP, "audio", "opus") + if err != nil { + return webrtc.SessionDescription{}, false, err + } + sameVideo, err := offerSendsCodec(offer.SDP, "video", receiver.Codec()) + if err != nil || !sameVideo || hasAudio != receiver.HasAudio() { + return webrtc.SessionDescription{}, false, err + } + receiver.mu.Lock() + closed := receiver.closed + receiver.iceServers = servers + receiver.mu.Unlock() + if closed { + return webrtc.SessionDescription{}, false, errors.New("native media receiver is closed") + } + if err = receiver.connection.SetRemoteDescription(offer); err != nil { + return webrtc.SessionDescription{}, false, err + } + answer, err := receiver.createAnswer() + return answer, err == nil, err +} + +func (receiver *Receiver) beginGathering() { + // Serialize candidate delivery with collector replacement, without holding + // the receiver state lock across the caller's event delivery. + receiver.candidateMu.Lock() + defer receiver.candidateMu.Unlock() + receiver.mu.Lock() + if receiver.closed { + receiver.mu.Unlock() + return + } + previous := receiver.localCandidates + var gathering *localCandidateGathering + current := func() bool { + receiver.mu.Lock() + defer receiver.mu.Unlock() + return !receiver.closed && receiver.localCandidates == gathering + } + gathering = newLocalCandidateGathering(receiver.engine, receiver.iceServers, receiver.prepareMapping, + func(candidate *webrtc.ICECandidateInit) { + receiver.candidateMu.Lock() + defer receiver.candidateMu.Unlock() + if current() && receiver.events.LocalCandidate != nil { + receiver.events.LocalCandidate(candidate, current) + } + }) + receiver.localCandidates = gathering + receiver.mu.Unlock() + if previous != nil { + previous.close() + } + receiver.connection.OnICECandidate(gathering.addPion) + gathering.start() +} + func (receiver *Receiver) createAnswer() (webrtc.SessionDescription, error) { receiver.mu.Lock() if receiver.closed { @@ -146,7 +229,6 @@ func (receiver *Receiver) createAnswer() (webrtc.SessionDescription, error) { if err = receiver.connection.SetLocalDescription(answer); err != nil { return webrtc.SessionDescription{}, err } - receiver.localCandidates.start() if receiver.connection.LocalDescription() == nil { return webrtc.SessionDescription{}, errors.New("native media receiver produced no SDP answer") } diff --git a/internal/app/mediaedge/receiver_test.go b/internal/app/mediaedge/receiver_test.go index 005b253..5e2fb91 100644 --- a/internal/app/mediaedge/receiver_test.go +++ b/internal/app/mediaedge/receiver_test.go @@ -112,6 +112,10 @@ func testReceiverForwarding(t *testing.T, codec string) { var receiverCandidates []*webrtc.ICECandidateInit var candidateMu sync.Mutex remoteReady := false + completed := make(chan struct{}, 4) + gatheringCurrent := make(chan func() bool, 4) + surveyCandidates := 0 + servers := []webrtc.ICEServer{{URLs: []string{localSurveyServer(t)}}} upstream.OnICECandidate(func(candidate *webrtc.ICECandidate) { if candidate == nil { return @@ -139,13 +143,19 @@ func testReceiverForwarding(t *testing.T, codec string) { nativeReceiver, answer, err := engine.NewReceiver(ReceiverOptions{ Offer: *upstream.LocalDescription(), + ICEServers: servers, EdgeCapacity: 1, Events: ReceiverEvents{ - LocalCandidate: func(candidate *webrtc.ICECandidateInit) { + LocalCandidate: func(candidate *webrtc.ICECandidateInit, current func() bool) { if candidate == nil { + gatheringCurrent <- current + completed <- struct{}{} return } candidateMu.Lock() + if strings.Contains(candidate.Candidate, "candidate:ns") { + surveyCandidates++ + } ready := remoteReady if !ready { receiverCandidates = append(receiverCandidates, candidate) @@ -267,6 +277,89 @@ func testReceiverForwarding(t *testing.T, codec string) { t.Fatalf("stream %d stopped delivering media after padding", index) } } + waitSignal(t, completed, "initial receiver gathering") + firstGatheringCurrent := <-gatheringCurrent + if !firstGatheringCurrent() { + t.Fatal("current receiver gathering lost delivery authority") + } + defer func() { + _ = nativeReceiver.Close() + if nativeReceiver.active() || firstGatheringCurrent() { + t.Error("retired receiver still authorizes queued events") + } + }() + source, audio := nativeReceiver.Source(), nativeReceiver.AudioSource() + for _, restart := range []bool{false, true} { + previousGathering := nativeReceiver.localCandidates + candidateMu.Lock() + remoteReady = false + candidateMu.Unlock() + offer, offerErr := upstream.CreateOffer(&webrtc.OfferOptions{ICERestart: restart}) + if offerErr != nil { + t.Fatal(offerErr) + } + if err = upstream.SetLocalDescription(offer); err != nil { + t.Fatal(err) + } + answer, reused, renegotiateErr := nativeReceiver.Renegotiate(offer, servers) + if renegotiateErr != nil || !reused { + t.Fatalf("receiver renegotiation: reused=%v, %v", reused, renegotiateErr) + } + if err = upstream.SetRemoteDescription(answer); err != nil { + t.Fatal(err) + } + candidateMu.Lock() + remoteReady = true + pendingUpstream, pendingReceiver = upstreamCandidates, receiverCandidates + upstreamCandidates, receiverCandidates = nil, nil + candidateMu.Unlock() + for _, candidate := range pendingUpstream { + _ = nativeReceiver.AddRemoteCandidate(candidate) + } + for _, candidate := range pendingReceiver { + _ = upstream.AddICECandidate(*candidate) + } + waitConnected(t, upstream, "renegotiated receiver") + if (previousGathering != nativeReceiver.localCandidates) != restart { + t.Fatalf("collector replaced=%v, ICE restart=%v", previousGathering != nativeReceiver.localCandidates, restart) + } + if restart { + waitSignal(t, completed, "restarted receiver gathering") + if firstGatheringCurrent() || !(<-gatheringCurrent)() { + t.Fatal("queued candidates did not retain their gathering owner") + } + previousGathering.addPion(nil) + select { + case <-completed: + t.Fatal("retired collector emitted completion") + default: + } + candidateMu.Lock() + count := surveyCandidates + candidateMu.Unlock() + if count != 2 { + t.Fatalf("STUN survey candidates=%d, want one per gathering", count) + } + } + if nativeReceiver.Source() != source || nativeReceiver.AudioSource() != audio || + downstream.connection.ConnectionState() != webrtc.PeerConnectionStateConnected { + t.Fatal("renegotiation retired the source or its healthy downstream") + } + for _, input := range []struct { + track *webrtc.TrackLocalStaticRTP + media *rtp.Packet + packets <-chan *rtp.Packet + }{{track, want, packets}, {audioTrack, wantAudio, audioPackets}} { + input.media.SequenceNumber += 10 + input.media.Timestamp += 90_000 + if err = input.track.WriteRTP(input.media); err != nil { + t.Fatal(err) + } + if packet := waitPacket(t, input.packets); string(packet.Payload) != string(input.media.Payload) { + t.Fatal("media did not resume on the retained downstream") + } + } + } } func connectedReceiverForSources( diff --git a/internal/app/nativecontrol/session.go b/internal/app/nativecontrol/session.go index db3f462..6a5b8c4 100644 --- a/internal/app/nativecontrol/session.go +++ b/internal/app/nativecontrol/session.go @@ -616,22 +616,27 @@ func (session *Session) receiveOffer( if err != nil { return nil, err } - answer, audio, codec, err := viewer.AcceptOffer( + result, err := viewer.AcceptOffer( request.ConnectionID, webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: request.SDP}, servers, + request.ReuseReceiver, ) if err != nil { return nil, err } - return receiveAnswerResponse{ + response := receiveAnswerResponse{ responseEnvelope: response(envelope, "receive-answer"), ShareID: request.ShareID, ConnectionID: request.ConnectionID, - SDP: answer.SDP, - Audio: audio, - Codec: codec, - }, nil + SDP: result.Answer.SDP, + Audio: result.Audio, + Codec: result.Codec, + } + if request.ReuseReceiver { + response.Reused = &result.Reused + } + return response, nil } func (session *Session) stopShare(shareID string) error { @@ -754,9 +759,13 @@ func (session *Session) relayEvents() { for { select { case event := <-session.hostEvents: - session.emit(eventMessage(event)) + var value any = eventMessage(event) + if event.Current != nil { + value = loopback.ControlEvent{Value: value, Current: event.Current} + } + session.emit(value) case event := <-session.viewerEvents: - session.emit(viewerEventMessage(event)) + session.emit(loopback.ControlEvent{Value: viewerEventMessage(event), Current: event.Current}) case <-session.ctx.Done(): return } diff --git a/internal/app/nativecontrol/session_test.go b/internal/app/nativecontrol/session_test.go index 849ecc9..a1b08d5 100644 --- a/internal/app/nativecontrol/session_test.go +++ b/internal/app/nativecontrol/session_test.go @@ -18,8 +18,47 @@ import ( "github.com/TNTcraftHIM/Piik/internal/app/loopback" "github.com/TNTcraftHIM/Piik/internal/app/nativecapture" "github.com/TNTcraftHIM/Piik/internal/app/nativehost" + "github.com/pion/webrtc/v4" ) +func TestReceiveOfferReusePreservesLegacyResponseShape(t *testing.T) { + peer, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = peer.Close() }) + if _, err = peer.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, + webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}); err != nil { + t.Fatal(err) + } + offer, err := peer.CreateOffer(nil) + if err != nil { + t.Fatal(err) + } + session := New("", nativecapture.Capabilities{}, false) + t.Cleanup(func() { _ = session.Close() }) + for _, reuse := range []bool{false, true} { + request := receiveOfferRequest{Version: 9, ID: "request_offer", Type: "receive-offer", ShareID: "viewer_session", + ConnectionID: "connection_1234", EdgeCapacity: 2, SDP: offer.SDP, ReuseReceiver: reuse} + payload, _ := json.Marshal(request) + value, err := session.Handle(t.Context(), payload) + if err != nil { + t.Fatal(err) + } + answer, ok := value.(receiveAnswerResponse) + if !ok { + t.Fatalf("receive offer returned %#v", value) + } + encoded, _ := json.Marshal(answer) + if bytes.Contains(encoded, []byte(`"reused"`)) != reuse { + t.Fatalf("new response field leaked across opt-in: %s", encoded) + } + if reuse && (answer.Reused == nil || !*answer.Reused) { + t.Fatal("opted-in answer did not retain its receiver") + } + } +} + func TestMain(tests *testing.M) { if marker := os.Getenv("PIIK_SOURCE_PROBE_FIXTURE"); marker != "" { if err := os.WriteFile(marker, nil, 0600); err != nil { diff --git a/internal/app/nativecontrol/wire.go b/internal/app/nativecontrol/wire.go index 5974863..281bd43 100644 --- a/internal/app/nativecontrol/wire.go +++ b/internal/app/nativecontrol/wire.go @@ -115,14 +115,15 @@ type prepareLocalEdgeRequest struct { } type receiveOfferRequest struct { - Version int `json:"version"` - ID string `json:"id"` - Type string `json:"type"` - ShareID string `json:"shareId"` - ConnectionID string `json:"connectionId"` - EdgeCapacity int `json:"edgeCapacity"` - ICEServers []iceServer `json:"iceServers"` - SDP string `json:"sdp"` + Version int `json:"version"` + ID string `json:"id"` + Type string `json:"type"` + ShareID string `json:"shareId"` + ConnectionID string `json:"connectionId"` + EdgeCapacity int `json:"edgeCapacity"` + ICEServers []iceServer `json:"iceServers"` + SDP string `json:"sdp"` + ReuseReceiver bool `json:"reuseReceiver,omitempty"` } type receiveCandidateRequest struct { @@ -223,6 +224,7 @@ type receiveAnswerResponse struct { SDP string `json:"sdp"` Audio bool `json:"audio"` Codec string `json:"codec"` + Reused *bool `json:"reused,omitempty"` } type eventEnvelope struct { diff --git a/internal/app/nativehost/publication.go b/internal/app/nativehost/publication.go index 207d927..c41bd26 100644 --- a/internal/app/nativehost/publication.go +++ b/internal/app/nativehost/publication.go @@ -20,15 +20,23 @@ func (session *Session) PreparePublication(generation, connectionID string, serv } source, audio := session.source, session.audioSource session.mu.Unlock() - publication, err := session.engine.NewPublication(source, mediaedge.EdgeOptions{ + var publication *mediaedge.Publication + current := func() bool { + session.mu.Lock() + defer session.mu.Unlock() + return !session.closed && publication != nil && session.publications[key] == publication + } + created, err := session.engine.NewPublication(source, mediaedge.EdgeOptions{ ConnectionID: connectionID, ICEServers: servers, Audio: audio, Events: mediaedge.EdgeEvents{ LocalCandidate: func(candidate *webrtc.ICECandidateInit) { session.emit(Event{Type: "publication-candidate", ShareID: session.shareID, + Current: current, PublicationGeneration: generation, ConnectionID: connectionID, Candidate: candidate}) }, ConnectionState: func(state webrtc.PeerConnectionState, _ *mediaedge.SelectedPair) { session.emit(Event{Type: "publication-state", ShareID: session.shareID, + Current: current, PublicationGeneration: generation, ConnectionID: connectionID, State: state.String()}) }, }, @@ -39,12 +47,13 @@ func (session *Session) PreparePublication(generation, connectionID string, serv session.mu.Lock() if session.closed || len(session.publications) >= 2 || session.publications[key] != nil { session.mu.Unlock() - _ = publication.Close() + _ = created.Close() return webrtc.SessionDescription{}, mediaedge.PublicationMedia{}, errors.New("native publication changed") } if session.publications == nil { session.publications = make(map[publicationKey]*mediaedge.Publication) } + publication = created session.publications[key] = publication session.mu.Unlock() offer, err := publication.CreateOffer() @@ -55,6 +64,14 @@ func (session *Session) PreparePublication(generation, connectionID string, serv return offer, publication.Media(), nil } +func (session *Session) ownsPublication(key publicationKey, publication *mediaedge.Publication) bool { + session.mu.Lock() + defer session.mu.Unlock() + // A current encoder failure still needs its terminal event. Retirement is + // determined by map ownership, not by the transport's closed state. + return !session.closed && publication != nil && session.publications[key] == publication +} + func (session *Session) Publication(generation, connectionID string) (*mediaedge.Publication, error) { session.mu.Lock() defer session.mu.Unlock() diff --git a/internal/app/nativehost/publication_test.go b/internal/app/nativehost/publication_test.go index f656838..f6612da 100644 --- a/internal/app/nativehost/publication_test.go +++ b/internal/app/nativehost/publication_test.go @@ -2,6 +2,7 @@ package nativehost import ( "testing" + "time" "github.com/TNTcraftHIM/Piik/internal/app/mediaedge" ) @@ -23,7 +24,8 @@ func TestPublicationGenerationFencesNativeControl(t *testing.T) { if err = source.ConfigureOutputs([]uint32{300_000}); err != nil { t.Fatal(err) } - session := &Session{shareID: "share_12345678", engine: engine, source: source, ctx: t.Context(), events: make(chan Event, 64)} + events := make(chan Event, 64) + session := &Session{shareID: "share_12345678", engine: engine, source: source, ctx: t.Context(), events: events} _, media, err := session.PreparePublication("generation_12345678", "connection_12345678", nil) if err != nil || media.Codec != "vp8" || len(media.Layers) != 1 { t.Fatalf("publication did not reuse capture metadata: %+v %v", media, err) @@ -45,8 +47,38 @@ func TestPublicationGenerationFencesNativeControl(t *testing.T) { if _, err = session.Publication("generation_12345678", "connection_12345678"); err != nil { t.Fatal("candidate rollback retired current publication") } + key := publicationKey{"generation_12345678", "connection_12345678"} + old, _ := session.Publication(key.generation, key.connectionID) + if err = old.Close(); err != nil || !session.ownsPublication(key, old) { + t.Fatal("current transport failure must still reach its owner", err) + } + var closedEvent Event + deadline := time.NewTimer(3 * time.Second) + defer deadline.Stop() + for closedEvent.Current == nil { + select { + case event := <-events: + if event.Type == "publication-state" && event.State == "closed" && + event.PublicationGeneration == key.generation && event.ConnectionID == key.connectionID { + if event.Current == nil || !event.Current() { + t.Fatal("current transport failure lost its delivery authority") + } + closedEvent = event + } + case <-deadline.C: + t.Fatal("publication did not emit its terminal state") + } + } session.ClosePublication("generation_12345678", "connection_12345678") if _, err = session.Publication("generation_12345678", "connection_12345678"); err == nil { t.Fatal("closed publication remained accessible") } + if _, _, err = session.PreparePublication(key.generation, key.connectionID, nil); err != nil { + t.Fatal(err) + } + defer session.ClosePublication(key.generation, key.connectionID) + replacement, _ := session.Publication(key.generation, key.connectionID) + if closedEvent.Current() || session.ownsPublication(key, old) || !session.ownsPublication(key, replacement) { + t.Fatal("same-key replacement accepted the retired publication") + } } diff --git a/internal/app/nativehost/session.go b/internal/app/nativehost/session.go index e62c198..513debf 100644 --- a/internal/app/nativehost/session.go +++ b/internal/app/nativehost/session.go @@ -42,6 +42,7 @@ type CaptureState struct { } type Event struct { + Current func() bool Type string ShareID string ConnectionID string @@ -1005,6 +1006,7 @@ func (session *Session) runQuality() { for key, publication := range publications { if sample, ok := publication.QualitySample(now); ok { session.emit(Event{Type: "publication-quality", ShareID: session.shareID, + Current: func() bool { return session.ownsPublication(key, publication) }, PublicationGeneration: key.generation, ConnectionID: key.connectionID, PublicationQuality: &sample}) } } diff --git a/internal/app/nativeviewer/session.go b/internal/app/nativeviewer/session.go index 60ae381..c48ad5d 100644 --- a/internal/app/nativeviewer/session.go +++ b/internal/app/nativeviewer/session.go @@ -12,6 +12,7 @@ import ( ) type Event struct { + Current func() bool Type string ShareID string ConnectionID string @@ -104,19 +105,36 @@ func (session *Session) UpdateProfile(profile nativecapture.VideoProfile) error return nil } +type OfferResult struct { + Answer webrtc.SessionDescription + Audio bool + Codec string + Reused bool +} + func (session *Session) AcceptOffer( connectionID string, offer webrtc.SessionDescription, iceServers []webrtc.ICEServer, -) (webrtc.SessionDescription, bool, string, error) { + reuse bool, +) (OfferResult, error) { if connectionID == "" || len(connectionID) > 256 { - return webrtc.SessionDescription{}, false, "", errors.New("native Viewer receiver identity is invalid") + return OfferResult{}, errors.New("native Viewer receiver identity is invalid") } session.mu.Lock() closed := session.closed session.mu.Unlock() if closed { - return webrtc.SessionDescription{}, false, "", errors.New("native Viewer session is closed") + return OfferResult{}, errors.New("native Viewer session is closed") + } + if receiver := session.receiver(connectionID); reuse && receiver != nil { + answer, reused, err := receiver.Renegotiate(offer, iceServers) + if err != nil { + return OfferResult{}, err + } + if reused { + return OfferResult{Answer: answer, Audio: receiver.HasAudio(), Codec: receiver.Codec(), Reused: true}, nil + } } session.CloseReceiver(connectionID) receiver, answer, err := session.engine.NewReceiver(mediaedge.ReceiverOptions{ @@ -125,20 +143,23 @@ func (session *Session) AcceptOffer( EdgeCapacity: session.edgeCapacity, Relay: session.relay, Events: mediaedge.ReceiverEvents{ - LocalCandidate: func(candidate *webrtc.ICECandidateInit) { + LocalCandidate: func(candidate *webrtc.ICECandidateInit, current func() bool) { session.emit(Event{ - Type: "edge-candidate", ShareID: session.shareID, + Current: current, + Type: "edge-candidate", ShareID: session.shareID, ConnectionID: connectionID, Candidate: candidate, }) }, - ConnectionState: func(state webrtc.PeerConnectionState, pair *mediaedge.SelectedPair) { + ConnectionState: func(state webrtc.PeerConnectionState, pair *mediaedge.SelectedPair, current func() bool) { session.emit(Event{ - Type: "edge-state", ShareID: session.shareID, + Current: current, + Type: "edge-state", ShareID: session.shareID, ConnectionID: connectionID, State: state.String(), }) if pair != nil { session.emit(Event{ - Type: "edge-path", ShareID: session.shareID, + Current: current, + Type: "edge-path", ShareID: session.shareID, ConnectionID: connectionID, LocalType: pair.Local.String(), RemoteType: pair.Remote.String(), @@ -149,24 +170,24 @@ func (session *Session) AcceptOffer( }, }) if err != nil { - return webrtc.SessionDescription{}, false, "", err + return OfferResult{}, err } session.mu.Lock() if session.closed || session.receivers[connectionID] != nil { session.mu.Unlock() _ = receiver.Close() - return webrtc.SessionDescription{}, false, "", errors.New("native Viewer receiver is unavailable") + return OfferResult{}, errors.New("native Viewer receiver is unavailable") } if session.profile != nil { if err = receiver.Source().SetRelayProfile(*session.profile); err != nil { session.mu.Unlock() _ = receiver.Close() - return webrtc.SessionDescription{}, false, "", err + return OfferResult{}, err } } session.receivers[connectionID] = receiver session.mu.Unlock() - return answer, receiver.HasAudio(), receiver.Codec(), nil + return OfferResult{Answer: answer, Audio: receiver.HasAudio(), Codec: receiver.Codec()}, nil } func (session *Session) AddReceiverCandidate( diff --git a/internal/app/nativeviewer/session_test.go b/internal/app/nativeviewer/session_test.go index 33a1876..b5aafbe 100644 --- a/internal/app/nativeviewer/session_test.go +++ b/internal/app/nativeviewer/session_test.go @@ -4,8 +4,68 @@ import ( "testing" "github.com/TNTcraftHIM/Piik/internal/app/nativecapture" + "github.com/pion/webrtc/v4" ) +func TestOfferReuseKeepsSourceAndChildOwnership(t *testing.T) { + offer := func(audio bool) webrtc.SessionDescription { + t.Helper() + peer, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = peer.Close() }) + if _, err = peer.AddTransceiverFromKind(webrtc.RTPCodecTypeVideo, + webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}); err != nil { + t.Fatal(err) + } + if audio { + if _, err = peer.AddTransceiverFromKind(webrtc.RTPCodecTypeAudio, + webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}); err != nil { + t.Fatal(err) + } + } + value, err := peer.CreateOffer(nil) + if err != nil { + t.Fatal(err) + } + return value + } + for _, reuse := range []bool{false, true} { + session, err := Start(t.Context(), Options{ShareID: "viewer-session", EdgeCapacity: 2}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = session.Close() }) + for _, audio := range []bool{false, true, false} { + input := offer(audio) + result, err := session.AcceptOffer("upstream", input, nil, reuse) + if err != nil || result.Reused || result.Audio != audio { + t.Fatalf("new media shape: %+v, %v", result, err) + } + if session.edge("local-playback") != nil { + t.Fatal("incompatible source retained its old edge") + } + receiver := session.receiver("upstream") + if _, err = session.PrepareLocalEdge("upstream", "local-playback"); err != nil { + t.Fatal(err) + } + child := session.edge("local-playback") + result, err = session.AcceptOffer("upstream", input, nil, reuse) + if err != nil || result.Reused != reuse { + t.Fatalf("repeat offer: %+v, %v", result, err) + } + if reuse { + if session.receiver("upstream") != receiver || session.edge("local-playback") != child { + t.Fatal("compatible offer retired the existing source or child") + } + } else if session.receiver("upstream") == receiver || session.edge("local-playback") != nil { + t.Fatal("legacy request did not retain its replacement behavior") + } + } + } +} + func TestViewerProfileHasNoImplicitCaptureDefault(t *testing.T) { session := &Session{} if session.profile != nil || session.UpdateProfile(nativecapture.VideoProfile{}) == nil { diff --git a/native/capture/windows/build.ps1 b/native/capture/windows/build.ps1 index d012ad6..5ed8642 100644 --- a/native/capture/windows/build.ps1 +++ b/native/capture/windows/build.ps1 @@ -56,12 +56,15 @@ Invoke-CaptureBuild ('{0} /out:"{1}" {2} "{3}" {4}' -f $linkCommand,$executableP if ($Check) { $vp8Object = Join-Path $outputPath 'vp8_encoder.obj' Invoke-CaptureBuild ('cl.exe {0} /DNDEBUG "{1}" /Fo:"{2}"' -f $compileFlags,(Join-Path $helperDirectory 'vp8_encoder.cpp'),$vp8Object) - foreach ($name in @('capture_geometry', 'capture_control', 'output_worker')) { + foreach ($name in @('capture_geometry', 'capture_control', 'capture_target', 'output_worker')) { $source = Join-Path $helperDirectory ($name + '.test.cpp') $object = Join-Path $outputPath ($name + '.test.obj') $executable = Join-Path $outputPath ($name + '.test.exe') Invoke-CaptureBuild ('cl.exe {0} "{1}" /Fo:"{2}"' -f $compileFlags,$source,$object) $codecObjects = if ($name -eq 'capture_control') { '"{0}" "{1}"' -f $vp8Object,$webrtc.Library } else { '' } + if ($name -eq 'capture_target') { + $codecObjects = '"{0}"' -f (Join-Path $outputPath 'capture_target.obj') + } if ($name -eq 'output_worker') { $codecObjects = (@('h264_encoder', 'process_audio', 'capture_target') | ForEach-Object { '"{0}"' -f (Join-Path $outputPath ($_ + '.obj')) diff --git a/native/capture/windows/capture_target.cpp b/native/capture/windows/capture_target.cpp index 9fec32e..3b1de55 100644 --- a/native/capture/windows/capture_target.cpp +++ b/native/capture/windows/capture_target.cpp @@ -577,11 +577,11 @@ HRESULT CaptureWithWgc(TargetKind kind, UINT64 source_id, } border.Close(); try { - session.Close(); + if (session) session.Close(); } catch (...) { } try { - pool.Close(); + if (pool) pool.Close(); } catch (...) { } // Release all WinRT/D3D wrappers before leaving the COM apartment. Their diff --git a/native/capture/windows/capture_target.test.cpp b/native/capture/windows/capture_target.test.cpp new file mode 100644 index 0000000..296a53b --- /dev/null +++ b/native/capture/windows/capture_target.test.cpp @@ -0,0 +1,30 @@ +#include "capture_target.h" + +#include +#include + +int main() { + // Enumeration and preview are separate operations. A once-listed window may + // become hidden before WGC creates its item. Unsupported capture may return + // an advisory error or a fallback preview, but cleanup must never crash. + HWND window = CreateWindowExW(0, L"STATIC", L"Piik hidden preview fixture", + WS_OVERLAPPEDWINDOW, 0, 0, 320, 180, nullptr, + nullptr, GetModuleHandleW(nullptr), nullptr); + assert(window != nullptr); + FILETIME created{}, exited{}, kernel{}, user{}; + assert(GetProcessTimes(GetCurrentProcess(), &created, &exited, &kernel, &user)); + const UINT64 creation_time = (static_cast(created.dwHighDateTime) << 32) | + created.dwLowDateTime; + HANDLE previous_output = GetStdHandle(STD_OUTPUT_HANDLE); + HANDLE sink = CreateFileW(L"NUL", GENERIC_WRITE, FILE_SHARE_WRITE, nullptr, + OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, nullptr); + assert(sink != INVALID_HANDLE_VALUE); + assert(SetStdHandle(STD_OUTPUT_HANDLE, sink)); + piik::capture::WriteSourcePreview(piik::capture::TargetKind::window, + static_cast(reinterpret_cast(window)), + GetCurrentProcessId(), creation_time); + SetStdHandle(STD_OUTPUT_HANDLE, previous_output); + CloseHandle(sink); + DestroyWindow(window); + std::cout << "Capture preview failure cleanup passed.\n"; +} diff --git a/src/client/App.tsx b/src/client/App.tsx index 96b7ff8..0e87a22 100644 --- a/src/client/App.tsx +++ b/src/client/App.tsx @@ -18,7 +18,9 @@ import type { RuntimeCapabilities } from "../shared/protocol"; import { parseAppRoute, readViewerRoute, + watchViewerInvites, takeClientLaunchBootstrap, + type ViewerRoute, } from "./lib/session"; import { AppHeader } from "./components/living/Header"; import { Btn, Pill } from "./components/living/primitives"; @@ -53,7 +55,7 @@ if (clientLaunchBootstrap?.presentation) { clientLaunchBootstrap.presentation.explicit?.includes("copy") ?? false); } const clientAccessBootstrap = clientLaunchBootstrap?.accessToken ?? null; -const viewerRoute = appRoute.kind === "viewer" ? readViewerRoute() : null; +const initialViewerRoute = appRoute.kind === "viewer" ? readViewerRoute() : null; const hostPageModule = appRoute.kind === "host" ? import("./pages/HostPage") : null; const joinPageModule = @@ -143,6 +145,10 @@ class RouteBoundary extends Component< } function AppRoute() { + const [viewerRoute, setViewerRoute] = useState(initialViewerRoute); + useEffect(() => { + if (appRoute.kind === "viewer") return watchViewerInvites(setViewerRoute); + }, []); if (PlaybackPreviewPage && window.location.pathname === "/__playback-preview") { return ; } @@ -158,11 +164,13 @@ function AppRoute() { if (appRoute.kind === "viewer" && viewerRoute) { return viewerRoute.viewerGrant ? ( ) : ( - + ); } if (appRoute.kind === "host") { @@ -246,11 +254,10 @@ function StaticRoute({ function SiteAccessGate({ surface, - invalidInvite, + viewerRoute, }: { surface: "host" | "join" | "viewer"; - /** The URL carried an invite fragment that could not be used. */ - invalidInvite?: boolean; + viewerRoute?: ViewerRoute; }) { const { t, vis } = useCopy(); const [access, setAccess] = useState({ kind: "checking" }); @@ -274,7 +281,7 @@ function SiteAccessGate({ setCapabilities(nextCapabilities); const next = stateFromStatus(status); setAccess( - next.kind === "required" && invalidInvite + next.kind === "required" && viewerRoute?.invalidGrant ? { kind: "required", error: t("viewer.msg.invalidInvite") } : next, ); diff --git a/src/client/lib/session.ts b/src/client/lib/session.ts index 95f0ffc..e19b27c 100644 --- a/src/client/lib/session.ts +++ b/src/client/lib/session.ts @@ -488,6 +488,18 @@ export function readViewerRoute(): ViewerRoute | null { return { roomId }; } +// Following another invitation to this page is a fragment navigation, not a +// document load. Consume it through the same authority owner as initial entry. +export function watchViewerInvites(changed: (route: ViewerRoute) => void): () => void { + const onHashChange = () => { + if (!window.location.hash.startsWith("#v=")) return; + const route = readViewerRoute(); + if (route) changed(route); + }; + window.addEventListener("hashchange", onHashChange); + return () => window.removeEventListener("hashchange", onHashChange); +} + export function replaceViewerInvite( roomId: string, inviteUrl: string | null, diff --git a/src/client/media/host-sfu-route.ts b/src/client/media/host-sfu-route.ts index b9e1d76..35f4092 100644 --- a/src/client/media/host-sfu-route.ts +++ b/src/client/media/host-sfu-route.ts @@ -593,7 +593,8 @@ export class HostSfuRoute { const pending = this.pending; this.pending = null; pending.failed = true; - void disconnectPublisher(pending.publisher); + const closing = disconnectPublisher(pending.publisher); + void this.queueTransition(() => closing); } private async retireActive(except?: HostPublisherSlot): Promise { diff --git a/src/client/native/client.ts b/src/client/native/client.ts index 7710b62..6db3c95 100644 --- a/src/client/native/client.ts +++ b/src/client/native/client.ts @@ -495,6 +495,7 @@ export class NativeClient { answer: { type: "answer"; sdp: string }; audio: boolean; codec: NativeVideoCodec; + reused: boolean; }> { if (offer.type !== "offer" || !offer.sdp) { throw new Error("Native receiver requires an SDP offer"); @@ -504,6 +505,7 @@ export class NativeClient { { shareId, connectionId, + ...(this.health.nativeMedia.receiverReuse ? { reuseReceiver: true } : {}), edgeCapacity, iceServers: iceConfig.iceServers.map((server) => ({ urls: Array.isArray(server.urls) ? server.urls : [server.urls], @@ -522,6 +524,7 @@ export class NativeClient { answer: { type: "answer", sdp: response.sdp }, audio: response.audio, codec: response.codec, + reused: response.reused === true, }; } diff --git a/src/client/native/native-sfu-publisher.ts b/src/client/native/native-sfu-publisher.ts index b8d6a89..314f4fd 100644 --- a/src/client/native/native-sfu-publisher.ts +++ b/src/client/native/native-sfu-publisher.ts @@ -26,6 +26,7 @@ export class NativeSfuPublisher implements HostPublisherTransport { private unsubscribe: (() => void) | null = null; private prepared = false; private closed = false; + private closing: Promise | null = null; private failure: SfuPublisherFailureStage | null = null; constructor( @@ -42,7 +43,6 @@ export class NativeSfuPublisher implements HostPublisherTransport { async connect(config: SfuConnectionConfig): Promise { if (this.config || this.closed) return false; this.config = config; - this.unsubscribe = this.client.onEvent((event) => this.onEvent(event)); return true; } @@ -58,7 +58,13 @@ export class NativeSfuPublisher implements HostPublisherTransport { async activate(): Promise { const config = this.config; - if (!config || this.prepared || this.closed) return false; + if (!config || this.prepared || this.closed) { + await this.closing?.catch(() => undefined); + return false; + } + // Activation runs after the route's retirement barrier. Subscribing in + // connect would let an old same-key publication reach this new instance. + this.unsubscribe = this.client.onEvent((event) => this.onEvent(event)); try { const result = await this.client.preparePublication( this.shareId, @@ -67,11 +73,7 @@ export class NativeSfuPublisher implements HostPublisherTransport { this.iceConfig, ); if (this.closed) { - await this.client.closePublication( - this.shareId, - config.publicationGeneration, - config.connectionId, - ); + await this.disconnect(); return false; } this.pending.unshift({ kind: "description", ...result }); @@ -79,7 +81,7 @@ export class NativeSfuPublisher implements HostPublisherTransport { this.flush(); return !this.closed; } catch { - this.fail("video-publish"); + await this.fail("video-publish"); return false; } } @@ -94,26 +96,32 @@ export class NativeSfuPublisher implements HostPublisherTransport { try { await this.client.acceptPublicationSignal(this.shareId, message); } catch { - this.fail("transport"); + await this.fail("transport"); } } async updateProfile(): Promise { const config = this.config; - if (!config || !this.prepared || this.closed) return false; + if (!config || !this.prepared || this.closed) { + await this.closing?.catch(() => undefined); + return false; + } try { const media = await this.client.publicationMedia( this.shareId, config.publicationGeneration, config.connectionId, ); - if (this.closed) return false; + if (this.closed) { + await this.closing?.catch(() => undefined); + return false; + } this.pending = this.pending.filter(({ kind }) => kind !== "media"); this.pending.push({ kind: "media", media }); this.flush(); return !this.closed; } catch { - this.fail("source"); + await this.fail("source"); return false; } } @@ -130,19 +138,19 @@ export class NativeSfuPublisher implements HostPublisherTransport { return true; } - async disconnect(): Promise { - if (this.closed) return; + disconnect(): Promise { + if (this.closing) return this.closing; this.closed = true; this.unsubscribe?.(); this.unsubscribe = null; this.pending = []; const config = this.config; - if (config) - await this.client.closePublication( + this.closing = config ? this.client.closePublication( this.shareId, config.publicationGeneration, config.connectionId, - ); + ) : Promise.resolve(); + return this.closing; } private onEvent(event: NativeClientEvent): void { @@ -179,7 +187,7 @@ export class NativeSfuPublisher implements HostPublisherTransport { }); } else if (event.type === "publication-candidate" && event.candidate) { if (this.pending.length >= 64) { - this.fail("transport"); + void this.fail("transport"); return; } this.pending.push({ kind: "candidate", candidate: event.candidate }); @@ -190,7 +198,7 @@ export class NativeSfuPublisher implements HostPublisherTransport { event.state === "failed" || event.state === "closed") ) { - this.fail("transport"); + void this.fail("transport"); } } @@ -211,10 +219,13 @@ export class NativeSfuPublisher implements HostPublisherTransport { } } - private fail(stage: SfuPublisherFailureStage): void { - if (this.closed || this.failure) return; + private async fail(stage: SfuPublisherFailureStage): Promise { + if (this.closed || this.failure) { + await this.closing?.catch(() => undefined); + return; + } this.failure = stage; - void this.disconnect().catch(() => undefined); + await this.disconnect().catch(() => undefined); this.events.onDisconnected(); } } diff --git a/src/client/native/native-viewer-peer.ts b/src/client/native/native-viewer-peer.ts index 5b9671b..3614a0d 100644 --- a/src/client/native/native-viewer-peer.ts +++ b/src/client/native/native-viewer-peer.ts @@ -54,7 +54,7 @@ export class NativeCapableViewerPeer implements ViewerMediaPeer { constructor( iceConfig: PeerIceConfig, private readonly events: ViewerPeerEvents, - private readonly options: ViewerPeerOptions, + private options: ViewerPeerOptions, private readonly acquireNativeClient: () => Promise, private readonly onNativeUnavailable: () => void, private readonly nativeSessionId: string, @@ -146,7 +146,11 @@ export class NativeCapableViewerPeer implements ViewerMediaPeer { : null; } stopDecodedFrameProof(): void { this.backend?.stopDecodedFrameProof(); } - activatePreparedRoute(): void { this.backend?.activatePreparedRoute(); } + activatePreparedRoute(): void { + // Promotion belongs to this route, including any later replacement backend. + this.options = { ...this.options, recoveryOwner: "viewer" }; + this.backend?.activatePreparedRoute(); + } dispose(): void { if (this.disposed) return; @@ -206,6 +210,7 @@ class NativeViewerPeer implements ViewerMediaPeer { private sourceGeneration = 0; private sourceReady = false; private sourceCodec: NativeVideoCodec | null = null; + private pendingOffer: { bridge: NativeMediaBridge | null; bridgeFailed: boolean } | null = null; get source(): NativeViewerSource | null { return this.connectionId && this.sourceReady && this.sourceCodec @@ -345,6 +350,8 @@ class NativeViewerPeer implements ViewerMediaPeer { const connectionId = payload.connectionId; const identityChanged = this.parentPeerId !== parentPeerId || this.connectionId !== connectionId; + const retainSource = !identityChanged && this.sourceReady && !this.failureNotified && + this.client.health?.nativeMedia.receiverReuse === true; if (identityChanged) { const previous = this.connectionId; this.disposeBridge(); @@ -357,24 +364,26 @@ class NativeViewerPeer implements ViewerMediaPeer { this.parentPeerId = parentPeerId; this.connectionId = connectionId; this.resetAutomaticRecovery(); - } else { + } else if (!retainSource) { this.disposeBridge(); - this.localCandidates?.discard(); } - this.state = "connecting"; - this.failureNotified = false; - this.sourceGeneration += 1; - this.sourceReady = false; - this.sourceCodec = null; - this.statsAccumulator = createStatsAccumulator(); - this.snapshot = { - peerId: parentPeerId, - connectionId, - connectionState: "connecting", - iceConnectionState: "checking", - metrics: { ...EMPTY_METRICS }, - error: null, - }; + this.localCandidates?.discard(); + if (!retainSource) { + this.state = "connecting"; + this.failureNotified = false; + this.sourceGeneration += 1; + this.sourceReady = false; + this.sourceCodec = null; + this.statsAccumulator = createStatsAccumulator(); + this.snapshot = { + peerId: parentPeerId, + connectionId, + connectionState: "connecting", + iceConnectionState: "checking", + metrics: { ...EMPTY_METRICS }, + error: null, + }; + } const predictionEnabled = this.natPredictionEnabled && natPredictionSurveyUrls( @@ -410,13 +419,16 @@ class NativeViewerPeer implements ViewerMediaPeer { : null; this.localCandidates = candidates; this.emit(); - const result = await this.client.receiveOffer( - this.sessionId, - connectionId, - payload.description, - nativeIceConfig, - this.edgeCapacity, - ); + const pending = { bridge: retainSource ? this.bridge : null, bridgeFailed: false }; + this.pendingOffer = pending; + let result: Awaited>; + try { + result = await this.client.receiveOffer( + this.sessionId, connectionId, payload.description, nativeIceConfig, this.edgeCapacity, + ); + } finally { + if (this.pendingOffer === pending) this.pendingOffer = null; + } if ( this.disposed || this.failureNotified || @@ -428,6 +440,26 @@ class NativeViewerPeer implements ViewerMediaPeer { ); return; } + if (retainSource && result.reused) { + // Renegotiation belongs to the current receiver. Network ICE recovery + // may pause upstream RTP, but it does not retire playback or relay edges. + if (pending.bridgeFailed) { + this.handleBridgeFailure(connectionId); + } else if (!this.events.sendSignal(parentPeerId, { + kind: "description", connectionId, description: result.answer, + })) { + this.scheduleRecoveryDeadline(); + } + return; + } + if (retainSource) { + // The App replaced an incompatible media shape. Old bridge failure + // during that decision must not close the newly installed receiver. + this.disposeBridge(); + this.sourceGeneration += 1; + this.statsAccumulator = createStatsAccumulator(); + if (this.snapshot) this.snapshot = { ...this.snapshot, metrics: { ...EMPTY_METRICS } }; + } this.sourceCodec = result.codec; this.sourceReady = true; if (!this.events.sendSignal(parentPeerId, { @@ -454,7 +486,7 @@ class NativeViewerPeer implements ViewerMediaPeer { const bridge = new NativeMediaBridge( this.sessionId, this.client, - () => this.handleBridgeFailure(connectionId), + () => this.handlePlaybackFailure(bridge, connectionId), result.audio, connectionId, ); @@ -467,10 +499,16 @@ class NativeViewerPeer implements ViewerMediaPeer { this.events.onStream(stream); this.startObservation(bridge, connectionId); }).catch(() => { - if (this.bridge === bridge) this.handleBridgeFailure(connectionId); + this.handlePlaybackFailure(bridge, connectionId); }); } + private handlePlaybackFailure(bridge: NativeMediaBridge, connectionId: string): void { + if (this.bridge !== bridge) return; + if (this.pendingOffer?.bridge === bridge) this.pendingOffer.bridgeFailed = true; + else this.handleBridgeFailure(connectionId); + } + private onEvent(event: NativeClientEvent): void { if (this.disposed || this.failureNotified || event.shareId !== this.sessionId) { return; diff --git a/src/client/native/wire.ts b/src/client/native/wire.ts index 2ab2c10..229244d 100644 --- a/src/client/native/wire.ts +++ b/src/client/native/wire.ts @@ -32,6 +32,7 @@ export const nativeHealthSchema = nativeDiscoveryIdentitySchema.extend({ .regex(/^[A-Za-z0-9_-]+$/), nativeMedia: z .object({ + receiverReuse: z.boolean().default(false), video: z.boolean().default(false), processAudio: z.boolean().default(false), systemAudio: z.boolean().default(false), @@ -195,6 +196,7 @@ export const receiveAnswerResponseSchema = z .max(48 * 1024), audio: z.boolean(), codec: nativeVideoCodecSchema, + reused: z.boolean().optional(), }) .strict(); diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 551f11a..6691b3b 100644 --- a/tests/client-state.test.ts +++ b/tests/client-state.test.ts @@ -33,6 +33,7 @@ import { readPreferredRoomId, readViewerGrant, readViewerRoute, + watchViewerInvites, replaceViewerInvite, roomRouteForExplicitEntry, roomRouteFromInput, @@ -564,6 +565,47 @@ describe("client session identity", () => { expect(replaceState).not.toHaveBeenCalled(); }); + it("consumes renewed same-page invitations and ignores unrelated fragments", () => { + const events = new EventTarget(); + const values = new Map(); + const location = new URL("https://share.test/r/1234"); + vi.stubGlobal("window", { + location, + addEventListener: events.addEventListener.bind(events), + removeEventListener: events.removeEventListener.bind(events), + history: { state: null, replaceState: (_state: unknown, _unused: string, url: string) => { + location.href = new URL(url, location).href; + } }, + sessionStorage: { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), + removeItem: (key: string) => values.delete(key), + }, + }); + const changed = vi.fn(); + const stop = watchViewerInvites(changed); + const first = `${"g".repeat(21)}g`; + const renewed = `${"h".repeat(21)}w`; + for (const grant of [first, renewed]) { + location.hash = `v=${grant}`; + events.dispatchEvent(new Event("hashchange")); + expect(changed).toHaveBeenLastCalledWith({ roomId: "1234", viewerGrant: grant }); + expect(readViewerGrant("1234")).toBe(grant); + expect(location.hash).toBe(""); + } + location.hash = "section"; + events.dispatchEvent(new Event("hashchange")); + expect(changed).toHaveBeenCalledTimes(2); + location.hash = "v=invalid"; + events.dispatchEvent(new Event("hashchange")); + expect(changed).toHaveBeenLastCalledWith({ roomId: "1234", invalidGrant: true }); + expect(readViewerGrant("1234")).toBeNull(); + stop(); + location.hash = `v=${renewed}`; + events.dispatchEvent(new Event("hashchange")); + expect(changed).toHaveBeenCalledTimes(3); + }); + it("persists a rotated Viewer invitation across reload and clears it on revoke", () => { const values = new Map(); vi.stubGlobal("window", { diff --git a/tests/native-client.test.ts b/tests/native-client.test.ts index 3b6979d..4eb3448 100644 --- a/tests/native-client.test.ts +++ b/tests/native-client.test.ts @@ -355,11 +355,12 @@ describe("native App private wire", () => { ...health, futureDescription: "ignored", nativeMedia: { video: true, futureFeature: true }, })).toEqual({ ...health, nativeMedia: { + receiverReuse: false, video: true, processAudio: false, systemAudio: false, captureBorderControl: false, hardwareH264: false, softwareVP8: false, }, }); expect(nativeHealthSchema.parse({ ...health, nativeMedia: undefined }).nativeMedia) - .toEqual({ video: false, processAudio: false, systemAudio: false, captureBorderControl: false, hardwareH264: false, softwareVP8: false }); + .toEqual({ receiverReuse: false, video: false, processAudio: false, systemAudio: false, captureBorderControl: false, hardwareH264: false, softwareVP8: false }); for (const invalid of [ { protocol: 0 }, { protocol: 9.5 }, { protocol: Number.MAX_SAFE_INTEGER + 1 }, { service: "other" }, { port: NATIVE_CLIENT_PORT_END + 1 }, { instanceToken: "short" }, @@ -423,6 +424,42 @@ describe("native App private wire", () => { } finally { client!.close(); } }); + it.each([false, true])("requests receiver reuse only from a capable App (%s)", async (supported) => { + const requests: Record[] = []; + class Socket extends EventTarget { + static readonly OPEN = 1; + static readonly CLOSING = 2; + readyState = Socket.OPEN; + protocol = `piik-client-v9.${health.instanceToken}`; + constructor() { super(); queueMicrotask(() => this.dispatchEvent(new Event("open"))); } + close() { this.readyState = Socket.CLOSING; } + send(payload: string) { + const request = JSON.parse(payload) as Record; + requests.push(request); + queueMicrotask(() => this.dispatchEvent(new MessageEvent("message", { data: JSON.stringify({ + version: 9, id: request.id, ...(request.type === "hello" ? { type: "ready" } : { + type: "receive-answer", shareId: request.shareId, connectionId: request.connectionId, + sdp: "answer", audio: false, codec: "vp8", ...(supported ? { reused: true } : {}), + }), + }) }))); + } + } + vi.stubGlobal("WebSocket", Socket); + vi.stubGlobal("window", { setTimeout, clearTimeout }); + vi.stubGlobal("fetch", vi.fn(async () => new Response(JSON.stringify({ + ...health, nativeMedia: { ...health.nativeMedia, ...(supported ? { receiverReuse: true } : {}) }, + })))); + const client = (await NativeClient.connect())!; + try { + const result = await client.receiveOffer("share_123456", "connection_1234", { type: "offer", sdp: "offer" }, { iceServers: [] }, 2); + expect(requests.at(-1)).toEqual({ version: 9, id: expect.any(String), type: "receive-offer", + shareId: "share_123456", connectionId: "connection_1234", sdp: "offer", iceServers: [], edgeCapacity: 2, + ...(supported ? { reuseReceiver: true } : {}), + }); + expect(result.reused).toBe(supported); + } finally { client.close(); } + }); + it("keeps 64-bit Windows identities as exact decimal strings", () => { expect( nativeCaptureTargetSchema.safeParse({ diff --git a/tests/native-sfu-publisher.test.ts b/tests/native-sfu-publisher.test.ts index 4e73792..95ecfaa 100644 --- a/tests/native-sfu-publisher.test.ts +++ b/tests/native-sfu-publisher.test.ts @@ -4,6 +4,47 @@ import type { NativeClientEvent } from "../src/client/native/wire"; import type { SfuSignalMessage } from "../src/shared/protocol"; describe("Native SFU publication", () => { + it.each(["transport", "pending-prepare"])("retires once before recovery: %s", async mode => { + const config = { revision: 3, publicationGeneration: "publication_generation", connectionId: "publication_connection" }; + const media = { codec: "vp8" as const, layers: [{ rid: "", width: 320, height: 180, bitrate: 300_000 }], audio: false, audioBitrate: 0 }; + let finishClose!: () => void, finishPrepare!: () => void; + const closing = new Promise(resolve => { finishClose = resolve; }); + const preparing = new Promise(resolve => { finishPrepare = resolve; }); + let listener: ((event: NativeClientEvent) => void) | null = null; + const control = { + preparePublication: vi.fn(async () => { + if (mode === "pending-prepare") await preparing; + return { description: { type: "offer" as const, sdp: "v=0\r\n" }, media }; + }), + publicationMedia: vi.fn(async () => media), acceptPublicationSignal: vi.fn(async () => {}), + closePublication: vi.fn(() => closing), + onEvent: vi.fn((callback: (event: NativeClientEvent) => void) => { + listener = callback; + return () => { listener = null; }; + }), + }; + const onDisconnected = vi.fn(); + const publisher = new NativeSfuPublisher(control, "native_share", { iceServers: [] }, { send: () => true, onDisconnected }); + await publisher.connect(config); + expect(control.onEvent).not.toHaveBeenCalled(); + const activation = publisher.activate(); + if (mode === "transport") { + await activation; + listener!({ version: 9, type: "publication-state", shareId: "native_share", ...config, state: "failed" }); + } + const firstClose = publisher.disconnect(); + expect(publisher.disconnect()).toBe(firstClose); + finishPrepare(); + await Promise.resolve(); + expect(onDisconnected).not.toHaveBeenCalled(); + expect(control.closePublication).toHaveBeenCalledOnce(); + finishClose(); + await firstClose; + await activation; + await Promise.resolve(); + expect(control.closePublication).toHaveBeenCalledOnce(); + expect(onDisconnected).toHaveBeenCalledTimes(mode === "transport" ? 1 : 0); + }); it("forwards existing Native media with exact identity and bounded signaling replay", async () => { const config = { revision: 3, @@ -136,7 +177,7 @@ describe("Native SFU publication", () => { ...config, state: "failed", }); - expect(onDisconnected).toHaveBeenCalledOnce(); + await vi.waitFor(() => expect(onDisconnected).toHaveBeenCalledOnce()); expect(control.closePublication).toHaveBeenCalledWith( "native_share", config.publicationGeneration, diff --git a/tests/native-viewer-peer.test.ts b/tests/native-viewer-peer.test.ts index 7e335e6..b754997 100644 --- a/tests/native-viewer-peer.test.ts +++ b/tests/native-viewer-peer.test.ts @@ -124,7 +124,7 @@ it.each([false, true])("ignores a late Native offer after control loss (disposed }); await vi.waitFor(() => expect(receiveOffer).toHaveBeenCalledOnce()); onClose(); - resolveOffer({ answer: { type: "answer", sdp: "v=0\r\n" }, audio: false, codec: "vp8" }); + resolveOffer({ answer: { type: "answer", sdp: "v=0\r\n" }, audio: false, codec: "vp8", reused: false }); await accepting; expect(updates.at(-1)?.connectionState).toBe("failed"); @@ -141,6 +141,70 @@ it.each([false, true])("ignores a late Native offer after control loss (disposed } }); +it.each(["reused", "replaced", "answer-lost", "control-lost"])("keeps Native receiver lifetimes coherent through renegotiation (%s)", async (outcome) => { + vi.useFakeTimers(); + vi.stubGlobal("window", globalThis); + vi.stubGlobal("MediaStream", class { getTracks() { return []; } }); + vi.stubGlobal("RTCPeerConnection", class { close() {} }); + const bridges: NativeMediaBridge[] = []; + const bridgeStart = vi.spyOn(NativeMediaBridge.prototype, "start").mockImplementation(async function (this: NativeMediaBridge) { + bridges.push(this); + return {} as MediaStream; + }); + let closeControl!: () => void; + let event!: (event: NativeClientEvent) => void; + const receiveOffer = vi.fn(async () => ({ + answer: { type: "answer" as const, sdp: "answer-sdp" }, audio: false, codec: "vp8" as const, reused: false, + })); + const closeReceiver = vi.fn(async () => undefined); + const closeEdge = vi.fn(async () => undefined); + const client = { + health: { nativeMedia: { receiverReuse: true } }, receiveOffer, closeReceiver, closeEdge, + onEvent: (listener: typeof event) => { event = listener; return () => {}; }, + onClose: (listener: typeof closeControl) => { closeControl = listener; return () => {}; }, + } as unknown as NativeClient; + const sendSignal = vi.fn(() => true); + const onStream = vi.fn(); + const unavailable = vi.fn(); + const peer = new NativeCapableViewerPeer({ iceServers: [] }, { + sendSignal, sendRestartRequest: () => true, onStream, onUpdate: () => {}, + }, {}, async () => client, unavailable, "session", 2); + const offer = (ice: string): SignalPayload => ({ + kind: "description", connectionId: "current", description: { type: "offer", + sdp: `v=0\r\na=ice-ufrag:${ice}\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=sendonly\r\na=rtpmap:96 VP8/90000\r\n` }, + }); + try { + await peer.acceptSignal("parent", offer("first")); + event({ version: 9, type: "edge-state", shareId: "session", connectionId: "current", state: "connected" }); + const source = peer.nativeSource; + receiveOffer.mockImplementationOnce(async () => { + if (outcome === "replaced") { + // Replacement closes the old local edge before its answer reaches JS. + (bridges[0] as unknown as { onFailed(): void }).onFailed(); + } + if (outcome === "control-lost") closeControl(); + return { answer: { type: "answer", sdp: "new-answer-sdp" }, audio: false, codec: "vp8", reused: outcome !== "replaced" }; + }); + if (outcome === "answer-lost") sendSignal.mockReturnValue(false); + await peer.acceptSignal("parent", offer("restarted")); + expect(receiveOffer).toHaveBeenCalledTimes(2); + if (outcome === "control-lost") { + expect(peer.nativeSource).toBeNull(); + expect(unavailable).toHaveBeenCalledOnce(); + } else { + expect(closeReceiver).not.toHaveBeenCalled(); + expect(unavailable).not.toHaveBeenCalled(); + expect(bridgeStart).toHaveBeenCalledTimes(outcome === "replaced" ? 2 : 1); + expect(onStream).toHaveBeenCalledTimes(outcome === "replaced" ? 2 : 1); + expect(peer.nativeSource?.generation).toBe(source!.generation + (outcome === "replaced" ? 1 : 0)); + if (outcome !== "replaced") { + expect(closeEdge).not.toHaveBeenCalled(); + expect(peer.isConnected()).toBe(true); + } + } + } finally { peer.dispose(); } +}); + it("admits one active sending H264 or VP8 video section, not unrelated SDP codec text", () => { const video = (codec: string, direction = "sendonly", port = 9, payload = 96) => [ `m=video ${port} UDP/TLS/RTP/SAVPF ${payload}`, @@ -386,3 +450,46 @@ it.each(["route", "viewer"] as const)("keeps %s bridge failure unavailable acros queued.dispose(); } }); + +it.each([false, true])("keeps committed route recovery across backend replacement (replaced=%s)", async (replaced) => { + vi.useFakeTimers(); + vi.stubGlobal("window", globalThis); + vi.stubGlobal("MediaStream", class { getTracks() { return []; } }); + vi.stubGlobal("RTCPeerConnection", class { close() {} }); + vi.spyOn(NativeMediaBridge.prototype, "start").mockResolvedValue({} as MediaStream); + vi.spyOn(NativeMediaBridge.prototype, "collectMetrics").mockResolvedValue({ ...EMPTY_METRICS }); + vi.spyOn(NativeMediaBridge.prototype, "decodedVideoFrames").mockResolvedValue(null); + let onEvent!: (event: NativeClientEvent) => void; + const client = { + receiveOffer: async () => ({ answer: { type: "answer", sdp: "v=0\r\n" }, audio: false, codec: "vp8" }), + onEvent: (fn: typeof onEvent) => { onEvent = fn; return () => undefined; }, + onClose: () => () => undefined, + closeReceiver: vi.fn(async () => undefined), closeEdge: vi.fn(async () => undefined), + } as unknown as NativeClient; + const restart = vi.fn(() => true); + const exhausted = vi.fn(() => true); + const options = { recoveryOwner: "route" as const }; + const peer = new NativeCapableViewerPeer({ iceServers: [] }, { + sendSignal: () => true, sendRestartRequest: restart, + onStream: () => undefined, onUpdate: () => undefined, onRecoveryExhausted: exhausted, + }, options, async () => client, () => undefined, "session", 2); + const offer = (connectionId: string) => ({ kind: "description" as const, connectionId, + description: { type: "offer" as const, sdp: "v=0\r\nm=video 9 UDP/TLS/RTP/SAVPF 96\r\na=sendonly\r\na=rtpmap:96 VP8/90000\r\n" }}); + const state = (connectionId: string, value: "connected" | "failed") => onEvent({ + version: 9, shareId: "session", connectionId, type: "edge-state", state: value, + }); + try { + await peer.acceptSignal("parent", offer("first")); + state("first", "connected"); + peer.activatePreparedRoute(); + expect(options.recoveryOwner).toBe("route"); // Caller options are not mutable route state. + const connectionId = replaced ? "replacement" : "first"; + if (replaced) { + await peer.acceptSignal("parent", offer(connectionId)); + state(connectionId, "connected"); + } + state(connectionId, "failed"); + expect(restart).toHaveBeenCalledExactlyOnceWith("parent", connectionId, false); + expect(exhausted).not.toHaveBeenCalled(); + } finally { peer.dispose(); } +}); diff --git a/tests/route-transition.test.ts b/tests/route-transition.test.ts index dae8b43..b1bf89f 100644 --- a/tests/route-transition.test.ts +++ b/tests/route-transition.test.ts @@ -975,6 +975,35 @@ describe("minimal route transition contracts", () => { expect(publishers[2]?.activate).toHaveBeenCalledOnce(); }); + it("waits for pending retirement before activating a same-key replacement", async () => { + let finishClose!: () => void; + const closing = new Promise(resolve => { finishClose = resolve; }); + const publishers: ReturnType[] = []; + 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((resolve) => {