From 671137af6d512d81e4cbcd9f134fe49d75826926 Mon Sep 17 00:00:00 2001 From: wingha Date: Fri, 16 Oct 2026 21:38:38 +0000 Subject: [PATCH 1/3] fix(huddle): forward subscriber keyframe requests to the publisher The SFU discarded each subscriber's RTPSender and never read its RTCP, so PLI/FIR never reached the publisher and pion's NACK responder never ran. Browsers only encode a keyframe when asked, so one lost packet froze a forwarded picture for good while audio continued. The single keyframe request made at track creation also landed before the subscriber had answered the renegotiation, so that keyframe was dropped by the unbound track. - read RTCP per subscriber track; forward PLI/FIR, throttled to 500ms - request a keyframe on the subscriber's first RTCP (track now bound) - request one when a publisher reconnects onto the same track ids - record the subscription only after AddTrack succeeds Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 11 +++ huddle/sfu/session.go | 4 +- huddle/sfu/sfu_test.go | 184 +++++++++++++++++++++++++++++++++++++++++ huddle/sfu/video.go | 140 +++++++++++++++++++++++++------ 4 files changed, 310 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 24f0388..e3135c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## [Unreleased] + +### Fixed + +- Fixes huddle video freezing for good after a lost packet on `ncli relay`: a + subscriber's keyframe requests (PLI/FIR) now reach the publisher, throttled, + and its NACKs are answered. The SFU also asks for a keyframe once a new + subscriber is actually receiving, and when a publisher reconnects, so the + first picture no longer waits for the encoder's next natural keyframe. Before + this, a 1:1 video call could lose one side's picture while audio carried on. + ## [0.8.0-rc.22] AltZap events move to kinds 9520/9521/9523. diff --git a/huddle/sfu/session.go b/huddle/sfu/session.go index 8402f2e..6be0d30 100644 --- a/huddle/sfu/session.go +++ b/huddle/sfu/session.go @@ -43,7 +43,7 @@ type session struct { hub *videoHub // videoTracks are this subscriber's outbound tracks, keyed by publication. videoMu sync.Mutex - videoTracks map[string]*webrtc.TrackLocalStaticRTP + videoTracks map[string]*subscription // writeMu serialises websocket writes. Signalling originates from three // places -- the read loop, ICE gathering, and renegotiation when a speaker @@ -86,7 +86,7 @@ type session struct { func (s *session) run(ctx context.Context) { defer func() { _ = s.conn.Close() }() s.done = make(chan struct{}) - s.videoTracks = make(map[string]*webrtc.TrackLocalStaticRTP) + s.videoTracks = make(map[string]*subscription) s.conn.SetReadLimit(wire.MaxControlBytes * 16) // SDP is far larger than a huddle control frame if !s.cfg.Enabled { diff --git a/huddle/sfu/sfu_test.go b/huddle/sfu/sfu_test.go index 7bbce2a..f4ed02b 100644 --- a/huddle/sfu/sfu_test.go +++ b/huddle/sfu/sfu_test.go @@ -1009,3 +1009,187 @@ func TestCandidatesBeforeTheOfferAreNotDropped(t *testing.T) { } assertNoDroppedCandidates(t, h) } + +// TestSubscriberKeyframeRequestReachesThePublisher covers the repair path a +// running call depends on. A subscriber that loses packets cannot decode until +// the next keyframe, and a browser encoder only produces one when asked, so the +// subscriber's own PLI has to travel through the SFU to the publisher. When it +// did not, one lost packet froze the picture for good while audio carried on. +func TestSubscriberKeyframeRequestReachesThePublisher(t *testing.T) { + h := newHarness(t, nil) + + publisher, _, videos := connectBrowserWith(t, h, "room-1", alicePriv, false, []string{"camera"}) + camera := videos["camera"] + + pli := make(chan struct{}, 64) + for _, sender := range publisher.pc.GetSenders() { + if track := sender.Track(); track == nil || track.ID() != "camera" { + continue + } + go func(s *webrtc.RTPSender) { + for { + packets, _, err := s.ReadRTCP() + if err != nil { + return + } + for _, p := range packets { + if _, ok := p.(*rtcp.PictureLossIndication); ok { + select { + case pli <- struct{}{}: + default: + } + } + } + } + }(sender) + } + + stop := make(chan struct{}) + defer close(stop) + go func() { + seq, ts := uint16(1), uint32(3000) + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + } + _ = camera.WriteRTP(&rtp.Packet{ + Header: rtp.Header{Version: 2, SequenceNumber: seq, Timestamp: ts, SSRC: 0xCAFE, PayloadType: 96}, + Payload: []byte{0x10, 0xAB, 0xCD, 0xEF}, + }) + seq++ + ts += 3000 + } + }() + + remoteSSRC := make(chan webrtc.SSRC, 1) + subscriber, _, _ := connectBrowserWith(t, h, "room-1", bobPriv, false, nil, + func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) { + if track.Kind() != webrtc.RTPCodecTypeVideo { + return + } + first := true + for { + if _, _, err := track.ReadRTP(); err != nil { + return + } + if first { + first = false + remoteSSRC <- track.SSRC() + } + } + }) + + var ssrc webrtc.SSRC + select { + case ssrc = <-remoteSSRC: + case <-time.After(window): + t.Fatal("the subscriber never received a forwarded video packet") + } + + // Let the setup-time requests (track creation, first receiver report) land + // and fall outside the SFU's throttle, so only the subscriber's own request + // below can satisfy the assertion. + time.Sleep(2 * time.Second) + for drained := false; !drained; { + select { + case <-pli: + default: + drained = true + } + } + + if err := subscriber.pc.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(ssrc)}}); err != nil { + t.Fatalf("subscriber PLI: %v", err) + } + select { + case <-pli: + case <-time.After(5 * time.Second): + t.Fatal("the subscriber's keyframe request never reached the publisher, so a lost packet freezes its picture for good") + } +} + +// TestKeyframeIsRequestedOnceTheSubscriberIsReceiving covers the first picture. +// The request made when the SFU creates a subscriber's track reaches the +// publisher before the subscriber has answered the renegotiation that adds it, +// so the keyframe it produces is written into a track with nowhere to send it. +// The SFU has to ask again once the subscriber is demonstrably receiving. +func TestKeyframeIsRequestedOnceTheSubscriberIsReceiving(t *testing.T) { + h := newHarness(t, nil) + + publisher, _, videos := connectBrowserWith(t, h, "room-1", alicePriv, false, []string{"camera"}) + camera := videos["camera"] + + var mu sync.Mutex + requests := 0 + for _, sender := range publisher.pc.GetSenders() { + if track := sender.Track(); track == nil || track.ID() != "camera" { + continue + } + go func(s *webrtc.RTPSender) { + for { + packets, _, err := s.ReadRTCP() + if err != nil { + return + } + for _, p := range packets { + if _, ok := p.(*rtcp.PictureLossIndication); ok { + mu.Lock() + requests++ + mu.Unlock() + } + } + } + }(sender) + } + + stop := make(chan struct{}) + defer close(stop) + go func() { + seq, ts := uint16(1), uint32(3000) + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + } + _ = camera.WriteRTP(&rtp.Packet{ + Header: rtp.Header{Version: 2, SequenceNumber: seq, Timestamp: ts, SSRC: 0xCAFE, PayloadType: 96}, + Payload: []byte{0x10, 0xAB, 0xCD, 0xEF}, + }) + seq++ + ts += 3000 + } + }() + + // The subscriber only reads; it never asks for a keyframe itself, so any + // request after the first comes from the SFU noticing it is receiving. + connectBrowserWith(t, h, "room-1", bobPriv, false, nil, + func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) { + for { + if _, _, err := track.ReadRTP(); err != nil { + return + } + } + }) + + deadline := time.After(window) + for { + mu.Lock() + got := requests + mu.Unlock() + if got >= 2 { + return + } + select { + case <-deadline: + t.Fatalf("keyframe requests = %d, want a second once the subscriber is receiving", got) + case <-time.After(50 * time.Millisecond): + } + } +} diff --git a/huddle/sfu/video.go b/huddle/sfu/video.go index 11de6ad..87be828 100644 --- a/huddle/sfu/video.go +++ b/huddle/sfu/video.go @@ -2,9 +2,11 @@ package sfu import ( "sync" + "time" "github.com/pion/rtcp" "github.com/pion/rtp" + "github.com/ohstr/nmilat/utils" "github.com/pion/webrtc/v4" ) @@ -109,60 +111,144 @@ func (s *session) forwardVideo(track *webrtc.TrackRemote) { } } +// keyframeInterval is the least time between two keyframe requests one +// subscriber makes for one publication. A browser asks again on every frame it +// cannot decode, and each request makes the publisher send a full, much larger +// frame; forwarding them all would flood the very link that is losing packets. +const keyframeInterval = 500 * time.Millisecond + +// subscription is one subscriber's outbound copy of a publication. +type subscription struct { + track *webrtc.TrackLocalStaticRTP + + mu sync.Mutex + // owner and ssrc name where keyframe requests go. They change when the + // publisher reconnects: a client keeps its track ids across sessions, so the + // new stream lands on this same subscription. + owner *session + ssrc webrtc.SSRC + lastRequest time.Time + heardFromPeer bool +} + +// adopt points the subscription at pub's source, reporting whether it moved. +func (sub *subscription) adopt(pub publication) bool { + sub.mu.Lock() + defer sub.mu.Unlock() + if sub.owner == pub.owner && sub.ssrc == pub.SSRC { + return false + } + sub.owner, sub.ssrc = pub.owner, pub.SSRC + return true +} + +// requestKeyframe asks the publisher for a keyframe, at most once per +// keyframeInterval unless force is set. +func (sub *subscription) requestKeyframe(force bool) { + sub.mu.Lock() + now := time.Now() + if !force && now.Sub(sub.lastRequest) < keyframeInterval { + sub.mu.Unlock() + return + } + sub.lastRequest = now + owner, ssrc := sub.owner, sub.ssrc + sub.mu.Unlock() + if owner != nil { + owner.requestKeyframe(ssrc) + } +} + // writeVideo delivers one forwarded packet to this subscriber, creating the // outbound track on first sight of the publication. func (s *session) writeVideo(pub publication, packet *rtp.Packet) { - track, fresh := s.videoTrackFor(pub) - if track == nil { + sub, fresh := s.subscriptionFor(pub) + if sub == nil { return } - if fresh { - // A subscriber joining mid-stream starts between keyframes, so its - // decoder has nothing to build on and shows nothing until the next one -- - // which can be seconds away. Asking the publisher for one now is the - // difference between video appearing at once and appearing eventually. - pub.owner.requestKeyframe(pub.SSRC) + // A subscriber joining mid-stream starts between keyframes, so its decoder has + // nothing to build on until the next one -- and so does one whose publisher + // just reconnected, since the new encoder's stream continues on this track. + if moved := sub.adopt(pub); fresh || moved { + sub.requestKeyframe(true) } - if err := track.WriteRTP(packet); err != nil { + if err := sub.track.WriteRTP(packet); err != nil { s.log.Debug().Err(err).Str("track", pub.TrackID).Msg("dropping a forwarded video packet") } } -// videoTrackFor returns this subscriber's outbound track for a publication, +// subscriptionFor returns this subscriber's outbound track for a publication, // creating and negotiating it on first use. fresh reports whether it was just // created. -func (s *session) videoTrackFor(pub publication) (track *webrtc.TrackLocalStaticRTP, fresh bool) { +func (s *session) subscriptionFor(pub publication) (sub *subscription, fresh bool) { s.videoMu.Lock() - if existing, ok := s.videoTracks[pub.key()]; ok { + defer func() { s.videoMu.Unlock() + if fresh { + s.renegotiate() + } + }() + if existing, ok := s.videoTracks[pub.key()]; ok { return existing, false } - s.videoMu.Unlock() // The publisher's own track and stream ids are carried through, so a receiver // can tell a camera from a screen share exactly as the publisher labelled // them, and group a peer's tracks by stream. - created, err := webrtc.NewTrackLocalStaticRTP(pub.Codec, pub.TrackID, pub.StreamID) + track, err := webrtc.NewTrackLocalStaticRTP(pub.Codec, pub.TrackID, pub.StreamID) if err != nil { s.log.Debug().Err(err).Msg("could not create an outbound video track") return nil, false } - - s.videoMu.Lock() - if existing, ok := s.videoTracks[pub.key()]; ok { - // Another packet from the same publication won the race. - s.videoMu.Unlock() - return existing, false - } - s.videoTracks[pub.key()] = created - s.videoMu.Unlock() - - if _, err := s.pc.AddTrack(created); err != nil { + sender, err := s.pc.AddTrack(track) + if err != nil { + // Not recorded, so the next packet tries again rather than writing into a + // track that was never added. s.log.Debug().Err(err).Msg("could not add an outbound video track") return nil, false } - s.renegotiate() - return created, true + sub = &subscription{track: track} + s.videoTracks[pub.key()] = sub + go s.readSubscriberRTCP(sender, sub, pub.TrackID) + return sub, true +} + +// readSubscriberRTCP carries this subscriber's feedback on one forwarded track +// back to its publisher. +// +// Reading is also what makes pion's interceptors see the feedback at all, so +// without this loop the NACK responder never retransmits either. Two signals +// matter here: +// +// - PLI and FIR, the subscriber saying it cannot decode: forwarded, throttled. +// - The first packet of any kind, which proves the track is now bound. The +// keyframe asked for when the track was created reached the publisher before +// the subscriber had even answered the renegotiation, so that keyframe was +// written into a track with nowhere to go. Asking again here is what makes +// the first picture appear without waiting for the subscriber to complain. +func (s *session) readSubscriberRTCP(sender *webrtc.RTPSender, sub *subscription, trackID string) { + defer utils.RecoverPanic(s.log) + for { + packets, _, err := sender.ReadRTCP() + if err != nil { + return + } + sub.mu.Lock() + first := !sub.heardFromPeer + sub.heardFromPeer = true + sub.mu.Unlock() + if first { + sub.requestKeyframe(true) + continue + } + for _, packet := range packets { + switch packet.(type) { + case *rtcp.PictureLossIndication, *rtcp.FullIntraRequest: + s.log.Debug().Str("track", trackID).Msg("forwarding a keyframe request") + sub.requestKeyframe(false) + } + } + } } // requestKeyframe asks this session's peer for a keyframe on one of its tracks. From 9758bd3af2aee9d114900dac9c56c0a6af04b433 Mon Sep 17 00:00:00 2001 From: naliyi <154817482+naliyi@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:18:30 +0000 Subject: [PATCH 2/3] changelog: cut 0.8.0-rc.23 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e3135c9..f310b88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## [Unreleased] +## [0.8.0-rc.23] ### Fixed From fffd90840f693ea3a392893597a80a49fa58d269 Mon Sep 17 00:00:00 2001 From: naliyi <154817482+naliyi@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:24:27 +0000 Subject: [PATCH 3/3] huddle/sfu: gofmt imports --- huddle/sfu/video.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/huddle/sfu/video.go b/huddle/sfu/video.go index 87be828..e51c50d 100644 --- a/huddle/sfu/video.go +++ b/huddle/sfu/video.go @@ -4,9 +4,9 @@ import ( "sync" "time" + "github.com/ohstr/nmilat/utils" "github.com/pion/rtcp" "github.com/pion/rtp" - "github.com/ohstr/nmilat/utils" "github.com/pion/webrtc/v4" )