fix: ensure transcoder correctly waits for source and capture persistence headers for new listeners

Author: Datanoise

relay/relay.go

@@ -87,6 +87,7 @@ type Stream struct {
 	CurrentSong  string
 	Public       bool
 	Visible      bool
+	IsTranscoded bool // True if this stream is an output of a transcoder
 
 	LastDataReceived time.Time
 
@@ -180,6 +181,15 @@ func (r *Relay) RemoveStream(mount string) {
 	}
 }
 
+func (r *Relay) GetStreamVisibility(mount string) bool {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	if s, ok := r.Streams[mount]; ok {
+		return s.Visible
+	}
+	return false
+}
+
 // GetStream safely retrieves a stream
 func (r *Relay) GetStream(mount string) (*Stream, bool) {
 	r.mu.RLock()
@@ -355,6 +365,7 @@ type StreamStats struct {
 	CurrentSong    string
 	Public         bool
 	Visible        bool
+	IsTranscoded   bool
 	ListenersCount int
 	Uptime         string
 	Health         float64
@@ -410,6 +421,7 @@ func (s *Stream) Snapshot() StreamStats {
 		CurrentSong:    s.CurrentSong,
 		Public:         s.Public,
 		Visible:        s.Visible,
+		IsTranscoded:   s.IsTranscoded,
 		ListenersCount: len(s.listeners),
 		Uptime:         s.uptimeLocked(),
 		Health:         health,

relay/transcode.go

@@ -110,14 +110,29 @@ func (tm *TranscoderManager) runTranscoder(ctx context.Context, inst *Transcoder
 }
 
 func (tm *TranscoderManager) performTranscode(ctx context.Context, inst *TranscoderInstance) {
-	input, ok := tm.relay.GetStream(inst.Config.InputMount)
-	if !ok {
-		return
+	var input *Stream
+	var ok bool
+
+	// 1. Wait for input stream to become available
+	for {
+		input, ok = tm.relay.GetStream(inst.Config.InputMount)
+		if ok {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(1 * time.Second):
+			// Keep waiting
+		}
 	}
 
-	// 1. Subscribe to input
+	logrus.Infof("Transcoder %s: Input stream %s found, initializing...", inst.Config.Name, inst.Config.InputMount)
+
+	// 2. Subscribe to input
 	id := fmt.Sprintf("transcoder-%s", inst.Config.Name)
-	offset, signal := input.Subscribe(id, 0) // No burst for transcoder
+	// We use a small burst to ensure the decoder gets enough data to start
+	offset, signal := input.Subscribe(id, 32*1024) 
 	defer input.Unsubscribe(id)
 
 	reader := &StreamReader{
@@ -128,29 +143,26 @@ func (tm *TranscoderManager) performTranscode(ctx context.Context, inst *Transco
 		ID:     id,
 	}
 
-	// 2. Decode (assuming MP3 input for now as standard)
+	// 3. Decode
 	decoder, err := mp3.NewDecoder(reader)
 	if err != nil {
 		logrus.WithError(err).Errorf("Transcoder %s: Failed to initialize decoder for input %s", inst.Config.Name, inst.Config.InputMount)
 		return
 	}
 
-	// 3. Create Output Stream
+	// 4. Create Output Stream
 	output := tm.relay.GetOrCreateStream(inst.Config.OutputMount)
 	output.Name = fmt.Sprintf("%s (%s %dK)", input.Name, inst.Config.Format, inst.Config.Bitrate)
 	output.Bitrate = fmt.Sprintf("%d", inst.Config.Bitrate)
-	output.ContentType = "audio/mpeg"
-	if inst.Config.Format == "opus" {
-		output.ContentType = "audio/ogg"
-	}
-
-	// 4. Encode & Broadcast
+	output.IsTranscoded = true
+	output.Visible = tm.relay.GetStreamVisibility(inst.Config.InputMount) // Follow input visibility
+	
 	if inst.Config.Format == "mp3" {
+		output.ContentType = "audio/mpeg"
 		tm.encodeMP3(ctx, inst, decoder, output)
 	} else if inst.Config.Format == "opus" {
+		output.ContentType = "audio/ogg"
 		tm.encodeOpus(ctx, inst, decoder, output)
-	} else {
-		logrus.Warnf("Transcoding format %s not yet implemented", inst.Config.Format)
 	}
 }
 

server/server.go

@@ -1660,6 +1660,7 @@ type streamEventInfo struct {
 	BytesDropped int64   `json:"bytes_dropped"`
 	CurrentSong  string  `json:"song"`
 	Health       float64 `json:"health"`
+	IsTranscoded bool    `json:"is_transcoded"`
 }
 
 type relayEventInfo struct {
@@ -1683,7 +1684,7 @@ func (s *Server) collectStatsPayload(user *config.User) ([]byte, error) {
 				Mount: st.MountName, Name: st.Name, Listeners: lc, Bitrate: st.Bitrate,
 				Uptime: st.Uptime, ContentType: st.ContentType, SourceIP: st.SourceIP,
 				BytesIn: st.BytesIn, BytesOut: st.BytesOut, BytesDropped: st.BytesDropped,
-				CurrentSong: st.CurrentSong, Health: st.Health,
+				CurrentSong: st.CurrentSong, Health: st.Health, IsTranscoded: st.IsTranscoded,
 			})
 			if st.SourceIP == "relay-pull" {
 				tr++

server/templates/admin.html

@@ -50,6 +50,7 @@
         .role-badge { font-size: 0.7rem; padding: 0.1rem 0.4rem; border-radius: 4px; font-weight: 800; text-transform: uppercase; margin-left: 0.5rem; }
         .role-super { background: var(--primary); color: var(--bg); }
         .role-admin { background: var(--border); color: var(--text); }
+        .badge-transcoded { background: rgba(168, 85, 247, 0.1); color: #a855f7; font-size: 0.6rem; padding: 0.1rem 0.4rem; border-radius: 4px; font-weight: 800; border: 1px solid rgba(168, 85, 247, 0.2); margin-left: 0.5rem; text-transform: uppercase; }
         .status-pill { display: inline-flex; align-items: center; gap: 0.4rem; background: rgba(16, 185, 129, 0.1); color: var(--success); padding: 0.2rem 0.6rem; border-radius: 100px; font-size: 0.75rem; font-weight: 700; }
         .status-pill.inactive { color: var(--text-dim); background: rgba(255,255,255,0.05); }
         .dot { width: 6px; height: 6px; background: currentColor; border-radius: 50%; }
@@ -564,7 +565,26 @@ <h2>Active Transcoders</h2>
 
                     data.streams.forEach(function(s) {
                         var prev = streamHistory[s.mount] || { in: s.bytes_in, out: s.bytes_out }; var curIn = (s.bytes_in - prev.in) / 1024, curOut = (s.bytes_out - prev.out) / 1024; streamHistory[s.mount] = { in: s.bytes_in, out: s.bytes_out };
-                        var clone = rowTemplate.content.cloneNode(true); clone.querySelector('.row-mount').textContent = s.mount; clone.querySelector('.row-ip').textContent = s.ip; clone.querySelector('.row-song').textContent = s.song;
+                        var clone = rowTemplate.content.cloneNode(true);
+                        
+                        var mountCell = clone.querySelector('.row-mount');
+                        mountCell.textContent = s.mount;
+                        if (s.ip === 'relay-pull') {
+                            var badge = document.createElement('span');
+                            badge.className = 'badge-transcoded';
+                            badge.style.background = 'rgba(56, 189, 248, 0.1)';
+                            badge.style.color = '#38bdf8';
+                            badge.style.borderColor = 'rgba(56, 189, 248, 0.2)';
+                            badge.textContent = 'RELAY';
+                            mountCell.parentElement.appendChild(badge);
+                        } else if (s.is_transcoded) {
+                            var badge = document.createElement('span');
+                            badge.className = 'badge-transcoded';
+                            badge.textContent = 'TRANSCODED';
+                            mountCell.parentElement.appendChild(badge);
+                        }
+
+                        clone.querySelector('.row-ip').textContent = s.ip; clone.querySelector('.row-song').textContent = s.song;
                         clone.querySelector('.row-listeners').textContent = s.listeners; clone.querySelector('.row-in').textContent = curIn.toFixed(1); clone.querySelector('.row-out').textContent = curOut.toFixed(1);
 
                         var health = s.health;