diff --git a/go.mod b/go.mod index 09c04ff40..0bd0c75c1 100644 --- a/go.mod +++ b/go.mod @@ -156,3 +156,5 @@ require ( google.golang.org/grpc v1.78.0 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) + +replace github.com/livekit/protocol => ../protocol diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 7705f742c..c45f809b2 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -518,6 +518,10 @@ func (t *MediaTrackReceiver) IsEncrypted() bool { return t.TrackInfo().Encryption != livekit.Encryption_NONE } +func (t *MediaTrackReceiver) HasUserTimestamp() bool { + return t.TrackInfo().GetHasUserTimestamp() +} + func (t *MediaTrackReceiver) AddOnClose(f func(isExpectedToResume bool)) { if f == nil { return diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 6f566b81e..bcfe6c7c5 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -129,18 +129,21 @@ func NewSubscribedTrack(params SubscribedTrackParams) (*SubscribedTrack, error) if isEncrypted { trailer = params.Subscriber.GetTrailer() } + stripUserTimestamp := params.MediaTrack.HasUserTimestamp() && + !params.Subscriber.ProtocolVersion().SupportsUserTimestampStripping() downTrack, err := sfu.NewDownTrack(sfu.DownTrackParams{ - Codecs: codecs, - IsEncrypted: isEncrypted, - Source: params.MediaTrack.Source(), - Receiver: params.WrappedReceiver, - BufferFactory: params.Subscriber.GetBufferFactory(), - SubID: params.Subscriber.ID(), - StreamID: streamID, - MaxTrack: maxTrack, - PlayoutDelayLimit: params.Subscriber.GetPlayoutDelayConfig(), - Pacer: params.Subscriber.GetPacer(), - Trailer: trailer, + Codecs: codecs, + IsEncrypted: isEncrypted, + Source: params.MediaTrack.Source(), + Receiver: params.WrappedReceiver, + BufferFactory: params.Subscriber.GetBufferFactory(), + SubID: params.Subscriber.ID(), + StreamID: streamID, + MaxTrack: maxTrack, + PlayoutDelayLimit: params.Subscriber.GetPlayoutDelayConfig(), + Pacer: params.Subscriber.GetPacer(), + Trailer: trailer, + StripUserTimestamp: stripUserTimestamp, Logger: LoggerWithTrack( params.Subscriber.GetLogger().WithComponent(sutils.ComponentSub), params.MediaTrack.ID(), diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 18a0181da..6de873ed6 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -761,6 +761,7 @@ type MediaTrack interface { ClearAllReceivers(isExpectedToResume bool) IsEncrypted() bool + HasUserTimestamp() bool } //counterfeiter:generate . LocalMediaTrack diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index e7f569854..b97c7bfae 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -16,7 +16,7 @@ package types type ProtocolVersion int -const CurrentProtocol = 16 +const CurrentProtocol = 17 func (v ProtocolVersion) SupportsPackedStreamId() bool { return v > 0 @@ -99,3 +99,7 @@ func (v ProtocolVersion) SupportsNonErrorSignalResponse() bool { func (v ProtocolVersion) SupportsMoving() bool { return v > 15 } + +func (v ProtocolVersion) SupportsUserTimestampStripping() bool { + return v > 16 +} diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index a06094ec2..81eeebd11 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -146,6 +146,16 @@ type FakeLocalMediaTrack struct { hasSignalCidReturnsOnCall map[int]struct { result1 bool } + HasUserTimestampStub func() bool + hasUserTimestampMutex sync.RWMutex + hasUserTimestampArgsForCall []struct { + } + hasUserTimestampReturns struct { + result1 bool + } + hasUserTimestampReturnsOnCall map[int]struct { + result1 bool + } IDStub func() livekit.TrackID iDMutex sync.RWMutex iDArgsForCall []struct { @@ -1072,6 +1082,59 @@ func (fake *FakeLocalMediaTrack) HasSignalCidReturnsOnCall(i int, result1 bool) }{result1} } +func (fake *FakeLocalMediaTrack) HasUserTimestamp() bool { + fake.hasUserTimestampMutex.Lock() + ret, specificReturn := fake.hasUserTimestampReturnsOnCall[len(fake.hasUserTimestampArgsForCall)] + fake.hasUserTimestampArgsForCall = append(fake.hasUserTimestampArgsForCall, struct { + }{}) + stub := fake.HasUserTimestampStub + fakeReturns := fake.hasUserTimestampReturns + fake.recordInvocation("HasUserTimestamp", []interface{}{}) + fake.hasUserTimestampMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) HasUserTimestampCallCount() int { + fake.hasUserTimestampMutex.RLock() + defer fake.hasUserTimestampMutex.RUnlock() + return len(fake.hasUserTimestampArgsForCall) +} + +func (fake *FakeLocalMediaTrack) HasUserTimestampCalls(stub func() bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = stub +} + +func (fake *FakeLocalMediaTrack) HasUserTimestampReturns(result1 bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = nil + fake.hasUserTimestampReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalMediaTrack) HasUserTimestampReturnsOnCall(i int, result1 bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = nil + if fake.hasUserTimestampReturnsOnCall == nil { + fake.hasUserTimestampReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.hasUserTimestampReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalMediaTrack) ID() livekit.TrackID { fake.iDMutex.Lock() ret, specificReturn := fake.iDReturnsOnCall[len(fake.iDArgsForCall)] diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index 95a4d0649..c815a5dfe 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -98,6 +98,16 @@ type FakeMediaTrack struct { getTemporalLayerForSpatialFpsReturnsOnCall map[int]struct { result1 int32 } + HasUserTimestampStub func() bool + hasUserTimestampMutex sync.RWMutex + hasUserTimestampArgsForCall []struct { + } + hasUserTimestampReturns struct { + result1 bool + } + hasUserTimestampReturnsOnCall map[int]struct { + result1 bool + } IDStub func() livekit.TrackID iDMutex sync.RWMutex iDArgsForCall []struct { @@ -742,6 +752,59 @@ func (fake *FakeMediaTrack) GetTemporalLayerForSpatialFpsReturnsOnCall(i int, re }{result1} } +func (fake *FakeMediaTrack) HasUserTimestamp() bool { + fake.hasUserTimestampMutex.Lock() + ret, specificReturn := fake.hasUserTimestampReturnsOnCall[len(fake.hasUserTimestampArgsForCall)] + fake.hasUserTimestampArgsForCall = append(fake.hasUserTimestampArgsForCall, struct { + }{}) + stub := fake.HasUserTimestampStub + fakeReturns := fake.hasUserTimestampReturns + fake.recordInvocation("HasUserTimestamp", []interface{}{}) + fake.hasUserTimestampMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) HasUserTimestampCallCount() int { + fake.hasUserTimestampMutex.RLock() + defer fake.hasUserTimestampMutex.RUnlock() + return len(fake.hasUserTimestampArgsForCall) +} + +func (fake *FakeMediaTrack) HasUserTimestampCalls(stub func() bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = stub +} + +func (fake *FakeMediaTrack) HasUserTimestampReturns(result1 bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = nil + fake.hasUserTimestampReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeMediaTrack) HasUserTimestampReturnsOnCall(i int, result1 bool) { + fake.hasUserTimestampMutex.Lock() + defer fake.hasUserTimestampMutex.Unlock() + fake.HasUserTimestampStub = nil + if fake.hasUserTimestampReturnsOnCall == nil { + fake.hasUserTimestampReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.hasUserTimestampReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeMediaTrack) ID() livekit.TrackID { fake.iDMutex.Lock() ret, specificReturn := fake.iDReturnsOnCall[len(fake.iDArgsForCall)] diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 849b9bad5..759616e00 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -293,6 +293,21 @@ var _ TrackSender = (*DownTrack)(nil) type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) +var userTimestampMagic = [4]byte{'L', 'K', 'T', 'S'} + +const userTimestampTrailerSize = 12 + +func stripUserTimestampTrailer(payload []byte, marker bool) int { + if !marker || len(payload) < userTimestampTrailerSize { + return 0 + } + tail := payload[len(payload)-4:] + if tail[0] == userTimestampMagic[0] && tail[1] == userTimestampMagic[1] && tail[2] == userTimestampMagic[2] && tail[3] == userTimestampMagic[3] { + return userTimestampTrailerSize + } + return 0 +} + type DownTrackParams struct { Codecs []webrtc.RTPCodecParameters IsEncrypted bool @@ -309,6 +324,7 @@ type DownTrackParams struct { RTCPWriter func([]rtcp.Packet) error DisableSenderReportPassThrough bool SupportsCodecChange bool + StripUserTimestamp bool Listener DownTrackListener } @@ -1057,6 +1073,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) int32 { } payload = payload[:len(tp.codecBytes)+n] + if d.params.StripUserTimestamp { + if strip := stripUserTimestampTrailer(payload, tp.marker); strip > 0 { + payload = payload[:len(payload)-strip] + } + } + // translate RTP header hdr := RTPHeaderFactory.Get().(*rtp.Header) *hdr = rtp.Header{ @@ -2175,6 +2197,12 @@ func (d *DownTrack) retransmitPacket(epm *extPacketMeta, sourcePkt []byte, isPro payload = payload[:rtxOffset+int(epm.numCodecBytesOut)+len(pkt.Payload)-int(epm.numCodecBytesIn)] } + if d.params.StripUserTimestamp { + if strip := stripUserTimestampTrailer(payload[rtxOffset:], epm.marker); strip > 0 { + payload = payload[:len(payload)-strip] + } + } + headerSize := hdr.MarshalSize() var ( payloadSize, paddingSize int diff --git a/pkg/sfu/downtrack_strip_test.go b/pkg/sfu/downtrack_strip_test.go new file mode 100644 index 000000000..86f2fa0b7 --- /dev/null +++ b/pkg/sfu/downtrack_strip_test.go @@ -0,0 +1,95 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sfu + +import ( + "encoding/binary" + "testing" +) + +func makePayloadWithTrailer(videoLen int, timestampUs int64) []byte { + payload := make([]byte, videoLen+userTimestampTrailerSize) + for i := 0; i < videoLen; i++ { + payload[i] = byte(i) + } + binary.BigEndian.PutUint64(payload[videoLen:], uint64(timestampUs)) + copy(payload[videoLen+8:], userTimestampMagic[:]) + return payload +} + +func TestStripUserTimestampTrailer(t *testing.T) { + tests := []struct { + name string + payload []byte + marker bool + wantStrip int + }{ + { + name: "marker set with valid trailer", + payload: makePayloadWithTrailer(20, 1700000000000000), + marker: true, + wantStrip: userTimestampTrailerSize, + }, + { + name: "marker not set with valid trailer", + payload: makePayloadWithTrailer(20, 1700000000000000), + marker: false, + wantStrip: 0, + }, + { + name: "marker set without magic", + payload: make([]byte, 32), + marker: true, + wantStrip: 0, + }, + { + name: "marker set but payload too short", + payload: []byte{0x4C, 0x4B, 0x54, 0x53, 0x00, 0x00, 0x00, 0x00}, + marker: true, + wantStrip: 0, + }, + { + name: "marker set with partial magic mismatch", + payload: func() []byte { + p := makePayloadWithTrailer(20, 1700000000000000) + p[len(p)-1] = 'x' // corrupt 'S' -> 'x' + return p + }(), + marker: true, + wantStrip: 0, + }, + { + name: "exactly trailer size with valid magic", + payload: makePayloadWithTrailer(0, 42), + marker: true, + wantStrip: userTimestampTrailerSize, + }, + { + name: "empty payload", + payload: []byte{}, + marker: true, + wantStrip: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := stripUserTimestampTrailer(tt.payload, tt.marker) + if got != tt.wantStrip { + t.Errorf("stripUserTimestampTrailer() = %d, want %d", got, tt.wantStrip) + } + }) + } +}