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) => {