From 8568d8c57cae5b6924a638d75aefff95a420818c Mon Sep 17 00:00:00 2001 From: t-animal Date: Sat, 21 Mar 2026 19:46:39 +0100 Subject: [PATCH] webrtc: fix random absolute timestamps with Opus, G711 and LPCM (#5597) When rewriting audio RTP timestamps in WebRTC egress, NTP was derived using regenerated packet timestamps minus the incoming RTP base timestamp. That mixed timestamp domains and could shift absolute time by an arbitrary offset while still exposing mapping as available. Fix by using a consistent outgoing RTP domain in rewritten audio paths: - snapshot outgoing base timestamp before rewriting each unit - compute NTP from (outgoing packet timestamp - outgoing base timestamp) This keeps RTP<->NTP mapping coherent for sender reports and prevents random absolute-time offsets in WebRTC loopback with useAbsoluteTimestamp. --------- Co-authored-by: aler9 <46489434+aler9@users.noreply.github.com> --- internal/protocols/webrtc/from_stream.go | 20 +- internal/protocols/webrtc/from_stream_test.go | 171 ++++++++++++++++++ 2 files changed, 185 insertions(+), 6 deletions(-) diff --git a/internal/protocols/webrtc/from_stream.go b/internal/protocols/webrtc/from_stream.go index f0b541f3..f127d98d 100644 --- a/internal/protocols/webrtc/from_stream.go +++ b/internal/protocols/webrtc/from_stream.go @@ -373,7 +373,10 @@ func setupAudioTrack( media, opusFormat, func(u *unit.Unit) error { + baseTimestamp := curTimestamp + for _, orig := range u.RTPPackets { + // create a copy of the packet that we can edit freely pkt := &rtp.Packet{ Header: orig.Header, Payload: orig.Payload, @@ -384,7 +387,7 @@ func setupAudioTrack( pkt.Timestamp = curTimestamp curTimestamp += uint32(opus.PacketDuration2(pkt.Payload)) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), 48000)) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 48000)) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -487,7 +490,10 @@ func setupAudioTrack( media, g711Format, func(u *unit.Unit) error { + baseTimestamp := curTimestamp + for _, orig := range u.RTPPackets { + // create a copy of the packet that we can edit freely pkt := &rtp.Packet{ Header: orig.Header, Payload: orig.Payload, @@ -498,7 +504,7 @@ func setupAudioTrack( pkt.Timestamp = curTimestamp curTimestamp += uint32(len(pkt.Payload)) / uint32(g711Format.ChannelCount) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), 8000)) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), 8000)) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -545,14 +551,15 @@ func setupAudioTrack( return nil //nolint:nilerr } + baseTimestamp := curTimestamp + for _, pkt := range packets { // recompute timestamp from scratch. // Chrome requires a precise timestamp that FFmpeg doesn't provide. pkt.Timestamp = curTimestamp curTimestamp += uint32(len(pkt.Payload)) / 2 / uint32(g711Format.ChannelCount) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), - g711Format.ClockRate())) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), g711Format.ClockRate())) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } @@ -619,14 +626,15 @@ func setupAudioTrack( return nil //nolint:nilerr } + baseTimestamp := curTimestamp + for _, pkt := range packets { // recompute timestamp from scratch. // Chrome requires a precise timestamp that FFmpeg doesn't provide. pkt.Timestamp = curTimestamp curTimestamp += uint32(len(pkt.Payload)) / 2 / uint32(lpcmFormat.ChannelCount) - ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-u.RTPPackets[0].Timestamp), - lpcmFormat.ClockRate())) + ntp := u.NTP.Add(timestampToDuration(int64(pkt.Timestamp-baseTimestamp), lpcmFormat.ClockRate())) track.WriteRTPWithNTP(pkt, ntp) //nolint:errcheck } diff --git a/internal/protocols/webrtc/from_stream_test.go b/internal/protocols/webrtc/from_stream_test.go index 7f40d956..48f5f23f 100644 --- a/internal/protocols/webrtc/from_stream_test.go +++ b/internal/protocols/webrtc/from_stream_test.go @@ -212,3 +212,174 @@ func TestFromStreamResampleOpus(t *testing.T) { <-done } + +func TestFromStreamResampleOpusAbsoluteTimestamp(t *testing.T) { + strm := &stream.Stream{ + Desc: &description.Session{Medias: []*description.Media{ + { + Type: description.MediaTypeAudio, + Formats: []format.Format{&format.Opus{ + ChannelCount: 2, + }}, + }, + }}, + WriteQueueSize: 512, + RTPMaxPayloadSize: 1450, + ReplaceNTP: false, + Parent: test.NilLogger, + } + err := strm.Initialize() + require.NoError(t, err) + + subStream := &stream.SubStream{ + Stream: strm, + UseRTPPackets: true, + } + err = subStream.Initialize() + require.NoError(t, err) + + pcReader := &PeerConnection{ + LocalRandomUDP: true, + IPsFromInterfaces: true, + Publish: false, + Log: test.NilLogger, + } + err = pcReader.Start() + require.NoError(t, err) + t.Cleanup(pcReader.Close) + + pcPublisher := &PeerConnection{ + LocalRandomUDP: true, + IPsFromInterfaces: true, + Publish: true, + Log: test.NilLogger, + } + + r := &stream.Reader{Parent: nil} + + err = FromStream(strm.Desc, r, pcPublisher) + require.NoError(t, err) + + err = pcPublisher.Start() + require.NoError(t, err) + t.Cleanup(pcPublisher.Close) + + offer, err := pcReader.CreatePartialOffer() + require.NoError(t, err) + + answer, err := pcPublisher.CreateFullAnswer(offer) + require.NoError(t, err) + + err = pcReader.SetAnswer(answer) + require.NoError(t, err) + + err = pcReader.WaitUntilConnected(10 * time.Second) + require.NoError(t, err) + + err = pcPublisher.WaitUntilConnected(10 * time.Second) + require.NoError(t, err) + + strm.AddReader(r) + t.Cleanup(func() { strm.RemoveReader(r) }) + + baseNTP := time.Unix(1710000000, 0) + step := 20 * time.Millisecond + + // prime the pipeline to allow track gathering + subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + PTS: 0, + NTP: baseNTP, + RTPPackets: []*rtp.Packet{{ + Header: rtp.Header{ + Version: 2, + Marker: true, + PayloadType: 111, + SequenceNumber: 1123, + Timestamp: 45343, + SSRC: 563424, + }, + Payload: []byte{1}, + }}, + }) + + err = pcReader.GatherIncomingTracks(2 * time.Second) + require.NoError(t, err) + + tracks := pcReader.IncomingTracks() + require.Len(t, tracks, 1) + + done := make(chan struct{}) + errCh := make(chan string, 1) + const startSeq = uint16(2000) + + expectedNTP := func(seq uint16) (time.Time, bool) { + if seq < startSeq { + return time.Time{}, false + } + return baseNTP.Add(time.Duration(seq-startSeq) * step), true + } + + tracks[0].OnPacketRTP = func(pkt *rtp.Packet) { + expected, ok := expectedNTP(pkt.SequenceNumber) + if !ok { + return + } + + ntp, avail := tracks[0].PacketNTP(pkt) + if !avail { + return + } + + if ntp.Sub(expected).Abs() > 50*time.Millisecond { + select { + case errCh <- fmt.Sprintf("absolute NTP mismatch for seq=%d: got=%v expected=%v", + pkt.SequenceNumber, ntp, expected): + default: + } + return + } + + select { + case done <- struct{}{}: + default: + } + } + + pcReader.StartReading() + + go func() { + ticker := time.NewTicker(step) + defer ticker.Stop() + + for i := range uint16(150) { + seq := startSeq + i + expected, _ := expectedNTP(seq) + + subStream.WriteUnit(strm.Desc.Medias[0], strm.Desc.Medias[0].Formats[0], &unit.Unit{ + PTS: 0, + NTP: expected, + RTPPackets: []*rtp.Packet{{ + Header: rtp.Header{ + Version: 2, + Marker: true, + PayloadType: 111, + SequenceNumber: seq, + Timestamp: 45343, + SSRC: 563424, + }, + Payload: []byte{1}, + }}, + }) + + <-ticker.C + } + }() + + select { + case <-done: + case err := <-errCh: + t.Fatal(err) + case <-time.After(8 * time.Second): + t.Fatal("absolute timestamp mapping did not become available") + } +}