Refactor audio pipeline for multi-client streaming and improved error handling

Refactor the audio pipeline to support multi-client streaming with new OggFlacStreamHandle and StreamingOggFlacSink.

- Replace DirectOggFlacSink with StreamingOggFlacSink in pipeline
- Update documentation and comments to reflect multi-client support
- Add detailed warning logs for buffer underruns and client disconnections
- Remove TimerBufferNode from pipeline as it's no longer needed
- Update connect() calls to subscribe() for new streaming behavior

This change enables multiple clients to subscribe to the same audio stream, with each subscription getting a live feed from the current point in time.

pmoaudio-ext/src/sinks/direct_ogg_flac_sink.rs

@@ -44,7 +44,7 @@ use pmoflac::{EncoderOptions, PcmFormat};
 use tokio::io::{AsyncRead, ReadBuf};
 use tokio::sync::{mpsc, watch, Mutex};
 use tokio_util::sync::CancellationToken;
-use tracing::debug;
+use tracing::{debug, warn};
 
 use crate::sinks::byte_stream_reader::{ByteStreamReader, PcmChunk};
 use crate::sinks::chunk_to_pcm::chunk_to_pcm_bytes;
@@ -228,7 +228,11 @@ impl NodeLogic for DirectOggFlacSinkLogic {
                                     duration_sec,
                                 };
                                 if tx.send(pcm_chunk).await.is_err() {
-                                    debug!("DirectOggFlacSink: pcm_tx send failed (client disconnected), clearing pcm_tx");
+                                    warn!(
+                                        ts = seg.timestamp_sec,
+                                        "DirectOggFlacSink: chunk dropped (client disconnected at {:.3}s), waiting for reconnect",
+                                        seg.timestamp_sec,
+                                    );
                                     *self.pcm_tx.lock().await = None;
                                 }
                             }
@@ -348,7 +352,13 @@ async fn run_ogg_encoder(
 
                     let ogg_page = Bytes::from(ogg.create_page(&frame, false, false, false));
                     if pipe_writer.write_all(&ogg_page).await.is_err() {
-                        // Client déconnecté
+                        // Client déconnecté — le pipe HTTP s'est rompu
+                        warn!(
+                            samples = encoded_samples,
+                            "DirectOggFlacSink: OGG pipe broken after {} samples ({:.3}s), client disconnected",
+                            encoded_samples,
+                            encoded_samples as f64 / DIRECT_OGG_FLAC_SAMPLE_RATE as f64,
+                        );
                         return Ok(());
                     }
                 }

pmoaudio/src/nodes/timer_buffer_node.rs

@@ -135,6 +135,14 @@ impl TimerBufferNodeLogic {
                 self.buffer.len()
             );
 
+            if self.buffer.is_empty() && self.buffered_time_sec < self.capacity_sec * 0.1 {
+                tracing::warn!(
+                    "TimerBufferNode: buffer underrun at ts={:.3}s (capacity={:.1}s) — source too slow or stalled",
+                    segment.timestamp_sec,
+                    self.capacity_sec,
+                );
+            }
+
             send_to_children(std::any::type_name::<Self>(), output, segment).await?;
         }
         Ok(())

pmowebrenderer/src/pipeline.rs

@@ -7,9 +7,12 @@
 
 use std::sync::Arc;
 
-use pmoaudio::{AudioSegment, PositionHandle, PositionTrackerNode, ResamplingNode, TimerBufferNode, ToI24Node};
+use pmoaudio::{AudioSegment, PositionTrackerNode, ResamplingNode, ToI24Node};
 use pmometadata::{MemoryTrackMetadata, TrackMetadata};
-use pmoaudio_ext::sinks::{DirectOggFlacHandle, DirectOggFlacSink, DIRECT_OGG_FLAC_SAMPLE_RATE};
+use pmoaudio_ext::sinks::{
+    OggFlacStreamHandle, StreamingOggFlacSink,
+    DIRECT_OGG_FLAC_BITS_PER_SAMPLE, DIRECT_OGG_FLAC_SAMPLE_RATE,
+};
 use pmoaudio_ext::UriSource;
 use pmoflac::EncoderOptions;
 use tokio::sync::mpsc;
@@ -56,10 +59,10 @@ impl PipelineHandle {
 /// Pipeline audio complet pour une instance WebRenderer.
 ///
 /// Créé au `POST /register`. Le flux OGG-FLAC est accessible via `flac_handle`
-/// (un seul client à la fois, reconnectable à chaque Play).
+/// (multi-clients, chaque `subscribe()` retourne un nouveau flux depuis le point courant).
 pub struct InstancePipeline {
-    /// Handle vers le sink OGG-FLAC — clonable, reconnectable à chaque Play.
-    pub flac_handle: DirectOggFlacHandle,
+    /// Handle vers le sink OGG-FLAC — clonable, chaque subscribe() donne un flux live.
+    pub flac_handle: OggFlacStreamHandle,
     pub pipeline_handle: PipelineHandle,
 }
 
@@ -79,21 +82,18 @@ impl InstancePipeline {
         // La backpressure remonte naturellement depuis le pipe duplex jusqu'à la source.
         use pmoaudio::pipeline::AudioPipelineNode;
 
-        let (sink, flac_handle) = DirectOggFlacSink::new(EncoderOptions::default());
+        let (sink, flac_handle) = StreamingOggFlacSink::new(
+            EncoderOptions::default(),
+            DIRECT_OGG_FLAC_BITS_PER_SAMPLE,
+        );
 
-        // Nœud de suivi de position : lit le timestamp des chunks sortant du buffer
+        // Nœud de suivi de position : lit le timestamp des chunks sortant vers le sink
         let (mut position_tracker, position_handle) = PositionTrackerNode::new();
         position_tracker.register(sink.boxed());
 
-        // Nœud de pacing : régule le débit pour éviter les rafales et pertes de segments
-        // 2s de buffer absorbe les irrégularités de la source réseau
-        let mut timer_buffer = TimerBufferNode::new(2.0);
-        timer_buffer.register(position_tracker.boxed());
-
         // Nœud de conversion de profondeur : tout type entier → I24
-        // Placé avant le buffer pour réduire la mémoire utilisée
         let mut to_i24 = ToI24Node::new();
-        to_i24.register(timer_buffer.boxed());
+        to_i24.register(position_tracker.boxed());
 
         // Nœud de rééchantillonnage : n'importe quel sample rate → 96 kHz
         let mut resampler = ResamplingNode::new(DIRECT_OGG_FLAC_SAMPLE_RATE);

pmowebrenderer/src/registry.rs

@@ -29,8 +29,8 @@ pub struct WebRendererInstance {
     pub udn: String,
     pub device_instance: Arc<DeviceInstance>,
     pub state: SharedState,
-    /// Handle vers le sink OGG-FLAC — clonable, reconnectable à chaque Play.
-    pub flac_handle: pmoaudio_ext::sinks::DirectOggFlacHandle,
+    /// Handle vers le sink OGG-FLAC — clonable, chaque subscribe() donne un flux live.
+    pub flac_handle: pmoaudio_ext::sinks::OggFlacStreamHandle,
     pub pipeline: PipelineHandle,
     pub created_at: SystemTime,
 }
@@ -116,11 +116,11 @@ impl RendererRegistry {
         Ok((stream_url, udn))
     }
 
-    /// Retourne le DirectOggFlacHandle pour l'endpoint /stream (clonable).
+    /// Retourne le OggFlacStreamHandle pour l'endpoint /stream (clonable).
     pub fn get_flac_handle(
         &self,
         instance_id: &str,
-    ) -> Option<pmoaudio_ext::sinks::DirectOggFlacHandle> {
+    ) -> Option<pmoaudio_ext::sinks::OggFlacStreamHandle> {
         self.instances
             .read()
             .get(instance_id)

pmowebrenderer/src/stream.rs

@@ -44,7 +44,7 @@ pub async fn stream_handler(
         }
     };
 
-    let stream = handle.connect().await;
+    let stream = handle.subscribe();
 
     info!(instance_id = %instance_id, "FLAC stream started");