From 9bc0c2544d7a578db22ba66b0a93a148e2ce2851 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Mon, 17 Nov 2025 03:03:06 +0100 Subject: [PATCH] Round 3 --- .DS_Store | Bin 14340 -> 14340 bytes .gitignore | 2 +- pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 32 +++++-- pmoaudio-ext/src/sinks/timed_broadcast.rs | 71 +++++++++------ pmoparadise/.pmomusic/config.yaml | 14 +++ pmoparadise/examples/single_channel_server.rs | 4 +- pmoplaylist/src/manager.rs | 8 ++ pmoplaylist/src/persistence/mod.rs | 86 ++++++++++++++++++ 8 files changed, 180 insertions(+), 37 deletions(-) create mode 100644 pmoparadise/.pmomusic/config.yaml diff --git a/.DS_Store b/.DS_Store index d0612186888b95432fb73f6070a85bad707a7e94..5f1c4e87cbed9515cedb6ca9c71b4a48073d462a 100644 GIT binary patch delta 24 gcmZoEXepSmpOJmzfqMB#5($&Fl)N^-RFD=20Dy1_Z~y=R delta 33 pcmZoEXepSmpOJIpfqMDLrxa7zP0Vx@j0`L%2kP2xey$)Z4glQS3&#Ke diff --git a/.gitignore b/.gitignore index 345d0102..a4119898 100644 --- a/.gitignore +++ b/.gitignore @@ -39,4 +39,4 @@ upmpdcli/ test_upnp*.cargo/ .cargo/ setup-env.sh -/cache \ No newline at end of file +cache \ No newline at end of file diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index 43a4c9fb..eff97442 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -79,7 +79,7 @@ use pmometadata::TrackMetadata; use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; use tokio::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, trace, warn}; +use tracing::{debug, error, info, trace, warn}; /// Default ICY metadata interval (bytes of audio between metadata blocks). /// Standard value used by most streaming servers. @@ -306,7 +306,9 @@ impl AsyncRead for FlacClientStream { self.state = FlacStreamState::Streaming; continue; // Now copy header to output buffer } else { - // Header not yet captured or can't acquire lock, skip to streaming + // Header not yet captured - client will receive it via broadcast + // Skip directly to streaming to avoid blocking + debug!("FLAC header not yet available, client will receive it via broadcast"); self.state = FlacStreamState::Streaming; } } @@ -483,7 +485,9 @@ impl AsyncRead for IcyClientStream { self.state = FlacStreamState::Streaming; continue; // Now copy header to output buffer } else { - // Header not yet captured or can't acquire lock, skip to streaming + // Header not yet captured - client will receive it via broadcast + // Skip directly to streaming to avoid blocking + debug!("FLAC header not yet available, ICY client will receive it via broadcast"); self.state = FlacStreamState::Streaming; } } @@ -605,6 +609,7 @@ struct StreamingFlacSinkLogic { encoder_state: Option, sample_rate: Option, broadcast_max_lead_time: f64, + first_chunk_timestamp_checked: bool, } impl StreamingFlacSinkLogic { @@ -748,6 +753,18 @@ impl NodeLogic for StreamingFlacSinkLogic { Some(seg) => { match &seg.segment { _AudioSegment::Chunk(chunk) => { + if !self.first_chunk_timestamp_checked { + self.first_chunk_timestamp_checked = true; + if seg.timestamp_sec.abs() > 1e-6 { + warn!( + "StreamingFlacSink: first chunk timestamp is {:.6}s (expected 0.0)", + seg.timestamp_sec + ); + } else { + trace!("StreamingFlacSink: first chunk timestamp verified at 0.0s"); + } + } + // Detect sample rate from first chunk and initialize encoder if self.sample_rate.is_none() { let sample_rate = chunk.sample_rate(); @@ -949,6 +966,8 @@ async fn broadcast_flac_stream( // ║ Cela crée la backpressure vers TimerBufferNode tout en ║ // ║ permettant de dropper les chunks vraiment périmés. ║ // ╚═══════════════════════════════════════════════════════════════╝ + + // Calculer le timestamp de cette FLAC frame 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; @@ -1012,9 +1031,9 @@ 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), not broadcasting", bytes.len()); - // Skip broadcasting the header: clients prepend it locally on subscribe - continue; + trace!("FLAC header captured ({} bytes), will also broadcast it", bytes.len()); + // Also broadcast the header so early-connecting clients receive it + // Later-connecting clients will get it from the cache } let num_receivers = broadcast_tx.receiver_count(); @@ -1143,6 +1162,7 @@ impl StreamingFlacSink { encoder_state: None, sample_rate: None, broadcast_max_lead_time: broadcast_max_lead_time.max(0.0), + first_chunk_timestamp_checked: false, }; let sink = Self { diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index 6afa68f5..238875e8 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -17,6 +17,9 @@ use std::{ use tokio::sync::Notify; use tracing::{trace, info, warn}; +/// Tolérance pour détecter un timestamp à zéro (TopZero). +const TOP_ZERO_EPSILON: f64 = 1e-9; + /// Paquet diffusé contenant la charge utile + méta timing. #[derive(Clone)] pub struct TimedPacket { @@ -102,9 +105,7 @@ impl State { } } - fn purge_expired(&mut self) -> bool { - let now = Instant::now(); - + fn purge_expired(&mut self, now: Instant) -> bool { // Throttling : purger au maximum toutes les 100ms if now.duration_since(self.last_purge) < Duration::from_millis(100) { return false; @@ -267,25 +268,27 @@ impl Sender { return Err(SendError(payload.expect("payload already consumed"))); } - // Détecter si c'est un TopZero - let is_top_zero = audio_timestamp == 0.0; + // Capturer le temps UNE SEULE FOIS pour cohérence temporelle + let now = Instant::now(); + + // Détecter si c'est un TopZero (avec tolérance pour éviter erreurs d'arrondi) + let is_top_zero = audio_timestamp.abs() < TOP_ZERO_EPSILON; // Gérer l'initialisation if !state.initialized { if !is_top_zero { warn!( - "TimedBroadcast: First packet should have timestamp=0.0, got {:.3}s", + "TimedBroadcast: First packet has non-zero timestamp {:.3}s, treating as epoch start anyway", audio_timestamp ); } - let now = Instant::now(); + // Initialiser TOUJOURS, quel que soit le timestamp du premier paquet state.epoch_start = now; state.epoch = 0; state.initialized = true; - info!("TimedBroadcast: initialized (epoch=0)"); + info!("TimedBroadcast: initialized (epoch=0, ts={:.3}s)", audio_timestamp); } else if is_top_zero { // TopZero = nouveau segment, toujours valide après l'initialisation - let now = Instant::now(); // Continuité temporelle : nouveau segment commence après le précédent state.epoch_start = state.last_segment_end.unwrap_or(now); state.epoch = state.epoch.wrapping_add(1); @@ -296,26 +299,34 @@ impl Sender { ); } - // 1. Calculer l'expiration du paquet actuel - let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration); - - // 2. Vérifier que le paquet n'est pas déjà expiré - let now = Instant::now(); - if expires_at <= now { - warn!( - "TimedBroadcast: packet already expired (ts={:.3}s, delta={}ms)", - audio_timestamp, - now.duration_since(expires_at).as_millis() - ); - // On peut décider de l'ignorer ou de continuer - // Pour l'instant on continue pour ne pas bloquer le flux + // 1. Purger d'abord les paquets expirés et consommés pour libérer l'espace + // (skip pour le tout premier paquet) + if state.buffer.len() > 0 { + let consumed = state.prune_consumed(); + let expired = state.purge_expired(now); + if consumed || expired { + self.inner.space_notify.notify_waiters(); + } } - // 3. Purger les paquets expirés et consommés - let consumed = state.prune_consumed(); - let expired = state.purge_expired(); - if consumed || expired { - self.inner.space_notify.notify_waiters(); + // 2. Calculer l'expiration du paquet actuel + let expires_at = state.epoch_start + Duration::from_secs_f64(audio_timestamp + segment_duration); + + // 3. Rejeter le paquet s'il est déjà expiré + // SAUF pour le premier paquet (initialisation) ou les paquets TopZero (nouveaux segments) + let is_first_packet = state.next_seq == 0; + if !is_first_packet && !is_top_zero && expires_at <= now { + // Tolérer une petite marge pour les latences d'initialisation + let grace_period = Duration::from_millis(50); + if now > expires_at + grace_period { + warn!( + "TimedBroadcast: rejecting already expired packet (ts={:.3}s, epoch={}, delta={}ms)", + audio_timestamp, + state.epoch, + now.duration_since(expires_at).as_millis() + ); + return Err(SendError(payload.expect("payload already consumed"))); + } } // 4. Vérifier la capacité et insérer @@ -330,7 +341,8 @@ impl Sender { state.next_seq += 1; state.buffer.push_back(entry); - // 5. Stocker la fin du segment SEULEMENT pour les paquets non-TopZero + // 5. Mettre à jour la fin du segment SEULEMENT pour les paquets non-TopZero + // (pour que le prochain segment commence à la fin du dernier paquet de données) if !is_top_zero { state.last_segment_end = Some(expires_at); } @@ -429,7 +441,8 @@ where return Err(TryRecvError::Closed); } - if state.purge_expired() { + let now = Instant::now(); + if state.purge_expired(now) { self.inner.space_notify.notify_waiters(); } diff --git a/pmoparadise/.pmomusic/config.yaml b/pmoparadise/.pmomusic/config.yaml new file mode 100644 index 00000000..87a89676 --- /dev/null +++ b/pmoparadise/.pmomusic/config.yaml @@ -0,0 +1,14 @@ +host: + http_port: '8080' + cover_cache: + directory: cache_covers + size: 2000 + audio_cache: + directory: cache_audio + size: 500 + logger: + buffer_capacity: 200 + enable_console: true + min_level: INFO +playlists: + directory: playlists diff --git a/pmoparadise/examples/single_channel_server.rs b/pmoparadise/examples/single_channel_server.rs index dd975bec..4d9859f1 100644 --- a/pmoparadise/examples/single_channel_server.rs +++ b/pmoparadise/examples/single_channel_server.rs @@ -34,7 +34,9 @@ struct AppState { #[tokio::main] async fn main() -> anyhow::Result<()> { - tracing_subscriber::fmt().with_env_filter("info").init(); + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); let descriptor = pick_descriptor(std::env::args().nth(1))?; info!( diff --git a/pmoplaylist/src/manager.rs b/pmoplaylist/src/manager.rs index 2daf75f8..ebe50ca0 100644 --- a/pmoplaylist/src/manager.rs +++ b/pmoplaylist/src/manager.rs @@ -35,6 +35,14 @@ impl PlaylistManager { // Initialiser la persistance let persistence = Arc::new(PersistenceManager::new(&db_path)?); + // Lancer la consolidation en arrière-plan + let persistence_clone = persistence.clone(); + tokio::spawn(async move { + if let Err(e) = persistence_clone.consolidate().await { + tracing::warn!("Failed to consolidate playlist database on startup: {}", e); + } + }); + let manager = Self { inner: Arc::new(ManagerInner { playlists: RwLock::new(HashMap::new()), diff --git a/pmoplaylist/src/persistence/mod.rs b/pmoplaylist/src/persistence/mod.rs index 5d53b8ba..a0f95624 100644 --- a/pmoplaylist/src/persistence/mod.rs +++ b/pmoplaylist/src/persistence/mod.rs @@ -245,4 +245,90 @@ impl PersistenceManager { })?; Ok(()) } + + /// Consolide la base de données des playlists + /// + /// Cette fonction nettoie les incohérences: + /// - Active les contraintes de clés étrangères + /// - Supprime les tracks orphelins (référençant des playlists inexistantes) + /// - Nettoie les tracks avec TTL expirés + pub async fn consolidate(&self) -> Result<()> { + let conn = self.conn.lock().unwrap(); + + // Activer les contraintes de clés étrangères (désactivées par défaut dans SQLite) + conn.execute("PRAGMA foreign_keys = ON", []) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to enable foreign keys: {}", e)) + })?; + + // Vérifier l'intégrité des clés étrangères + let mut stmt = conn + .prepare("PRAGMA foreign_key_check") + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to prepare FK check: {}", e)) + })?; + + let violations: Vec<(String, i64, String, i64)> = stmt + .query_map([], |row| { + Ok(( + row.get(0)?, + row.get(1)?, + row.get(2)?, + row.get(3)?, + )) + }) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to check foreign keys: {}", e)) + })? + .collect::, _>>() + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to read FK violations: {}", e)) + })?; + + if !violations.is_empty() { + tracing::warn!( + "Found {} foreign key violations, cleaning up orphaned tracks", + violations.len() + ); + + // Supprimer les tracks orphelins (ceux qui référencent des playlists inexistantes) + let deleted = conn + .execute( + "DELETE FROM tracks WHERE playlist_id NOT IN (SELECT id FROM playlists)", + [], + ) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to delete orphaned tracks: {}", e)) + })?; + + if deleted > 0 { + tracing::info!("Removed {} orphaned tracks during consolidation", deleted); + } + } + + // Nettoyer les tracks avec TTL expirés + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + + let deleted_expired = conn + .execute( + "DELETE FROM tracks WHERE ttl_secs IS NOT NULL AND (added_at + ttl_secs) < ?1", + params![now], + ) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to delete expired tracks: {}", e)) + })?; + + if deleted_expired > 0 { + tracing::info!( + "Removed {} expired tracks during consolidation", + deleted_expired + ); + } + + tracing::info!("Playlist database consolidation completed successfully"); + Ok(()) + } }