diff --git a/CHANGELOG.md b/CHANGELOG.md index 24f0388..f310b88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## [0.8.0-rc.23] + +### 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..e51c50d 100644 --- a/huddle/sfu/video.go +++ b/huddle/sfu/video.go @@ -2,7 +2,9 @@ package sfu import ( "sync" + "time" + "github.com/ohstr/nmilat/utils" "github.com/pion/rtcp" "github.com/pion/rtp" "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.