fix(video): checkpoint on encoder reconfig + real PTS-derived segment duration

Two related issues with live RTMP video.

1. "Very old video then stops"
   When OBS reconfigures its encoder mid-session (resolution change,
   preset swap, restart) it re-sends the AVCDecoderConfigurationRecord
   with new SPS/PPS. Any bytes already in the 8 MB video ring buffer
   were encoded under the *old* parameters, and the cached keyframe
   offsets point into them. A listener that subscribed after the
   reconfig would replay those old frames with the new SPS/PPS active
   and decode blew up, then stalled.

   parseAVCConfig now detects that it already had SPS/PPS (i.e. this
   is a reconfig, not first init) and calls
   videoStream.CheckpointAtHead(), which marks the current
   Buffer.Head as a new MinListenerOffset and clears the keyframe
   index (Buffer.ResetKeyframes). serveStreamData bumps any new
   listener's start offset up to MinListenerOffset before seeking
   to the latest keyframe, so reconfigs no longer replay stale
   content. Stream.GetMinListenerOffset / new CheckpointAtHead
   encapsulate the read/write under the stream mutex.

2. "Buffering at 90%, 5s cache"
   The HLS ring previously recorded every segment with the
   configured SegmentDuration regardless of the actual wall time of
   the frames inside — a mismatch with OBS's keyframe interval made
   the player wait for phantom content that was never coming.

   segmentLoopFramed now derives segment duration from the span of
   PTS values it actually included (pesBatchRange returns first +
   last-end PTS across the audio and video batches) and pushes that
   into the ring. EXTINF is now accurate and TARGETDURATION adapts
   correctly via math.Ceil over the real durations.

Author: DatanoiseTV

relay/buffer.go

@@ -147,6 +147,17 @@ func (cb *CircularBuffer) HeadOffset() int64 {
 	return cb.Head
 }
 
+// ResetKeyframes drops every recorded keyframe offset. Use when the
+// source's codec config changes mid-stream — old keyframe offsets point
+// at bytes the new decoder state can't interpret, so they must not be
+// handed out as safe starting points for new listeners.
+func (cb *CircularBuffer) ResetKeyframes() {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+	cb.kfHead = 0
+	cb.kfCount = 0
+}
+
 // FindNextPageBoundaryLocked holds the buffer's RWMutex for the duration of
 // the scan, so it's safe to use against a buffer that may be concurrently
 // written by a source goroutine. Returns the absolute offset of the next

relay/ingest_rtmp.go

@@ -438,6 +438,13 @@ func (h *rtmpHandler) parseAVCConfig(data []byte) {
 	if len(data) < 8 {
 		return
 	}
+	// If we already had SPS/PPS from an earlier config, the publisher has
+	// reconfigured its encoder (different resolution, profile, GOP size,
+	// etc.). Bytes already in the video buffer were encoded under the
+	// old parameters and would blow up any new listener's decoder once
+	// the new SPS/PPS is active. Checkpoint the current Head so new
+	// listeners only see bytes written from here on.
+	reconfig := len(h.sps) > 0 || len(h.pps) > 0
 	// AVCDecoderConfigurationRecord structure:
 	// [0] configurationVersion
 	// [1] AVCProfileIndication
@@ -492,11 +499,19 @@ func (h *rtmpHandler) parseAVCConfig(data []byte) {
 		h.videoStream.StoreVideoHeaders(out)
 	}
 
+	if reconfig && h.videoStream != nil {
+		h.videoStream.CheckpointAtHead()
+		logger.L.Infow("RTMP: Encoder reconfig — checkpointed video buffer",
+			"mount", h.mount,
+		)
+	}
+
 	logger.L.Infow("RTMP: Parsed AVC config",
 		"mount", h.mount,
 		"sps_len", len(h.sps),
 		"pps_len", len(h.pps),
 		"nalu_len_size", h.naluLenSize,
+		"reconfig", reconfig,
 	)
 }
 

relay/output_hls.go

@@ -176,18 +176,24 @@ func (h *HLSOutput) segmentLoopFramed(ctx context.Context, audio, video *Track)
 			return
 		}
 		tsData := h.buildAVSegment(audioBatch, videoBatch, audioStreamType)
-		// Use the first video PTS (or audio PTS as a fallback) as the
-		// segment's nominal start so the ring records a sensible time.
-		startPTS := int64(0)
-		if len(videoBatch) > 0 {
-			startPTS = videoBatch[0].pts
-		} else if len(audioBatch) > 0 {
-			startPTS = audioBatch[0].pts
+		// Use the first PTS as the segment's nominal start, and derive
+		// the segment duration from the actual span of PTS values we
+		// included. Previously we always declared SegmentDuration
+		// regardless of what the content contained, which made the
+		// player buffer whenever keyframe interval didn't match — e.g.
+		// an OBS keyframe every 5 s into a 4 s advertised window left
+		// 1 s of "phantom" duration the player waited to arrive.
+		firstPTS, lastEndPTS := pesBatchRange(videoBatch, audioBatch)
+		startPTS := firstPTS
+		segDur := h.config.SegmentDuration
+		if lastEndPTS > firstPTS {
+			segDur = time.Duration(lastEndPTS-firstPTS) * time.Second / 90000
 		}
-		h.ring.Push(tsData, h.config.SegmentDuration, startPTS, false)
+		h.ring.Push(tsData, segDur, startPTS, false)
 		logger.L.Debugw("HLS: framed segment",
 			"mount", h.mount,
 			"bytes", len(tsData),
+			"duration", segDur,
 			"audio_frames", len(audioBatch),
 			"video_frames", len(videoBatch),
 			"sequence", h.ring.Sequence()-1,
@@ -238,6 +244,34 @@ func (h *HLSOutput) segmentLoopFramed(ctx context.Context, audio, video *Track)
 	}
 }
 
+// pesBatchRange returns the earliest first-PTS across audio + video and the
+// latest end-PTS (last frame's PTS rounded up to include its own duration
+// by using the following frame's start, or falling back to the last PTS
+// itself when there's only one frame).
+func pesBatchRange(videoBatch, audioBatch []pesUnit) (first, end int64) {
+	first = -1
+	end = -1
+	consider := func(batch []pesUnit) {
+		for _, u := range batch {
+			if first < 0 || u.pts < first {
+				first = u.pts
+			}
+			if u.pts > end {
+				end = u.pts
+			}
+		}
+	}
+	consider(videoBatch)
+	consider(audioBatch)
+	if first < 0 {
+		first = 0
+	}
+	if end < first {
+		end = first
+	}
+	return first, end
+}
+
 // buildAVSegment muxes a list of audio + video access units into a single
 // MPEG-TS segment. Each frame becomes its own PES packet with its own
 // PTS.

relay/stream.go

@@ -84,6 +84,15 @@ type Stream struct {
 	// bytes, frame-level consumers subscribe to the hub. nil on streams
 	// that never see a frame-level producer (e.g. relay pull).
 	Frames *FrameHub
+
+	// MinListenerOffset is a hard lower bound for listener reads of the
+	// byte Buffer. Used when the source changes its codec config mid-
+	// stream (e.g. OBS reconfigures and a new AVCDecoderConfigurationRecord
+	// arrives) — any bytes written before that point were encoded with
+	// parameters the new SPS/PPS no longer describes, so the decoder can
+	// only crash on them. Subscribe() bumps a new listener's start
+	// offset up to this value.
+	MinListenerOffset int64
 	LastPageOffset  int64   // Absolute offset of the last valid Ogg page start
 	PageOffsets     []int64 // Circular list of last ~100 page starts
 	PageIndex       int     // Index for managing PageOffsets circular list
@@ -138,6 +147,29 @@ func (s *Stream) StoreVideoHeaders(headers []byte) {
 	s.VideoHeaders = headers
 }
 
+// GetMinListenerOffset returns MinListenerOffset under the stream mutex.
+func (s *Stream) GetMinListenerOffset() int64 {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.MinListenerOffset
+}
+
+// CheckpointAtHead marks the current Buffer.Head as the new minimum
+// valid offset for listener reads, and clears any previously recorded
+// keyframe offsets (they pointed at bytes encoded under the old config).
+// Called by ingest paths when they observe a codec-parameter change
+// mid-stream — OBS restarting its encoder, resolution switch, etc.
+func (s *Stream) CheckpointAtHead() {
+	if s.Buffer == nil {
+		return
+	}
+	head := s.Buffer.HeadOffset()
+	s.mu.Lock()
+	s.MinListenerOffset = head
+	s.mu.Unlock()
+	s.Buffer.ResetKeyframes()
+}
+
 // VideoInfo returns the flags that the HTTP listener path needs to
 // decide whether a stream is video and what headers to prepend, under
 // the stream mutex. Callers outside the relay package can't read the

server/handlers_stream.go

@@ -397,6 +397,15 @@ func (s *Server) serveStreamData(w http.ResponseWriter, r *http.Request, stream
 		logger.L.Debugf("Ogg Listener %s: Sending stored headers (%d bytes), then starting burst at %d", id, len(stream.OggHead), offset)
 	}
 
+	// Respect the stream's minimum valid offset — set whenever the
+	// source reconfigures its codec mid-stream. Without this a listener
+	// that subscribes right after an OBS reconfig replays pre-reconfig
+	// bytes (old SPS/PPS) and the decoder dies as soon as the new
+	// parameters activate.
+	if min := stream.GetMinListenerOffset(); min > offset {
+		offset = min
+	}
+
 	// For raw H.264 video listeners, seek back to the most recent
 	// keyframe (so playback doesn't start on a P-frame whose reference
 	// frames aren't in the listener's buffer) and prepend the cached