relay: fix stale-burst replay and false Ogg page alignment

Several related bugs in the audio path together caused two user-visible
symptoms: new listeners hearing old audio when the source was idle, and
glitched / shuffled audio on otherwise-healthy streams.

  - Stream.Subscribe was bursting from Buffer.Head minus burst-size
    unconditionally, so when a producer had been silent for a while a
    new listener received seconds of stale buffered audio. Add a
    FreshHead watermark plus a LastDataReceived > 2s freshness gate,
    and have all producer entry points (icecast SOURCE, relay pull,
    WebRTC source, transcoder, AutoDJ MP3/Opus encoders) call a new
    Stream.BeginSession() that advances FreshHead and wipes stale Ogg
    state at the start of each session.

  - Stream.Broadcast tracked any byte sequence matching 'OggS' as a
    page boundary, including false positives inside Opus packet
    payloads. Validate the Ogg version byte (must be 0) before
    recording an offset; same validation in the new
    CircularBuffer.AlignToOggPage replacing the unsafe free function
    in relay/ogg.go.

  - Transcoder restarts re-initialised the Opus encoder with a fresh
    serial number but left the old PageOffsets / OggHead behind, so
    listeners could be aligned to bytes from a previous Ogg serial.
    BeginSession now resets that state.

  - OggHead / OggHeaderOffset and Buffer.Head / Buffer.Data were read
    and written without locking from multiple goroutines. Add
    Stream.SetOggHead / GetOggHead (defensive copy under lock) and
    CircularBuffer.SnapshotHead / AlignToOggPage (under the buffer's
    own lock) and route all callers through them.

  - Subscribe's page-alignment fallback initialised bestAlign to
    LastPageOffset and could end up walking the listener backwards
    past the requested start. Search for the smallest tracked offset
    >= start and otherwise fall back to Head (never backwards).

Author: Datanoise

relay/buffer.go

@@ -9,6 +9,9 @@ import (
 	"sync"
 )
 
+// oggMagic is the 4-byte capture pattern at the start of every Ogg page header.
+var oggMagic = [4]byte{'O', 'g', 'g', 'S'}
+
 // CircularBuffer is a thread-safe fixed-size ring buffer for stream data.
 //
 // This buffer implementation is optimized for audio streaming scenarios where
@@ -128,3 +131,82 @@ func (cb *CircularBuffer) ReadAt(start int64, p []byte) (int, int64, bool) {
 	actual := copy(p, cb.Data[pos:pos+n])
 	return actual, start + int64(actual), skipped
 }
+
+// SnapshotHead returns the current write position under the buffer lock.
+// Callers that need a consistent Head value without copying data should use
+// this rather than reading cb.Head directly, which races with Write.
+func (cb *CircularBuffer) SnapshotHead() int64 {
+	cb.mu.RLock()
+	defer cb.mu.RUnlock()
+	return cb.Head
+}
+
+// AlignToOggPage scans the buffer for the next valid Ogg page header at or
+// after the absolute offset 'start' and returns its offset. If no valid page
+// is found within the buffer window, it returns the current Head (i.e. "start
+// from now").
+//
+// A valid header here means the 4-byte "OggS" capture pattern followed by a
+// version byte of 0. This rejects the false positives that occur when the
+// byte sequence 0x4F 0x67 0x67 0x53 appears inside Opus packet payloads and
+// would otherwise misalign listeners mid-packet.
+//
+// The whole scan runs under the buffer's read lock so the data isn't being
+// overwritten beneath us while we search.
+func (cb *CircularBuffer) AlignToOggPage(start int64) int64 {
+	cb.mu.RLock()
+	defer cb.mu.RUnlock()
+
+	head := cb.Head
+	size := cb.Size
+
+	// Clamp to the valid buffer window.
+	if start < head-size {
+		start = head - size
+	}
+	if start < 0 {
+		start = 0
+	}
+	// Need at least the 4-byte magic plus the version byte.
+	if start >= head-5 {
+		return head
+	}
+
+	for i := start; i < head-5; {
+		pos := i % size
+		// Scan in segments that don't cross the physical buffer boundary.
+		end := size
+		if i+size > head {
+			end = pos + (head - i)
+		}
+		if end > size {
+			end = size
+		}
+
+		segment := cb.Data[pos:end]
+		// Find "OggS" within the segment. Don't include the last 4 bytes
+		// because we need to peek at the version byte that follows.
+		searchEnd := len(segment) - 4
+		if searchEnd < 0 {
+			searchEnd = 0
+		}
+		for j := 0; j < searchEnd; j++ {
+			if segment[j] == oggMagic[0] &&
+				segment[j+1] == oggMagic[1] &&
+				segment[j+2] == oggMagic[2] &&
+				segment[j+3] == oggMagic[3] &&
+				segment[j+4] == 0 { // Ogg version byte
+				return i + int64(j)
+			}
+		}
+
+		// Step forward, overlapping by 4 bytes so an "OggS" straddling the
+		// segment boundary still gets caught next iteration.
+		advance := int64(len(segment)) - 4
+		if advance <= 0 {
+			break
+		}
+		i += advance
+	}
+	return head
+}

relay/client.go

@@ -151,6 +151,10 @@ func (rm *RelayManager) performPull(ctx context.Context, inst *RelayInstance) {
 
 	stream := rm.relay.GetOrCreateStream(inst.Mount)
 	stream.SourceIP = "relay-pull"
+	// Each (re)connect is a new producer session — wipe any stale audio /
+	// Ogg state left over from a previous pull so reconnecting listeners
+	// don't get bursted with content from before the disconnect.
+	stream.BeginSession()
 
 	// Metadata
 	name := resp.Header.Get("Ice-Name")

relay/io.go

@@ -111,9 +111,10 @@ func (r *StreamReader) Read(p []byte) (int, error) {
 
 		// Handle Ogg synchronization if enabled
 		if skipped && r.oggSync && r.stream != nil {
-			r.stream.mu.RLock()
-			r.offset = FindNextPageBoundary(r.stream.Buffer.Data, r.stream.Buffer.Size, r.stream.Buffer.Head, next)
-			r.stream.mu.RUnlock()
+			// AlignToOggPage scans under the buffer's own read lock and
+			// validates the Ogg version byte, so we can't drift into a
+			// false positive inside a payload.
+			r.offset = r.buffer.AlignToOggPage(next)
 			continue // Retry read at aligned offset
 		}
 

relay/ogg.go

@@ -76,6 +76,12 @@ type Stream struct {
 	PageOffsets     []int64 // Circular list of last ~100 page starts
 	PageIndex       int     // Index for managing PageOffsets circular list
 
+	// FreshHead marks the oldest buffer offset that belongs to the current
+	// playback session. Subscribe() will never hand out a start offset older
+	// than this, which prevents bursting stale audio from a prior source.
+	// Updated by BeginSession() whenever a new source/encoder takes over.
+	FreshHead int64
+
 	// Core streaming infrastructure
 	Buffer    *CircularBuffer          // Audio data buffer (typically 2MB)
 	listeners map[string]chan struct{} // Signal channels for connected listeners
@@ -146,6 +152,58 @@ func (s *Stream) DisconnectListeners() {
 	}
 }
 
+// BeginSession marks the start of a new producer session on this stream
+// (new source client, new transcoder run, new AutoDJ encoder, ...). It:
+//
+//   - advances FreshHead so Subscribe never bursts audio from before this point,
+//   - clears stale Ogg page tracking (different Ogg serial / different format
+//     would otherwise leave PageOffsets pointing at unrelated bytes),
+//   - clears OggHead until the new producer captures fresh headers.
+//
+// Producers should call this once, before they start writing audio to the
+// stream's buffer.
+func (s *Stream) BeginSession() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	head := s.Buffer.SnapshotHead()
+	s.FreshHead = head
+	s.OggHeaderOffset = head
+	s.OggHead = nil
+	s.LastPageOffset = 0
+	s.PageIndex = 0
+	for i := range s.PageOffsets {
+		s.PageOffsets[i] = 0
+	}
+}
+
+// SetOggHead stores the captured Ogg ID/Tags headers and the buffer offset
+// at which the first audio page starts. Stored under lock so concurrent
+// listener subscriptions see a consistent snapshot (no torn slice).
+func (s *Stream) SetOggHead(headers []byte, offset int64) {
+	cp := make([]byte, len(headers))
+	copy(cp, headers)
+
+	s.mu.Lock()
+	s.OggHead = cp
+	s.OggHeaderOffset = offset
+	s.mu.Unlock()
+}
+
+// GetOggHead returns a defensive copy of the stored Ogg headers, or nil if
+// none have been captured yet. The copy means callers can safely write the
+// bytes to a network connection without racing the next SetOggHead.
+func (s *Stream) GetOggHead() []byte {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if s.OggHead == nil {
+		return nil
+	}
+	cp := make([]byte, len(s.OggHead))
+	copy(cp, s.OggHead)
+	return cp
+}
+
 // Broadcast sends data to all listeners via the shared circular buffer.
 //
 // This is the core method that distributes audio data from sources to all connected
@@ -188,11 +246,20 @@ func (s *Stream) Broadcast(data []byte, relay *Relay) {
 	atomic.AddInt64(&relay.BytesIn, int64(len(data)))
 	atomic.AddInt64(&s.BytesIn, int64(len(data)))
 
-	// Track Ogg Page boundaries for alignment
-	// This enables new Opus listeners to start at proper page boundaries
+	// Track Ogg page boundaries for listener alignment.
+	//
+	// We MUST validate the byte that follows the "OggS" magic: the four
+	// bytes 0x4F 0x67 0x67 0x53 also occur inside Opus packet payloads, and
+	// without validation those false positives end up in PageOffsets and
+	// cause new listeners to start mid-packet (audible as glitched/shuffled
+	// audio). Byte index 4 of a real Ogg page header is the version, which
+	// is always 0.
+	//
+	// If the magic lands at the very tail of `data` (no version byte yet),
+	// we skip it — the next Broadcast call will see the full header.
 	if s.IsOggStream {
-		for i := 0; i <= len(data)-4; i++ {
-			if data[i] == 'O' && data[i+1] == 'g' && data[i+2] == 'g' && data[i+3] == 'S' {
+		for i := 0; i <= len(data)-5; i++ {
+			if data[i] == 'O' && data[i+1] == 'g' && data[i+2] == 'g' && data[i+3] == 'S' && data[i+4] == 0 {
 				offset := s.Buffer.Head + int64(i)
 				s.LastPageOffset = offset
 				s.PageOffsets[s.PageIndex%len(s.PageOffsets)] = offset
@@ -265,53 +332,77 @@ func (s *Stream) Subscribe(id string, burstSize int) (int64, chan struct{}) {
 	ch := make(chan struct{}, 1)
 	s.listeners[id] = ch
 
-	// Start at current head minus burst size for instant playback
-	// This gives the listener immediate audio data instead of waiting for new data
-	start := s.Buffer.Head - int64(burstSize)
+	head := s.Buffer.SnapshotHead()
+
+	// If the source has been silent for a while, the buffer holds stale
+	// audio from a previous session. Don't burst it — start at "now" and
+	// the listener will get fresh data as soon as the source resumes.
+	// 2s is long enough to span normal between-frame gaps but short enough
+	// that pauses between songs/sources don't replay old audio.
+	if !s.LastDataReceived.IsZero() && time.Since(s.LastDataReceived) > 2*time.Second {
+		return head, ch
+	}
+
+	// Start at current head minus burst size for instant playback.
+	start := head - int64(burstSize)
 	if start < 0 {
 		start = 0
 	}
+	// Never reach back past the current session's first byte.
+	if start < s.FreshHead {
+		start = s.FreshHead
+	}
 
-	// For Ogg/Opus, align to the oldest known page boundary within the valid buffer range
-	// This is crucial for proper Opus decoding - listeners MUST start at page boundaries
+	// For Ogg/Opus, align to the oldest known page boundary within the valid
+	// buffer range. Listeners MUST start at a page boundary or the decoder
+	// will produce garbage.
 	if s.IsOggStream {
-		validStart := s.Buffer.Head - s.Buffer.Size
+		validStart := head - s.Buffer.Size
+		if validStart < s.FreshHead {
+			validStart = s.FreshHead
+		}
 		if validStart < 0 {
 			validStart = 0
 		}
 
-		// If we have an OggHead persistent storage, we want to start reading
-		// from the Buffer AFTER the initial headers to avoid duplicates.
+		// The Ogg ID/Tags headers live before OggHeaderOffset; we prepend
+		// them separately, so the buffer read must start at or after them.
 		if s.OggHeaderOffset > start {
 			start = s.OggHeaderOffset
 		}
-
 		if start < validStart {
 			start = validStart
 		}
 
-		// Find the best page boundary that is >= start and still valid
-		bestAlign := s.LastPageOffset
-		found := false
+		// Find the smallest tracked page offset that is >= start. This is
+		// the oldest valid alignment we can hand the listener that still
+		// honours their burst request.
+		const noAlign = int64(-1)
+		bestAlign := noAlign
 		for _, po := range s.PageOffsets {
-			// Find the smallest PO that is >= start AND is still valid
-			if po >= start && po >= validStart && po < bestAlign {
-				bestAlign = po
-				found = true
+			if po >= start && po >= validStart && po < head {
+				if bestAlign == noAlign || po < bestAlign {
+					bestAlign = po
+				}
 			}
 		}
-		if found {
-			start = bestAlign
-		} else if bestAlign >= validStart && bestAlign > 0 {
+		if bestAlign != noAlign {
 			start = bestAlign
+		} else if s.LastPageOffset >= start && s.LastPageOffset < head {
+			// Tracked offsets all wrapped out of the window, but the most
+			// recent one is still valid and at/after our start — use it.
+			start = s.LastPageOffset
 		} else {
-			start = s.Buffer.Head // Fallback to now if nothing valid found
+			// No known page boundary at or after `start` — start from head
+			// and the StreamReader's Ogg sync will re-align once new pages
+			// arrive. Never walk the listener backwards.
+			start = head
 		}
 	}
 
 	// Ensure we don't go back further than the buffer allows
-	if s.Buffer.Head-start > s.Buffer.Size {
-		start = s.Buffer.Head - s.Buffer.Size
+	if head-start > s.Buffer.Size {
+		start = head - s.Buffer.Size
 	}
 
 	return start, ch

relay/stream.go

@@ -169,6 +169,9 @@ func EncodeMP3(ctx context.Context, relay *Relay, output *Stream, decoder io.Rea
 	if sampleRate <= 0 {
 		sampleRate = 44100 // Fallback
 	}
+	// Fresh producer session: stops listeners that join during this run from
+	// being bursted with audio buffered by a previous source.
+	output.BeginSession()
 	// Shine MP3 initialization
 	encoder := shine.NewEncoder(sampleRate, 2)
 
@@ -233,6 +236,11 @@ func EncodeOpus(ctx context.Context, relay *Relay, output *Stream, decoder io.Re
 		enc.SetBitrate(bitrate * 1000)
 	}
 
+	// New Opus session: fresh serial number, fresh headers. Wipe any Ogg
+	// state left over from a prior producer so subscribers don't get aligned
+	// to page offsets that now point at bytes from a different stream.
+	output.BeginSession()
+
 	// Ogg encapsulation
 	writer := &streamWriter{stream: output, relay: relay, stats: stats, capture: true}
 	serial := uint32(time.Now().UnixNano())
@@ -254,9 +262,9 @@ func EncodeOpus(ctx context.Context, relay *Relay, output *Stream, decoder io.Re
 	pw.WritePacket(tagsPacket, 0, false, false)
 	pw.Flush()
 
-	// Store for mid-stream listeners
-	output.OggHead = writer.headerBuf.Bytes()
-	output.OggHeaderOffset = output.Buffer.Head
+	// Publish captured headers under lock so concurrent listener
+	// subscriptions see a consistent (headers, offset) pair.
+	output.SetOggHead(writer.headerBuf.Bytes(), output.Buffer.SnapshotHead())
 	writer.capture = false // Stop capturing headers
 
 	pcmBuf := make([]byte, frameSize*channels*2)