diff --git a/sender/capture_timestamp.go b/sender/capture_timestamp.go index 35981f4..0a969f3 100644 --- a/sender/capture_timestamp.go +++ b/sender/capture_timestamp.go @@ -28,10 +28,26 @@ import ( type captureTimestampInterceptor struct { interceptor.NoOp + // rewrite gates the RTP-timestamp overwrite. It is DISABLED by default: + // replacing the packetizer's clock with an external wall clock breaks any + // track whose sending is gated outside the RTP stack (e.g. a simulcast-like + // setup where a track is allocated zero bitrate and starts sending later) — + // such a track begins mid-session with a base derived from wall time and the + // receiver never renders it. Enable with sender.CaptureTimestampRewrite() + // only when every track sends continuously for the session's lifetime. + rewrite atomic.Bool + mu sync.Mutex slots map[uint32]*atomic.Int64 // ssrc -> capture time (unix microseconds), 0 = none } +// SetRewriteEnabled turns the RTP-timestamp overwrite on or off. Capture times +// supplied via SetCaptureTSUs are still recorded when disabled, so callers that +// only want capture-time telemetry are unaffected. +func (it *captureTimestampInterceptor) SetRewriteEnabled(enabled bool) { + it.rewrite.Store(enabled) +} + func newCaptureTimestampInterceptor() *captureTimestampInterceptor { return &captureTimestampInterceptor{slots: make(map[uint32]*atomic.Int64)} } @@ -102,8 +118,9 @@ func (it *captureTimestampInterceptor) BindLocalStream( } // Apply the capture-derived timestamp to every packet of the frame, - // keeping it constant across the frame's packets. - if frameValid { + // keeping it constant across the frame's packets. Skipped unless the + // rewrite was explicitly enabled — see the `rewrite` field. + if frameValid && it.rewrite.Load() { header.Timestamp = frameRTPTS } diff --git a/sender/capture_timestamp_test.go b/sender/capture_timestamp_test.go index 39a68b4..c8c0bdc 100644 --- a/sender/capture_timestamp_test.go +++ b/sender/capture_timestamp_test.go @@ -37,6 +37,7 @@ func bindCapture(it *captureTimestampInterceptor, sink interceptor.RTPWriter) in // is overwritten with captureUs*9/100 (mod 2^32), constant across a frame. func TestCaptureTimestampInterceptor_EncodesCaptureTime(t *testing.T) { it := newCaptureTimestampInterceptor() + it.SetRewriteEnabled(true) sink := &captureCollector{} w := bindCapture(it, sink) @@ -82,3 +83,42 @@ func TestCaptureTimestampInterceptor_RemoveSSRC(t *testing.T) { it.RemoveSSRC(0xDEAD) assert.Len(t, it.slots, 1) } + +// TestCaptureTimestampInterceptor_DisabledByDefault asserts the rewrite is off +// unless explicitly enabled, so a capture time supplied for telemetry does not +// silently replace the packetizer's clock. Overwriting it breaks any track whose +// sending is gated outside the RTP stack: such a track starts mid-session with a +// wall-clock-derived base and receivers never render it. +func TestCaptureTimestampInterceptor_DisabledByDefault(t *testing.T) { + it := newCaptureTimestampInterceptor() + sink := &captureCollector{} + w := bindCapture(it, sink) + + it.SetCaptureTSUs(testCaptureSSRC, 1_751_000_000_000_000) + _, _ = w.Write(&rtp.Header{Timestamp: 4242}, nil, nil) + + // Packetizer timestamp preserved, capture time still recorded for callers + // that read it for telemetry. + assert.Equal(t, []uint32{4242}, sink.timestamps) + assert.Equal(t, int64(1_751_000_000_000_000), it.slot(testCaptureSSRC).Load()) +} + +// TestCaptureTimestampInterceptor_RewriteToggle asserts the gate takes effect +// per frame, so enabling and disabling at runtime is honored. +func TestCaptureTimestampInterceptor_RewriteToggle(t *testing.T) { + it := newCaptureTimestampInterceptor() + sink := &captureCollector{} + w := bindCapture(it, sink) + + captureUs := int64(1_751_000_000_000_000) + it.SetCaptureTSUs(testCaptureSSRC, captureUs) + want := uint32(captureUs * 9 / 100) //nolint:gosec // intentional 32-bit wrap + + _, _ = w.Write(&rtp.Header{Timestamp: 100}, nil, nil) + it.SetRewriteEnabled(true) + _, _ = w.Write(&rtp.Header{Timestamp: 200}, nil, nil) + it.SetRewriteEnabled(false) + _, _ = w.Write(&rtp.Header{Timestamp: 300}, nil, nil) + + assert.Equal(t, []uint32{100, want, 300}, sink.timestamps) +} diff --git a/sender/option.go b/sender/option.go index 2c94a8a..ed19d35 100644 --- a/sender/option.go +++ b/sender/option.go @@ -7,6 +7,7 @@ package sender import ( + "errors" "io" "time" @@ -65,6 +66,31 @@ func DefaultInterceptors() Option { } } +var errCaptureTimestampUnsupported = errors.New( + "CaptureTimestampRewrite requires an *RTCSender") + +// CaptureTimestampRewrite returns an Option that encodes each frame's capture +// time (supplied via SetCaptureTSUs) into the outgoing RTP timestamp, so the +// capture instant survives an SFU that strips header extensions on egress. +// +// Off by default, and unsafe for tracks whose sending is gated outside the RTP +// stack: a track that is idle and then starts sending mid-session begins with a +// timestamp base derived from wall time, and receivers do not render it. There +// is no way to both preserve the packetizer's timeline and carry an absolute +// capture instant in the same field, so prefer an out-of-band channel (e.g. a +// data channel keyed by RTP timestamp) when tracks may start or stop. +func CaptureTimestampRewrite() Option { + return func(sender ConfigurableWebRTCSender) error { + rtcSender, ok := sender.(*RTCSender) + if !ok { + return errCaptureTimestampUnsupported + } + rtcSender.captureTimestamp.SetRewriteEnabled(true) + + return nil + } +} + // CCLogWriter returns an Option that configures congestion control logging. func CCLogWriter(w io.Writer) Option { return func(sender ConfigurableWebRTCSender) error {