From 5dac0aab20a7e8d296e1a4060966f3102c79a365 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Mon, 12 Oct 2026 20:22:15 +0530 Subject: [PATCH] sfu: send padding-only packets with no payload (#4993) --- pkg/sfu/downtrack.go | 31 ++++----------- .../downtrack_downstream_integration_test.go | 38 ++++++++++++------- pkg/sfu/export_test.go | 6 +++ pkg/sfu/pacer/base.go | 2 +- 4 files changed, 39 insertions(+), 38 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 346580149..822387313 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -86,7 +86,7 @@ var ( ) // padding of a dropped packet's slot, one byte: the count byte itself -var droppedSlotPadding = []byte{1} +const droppedSlotPaddingSize = 1 var ( VP8KeyFrame8x8 = []byte{ @@ -1253,7 +1253,7 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta initPooledRTPHeader(hdr, rtp.Header{ Version: 2, Padding: true, - PaddingSize: byte(len(droppedSlotPadding)), + PaddingSize: droppedSlotPaddingSize, Marker: false, PayloadType: uint8(d.payloadType.Load()), SequenceNumber: uint16(extSequenceNumber), @@ -1270,7 +1270,7 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta hdr.Marker, hdrSize, 0, - len(droppedSlotPadding), + droppedSlotPaddingSize, isOutOfOrder, ) @@ -1279,7 +1279,6 @@ func (d *DownTrack) writeDroppedSlotPadding(extSequenceNumber uint64, extTimesta Header: hdr, HeaderPool: RTPHeaderFactory, HeaderSize: hdrSize, - Payload: droppedSlotPadding, ProbeClusterId: ccutils.ProbeClusterIdInvalid, // not probe traffic AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), @@ -1354,7 +1353,6 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa } bytesSent := 0 - payloads := make([]byte, RTPPaddingMaxPayloadSize*len(snts)) for i := range snts { hdr := RTPHeaderFactory.Get().(*rtp.Header) initPooledRTPHeader(hdr, rtp.Header{ @@ -1369,12 +1367,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa }) d.addDummyExtensions(hdr) - payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize] - // last byte of padding has padding size including that byte - payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize) - hdrSize := hdr.MarshalSize() - payloadSize := len(payload) d.rtpStats.Update( mono.UnixNano(), snts[i].extSequenceNumber, @@ -1382,7 +1375,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa hdr.Marker, hdrSize, 0, - payloadSize, + RTPPaddingMaxPayloadSize, false, ) @@ -1391,7 +1384,6 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa Header: hdr, HeaderPool: RTPHeaderFactory, HeaderSize: hdrSize, - Payload: payload, ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()), IsProbe: true, AbsSendTimeExtID: uint8(d.absSendTimeExtID), @@ -1400,7 +1392,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa } d.pacer.Enqueue(pacerPacket) - bytesSent += hdrSize + payloadSize + bytesSent += hdrSize + RTPPaddingMaxPayloadSize } return bytesSent @@ -2528,8 +2520,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { return 0 } - payloads := make([]byte, RTPPaddingMaxPayloadSize*num) - for i := range num { + for range num { rtxExtSequenceNumber := d.rtxSequenceNumber.Inc() hdr := RTPHeaderFactory.Get().(*rtp.Header) initPooledRTPHeader(hdr, rtp.Header{ @@ -2544,12 +2535,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { }) d.addDummyExtensions(hdr) - payload := payloads[i*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize : (i+1)*RTPPaddingMaxPayloadSize] - // last byte of padding has padding size including that byte - payload[RTPPaddingMaxPayloadSize-1] = byte(RTPPaddingMaxPayloadSize) - hdrSize := hdr.MarshalSize() - payloadSize := len(payload) d.rtpStatsRTX.Update( mono.UnixNano(), rtxExtSequenceNumber, @@ -2557,7 +2543,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { hdr.Marker, hdrSize, 0, - payloadSize, + RTPPaddingMaxPayloadSize, false, ) pacerPacket := pacer.PacketFactory.Get().(*pacer.Packet) @@ -2565,7 +2551,6 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { Header: hdr, HeaderPool: RTPHeaderFactory, HeaderSize: hdrSize, - Payload: payload, ProbeClusterId: ccutils.ProbeClusterId(d.probeClusterId.Load()), IsProbe: true, AbsSendTimeExtID: uint8(d.absSendTimeExtID), @@ -2574,7 +2559,7 @@ func (d *DownTrack) WriteProbePackets(bytesToSend int, usePadding bool) int { } d.pacer.Enqueue(pacerPacket) - bytesSent += hdrSize + payloadSize + bytesSent += hdrSize + RTPPaddingMaxPayloadSize } } else { src := PacketFactory.Get().(*[]byte) diff --git a/pkg/sfu/downtrack_downstream_integration_test.go b/pkg/sfu/downtrack_downstream_integration_test.go index 41282e46c..73f2c4c43 100644 --- a/pkg/sfu/downtrack_downstream_integration_test.go +++ b/pkg/sfu/downtrack_downstream_integration_test.go @@ -584,10 +584,10 @@ func lktsTrailer() []byte { // the primary SSRC/payload type, carries the padding bit, and declares a full // padding block. // -// A padding-only packet is entirely padding; pion/srtp serializes the padding block -// from header.PaddingSize (via rtp.MarshalPacketTo), and the pion receiver strips it -// and reports both the padding bit and the padding size, so the received payload -// length equals the declared padding size (RTPPaddingMaxPayloadSize). +// A padding-only packet is entirely padding: pion/srtp serializes the padding block +// from header.PaddingSize (via rtp.MarshalPacketTo) after the payload, so the packet +// has no payload. A receiver takes a packet with any payload left after the padding +// as media. func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) { h := vnettest.NewHosts(t) factory := buffer.NewFactoryOfBufferFactory(500, 500).CreateBufferFactory() @@ -665,12 +665,27 @@ func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) { // a padding-only packet: it must declare a full padding block. require.EqualValues(t, RTPPaddingMaxPayloadSize, pkt.Header.PaddingSize, "padding-only packet must declare full padding size (sn=%d)", pkt.SequenceNumber) - require.Equal(t, RTPPaddingMaxPayloadSize, len(pkt.Payload), - "padding-only packet must serialize a full padding block (sn=%d)", pkt.SequenceNumber) + require.Empty(t, pkt.Payload, "padding-only packet must have no payload (sn=%d)", pkt.SequenceNumber) sawPadding++ } require.Equal(t, 1, sawKeyframe, "expected exactly one forwarded media keyframe") require.GreaterOrEqual(t, sawPadding, numPadding, "expected the padding-only packets to be observed") + + // the padding in the slot of a dropped packet has no payload either + dh.dt.WriteDroppedSlotPaddingForTest(6000, 190000) + var pkt *rtp.Packet + require.Eventually(t, func() bool { + for _, p := range dh.capture.all() { + if p.SequenceNumber == 6000 { + pkt = p + return true + } + } + return false + }, 10*time.Second, 20*time.Millisecond, "expected the dropped slot padding to arrive on the far side") + require.True(t, pkt.Header.Padding) + require.EqualValues(t, DroppedSlotPaddingSizeForTest, pkt.Header.PaddingSize) + require.Empty(t, pkt.Payload) } // ----------------------------------------------------------------------------- @@ -681,7 +696,6 @@ func TestDownTrackSendsPaddingOnlyPackets(t *testing.T) { type capturingProbe struct { hdr rtp.Header payloadLen int - lastByte byte } // capturedRTX records what the DownTrack handed the pacer for an RTX packet. @@ -706,9 +720,6 @@ func (c *capturingPacer) Enqueue(p *pacer.Packet) { switch { case p.IsProbe: cp := capturingProbe{hdr: *p.Header, payloadLen: len(p.Payload)} - if len(p.Payload) > 0 { - cp.lastByte = p.Payload[len(p.Payload)-1] - } c.lock.Lock() c.probes = append(c.probes, cp) c.lock.Unlock() @@ -900,16 +911,15 @@ func TestDownTrackSendsProbePackets(t *testing.T) { }, 5*time.Second, 20*time.Millisecond, "expected WriteProbePackets to emit RTX probe packets") // every probe packet must be on the RTX stream, carry the padding bit, declare - // the full padding size, and serialize a full padding block ending in the - // padding count byte. + // the full padding size, and have no payload: pion serializes the padding from + // the header. for _, pr := range cp.probePackets() { require.EqualValues(t, 2, pr.hdr.Version, "probe packet version") require.EqualValues(t, dh.dt.SSRCRTX(), pr.hdr.SSRC, "probe packet must be on the RTX stream") require.EqualValues(t, dh.dt.PayloadTypeRTXForTest(), pr.hdr.PayloadType, "probe packet must use the RTX payload type") require.True(t, pr.hdr.Padding, "probe packet must carry padding bit") require.EqualValues(t, RTPPaddingMaxPayloadSize, pr.hdr.PaddingSize, "probe packet must set header.PaddingSize") - require.Equal(t, RTPPaddingMaxPayloadSize, pr.payloadLen, "probe packet must carry a full padding block") - require.EqualValues(t, RTPPaddingMaxPayloadSize, pr.lastByte, "probe packet trailing byte must be the padding count") + require.Zero(t, pr.payloadLen, "probe packet must have no payload") } // the BWE is wired and exercised (the pacer records every send into it); diff --git a/pkg/sfu/export_test.go b/pkg/sfu/export_test.go index b6536362a..9812b2c33 100644 --- a/pkg/sfu/export_test.go +++ b/pkg/sfu/export_test.go @@ -32,6 +32,12 @@ func (d *DownTrack) PayloadTypeRTXForTest() uint32 { return d.payloadTypeRTX.Loa func (d *DownTrack) RTPStatsActiveForTest() bool { return d.rtpStats.IsActive() } +const DroppedSlotPaddingSizeForTest = droppedSlotPaddingSize + +func (d *DownTrack) WriteDroppedSlotPaddingForTest(extSequenceNumber uint64, extTimestamp uint64) { + d.writeDroppedSlotPadding(extSequenceNumber, extTimestamp, false) +} + func (d *DownTrack) LastReceiverReportTimeForTest() int64 { return d.rtpStats.LastReceiverReportTime() } diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index b5a35e9e8..b93362a00 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -110,7 +110,7 @@ func (b *Base) patchRTPHeaderExtensions(p *Packet) error { b.lastPacketSentAt.Store(sendingAt.UnixNano()) } - packetSize := p.HeaderSize + len(p.Payload) + packetSize := p.HeaderSize + len(p.Payload) + int(p.Header.PaddingSize) if p.TransportWideExtID != 0 && b.bwe != nil { twccSN := b.bwe.RecordPacketSendAndGetSequenceNumber( sendingAt.UnixMicro(),