fix: larger audio burst + Ogg page-align fallback + SSE bandwidth rate

- Bandwidth counters in the /admin SSE feed were sending cumulative
  byte totals, which the dashboard formatted as MB/s (e.g. "636.9 MB/s
  inbound" on any server with traffic). Compute a per-tick rate from
  consecutive samples and send that.
- Bump listener default burst 128 → 512 KiB, transcoder input burst
  32 → 256 KiB, and the listener read chunk 4 → 64 KiB so short
  network stalls don't drain client caches.
- Track 2048 Ogg page offsets (was 128) — with a 512 KiB burst the
  intended start landed beyond any tracked page, and the page-align
  fallback used LastPageOffset (near head), dropping the real burst
  to ~0 bytes. That was the "underrun every 4–5 s on reconnect".
- Rework Subscribe Ogg page-align logic: pick the oldest tracked page
  >= intended start; if none, fall back to the oldest tracked page
  overall, then LastPageOffset, then raw burst offset.

Author: DatanoiseTV

relay/pipeline.go

@@ -90,7 +90,7 @@ func NewVideoTrack(codec string, bufferSize int) *Track {
 			Buffer:      NewCircularBuffer(bufferSize),
 			Started:     time.Now(),
 			Enabled:     true,
-			PageOffsets: make([]int64, 128),
+			PageOffsets: make([]int64, 2048),
 		},
 	}
 }

relay/relay.go

@@ -106,7 +106,11 @@ func (r *Relay) GetOrCreateStreamSized(mount string, bufferSize int) *Stream {
 		IsOggStream: false,
 		Enabled:     true,
 		CurrentSong: "N/A",
-		PageOffsets: make([]int64, 128), // Track last 128 Ogg pages
+		// Track last 2048 Ogg pages so Subscribe can find a page-boundary
+		// alignment anywhere in the buffer — 128 covered barely 3–5 s at
+		// typical Opus framing and truncated the actual burst delivered to
+		// listeners ("underrun every 4–5 s on reconnect").
+		PageOffsets: make([]int64, 2048),
 		Frames:      NewFrameHub(),
 	}
 	r.Streams[mount] = s

relay/stream.go

@@ -389,22 +389,37 @@ func (s *Stream) Subscribe(id string, burstSize int) (int64, chan struct{}) {
 			start = validStart
 		}
 
-		// Find the best page boundary that is >= start and still valid
-		bestAlign := s.LastPageOffset
-		found := false
+		// Prefer the oldest tracked page boundary that is still >= start and
+		// within the valid buffer range — this maximises the burst delivered
+		// to the listener. Falling back to LastPageOffset (newest page) used
+		// to cap the burst at near-zero whenever PageOffsets didn't happen
+		// to cover far enough back, which manifested as "client underrun
+		// every few seconds on reconnect".
+		bestAlign := int64(-1)
+		oldestValid := int64(-1)
 		for _, po := range s.PageOffsets {
-			// Find the smallest PO that is >= start AND is still valid
-			if po >= start && po >= validStart && po < bestAlign {
+			if po < validStart || po == 0 {
+				continue
+			}
+			if oldestValid < 0 || po < oldestValid {
+				oldestValid = po
+			}
+			if po >= start && (bestAlign < 0 || po < bestAlign) {
 				bestAlign = po
-				found = true
 			}
 		}
-		if found {
+		switch {
+		case bestAlign >= 0:
 			start = bestAlign
-		} else if bestAlign >= validStart && bestAlign > 0 {
-			start = bestAlign
-		} else {
-			// No valid Ogg page boundaries found — fall back to burst-based offset
+		case oldestValid >= 0:
+			// Intended start is older than any tracked page — use the
+			// oldest we have, which is still newer than the buffer tail.
+			start = oldestValid
+		case s.LastPageOffset >= validStart && s.LastPageOffset > 0:
+			// Absolute last resort: at least align on the most recent page.
+			start = s.LastPageOffset
+		default:
+			// Nothing tracked yet — fall back to burst-based offset.
 			start = s.Buffer.Head - int64(burstSize)
 			if start < validStart {
 				start = validStart

relay/transcode.go

@@ -154,8 +154,10 @@ func (tm *TranscoderManager) performTranscode(ctx context.Context, inst *Transco
 
 	// 2. Subscribe to input
 	id := fmt.Sprintf("transcoder-%s", inst.Config.Name)
-	// We use a small burst to ensure the decoder gets enough data to start
-	offset, signal := input.Subscribe(id, 32*1024)
+	// Burst 256 KiB so the decoder has several seconds of data to initialise
+	// even when the source bitrate is low — 32 KiB left strict Opus / FLAC
+	// decoders short on bytes at startup, producing a noisy first packet.
+	offset, signal := input.Subscribe(id, 256*1024)
 	defer input.Unsubscribe(id)
 
 	// For live Ogg inputs, the subscribe offset lands mid-stream. The decoder

relay/webrtc.go

@@ -167,7 +167,7 @@ func (wm *WebRTCManager) HandleSourceOffer(mount string, offer webrtc.SessionDes
 		stream.IsOggStream = true
 		stream.SourceIP = "webrtc-source"
 		// Reset page offsets so we don't sync to old pages from a previous session
-		stream.PageOffsets = make([]int64, 128)
+		stream.PageOffsets = make([]int64, 2048)
 		stream.PageIndex = 0
 		stream.OggHeaderOffset = headOffset
 		stream.mu.Unlock()