fix(hls): carry DTS for B-frame reordering — no more video "boomerang"

The PES header only advertised PTS (PTS_DTS_flags = 0b10), which means
decoders displayed frames in decode order. B-frames in H.264 reference
frames both before and after them in display order, so without a
separate DTS the picture jumps back and forth between the B-frame pair
and the P-frame they reference — the "boomerang" mpv and the web
player were both showing.

Changes:

- Frame struct gains a DTS field (90 kHz). Audio always sets DTS ==
  PTS (no B-frames in audio).
- RTMP OnVideo now reads videoTag.CompositionTime (int32 ms, signed)
  and splits it into DTS = ts*90, PTS = (ts+comp)*90, matching how
  FLV already encodes A/V timing. OBS and most other H.264 RTMP
  publishers populate CompositionTime whenever B-frames are
  enabled.
- buildPESHeader takes (pts, dts) and emits both when they differ,
  using the spec-correct prefix nibbles (0011 for PTS, 0001 for DTS)
  via a shared encodePTSWithPrefix helper. When dts == pts it emits
  only the 5-byte PTS header exactly as before.
- writeVideoPES now takes a dts parameter and derives PCR from the
  lower of DTS/PTS (PCR must never be ahead of any DTS).
- writeAudioPES passes dts == pts (no change in PES size).
- pesUnit in the HLS framed segmenter carries dts too; videoBatch
  and audioBatch entries both populate it on frame dispatch.
- MuxAVSegment's legacy byte-buffer path (single PES per segment)
  keeps dts == pts since it doesn't have CompositionTime info.

Author: DatanoiseTV

relay/frames.go

@@ -23,9 +23,10 @@ const (
 // single non-IDR slice). For AAC it's one ADTS frame. For MP3 it's one
 // MPEG audio frame.
 type Frame struct {
-	Kind     FrameKind
-	PTS      int64 // 90 kHz units
-	Data     []byte
+	Kind FrameKind
+	PTS  int64 // presentation timestamp, 90 kHz units
+	DTS  int64 // decode timestamp, 90 kHz units (== PTS for audio and B-frame-less video)
+	Data []byte
 	Keyframe bool // video only; ignored for audio
 }
 

relay/ingest_rtmp.go

@@ -343,14 +343,16 @@ func (h *rtmpHandler) OnAudio(timestamp uint32, payload io.Reader) error {
 
 	// Broadcast the raw audio data to the stream (for byte-level
 	// Icecast listeners) and publish a Frame to the hub (for HLS /
-	// other per-frame consumers).
+	// other per-frame consumers). Audio has no B-frames so DTS == PTS.
 	h.stream.Broadcast(data, h.relay)
 	if h.stream.Frames != nil {
 		frameData := make([]byte, len(data))
 		copy(frameData, data)
+		pts := ptsFromFLVTimestamp(timestamp)
 		h.stream.Frames.Publish(Frame{
 			Kind: FrameAudio,
-			PTS:  ptsFromFLVTimestamp(timestamp),
+			PTS:  pts,
+			DTS:  pts,
 			Data: frameData,
 		})
 	}
@@ -407,15 +409,20 @@ func (h *rtmpHandler) OnVideo(timestamp uint32, payload io.Reader) error {
 			}
 
 			h.videoStream.Broadcast(annexB, h.relay)
-			// Also publish a per-frame record with the real FLV timestamp
-			// so the HLS segmenter can emit one PES per frame with the
-			// right PTS instead of faking a single timestamp per segment.
+			// Publish a per-frame record carrying PTS *and* DTS. FLV
+			// video tags have CompositionTime (int32 ms, signed) which
+			// is PTS - DTS; without splitting them, B-frames display
+			// in decode order and the picture boomerangs between I/P
+			// and the two Bs either side of it.
 			if h.videoStream.Frames != nil {
 				frameData := make([]byte, len(annexB))
 				copy(frameData, annexB)
+				dts := ptsFromFLVTimestamp(timestamp)
+				pts := dts + int64(videoTag.CompositionTime)*90
 				h.videoStream.Frames.Publish(Frame{
 					Kind:     FrameVideo,
-					PTS:      ptsFromFLVTimestamp(timestamp),
+					PTS:      pts,
+					DTS:      dts,
 					Data:     frameData,
 					Keyframe: isKeyframe,
 				})

relay/mux_mpegts.go

@@ -72,9 +72,12 @@ func (m *TSMuxer) MuxAVSegment(audioData []byte, videoData []byte, audioPTS, vid
 	// Write PMT (updated for A/V, with the correct audio stream_type)
 	m.writePMTAV(&buf, audioStreamType)
 
-	// Write video PES first (usually larger, keyframe-aligned)
+	// Write video PES first (usually larger, keyframe-aligned). This is
+	// the legacy one-PES-per-segment path (kept for byte-buffer HLS);
+	// with no CompositionTime information we pass DTS == PTS and
+	// buildPESHeader will emit just the 5-byte PTS form.
 	if len(videoData) > 0 {
-		m.writeVideoPES(&buf, videoData, videoPTS)
+		m.writeVideoPES(&buf, videoData, videoPTS, videoPTS)
 	}
 
 	// Write audio PES
@@ -135,13 +138,20 @@ func (m *TSMuxer) writePMTAV(buf *bytes.Buffer, audioStreamType byte) {
 	buf.Write(packet)
 }
 
-func (m *TSMuxer) writeVideoPES(buf *bytes.Buffer, data []byte, pts int64) {
-	pesHeader := buildPESHeader(h264StreamID, len(data), pts)
+// writeVideoPES emits one video PES. Pass dts == pts (or dts < 0) when
+// the stream has no B-frames; pass the real decode timestamp otherwise
+// so the decoder schedules B-frames correctly instead of the picture
+// "boomeranging" between frames.
+func (m *TSMuxer) writeVideoPES(buf *bytes.Buffer, data []byte, pts, dts int64) {
+	pesHeader := buildPESHeader(h264StreamID, len(data), pts, dts)
 	payload := append(pesHeader, data...)
-	// The PMT for A/V declares video as PCR_PID, so we emit the PCR on
-	// the first TS packet of the video PES. pts is already on the 90 kHz
-	// clock; PCR base is also 90 kHz so pass it through directly.
-	m.writePESPackets(buf, videoPID, &m.videoContinuity, payload, pts)
+	// PCR uses DTS (the earliest timestamp in the stream) when we have
+	// both, otherwise PTS. PCR on the PID declared as PCR_PID in the PMT.
+	pcr := dts
+	if pcr < 0 || pcr > pts {
+		pcr = pts
+	}
+	m.writePESPackets(buf, videoPID, &m.videoContinuity, payload, pcr)
 }
 
 func (m *TSMuxer) writePAT(buf *bytes.Buffer) {
@@ -232,7 +242,9 @@ func (m *TSMuxer) writePMT(buf *bytes.Buffer) {
 }
 
 func (m *TSMuxer) writeAudioPES(buf *bytes.Buffer, data []byte, pts int64) {
-	pesHeader := buildPESHeader(mp3StreamID, len(data), pts)
+	// Audio has no B-frames, so DTS == PTS and the PES header only
+	// advertises PTS (the 5-byte shorter form).
+	pesHeader := buildPESHeader(mp3StreamID, len(data), pts, pts)
 	payload := append(pesHeader, data...)
 	// Audio-only PMT declares audio as PCR_PID, so the audio PES carries
 	// the PCR. For A/V the PMT uses video as PCR_PID and writeVideoPES
@@ -342,8 +354,12 @@ func writePCR(dst []byte, base90k uint64) {
 	dst[5] = 0x00
 }
 
-// buildPESHeader creates a PES packet header with PTS.
-func buildPESHeader(streamID byte, dataLen int, pts int64) []byte {
+// buildPESHeader creates a PES packet header carrying PTS (and optionally
+// DTS). When dts != pts, both are emitted — the decoder needs DTS to
+// schedule decoding of B-frames, and PTS to decide display order. For
+// audio streams and for I/P-only video, pass dts == pts and the PES will
+// advertise PTS only.
+func buildPESHeader(streamID byte, dataLen int, pts, dts int64) []byte {
 	var buf bytes.Buffer
 
 	// PES start code: 00 00 01
@@ -352,33 +368,53 @@ func buildPESHeader(streamID byte, dataLen int, pts int64) []byte {
 	// Stream ID
 	buf.WriteByte(streamID)
 
-	// PES packet length (0 = unbounded, but for segments we set it)
-	headerDataLen := 5 // PTS is 5 bytes
+	includeDTS := dts >= 0 && dts != pts
+	headerDataLen := 5
+	ptsDtsFlags := byte(0x80) // PTS only
+	if includeDTS {
+		headerDataLen = 10
+		ptsDtsFlags = 0xC0 // PTS + DTS
+	}
 	pesLen := 3 + headerDataLen + dataLen // 3 = flags(2) + header_data_length(1)
 	if pesLen > 0xFFFF {
-		pesLen = 0 // unbounded for large segments
+		pesLen = 0
 	}
 	buf.WriteByte(byte(pesLen >> 8))
 	buf.WriteByte(byte(pesLen & 0xFF))
 
-	// Flags: marker bits, PTS present
 	buf.WriteByte(0x80) // 10 marker, no scrambling, no priority, no alignment, no copyright, no original
-	buf.WriteByte(0x80) // PTS only (no DTS)
-
-	// PES header data length
+	buf.WriteByte(ptsDtsFlags)
 	buf.WriteByte(byte(headerDataLen))
 
-	// PTS (5 bytes)
-	buf.Write(encodePTS(pts))
+	if includeDTS {
+		// Spec: when both flags are set, the PTS prefix nibble is 0x3
+		// and the DTS prefix nibble is 0x1. encodePTSWithPrefix builds
+		// the 5-byte bitfield with the right first-nibble markers.
+		buf.Write(encodePTSWithPrefix(pts, 0x3))
+		buf.Write(encodePTSWithPrefix(dts, 0x1))
+	} else {
+		buf.Write(encodePTS(pts))
+	}
 
 	return buf.Bytes()
 }
 
-// encodePTS encodes a 33-bit PTS value into the 5-byte MPEG-TS format.
+// encodePTS encodes a 33-bit PTS value into the 5-byte MPEG-TS format
+// (PTS-only PES header — first nibble is 0b0010).
 func encodePTS(pts int64) []byte {
+	return encodePTSWithPrefix(pts, 0x2)
+}
+
+// encodePTSWithPrefix encodes a 33-bit timestamp into the 5-byte
+// MPEG-TS PES-header format with a caller-supplied 4-bit prefix nibble.
+// The PES spec uses different prefix values for PTS-only (0010), PTS
+// when DTS also present (0011), and DTS (0001) — passing the prefix
+// lets a single routine produce all three.
+func encodePTSWithPrefix(pts int64, prefix byte) []byte {
 	p := make([]byte, 5)
-	// Format: 0010 xxx1 | xxxxxxxx | xxxxxxx1 | xxxxxxxx | xxxxxxx1
-	p[0] = 0x21 | byte((pts>>29)&0x0E)
+	prefix &= 0x0F
+	// Format: <prefix[3:0]> xxx1 | xxxxxxxx | xxxxxxx1 | xxxxxxxx | xxxxxxx1
+	p[0] = (prefix << 4) | 0x01 | byte((pts>>29)&0x0E)
 	p[1] = byte(pts >> 22)
 	p[2] = 0x01 | byte((pts>>14)&0xFE)
 	p[3] = byte(pts >> 7)

relay/output_hls.go

@@ -11,9 +11,13 @@ import (
 )
 
 // pesUnit is a single access unit queued for inclusion in the next HLS
-// segment, already tagged with its 90 kHz PTS.
+// segment, with both its presentation (PTS) and decode (DTS) timestamps
+// on the 90 kHz MPEG clock. For audio and for video without B-frames
+// dts == pts; for video with B-frames dts comes from the FLV
+// CompositionTime adjustment.
 type pesUnit struct {
 	pts  int64
+	dts  int64
 	data []byte
 }
 
@@ -207,7 +211,7 @@ func (h *HLSOutput) segmentLoopFramed(ctx context.Context, audio, video *Track)
 			if !ok {
 				return
 			}
-			audioBatch = append(audioBatch, pesUnit{pts: f.PTS, data: f.Data})
+			audioBatch = append(audioBatch, pesUnit{pts: f.PTS, dts: f.DTS, data: f.Data})
 		case f, ok := <-videoFrames:
 			if !ok {
 				return
@@ -224,7 +228,7 @@ func (h *HLSOutput) segmentLoopFramed(ctx context.Context, audio, video *Track)
 				}
 				segHasIDR = true
 			}
-			videoBatch = append(videoBatch, pesUnit{pts: f.PTS, data: f.Data})
+			videoBatch = append(videoBatch, pesUnit{pts: f.PTS, dts: f.DTS, data: f.Data})
 		case <-timer.C:
 			// Time-based flush safety net.
 			if time.Since(segStart) >= h.config.SegmentDuration {
@@ -242,7 +246,7 @@ func (h *HLSOutput) buildAVSegment(audioBatch, videoBatch []pesUnit, audioStream
 	h.muxer.writePAT(&buf)
 	h.muxer.writePMTAV(&buf, audioStreamType)
 	for _, v := range videoBatch {
-		h.muxer.writeVideoPES(&buf, v.data, v.pts)
+		h.muxer.writeVideoPES(&buf, v.data, v.pts, v.dts)
 	}
 	for _, a := range audioBatch {
 		h.muxer.writeAudioPES(&buf, a.data, a.pts)