From 3957e17f64ce5d037fdcab3f631855c84e26b11d Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 24 Nov 2025 08:06:56 +0100 Subject: [PATCH] encore des debug... --- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 240 +++++------------- .../src/sinks/streaming_sink_common.rs | 23 +- pmoaudiocache/src/streaming.rs | 46 ++-- pmocache/src/cache.rs | 35 +++ pmoflac/src/encoder.rs | 6 + pmoparadise/examples/single_channel_server.rs | 11 +- 6 files changed, 148 insertions(+), 213 deletions(-) diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index 26598090..1db1471b 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -8,7 +8,6 @@ use pmoaudio::{ }; use pmoaudiocache::AudioTrackMetadataExt; use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat}; -use serde_json::{Number, Value}; use std::{ collections::VecDeque, pin::Pin, @@ -35,13 +34,6 @@ use tokio_util::sync::CancellationToken; // FlacCacheSinkLogic - Logique métier pure // ═══════════════════════════════════════════════════════════════════════════ -/// Signal retourné par pump_segments indiquant pourquoi l'encodage s'est arrêté. -enum StopReason { - TrackBoundary(Arc>), - EndOfStream, - ChannelClosed, -} - /// Logique pure d'encodage FLAC vers le cache pub struct FlacCacheSinkLogic { cache: Arc, @@ -291,7 +283,64 @@ impl NodeLogic for FlacCacheSinkLogic { }; if let Some(transform) = self.cache.transform_metadata(&pk).await { - persist_transform_streaminfo(&self.cache, &pk, &transform); + tracing::debug!( + "FlacCacheSink: Got transform metadata for pk {}: sr={:?}, bps={:?}, ch={:?}, ts={:?}", + pk, + transform.sample_rate, + transform.bits_per_sample, + transform.channels, + transform.total_samples + ); + + // Persister les métadonnées techniques via l'interface TrackMetadata + let track_meta = self.cache.track_metadata(&pk); + let mut meta = track_meta.write().await; + + if let Some(sr) = transform.sample_rate { + if let Err(e) = meta.set_sample_rate(Some(sr)).await { + tracing::error!("FlacCacheSink: Failed to set sample_rate for pk {}: {:?}", pk, e); + } else { + tracing::debug!("FlacCacheSink: Set sample_rate={} for pk {}", sr, pk); + } + } + if let Some(bps) = transform.bits_per_sample { + if let Err(e) = meta.set_bits_per_sample(Some(bps)).await { + tracing::error!("FlacCacheSink: Failed to set bits_per_sample for pk {}: {:?}", pk, e); + } else { + tracing::debug!("FlacCacheSink: Set bits_per_sample={} for pk {}", bps, pk); + } + } + if let Some(ch) = transform.channels { + if let Err(e) = meta.set_channels(Some(ch)).await { + tracing::error!("FlacCacheSink: Failed to set channels for pk {}: {:?}", pk, e); + } else { + tracing::debug!("FlacCacheSink: Set channels={} for pk {}", ch, pk); + } + } + if let Some(ts) = transform.total_samples { + if let Err(e) = meta.set_total_samples(Some(ts)).await { + tracing::error!("FlacCacheSink: Failed to set total_samples for pk {}: {:?}", pk, e); + } else { + tracing::debug!("FlacCacheSink: Set total_samples={} for pk {}", ts, pk); + } + + // Calculer la durée à partir de total_samples et sample_rate + if let Some(sr) = transform.sample_rate { + if sr > 0 { + use std::time::Duration; + let secs = (ts as f64 / sr as f64).round() as u64; + if let Err(e) = meta.set_duration(Some(Duration::from_secs(secs))).await { + tracing::error!("FlacCacheSink: Failed to set duration for pk {}: {:?}", pk, e); + } else { + tracing::debug!("FlacCacheSink: Set duration={} secs for pk {}", secs, pk); + } + } + } + } + + drop(meta); // Libérer le lock explicitement + } else { + tracing::warn!("FlacCacheSink: No transform metadata available for pk {}", pk); } // Phase 2: Prebuffer terminé! Copier les métadonnées et pusher à la playlist @@ -701,144 +750,6 @@ async fn wait_for_first_audio_chunk_with_metadata( } } -/// Draine tous les segments jusqu'au prochain TrackBoundary ou EndOfStream -/// -/// Cette fonction est utilisée quand le fichier était déjà en cache et que -/// nous devons ignorer les segments restants pour rester synchronisé avec la source. -async fn drain_until_track_boundary( - rx: &mut mpsc::Receiver>, - stop_token: &CancellationToken, -) -> Result { - loop { - let segment = tokio::select! { - result = rx.recv() => { - match result { - Some(seg) => seg, - None => { - return Ok(StopReason::ChannelClosed); - } - } - } - _ = stop_token.cancelled() => { - return Ok(StopReason::ChannelClosed); - } - }; - - match &segment.segment { - _AudioSegment::Chunk(_) => { - // Ignorer les chunks audio - continue; - } - _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { metadata, .. } => { - return Ok(StopReason::TrackBoundary(metadata.clone())); - } - SyncMarker::EndOfStream => { - return Ok(StopReason::EndOfStream); - } - _ => { - // Ignorer les autres syncmarkers - continue; - } - }, - } - } -} - -/// Pompe les segments pour une seule track (s'arrête au TrackBoundary). -async fn pump_track_segments( - first_segment: Arc, - rx: &mut mpsc::Receiver>, - pcm_tx: mpsc::Sender>, - bits_per_sample: u8, - expected_rate: u32, - stop_token: &CancellationToken, -) -> Result<(u64, u64, f64, StopReason), AudioError> { - let mut chunks = 0u64; - let mut samples = 0u64; - let mut duration_sec = 0.0f64; - - // Traiter le premier segment - if let Some(chunk) = first_segment.as_chunk() { - let pcm_bytes = chunk_to_pcm_bytes(chunk, bits_per_sample)?; - if !pcm_bytes.is_empty() { - // Si le send échoue, c'est que le receiver est fermé - // (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement) - if pcm_tx.send(pcm_bytes).await.is_err() { - drop(pcm_tx); - return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); - } - chunks += 1; - samples += chunk.len() as u64; - duration_sec += chunk.len() as f64 / expected_rate as f64; - } - } - - // Boucle sur les segments suivants - loop { - let segment = tokio::select! { - result = rx.recv() => { - match result { - Some(seg) => seg, - None => { - drop(pcm_tx); // Fermer le channel PCM - return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); - } - } - } - _ = stop_token.cancelled() => { - drop(pcm_tx); // Fermer le channel PCM - return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); - } - }; - - match &segment.segment { - _AudioSegment::Chunk(chunk) => { - // Vérifier la cohérence du sample rate - if chunk.sample_rate() != expected_rate { - return Err(AudioError::ProcessingError(format!( - "FlacCacheSink: inconsistent sample rate ({} vs {})", - chunk.sample_rate(), - expected_rate - ))); - } - - let pcm_bytes = chunk_to_pcm_bytes(&chunk, bits_per_sample)?; - if pcm_bytes.is_empty() { - continue; - } - - // Si le send échoue, c'est que le receiver est fermé - // (par exemple, le fichier était déjà en cache et add_from_reader a retourné immédiatement) - if pcm_tx.send(pcm_bytes).await.is_err() { - drop(pcm_tx); - return Ok((chunks, samples, duration_sec, StopReason::ChannelClosed)); - } - - chunks += 1; - samples += chunk.len() as u64; - duration_sec += chunk.len() as f64 / expected_rate as f64; - } - _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { metadata, .. } => { - drop(pcm_tx); // Fermer le channel PCM - return Ok(( - chunks, - samples, - duration_sec, - StopReason::TrackBoundary(metadata.clone()), - )); - } - SyncMarker::EndOfStream => { - drop(pcm_tx); // Fermer le channel PCM - return Ok((chunks, samples, duration_sec, StopReason::EndOfStream)); - } - _ => {} // Ignorer les autres syncmarkers - }, - } - } -} - /// Pompe les segments pour une seule track depuis un channel dédié. /// /// Cette version permet d'avoir plusieurs pumps en parallèle (pour cache progressif), @@ -915,41 +826,6 @@ async fn pump_track_segments_from_channel( } } -fn persist_transform_streaminfo( - cache: &pmoaudiocache::Cache, - pk: &str, - tmeta: &pmocache::download::TransformMetadata, -) { - if let Some(sr) = tmeta.sample_rate { - let _ = cache - .db - .set_a_metadata(pk, "sample_rate", Value::Number(Number::from(sr))); - } - if let Some(bps) = tmeta.bits_per_sample { - let _ = cache - .db - .set_a_metadata(pk, "bits_per_sample", Value::Number(Number::from(bps))); - } - if let Some(ch) = tmeta.channels { - let _ = cache - .db - .set_a_metadata(pk, "channels", Value::Number(Number::from(ch))); - } - if let Some(ts) = tmeta.total_samples { - let _ = cache - .db - .set_a_metadata(pk, "total_samples", Value::Number(Number::from(ts))); - if let Some(sr) = tmeta.sample_rate { - if sr > 0 { - let secs = (ts as f64 / sr as f64).round() as u64; - let _ = cache - .db - .set_a_metadata(pk, "duration_secs", Value::Number(Number::from(secs))); - } - } - } -} - /// Détermine la profondeur de bit d'un chunk audio fn get_chunk_bit_depth(chunk: &AudioChunk) -> u8 { match chunk { diff --git a/pmoaudio-ext/src/sinks/streaming_sink_common.rs b/pmoaudio-ext/src/sinks/streaming_sink_common.rs index e8748404..bd4d6589 100644 --- a/pmoaudio-ext/src/sinks/streaming_sink_common.rs +++ b/pmoaudio-ext/src/sinks/streaming_sink_common.rs @@ -376,15 +376,6 @@ impl SharedSinkContext { // Always pass the metadata handle to the encoder so Vorbis comments are emitted. self.encoder_options.metadata = Some(metadata_lock.clone()); - // In raw FLAC live streaming we must NOT advertise a total_samples value, - // otherwise players think the stream ends after the first track. - if !self.enable_total_samples { - self.pending_track_duration = None; - self.pending_total_samples = None; - self.encoder_options.total_samples = None; - return Ok(()); - } - // Capture duration (if any) to set total_samples. let duration_opt = { let metadata = metadata_lock.read().await; @@ -396,6 +387,20 @@ impl SharedSinkContext { metadata.get_total_samples().await.ok().flatten() }; + debug!( + "Encoder metadata: from TrackBoundary - duration={:?}s, total_samples={:?}", + self.pending_track_duration.as_ref().map(|d| d.as_secs_f64()), + self.pending_total_samples + ); + + // In raw FLAC live streaming we must NOT advertise a total_samples value, + // otherwise players think the stream ends after the first track. + if !self.enable_total_samples { + self.encoder_options.total_samples = None; + debug!("Encoder metadata: total_samples disabled for this sink (enable_total_samples=false)"); + return Ok(()); + } + // Compute total_samples only when we know the sample rate. self.refresh_total_samples_with_sample_rate(); diff --git a/pmoaudiocache/src/streaming.rs b/pmoaudiocache/src/streaming.rs index c086de30..e011c3bf 100644 --- a/pmoaudiocache/src/streaming.rs +++ b/pmoaudiocache/src/streaming.rs @@ -33,25 +33,33 @@ pub fn create_streaming_flac_transformer() -> StreamTransformer { "transcode" }; - context - .set_metadata(TransformMetadata { - mode: Some(mode.to_string()), - input_codec: Some(codec_to_string(codec)), - details: Some( - json!({ - "sample_rate": info.sample_rate, - "bits_per_sample": info.bits_per_sample, - "channels": info.channels, - "total_samples": info.total_samples, - }) - .to_string(), - ), - sample_rate: Some(info.sample_rate), - bits_per_sample: Some(info.bits_per_sample), - channels: Some(info.channels), - total_samples: info.total_samples, - }) - .await; + let metadata = TransformMetadata { + mode: Some(mode.to_string()), + input_codec: Some(codec_to_string(codec)), + details: Some( + json!({ + "sample_rate": info.sample_rate, + "bits_per_sample": info.bits_per_sample, + "channels": info.channels, + "total_samples": info.total_samples, + }) + .to_string(), + ), + sample_rate: Some(info.sample_rate), + bits_per_sample: Some(info.bits_per_sample), + channels: Some(info.channels), + total_samples: info.total_samples, + }; + + tracing::debug!( + "Transformer setting metadata: sr={:?}, bps={:?}, ch={:?}, ts={:?}", + metadata.sample_rate, + metadata.bits_per_sample, + metadata.channels, + metadata.total_samples + ); + + context.set_metadata(metadata).await; let mut flac_stream = transcode.into_stream(); let mut buffer = vec![0u8; 64 * 1024]; diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index e3429c44..b7e33f50 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -189,6 +189,41 @@ impl Cache { self.db.set_origin_url(pk, url)?; } + // Sauvegarder les métadonnées techniques du transformer + if let Some(transform) = download.transform_metadata().await { + tracing::debug!( + "Cache: Got transform metadata for pk {}: sr={:?}, bps={:?}, ch={:?}, ts={:?}", + pk, + transform.sample_rate, + transform.bits_per_sample, + transform.channels, + transform.total_samples + ); + + if let Some(sr) = transform.sample_rate { + self.db.set_a_metadata(pk, "sample_rate", serde_json::json!(sr))?; + } + if let Some(bps) = transform.bits_per_sample { + self.db.set_a_metadata(pk, "bits_per_sample", serde_json::json!(bps))?; + } + if let Some(ch) = transform.channels { + self.db.set_a_metadata(pk, "channels", serde_json::json!(ch))?; + } + if let Some(ts) = transform.total_samples { + self.db.set_a_metadata(pk, "total_samples", serde_json::json!(ts))?; + + // Calculer la durée à partir de total_samples et sample_rate + if let Some(sr) = transform.sample_rate { + if sr > 0 { + let secs = (ts as f64 / sr as f64).round() as u64; + self.db.set_a_metadata(pk, "duration_secs", serde_json::json!(secs))?; + } + } + } + } else { + tracing::debug!("Cache: No transform metadata available for pk {}", pk); + } + if let Err(e) = self.enforce_limit().await { tracing::warn!("Error enforcing cache limit: {}", e); } diff --git a/pmoflac/src/encoder.rs b/pmoflac/src/encoder.rs index 444de9b9..497f565e 100755 --- a/pmoflac/src/encoder.rs +++ b/pmoflac/src/encoder.rs @@ -525,10 +525,16 @@ fn run_encoder( "set_verify failed", )?; if let Some(total) = options.total_samples { + tracing::debug!( + "FLAC encoder: setting total_samples_estimate = {} before init", + total + ); ensure( FLAC__stream_encoder_set_total_samples_estimate(encoder, total), "set_total_samples_estimate failed", )?; + } else { + tracing::warn!("FLAC encoder: total_samples is None, STREAMINFO will have total_samples=0"); } if let Some(block_size) = options.block_size { ensure( diff --git a/pmoparadise/examples/single_channel_server.rs b/pmoparadise/examples/single_channel_server.rs index 82e2ca2f..722e0931 100644 --- a/pmoparadise/examples/single_channel_server.rs +++ b/pmoparadise/examples/single_channel_server.rs @@ -64,10 +64,15 @@ async fn main() -> anyhow::Result<()> { let history_opts = history_builder.build_for_channel(&descriptor).await?; let mut channel_config = ParadiseStreamChannelConfig::default(); - channel_config.flac_options = StreamingSinkOptions::flac_defaults() + // Configuration commune pour FLAC et OGG + let common_options = StreamingSinkOptions::flac_defaults() .with_default_artist(Some("Radio Paradise".to_string())) - .with_default_title(descriptor.display_name.to_string()) - .with_only_default_metadata(true); + .with_default_title(descriptor.display_name.to_string()); + + channel_config.flac_options = common_options.clone(); + channel_config.ogg_options = StreamingSinkOptions::ogg_defaults() + .with_default_artist(Some("Radio Paradise".to_string())) + .with_default_title(descriptor.display_name.to_string()); let channel = Arc::new( ParadiseStreamChannel::new(