From 765070c4b0c27fe8638d5d8a7a4e935417d75b45 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sun, 16 Nov 2025 22:02:00 +0100 Subject: [PATCH] =?UTF-8?q?Nouveau=20broadcast=20strat=C3=A9gie?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pmoaudio-ext/src/sinks/broadcast_pacing.rs | 83 ++---------- pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 112 ++++++++++++---- .../src/sinks/streaming_ogg_flac_sink.rs | 75 ++++++++--- pmoaudio-ext/src/sinks/timed_broadcast.rs | 124 ++++++++++++++---- 4 files changed, 251 insertions(+), 143 deletions(-) diff --git a/pmoaudio-ext/src/sinks/broadcast_pacing.rs b/pmoaudio-ext/src/sinks/broadcast_pacing.rs index 0daeb0de..32666c6b 100644 --- a/pmoaudio-ext/src/sinks/broadcast_pacing.rs +++ b/pmoaudio-ext/src/sinks/broadcast_pacing.rs @@ -40,85 +40,20 @@ impl BroadcastPacer { } } - /// Check timing and apply pacing + /// Check timing and apply pacing - NO-OP VERSION /// - /// This function: - /// 1. Detects TopZeroSync (audio_timestamp < 0.1 after >1s) and marks pending reset - /// 2. On next chunk, resets timer with elapsed=0 guarantee - /// 3. Drops frames that are late (audio_ts < elapsed) - /// 4. Sleeps if too far ahead (lead_time > max_lead_time) + /// Pacing is now handled entirely by the expiration-based system in + /// TimedBroadcast. This method is kept for backward compatibility + /// but always returns Ok(()). /// /// # Returns /// - /// - `Ok(())` if frame is on time or successfully paced - /// - `Err(SkipFrame)` if frame is too late and should be dropped + /// - Always returns `Ok(())` pub async fn check_and_pace(&mut self, audio_timestamp: f64) -> Result<(), SkipFrame> { - // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 1. DÉTECTION timestamp proche de 0 → marquer reset ║ - // ║ Quand timestamp < 0.1s, c'est un nouveau morceau ║ - // ╚═══════════════════════════════════════════════════════════════╝ - if audio_timestamp < 0.1 && !self.pending_reset { - trace!( - "{} broadcaster: Timestamp near zero detected, will reset timer on next chunk", - self.label - ); - self.pending_reset = true; - } - - // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 2. RESET TIMER si pending ║ - // ║ Le reset se fait AVANT le calcul d'elapsed pour garantir ║ - // ║ elapsed=0 pour le premier chunk du nouveau morceau ║ - // ╚═══════════════════════════════════════════════════════════════╝ - let elapsed = if self.pending_reset { - self.start_time = Instant::now(); - self.pending_reset = false; - trace!( - "{} broadcaster: Timer reset at audio_ts={:.3}s", - self.label, audio_timestamp - ); - 0.0 // Garantit elapsed=0 pour ce chunk - } else { - self.start_time.elapsed().as_secs_f64() - }; - - // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 3. CALCUL DU LEAD TIME ║ - // ║ lead_time > 0 : en avance (OK) ║ - // ║ lead_time < 0 : en retard (SKIP) ║ - // ╚═══════════════════════════════════════════════════════════════╝ - let lead_time = audio_timestamp - elapsed; - - // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 4. DROP FRAMES EN RETARD ║ - // ╚═══════════════════════════════════════════════════════════════╝ - if lead_time < 0.0 { - warn!( - "{}: Dropping late frame: audio_ts={:.3}s, elapsed={:.3}s, lag={:.3}s", - self.label, audio_timestamp, elapsed, -lead_time - ); - return Err(SkipFrame); - } - - // ╔═══════════════════════════════════════════════════════════════╗ - // ║ 5. BACKPRESSURE NATURELLE - Pas de sleep ! ║ - // ║ ║ - // ║ Le pacing vient de : ║ - // ║ - TimerBufferNode en amont (envoi régulier à 50ms/chunk) ║ - // ║ - Capacité limitée du broadcast channel ║ - // ║ - Client HTTP qui lit à vitesse réelle ║ - // ║ ║ - // ║ Pas besoin de sleep explicite qui causerait des bursts ║ - // ╚═══════════════════════════════════════════════════════════════╝ - - // Log pour info si on est très en avance, mais on ne dort PAS - if self.max_lead_time > 0.0 && lead_time > self.max_lead_time { - trace!( - "{} broadcaster: lead_time={:.3}s > max={:.3}s (audio_ts={:.3}s, elapsed={:.3}s) - relying on natural backpressure", - self.label, lead_time, self.max_lead_time, audio_timestamp, elapsed - ); - } - + trace!( + "{} broadcaster: check_and_pace called with audio_ts={:.3}s (no-op - pacing handled by TimedBroadcast)", + self.label, audio_timestamp + ); Ok(()) } } diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index 25c66364..43a4c9fb 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -116,6 +116,8 @@ struct PcmChunk { bytes: Vec, /// Timestamp in seconds (from AudioSegment) timestamp_sec: f64, + /// Duration in seconds of this PCM chunk (samples / sample_rate) + duration_sec: f64, } /// Snapshot of track metadata at a point in time. @@ -259,6 +261,12 @@ enum FlacStreamState { } /// Pure FLAC client stream (implements AsyncRead). +/// +/// Each read pulls bytes out of a [`timed_broadcast`] receiver. +/// If the receiver reports [`TryRecvError::Lagged`] it means the underlying +/// queue expired packets before the client consumed them; we log the skip +/// and immediately keep draining so that a late client can resynchronise +/// with the latest epoch instead of stalling forever. pub struct FlacClientStream { rx: timed_broadcast::Receiver, buffer: VecDeque, @@ -369,6 +377,9 @@ impl Drop for FlacClientStream { /// /// This stream injects ICY metadata blocks at regular intervals, /// allowing clients to display "Now Playing" information. +/// As with [`FlacClientStream`], hitting [`TryRecvError::Lagged`] +/// simply indicates that the timed broadcast discarded a stale chunk; +/// the client resumes with fresh data to avoid wedging the HTTP response. pub struct IcyClientStream { rx: timed_broadcast::Receiver, metadata: Arc>, @@ -614,11 +625,13 @@ impl StreamingFlacSinkLogic { .take() .ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?; - // Create shared timestamp for pacing + // Create shared timestamp and duration for pacing let current_timestamp = Arc::new(RwLock::new(0.0f64)); + let current_duration = Arc::new(RwLock::new(0.0f64)); // Create ByteStreamReader for the encoder - let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); + let pcm_reader = + ByteStreamReader::new(pcm_rx, current_timestamp.clone(), current_duration.clone()); // Create PCM format let pcm_format = PcmFormat { @@ -636,7 +649,7 @@ impl StreamingFlacSinkLogic { debug!("FLAC encoder initialized successfully"); - // Spawn broadcaster task with timestamp for pacing + // Spawn broadcaster task with timestamp and duration for pacing let flac_broadcast = self.flac_broadcast.clone(); let flac_header = self.flac_header.clone(); let max_lead = self.broadcast_max_lead_time; @@ -646,7 +659,9 @@ impl StreamingFlacSinkLogic { flac_broadcast, flac_header, current_timestamp, + current_duration, max_lead, + sample_rate, ) .await { @@ -746,17 +761,25 @@ impl NodeLogic for StreamingFlacSinkLogic { // Convert chunk to PCM bytes let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?; + // Calculate exact duration from samples and sample rate + let sample_rate = self + .sample_rate + .expect("sample_rate should be initialized"); + let duration_sec = chunk.len() as f64 / sample_rate as f64; + trace!( - "Sending PCM chunk: {} bytes, {} samples @ {:.2}s", + "Sending PCM chunk: {} bytes, {} samples @ {:.2}s (duration={:.3}s)", pcm_bytes.len(), chunk.len(), - seg.timestamp_sec + seg.timestamp_sec, + duration_sec ); - // Send to FLAC encoder with timestamp + // Send to FLAC encoder with timestamp and duration let pcm_chunk = PcmChunk { bytes: pcm_bytes, timestamp_sec: seg.timestamp_sec, + duration_sec, }; let send_start = std::time::Instant::now(); if let Err(e) = self.pcm_tx.send(pcm_chunk).await { @@ -786,11 +809,6 @@ impl NodeLogic for StreamingFlacSinkLogic { break; } - SyncMarker::TopZeroSync => { - self.flac_broadcast.mark_top_zero(); - trace!("TopZeroSync propagated to FLAC broadcast"); - } - _ => { trace!("Received other sync marker"); } @@ -826,7 +844,9 @@ async fn broadcast_flac_stream( broadcast_tx: timed_broadcast::Sender, header_cache: Arc>>, current_timestamp: Arc>, + current_duration: Arc>, broadcast_max_lead_time: f64, + sample_rate: u32, ) -> Result<(), AudioError> { trace!( "Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", @@ -847,6 +867,8 @@ async fn broadcast_flac_stream( let mut broadcast_count = 0u64; let mut total_read_time = 0.0f64; let mut read_count = 0u64; + let mut encoded_samples = 0u64; + let sample_rate_f64 = sample_rate as f64; loop { let read_start = std::time::Instant::now(); @@ -856,7 +878,12 @@ async fn broadcast_flac_stream( if !accumulator.is_empty() { let bytes = Bytes::from(std::mem::take(&mut accumulator)); let audio_ts = *current_timestamp.read().await; - if broadcast_tx.send(bytes.clone(), audio_ts).await.is_err() { + let segment_dur = *current_duration.read().await; + if broadcast_tx + .send(bytes.clone(), audio_ts, segment_dur) + .await + .is_err() + { trace!("Broadcast closed before sending final FLAC data"); break; } @@ -897,19 +924,20 @@ async fn broadcast_flac_stream( n ); - // Find where to split: position of last sync code (start of last incomplete frame) - // Everything before this position contains only complete frames - let boundary = flac_frame_utils::find_complete_frames_boundary(&accumulator); + // Locate complete frames and total samples they represent + let (boundary, total_samples) = + flac_frame_utils::find_complete_frames_with_samples(&accumulator); trace!( - "Buffer state: accumulator={} bytes, boundary={} bytes, will_send={}", + "Buffer state: accumulator={} bytes, boundary={} bytes, total_samples={}, will_send={}", accumulator.len(), boundary, - boundary >= 1024 + total_samples, + boundary >= 1024 && total_samples > 0 ); - // Only broadcast if we have at least one complete frame (1KB minimum to avoid excessive small sends) - if boundary >= 1024 { + // Only broadcast if we have at least one complete frame (keep 1KB minimum to avoid tiny sends) + if boundary >= 1024 && total_samples > 0 { // ╔═══════════════════════════════════════════════════════════════╗ // ║ BACKPRESSURE INTELLIGENTE BASÉE SUR LE TIMING ║ // ║ ║ @@ -921,13 +949,17 @@ async fn broadcast_flac_stream( // ║ Cela crée la backpressure vers TimerBufferNode tout en ║ // ║ permettant de dropper les chunks vraiment périmés. ║ // ╚═══════════════════════════════════════════════════════════════╝ - let audio_timestamp = *current_timestamp.read().await; + let frame_start_samples = encoded_samples; + encoded_samples = encoded_samples.saturating_add(total_samples); + let audio_timestamp = frame_start_samples as f64 / sample_rate_f64; + let segment_duration = total_samples as f64 / sample_rate_f64; if stats_last_log.elapsed() >= Duration::from_secs(1) { trace!( - "Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={}", + "Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={} samples={} ", audio_timestamp, - accumulator.len() + accumulator.len(), + total_samples ); stats_last_log = std::time::Instant::now(); } @@ -939,6 +971,13 @@ async fn broadcast_flac_stream( continue; } + if let Ok(mut ts) = current_timestamp.try_write() { + *ts = audio_timestamp; + } + if let Ok(mut dur) = current_duration.try_write() { + *dur = segment_duration; + } + // Split at boundary to avoid copying - extract prefix, keep suffix let remaining = accumulator.split_off(boundary); let to_send = std::mem::replace(&mut accumulator, remaining); @@ -973,17 +1012,24 @@ async fn broadcast_flac_stream( if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" { *header_cache.write().await = Some(bytes.clone()); header_captured = true; - trace!("FLAC header captured ({} bytes)", bytes.len()); + trace!("FLAC header captured ({} bytes), not broadcasting", bytes.len()); + // Skip broadcasting the header: clients prepend it locally on subscribe + continue; } let num_receivers = broadcast_tx.receiver_count(); - match broadcast_tx.send(bytes.clone(), audio_timestamp).await { + match broadcast_tx + .send(bytes.clone(), audio_timestamp, segment_duration) + .await + { Ok(_) => { if num_receivers > 0 { trace!( - "Broadcasted {} bytes to {} receivers", + "Broadcasted {} bytes to {} receivers (ts={:.3}s, dur={:.3}s)", bytes.len(), - num_receivers + num_receivers, + audio_timestamp, + segment_duration ); } } @@ -1237,15 +1283,22 @@ struct ByteStreamReader { finished: bool, /// Shared timestamp for broadcaster pacing current_timestamp: Arc>, + /// Shared duration for broadcaster pacing + current_duration: Arc>, } impl ByteStreamReader { - fn new(rx: mpsc::Receiver, current_timestamp: Arc>) -> Self { + fn new( + rx: mpsc::Receiver, + current_timestamp: Arc>, + current_duration: Arc>, + ) -> Self { Self { rx, buffer: VecDeque::new(), finished: false, current_timestamp, + current_duration, } } } @@ -1278,10 +1331,13 @@ impl AsyncRead for ByteStreamReader { if chunk.bytes.is_empty() { continue; } - // Update shared timestamp for broadcaster pacing + // Update shared timestamp and duration for broadcaster pacing if let Ok(mut ts) = self.current_timestamp.try_write() { *ts = chunk.timestamp_sec; } + if let Ok(mut dur) = self.current_duration.try_write() { + *dur = chunk.duration_sec; + } self.buffer.extend(chunk.bytes); } Poll::Ready(None) => { diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 3710a799..3a1dc156 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -92,6 +92,8 @@ struct PcmChunk { bytes: Vec, /// Timestamp in seconds (from AudioSegment) timestamp_sec: f64, + /// Duration in seconds of this PCM chunk (samples / sample_rate) + duration_sec: f64, } /// Snapshot of track metadata (reuse from streaming_flac_sink) @@ -301,11 +303,13 @@ impl StreamingOggFlacSinkLogic { .take() .ok_or_else(|| AudioError::ProcessingError("PCM receiver already consumed".into()))?; - // Create shared timestamp for pacing + // Create shared timestamp and duration for pacing let current_timestamp = Arc::new(RwLock::new(0.0f64)); + let current_duration = Arc::new(RwLock::new(0.0f64)); // Create ByteStreamReader for the encoder - let pcm_reader = ByteStreamReader::new(pcm_rx, current_timestamp.clone()); + let pcm_reader = + ByteStreamReader::new(pcm_rx, current_timestamp.clone(), current_duration.clone()); // Create PCM format let pcm_format = PcmFormat { @@ -323,7 +327,7 @@ impl StreamingOggFlacSinkLogic { debug!("OGG-FLAC encoder initialized successfully"); - // Spawn OGG wrapper + broadcaster task with timestamp for pacing + // Spawn OGG wrapper + broadcaster task with timestamp and duration for pacing let ogg_broadcast = self.ogg_broadcast.clone(); let ogg_header = self.ogg_header.clone(); let max_lead = self.broadcast_max_lead_time; @@ -333,6 +337,7 @@ impl StreamingOggFlacSinkLogic { ogg_broadcast, ogg_header, current_timestamp, + current_duration, max_lead, ) .await @@ -432,17 +437,23 @@ impl NodeLogic for StreamingOggFlacSinkLogic { // Convert chunk to PCM bytes let pcm_bytes = chunk_to_pcm_bytes(&chunk, self.bits_per_sample)?; + // Calculate exact duration from samples and sample rate + let sample_rate = self.sample_rate.expect("sample_rate should be initialized"); + let duration_sec = chunk.len() as f64 / sample_rate as f64; + trace!( - "Sending PCM chunk: {} bytes, {} samples @ {:.2}s", + "Sending PCM chunk: {} bytes, {} samples @ {:.2}s (duration={:.3}s)", pcm_bytes.len(), chunk.len(), - seg.timestamp_sec + seg.timestamp_sec, + duration_sec ); - // Send to FLAC encoder with timestamp + // Send to FLAC encoder with timestamp and duration let pcm_chunk = PcmChunk { bytes: pcm_bytes, timestamp_sec: seg.timestamp_sec, + duration_sec, }; if let Err(e) = self.pcm_tx.send(pcm_chunk).await { warn!("Failed to send PCM data to encoder: {}", e); @@ -626,15 +637,22 @@ struct ByteStreamReader { finished: bool, /// Shared timestamp for broadcaster pacing current_timestamp: Arc>, + /// Shared duration for broadcaster pacing + current_duration: Arc>, } impl ByteStreamReader { - fn new(rx: mpsc::Receiver, current_timestamp: Arc>) -> Self { + fn new( + rx: mpsc::Receiver, + current_timestamp: Arc>, + current_duration: Arc>, + ) -> Self { Self { rx, buffer: VecDeque::new(), finished: false, current_timestamp, + current_duration, } } } @@ -667,10 +685,13 @@ impl AsyncRead for ByteStreamReader { if chunk.bytes.is_empty() { continue; } - // Update shared timestamp for broadcaster pacing + // Update shared timestamp and duration for broadcaster pacing if let Ok(mut ts) = self.current_timestamp.try_write() { *ts = chunk.timestamp_sec; } + if let Ok(mut dur) = self.current_duration.try_write() { + *dur = chunk.duration_sec; + } self.buffer.extend(chunk.bytes); } Poll::Ready(None) => { @@ -784,6 +805,7 @@ async fn broadcast_ogg_flac_stream( broadcast_tx: timed_broadcast::Sender, header_cache: Arc>>, current_timestamp: Arc>, + current_duration: Arc>, broadcast_max_lead_time: f64, ) -> Result<(), AudioError> { trace!( @@ -840,13 +862,21 @@ async fn broadcast_ogg_flac_stream( bos_bytes.len() + comment_bytes.len() ); - // Broadcast header - if broadcast_tx.send(bos_bytes.clone(), 0.0).await.is_err() { + // Broadcast header (BOS and comment are metadata, not audio, so duration=0.0) + if broadcast_tx + .send(bos_bytes.clone(), 0.0, 0.0) + .await + .is_err() + { trace!("No receivers for BOS page, terminating broadcast"); return Ok(()); } total_ogg_bytes += comment_bytes.len() as u64; - if broadcast_tx.send(comment_bytes.clone(), 0.0).await.is_err() { + if broadcast_tx + .send(comment_bytes.clone(), 0.0, 0.0) + .await + .is_err() + { trace!("No receivers for comment page, terminating broadcast"); return Ok(()); } @@ -867,7 +897,12 @@ async fn broadcast_ogg_flac_stream( let eos_bytes = Bytes::from(eos_page); total_ogg_bytes += eos_bytes.len() as u64; let eos_ts = *current_timestamp.read().await; - if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { + let eos_dur = *current_duration.read().await; + if broadcast_tx + .send(eos_bytes.clone(), eos_ts, eos_dur) + .await + .is_err() + { trace!("Broadcast closed before sending final EOS page"); break; } @@ -876,12 +911,16 @@ async fn broadcast_ogg_flac_stream( flac_accumulator.len() ); } else { - // Send empty EOS page + // Send empty EOS page (metadata page, duration=0.0) let eos_page = ogg_writer.create_page(&[], false, true, false); let eos_bytes = Bytes::from(eos_page); total_ogg_bytes += eos_bytes.len() as u64; let eos_ts = *current_timestamp.read().await; - if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() { + if broadcast_tx + .send(eos_bytes.clone(), eos_ts, 0.0) + .await + .is_err() + { trace!("Broadcast closed before sending empty EOS page"); break; } @@ -1023,9 +1062,13 @@ async fn broadcast_ogg_flac_stream( } // Envoyer au broadcast - match broadcast_tx.send(ogg_bytes.clone(), audio_timestamp).await { + let segment_duration = *current_duration.read().await; + match broadcast_tx + .send(ogg_bytes.clone(), audio_timestamp, segment_duration) + .await + { Ok(n) => { - trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers", first_frame.len(), first_frame_samples, ogg_bytes.len(), n); + trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers (ts={:.3}s, dur={:.3}s)", first_frame.len(), first_frame_samples, ogg_bytes.len(), n, audio_timestamp, segment_duration); } Err(_) => { trace!("No active receivers for OGG-FLAC broadcast, terminating loop"); diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index 86d872f6..8fb9bc30 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -15,7 +15,7 @@ use std::{ }; use tokio::sync::Notify; -use tracing::{trace, warn}; +use tracing::{trace, info, warn}; /// Paquet diffusé contenant la charge utile + méta timing. #[derive(Clone)] @@ -40,8 +40,17 @@ impl fmt::Debug for TimedPacket { /// Erreur remontée par `Receiver::try_recv`. #[derive(Debug)] pub enum TryRecvError { + /// Aucun paquet n'est disponible pour le moment. Empty, + /// Le receiver est en retard : le champ contient combien de paquets ont expiré + /// ou ont déjà été consommés par les autres abonnés. + /// + /// Ce cas survient lorsque `purge_expired()` avance `head_seq` et que ce + /// `Receiver` réclamait encore l'un des numéros supprimés. Le client doit + /// donc ignorer les données perdues et se resynchroniser sur les paquets + /// courants. Lagged(u64), + /// Le channel est fermé et plus aucun paquet n'est disponible. Closed, } @@ -71,7 +80,10 @@ struct State { closed: bool, epoch: u64, epoch_start: Instant, + last_segment_end: Option, cursors: Vec>, + initialized: bool, + saw_positive_timestamp: bool, } impl State { @@ -83,14 +95,19 @@ impl State { closed: false, epoch: 0, epoch_start, + last_segment_end: None, cursors: Vec::new(), + initialized: false, + saw_positive_timestamp: false, } } - fn purge_expired(&mut self, now: Instant) -> bool { + fn purge_expired(&mut self) -> bool { let mut purged = 0u64; while let Some(entry) = self.buffer.front() { - if entry.expires_at <= now { + if entry.expires_at <= Instant::now() { + let delta=Instant::now()-entry.expires_at; + info!("TimedBroadcast: purging expired packet (epoch={},delta={})",entry.epoch,delta.as_millis()); self.buffer.pop_front(); self.head_seq += 1; purged += 1; @@ -220,7 +237,11 @@ impl Clone for Sender { impl Sender { /// Diffuse un paquet. Bloque si la capacité est atteinte avec des paquets non périmés. - pub async fn send(&self, payload: T, audio_timestamp: f64) -> Result> + /// + /// Le TTL de chaque paquet est calculé à partir du `epoch_start` courant et du + /// `audio_timestamp` fournis, ce qui signifie qu’un receiver en retard finira + /// par recevoir un [`TryRecvError::Lagged`] lorsque `expires_at` est dépassé. + pub async fn send(&self, payload: T, audio_timestamp: f64, segment_duration: f64) -> Result> where T: Clone, { @@ -228,7 +249,6 @@ impl Sender { loop { let mut wait_deadline = None; { - let now = Instant::now(); let mut state = self .inner .state @@ -239,17 +259,64 @@ impl Sender { return Err(SendError(payload.expect("payload already consumed"))); } - if state.purge_expired(now) { - self.inner.space_notify.notify_waiters(); + // Détecter si c'est un TopZero + let is_top_zero = audio_timestamp == 0.0; + + // Gérer l'initialisation ET le TopZero ensemble + if !state.initialized { + let now = Instant::now(); + if is_top_zero { + // Premier paquet = TopZero → epoch commence à 0 + state.epoch_start = now; + state.epoch = 0; + info!("TimedBroadcast: initialized with TopZero (epoch=0)"); + } else { + // Premier paquet avec ts > 0 → calculer epoch_start rétroactif + let offset = Duration::from_secs_f64(audio_timestamp); + state.epoch_start = now.checked_sub(offset).unwrap_or(now); + state.epoch = 0; + info!( + "TimedBroadcast: initialized with ts={:.3}s (epoch=0)", + audio_timestamp + ); + } + state.initialized = true; + state.saw_positive_timestamp = !is_top_zero; + } else if is_top_zero { + // TopZero sur un channel déjà initialisé + let allow_reset = state.saw_positive_timestamp; + if allow_reset { + let now = Instant::now(); + state.epoch_start = state + .last_segment_end + .map(|end| end.max(now)) + .unwrap_or(now); + state.epoch = state.epoch.wrapping_add(1); + state.saw_positive_timestamp = false; // Reset pour le prochain cycle + info!( + "TimedBroadcast: TopZero detected, new epoch={} (had_last_segment={})", + state.epoch, + state.last_segment_end.is_some() + ); + } else { + warn!( + "TimedBroadcast: Ignoring duplicate TopZero (epoch={})", + state.epoch + ); + } + } else if audio_timestamp > 0.0 { + state.saw_positive_timestamp = true; } - if state.prune_consumed() { + let consumed = state.prune_consumed(); + let expired = state.purge_expired(); + if consumed || expired { self.inner.space_notify.notify_waiters(); } if state.buffer.len() < self.inner.capacity { - let audio_offset = Duration::from_secs_f64(audio_timestamp.max(0.0)); - let expires_at = state.epoch_start + audio_offset; + // Le paquet expire à la fin de son segment audio + let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration); let entry = Entry { seq: state.next_seq, expires_at, @@ -259,6 +326,10 @@ impl Sender { }; state.next_seq += 1; state.buffer.push_back(entry); + + // Stocker la fin de ce segment pour la continuité temporelle + state.last_segment_end = Some(expires_at); + let receivers = self.inner.receiver_count.load(Ordering::SeqCst); drop(state); self.inner.data_notify.notify_waiters(); @@ -304,21 +375,15 @@ impl Sender { } } - /// Marque un TopZero : incrémente l'epoch pour les paquets suivants. + /// Marque un TopZero : DEPRECATED - no-op pour compatibilité. /// - /// Reset le timer epoch_start sans effacer le buffer. Les paquets - /// du morceau précédent continueront à être distribués naturellement. - /// Cela évite de perdre les dernières frames FLAC à la transition entre morceaux. + /// Le vrai TopZero est maintenant détecté automatiquement dans send() + /// quand audio_timestamp == 0.0. Cette méthode est conservée pour + /// compatibilité avec le code existant mais ne fait rien. pub fn mark_top_zero(&self) { - let mut state = self - .inner - .state - .lock() - .expect("timed broadcast mutex poisoned"); - state.epoch = state.epoch.wrapping_add(1); - state.epoch_start = Instant::now(); - // Ne PAS effacer le buffer - laisser les paquets du morceau précédent - // se vider naturellement pour éviter de perdre les dernières frames + trace!("TimedBroadcast: mark_top_zero() called but ignored (auto-detection active)"); + // No-op - TopZero est maintenant détecté automatiquement dans send() + // quand audio_timestamp == 0.0 } /// Nombre actuel de receivers abonnés. @@ -341,6 +406,10 @@ impl Drop for Sender { } /// Receiver côté consommateur. +/// +/// Chaque receiver garde son propre curseur `next_seq`. Si le producteur +/// recycle un paquet via `purge_expired()` avant que ce curseur ne l’ait lu, +/// la prochaine tentative de lecture retournera [`TryRecvError::Lagged`]. pub struct Receiver { inner: Arc>, next_seq: u64, @@ -366,8 +435,7 @@ where return Err(TryRecvError::Closed); } - let now = Instant::now(); - if state.purge_expired(now) { + if state.purge_expired() { self.inner.space_notify.notify_waiters(); } @@ -401,6 +469,12 @@ where } /// Version synchrone utilisée dans `poll_read`. + /// + /// # Erreurs + /// + /// * [`TryRecvError::Lagged`] — des paquets ont expiré avant d'être consommés. + /// * [`TryRecvError::Empty`] — la file est vide pour l'instant. + /// * [`TryRecvError::Closed`] — plus aucun paquet n'arrivera. pub fn try_recv(&mut self) -> Result, TryRecvError> { self.poll_entry() }