Fix REAL bug: HTTP timeout was truncating Radio Paradise blocks

ROOT CAUSE IDENTIFIED:
The previous "wait for playback duration" workaround was masking the real
issue. Radio Paradise blocks last ~20 minutes (1200s), but the HTTP timeout
was only 180 seconds, causing premature stream termination.

With backpressure from the audio pipeline, HTTP download proceeds at real-time
pace. A 20-minute block takes ~20 minutes to download. The 180s timeout
was killing the connection after 3 minutes, resulting in incomplete blocks.

Changes:
1. **Increase HTTP block_timeout: 180s → 7200s (2 hours)**
   - Allows complete download of even the longest blocks
   - Comment explains why such a long timeout is needed

2. **Increase MPSC channel sizes: 16 → 60 chunks**
   - Matches TimerNode max_lead_time (3.0s / 0.05s = 60 chunks)
   - Prevents stop-and-go backpressure pattern
   - Allows smooth buffering as intended

3. **Replace workaround with proper channel drainage**
   - Use tx.closed().await instead of sleep()
   - Guarantees all buffered chunks are processed
   - More architecturally sound solution

4. **Add comprehensive diagnostic traces**
   - Log expected vs actual block duration
   - Detect premature EOF (< 95% of expected duration)
   - Track bytes decoded and HTTP Content-Length
   - Monitor backpressure blocking with timing

This fixes the streaming completely. The block will now:
- Download for the full ~20 minutes (real-time with backpressure)
- Decode all audio data without truncation
- Process all chunks before pipeline shutdown

pmoparadise/examples/stream_block.rs

@@ -208,11 +208,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     source_flac.push_block_id(block.event);
     tracing::debug!("RadioParadiseStreamSource (FLAC) created with block {}", block.event);
 
-    let mut timer_flac = TimerNode::new(3.0);
-    tracing::debug!("TimerNode (FLAC) created with 3.0s max lead time");
+    // Calculate channel size to match max_lead_time
+    // With 50ms chunks and 3.0s lead time: 3.0 / 0.05 = 60 chunks
+    let max_lead_time = 3.0;
+    let chunk_duration_sec = 0.05;
+    let channel_size = ((max_lead_time / chunk_duration_sec) as usize).max(16);
+    tracing::debug!("Calculated channel size: {} chunks ({:.1}s buffer)", channel_size, channel_size as f64 * chunk_duration_sec);
 
-    let (streaming_sink, stream_handle) = StreamingFlacSink::new(encoder_options.clone(), 16);
-    tracing::debug!("StreamingFlacSink created");
+    let mut timer_flac = TimerNode::with_channel_size(max_lead_time, channel_size);
+    tracing::debug!("TimerNode (FLAC) created with {:.1}s max lead time, {} chunk buffer", max_lead_time, channel_size);
+
+    let (streaming_sink, stream_handle) = StreamingFlacSink::new(encoder_options.clone(), channel_size.min(255) as u8);
+    tracing::debug!("StreamingFlacSink created with {} chunk buffer", channel_size);
 
     timer_flac.register(Box::new(streaming_sink));
     source_flac.register(Box::new(timer_flac));
@@ -226,11 +233,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     source_ogg.push_block_id(block.event);
     tracing::debug!("RadioParadiseStreamSource (OGG) created with block {}", block.event);
 
-    let mut timer_ogg = TimerNode::new(3.0);
-    tracing::debug!("TimerNode (OGG) created with 3.0s max lead time");
+    let mut timer_ogg = TimerNode::with_channel_size(max_lead_time, channel_size);
+    tracing::debug!("TimerNode (OGG) created with {:.1}s max lead time, {} chunk buffer", max_lead_time, channel_size);
 
-    let (ogg_sink, ogg_handle) = StreamingOggFlacSink::new(encoder_options, 16);
-    tracing::debug!("StreamingOggFlacSink created");
+    let (ogg_sink, ogg_handle) = StreamingOggFlacSink::new(encoder_options, channel_size.min(255) as u8);
+    tracing::debug!("StreamingOggFlacSink created with {} chunk buffer", channel_size);
 
     timer_ogg.register(Box::new(ogg_sink));
     source_ogg.register(Box::new(timer_ogg));

pmoparadise/src/client.rs

@@ -19,7 +19,10 @@ pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/";
 pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 30;
 
 /// Default timeout for large block downloads/streams
-pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180;
+/// IMPORTANT: Radio Paradise blocks can be ~20 minutes long, and with backpressure
+/// from the audio pipeline, the HTTP stream must stay open for the entire duration.
+/// Setting this to 2 hours to safely handle even the longest blocks.
+pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 7200; // 2 hours
 
 /// Default User-Agent
 pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0";

pmoparadise/src/radio_paradise_stream_source.rs

@@ -89,7 +89,11 @@ impl RadioParadiseStreamSourceLogic {
         order: &mut u64,
     ) -> Result<(f64, Instant), AudioError> {
         // Télécharger le FLAC
-        tracing::debug!("Sending HTTP GET request for block FLAC");
+        tracing::info!(
+            "Sending HTTP GET request for block FLAC (expected duration: {:.1}min, url: {})",
+            block.length as f64 / 60000.0,
+            block.url
+        );
         let response = self.client.client
             .get(&block.url)
             .timeout(self.client.block_timeout)
@@ -105,6 +109,17 @@ impl RadioParadiseStreamSourceLogic {
             )));
         }
 
+        // Vérifier la taille du contenu si disponible
+        if let Some(content_length) = response.content_length() {
+            tracing::info!(
+                "HTTP Content-Length: {} bytes ({:.1} MB)",
+                content_length,
+                content_length as f64 / 1_048_576.0
+            );
+        } else {
+            tracing::warn!("HTTP response has no Content-Length header");
+        }
+
         // Créer un stream reader
         tracing::debug!("Creating byte stream reader");
         let byte_stream = response.bytes_stream().map(|result| {
@@ -176,14 +191,19 @@ impl RadioParadiseStreamSourceLogic {
 
         // Traiter les chunks audio
         let mut chunk_count = 0;
+        let mut total_bytes_decoded = 0u64;
+        let expected_duration_sec = block.length as f64 / 1000.0;
+
         loop {
             // Vérifier stop_token
             if stop_token.is_cancelled() {
                 // Retourner le timestamp actuel et start_instant si on est interrompu
                 let current_timestamp = total_samples as f64 / sample_rate as f64;
-                tracing::debug!(
-                    "Block decode cancelled: sent {} chunks, {:.2}s duration",
-                    chunk_count, current_timestamp
+                tracing::warn!(
+                    "Block decode CANCELLED: sent {} chunks, {:.2}s duration ({:.1}% of expected {:.2}s), decoded {} bytes",
+                    chunk_count, current_timestamp,
+                    (current_timestamp / expected_duration_sec) * 100.0,
+                    expected_duration_sec, total_bytes_decoded
                 );
                 return Ok((current_timestamp, start_instant));
             }
@@ -194,12 +214,23 @@ impl RadioParadiseStreamSourceLogic {
                     .map_err(|e| AudioError::ProcessingError(format!("Read error: {}", e)))?;
 
                 if read == 0 {
-                    tracing::debug!(
-                        "FLAC decode EOF reached: sent {} chunks, {:.2}s total",
-                        chunk_count, total_samples as f64 / sample_rate as f64
-                    );
+                    let actual_duration = total_samples as f64 / sample_rate as f64;
+                    let percentage = (actual_duration / expected_duration_sec) * 100.0;
+
+                    if percentage < 95.0 {
+                        tracing::error!(
+                            "FLAC decode EOF PREMATURE: sent {} chunks, {:.2}s actual vs {:.2}s expected ({:.1}%), decoded {} bytes",
+                            chunk_count, actual_duration, expected_duration_sec, percentage, total_bytes_decoded
+                        );
+                    } else {
+                        tracing::info!(
+                            "FLAC decode EOF reached: sent {} chunks, {:.2}s duration ({:.1}% of expected), decoded {} bytes",
+                            chunk_count, actual_duration, percentage, total_bytes_decoded
+                        );
+                    }
                     break; // EOF
                 }
+                total_bytes_decoded += read as u64;
                 pending.extend_from_slice(&read_buf[..read]);
             }
 
@@ -540,7 +571,7 @@ impl NodeLogic for RadioParadiseStreamSourceLogic {
         }
 
         // Envoyer EndOfStream avec le timestamp du dernier chunk
-        tracing::debug!("Sending EndOfStream with timestamp {:.2}s", last_timestamp);
+        tracing::info!("Sending EndOfStream with timestamp {:.2}s to {} outputs", last_timestamp, output.len());
         let eos = AudioSegment::new_end_of_stream(order, last_timestamp);
         for tx in &output {
             tx.send(eos.clone())
@@ -548,31 +579,23 @@ impl NodeLogic for RadioParadiseStreamSourceLogic {
                 .map_err(|_| AudioError::ChildDied)?;
         }
 
-        // IMPORTANT: Attendre que la durée réelle du bloc soit écoulée avant de fermer le channel
-        // Sinon, le TimerNode reçoit EOF et se termine avant d'avoir fini de diffuser tous les chunks
-        if let Some(start_instant) = last_start_instant {
-            let elapsed = start_instant.elapsed().as_secs_f64();
-            if elapsed < last_timestamp {
-                let remaining = last_timestamp - elapsed;
-                tracing::info!(
-                    "Waiting {:.2}s for block playback to complete (elapsed={:.2}s, duration={:.2}s)",
-                    remaining, elapsed, last_timestamp
-                );
+        // IMPORTANT: Attendre que tous les channels soient fermés par les enfants
+        // Cela garantit que tous les chunks (y compris ceux en attente dans les buffers MPSC)
+        // ont été traités avant que nous ne fermions notre bout
+        tracing::info!("Waiting for all child nodes to close their channels...");
+        for (i, tx) in output.iter().enumerate() {
+            tracing::debug!("Waiting for child {} to close channel...", i);
+            tx.closed().await;
+            tracing::debug!("Child {} channel closed", i);
+        }
+        tracing::info!("All child channels closed, pipeline complete");
 
-                tokio::select! {
-                    _ = tokio::time::sleep(Duration::from_secs_f64(remaining)) => {
-                        tracing::debug!("Block playback duration complete");
-                    }
-                    _ = stop_token.cancelled() => {
-                        tracing::debug!("Cancelled while waiting for playback completion");
-                    }
-                }
-            } else {
-                tracing::debug!(
-                    "Block already played in real-time (elapsed={:.2}s >= duration={:.2}s)",
-                    elapsed, last_timestamp
-                );
-            }
+        if let Some(start_instant) = last_start_instant {
+            let total_elapsed = start_instant.elapsed().as_secs_f64();
+            tracing::info!(
+                "Block processing complete: duration={:.2}s, total_elapsed={:.2}s ({:.1}% of real-time)",
+                last_timestamp, total_elapsed, (total_elapsed / last_timestamp) * 100.0
+            );
         }
 
         Ok(())