relay: add tests covering freshness gate and Ogg page validation

Covers the regressions just fixed:
  - Subscribe on an idle stream returns Head, not a stale burst.
  - Subscribe on an active stream still bursts as requested.
  - FreshHead clamps the burst window after BeginSession.
  - AlignToOggPage rejects 'OggS' with a non-zero version byte and
    finds the real page that follows.
  - Broadcast does not poison PageOffsets with false positives but
    does record a real page when the version byte is 0.
  - BeginSession wipes OggHead, LastPageOffset, PageOffsets,
    PageIndex and advances FreshHead.

Author: Datanoise

relay/stream_test.go

@@ -0,0 +1,192 @@
+package relay
+
+import (
+	"sync"
+	"testing"
+	"time"
+)
+
+// newTestStream builds a Stream of the same shape that GetOrCreateStream
+// produces, but without needing the global Relay wired up.
+func newTestStream(ogg bool) *Stream {
+	return &Stream{
+		MountName:   "/test",
+		listeners:   make(map[string]chan struct{}),
+		Buffer:      NewCircularBuffer(64 * 1024),
+		Started:     time.Now(),
+		ContentType: "audio/ogg",
+		IsOggStream: ogg,
+		Enabled:     true,
+		PageOffsets: make([]int64, 32),
+		mu:          sync.RWMutex{},
+	}
+}
+
+// fakeRelay lets Broadcast update its atomic counters without dragging the
+// HistoryManager / database into a unit test.
+func fakeRelay() *Relay { return &Relay{} }
+
+// TestSubscribeSkipsBurstWhenSourceIdle verifies bug A's fix: if the source
+// hasn't written audio for >2s, a new subscriber must start at Head and not
+// be bursted with stale buffer contents.
+func TestSubscribeSkipsBurstWhenSourceIdle(t *testing.T) {
+	s := newTestStream(false)
+	r := fakeRelay()
+
+	// Pretend a source was active in the past and left audio in the buffer.
+	s.Broadcast(make([]byte, 8192), r)
+	s.mu.Lock()
+	s.LastDataReceived = time.Now().Add(-10 * time.Second)
+	s.mu.Unlock()
+
+	head := s.Buffer.SnapshotHead()
+	start, _ := s.Subscribe("listener-1", 4096)
+	if start != head {
+		t.Fatalf("idle-source subscribe should start at Head=%d, got %d (would replay %d stale bytes)",
+			head, start, head-start)
+	}
+}
+
+// TestSubscribeBurstsWhenSourceFresh verifies the burst still works on an
+// active stream — the freshness gate must not break the normal "instant
+// playback" path.
+func TestSubscribeBurstsWhenSourceFresh(t *testing.T) {
+	s := newTestStream(false)
+	r := fakeRelay()
+
+	// LastDataReceived is set by Broadcast itself to time.Now().
+	s.Broadcast(make([]byte, 8192), r)
+
+	head := s.Buffer.SnapshotHead()
+	start, _ := s.Subscribe("listener-1", 4096)
+	if start >= head {
+		t.Fatalf("fresh-source subscribe should burst, start=%d head=%d", start, head)
+	}
+	if head-start != 4096 {
+		t.Fatalf("expected 4096-byte burst, got %d", head-start)
+	}
+}
+
+// TestSubscribeNeverWalksBackPastFreshHead verifies bug E's fix: even if the
+// Ogg page-alignment fallback would otherwise pick an offset older than the
+// freshness boundary, Subscribe must clamp it.
+func TestSubscribeNeverWalksBackPastFreshHead(t *testing.T) {
+	s := newTestStream(true)
+	r := fakeRelay()
+
+	// Stale audio from a previous session.
+	s.Broadcast(make([]byte, 16*1024), r)
+
+	// New session starts — FreshHead advances to the current Head, all old
+	// page offsets must be invalidated.
+	s.BeginSession()
+	freshHead := s.Buffer.SnapshotHead()
+
+	// Subscribe before any new audio has arrived. LastDataReceived is still
+	// from the old session, so the >2s freshness gate alone won't trip; we
+	// rely on FreshHead.
+	s.mu.Lock()
+	s.LastDataReceived = time.Now() // pretend source is "live"
+	s.mu.Unlock()
+
+	start, _ := s.Subscribe("listener-1", 8192)
+	if start < freshHead {
+		t.Fatalf("subscribe returned start=%d which is BEFORE FreshHead=%d (would replay stale audio)",
+			start, freshHead)
+	}
+}
+
+// TestAlignToOggPageRejectsFalseMagic verifies bug B's fix: bare "OggS"
+// bytes inside a payload are not treated as page boundaries — only the
+// magic followed by a 0 version byte counts.
+func TestAlignToOggPageRejectsFalseMagic(t *testing.T) {
+	buf := NewCircularBuffer(4096)
+
+	// False positive: "OggS" followed by version byte 0xFF (could be Opus
+	// payload data). Must NOT be picked as a page boundary.
+	false1 := []byte{0x00, 0x11, 'O', 'g', 'g', 'S', 0xFF, 0x22, 0x33, 0x44, 0x55, 0x66}
+	buf.Write(false1)
+
+	// Real page: "OggS" followed by version byte 0x00.
+	realHeaderOffset := buf.SnapshotHead() + 4
+	real := []byte{0xAA, 0xBB, 0xCC, 0xDD, 'O', 'g', 'g', 'S', 0x00, 0x00, 0x00, 0x00}
+	buf.Write(real)
+
+	got := buf.AlignToOggPage(0)
+	if got != realHeaderOffset {
+		t.Fatalf("AlignToOggPage should have skipped the version!=0 false positive and "+
+			"returned the real page at offset %d, got %d", realHeaderOffset, got)
+	}
+}
+
+// TestBroadcastIgnoresFalseOggMagic verifies the same validation inside the
+// Broadcast page-tracking path: a false-positive "OggS" in payload bytes
+// must not poison PageOffsets.
+func TestBroadcastIgnoresFalseOggMagic(t *testing.T) {
+	s := newTestStream(true)
+	r := fakeRelay()
+
+	// All-zeros payload with a "OggS" sequence followed by 0xFF (not a real
+	// Ogg version). No tracked page should be recorded.
+	payload := make([]byte, 64)
+	payload[10] = 'O'
+	payload[11] = 'g'
+	payload[12] = 'g'
+	payload[13] = 'S'
+	payload[14] = 0xFF
+	s.Broadcast(payload, r)
+
+	for i, po := range s.PageOffsets {
+		if po != 0 {
+			t.Fatalf("PageOffsets[%d]=%d — a false-positive OggS was tracked", i, po)
+		}
+	}
+
+	// Now a real page (version byte == 0). Must be tracked.
+	real := make([]byte, 32)
+	real[5] = 'O'
+	real[6] = 'g'
+	real[7] = 'g'
+	real[8] = 'S'
+	real[9] = 0x00
+	expected := s.Buffer.SnapshotHead() + 5
+	s.Broadcast(real, r)
+
+	if s.LastPageOffset != expected {
+		t.Fatalf("LastPageOffset=%d, expected %d", s.LastPageOffset, expected)
+	}
+}
+
+// TestBeginSessionWipesOggState verifies bug C's fix: after a producer
+// restart, stale Ogg headers/page tracking must not survive.
+func TestBeginSessionWipesOggState(t *testing.T) {
+	s := newTestStream(true)
+
+	s.SetOggHead([]byte{1, 2, 3, 4}, 100)
+	s.mu.Lock()
+	s.LastPageOffset = 999
+	s.PageOffsets[0] = 999
+	s.PageIndex = 7
+	s.mu.Unlock()
+
+	s.BeginSession()
+
+	if got := s.GetOggHead(); got != nil {
+		t.Fatalf("OggHead should be nil after BeginSession, got %v", got)
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if s.LastPageOffset != 0 {
+		t.Fatalf("LastPageOffset should be reset, got %d", s.LastPageOffset)
+	}
+	if s.PageOffsets[0] != 0 {
+		t.Fatalf("PageOffsets[0] should be reset, got %d", s.PageOffsets[0])
+	}
+	if s.PageIndex != 0 {
+		t.Fatalf("PageIndex should be reset, got %d", s.PageIndex)
+	}
+	head := s.Buffer.SnapshotHead()
+	if s.FreshHead != head {
+		t.Fatalf("FreshHead=%d should equal current Head=%d after BeginSession", s.FreshHead, head)
+	}
+}