From f23e43b5ea1b76ec163c6226ee0f0678beb1eae6 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Nov 2025 05:43:25 +0000 Subject: [PATCH] Implement completion marker system for cache files - Add .complete marker files to track completed downloads - Check marker instead of file size for completion detection - Drain segments when file already in cache to avoid pipeline errors - Consolidate() now removes incomplete files without markers - Add new_cache_with_consolidation() for automatic cleanup on startup --- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 53 +++++++++++++++ pmoaudiocache/src/cache.rs | 40 +++++++++++ pmocache/src/cache.rs | 81 +++++++++++++++++------ 3 files changed, 153 insertions(+), 21 deletions(-) diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index 150572f6..79cd09cb 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -164,6 +164,15 @@ impl NodeLogic for FlacCacheSinkLogic { let (_chunks, _samples, _duration_sec, stop_reason) = pump_result?; + // Si le fichier était déjà en cache (ChannelClosed), drainer les segments restants + // jusqu'au prochain TrackBoundary ou EndOfStream + let stop_reason = if matches!(stop_reason, StopReason::ChannelClosed) { + tracing::debug!("File was already in cache, draining remaining segments"); + drain_until_track_boundary(&mut rx, &stop_token).await? + } else { + stop_reason + }; + // Copier les métadonnées du TrackBoundary dans le cache if let Some(src_metadata) = track_metadata { let dest_metadata = self.cache.track_metadata(&pk); @@ -346,6 +355,50 @@ 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, diff --git a/pmoaudiocache/src/cache.rs b/pmoaudiocache/src/cache.rs index 5f168ebb..8c117856 100644 --- a/pmoaudiocache/src/cache.rs +++ b/pmoaudiocache/src/cache.rs @@ -56,6 +56,46 @@ pub fn new_cache(dir: &str, limit: usize) -> Result { Cache::with_transformer(dir, limit, Some(transformer_factory)) } +/// Crée un cache audio et lance la consolidation en arrière-plan +/// +/// Cette fonction crée le cache et lance immédiatement une consolidation +/// pour nettoyer les fichiers incomplets (sans marker de complétion). +/// +/// # Arguments +/// +/// * `dir` - Répertoire de stockage du cache +/// * `limit` - Limite de taille du cache (nombre de pistes) +/// +/// # Returns +/// +/// Arc vers l'instance du cache configurée pour la conversion FLAC automatique +/// +/// # Exemple +/// +/// ```rust,no_run +/// use pmoaudiocache::cache; +/// +/// # async fn example() -> anyhow::Result<()> { +/// let cache = cache::new_cache_with_consolidation("./audio_cache", 1000).await?; +/// # Ok(()) +/// # } +/// ``` +pub async fn new_cache_with_consolidation(dir: &str, limit: usize) -> Result> { + let cache = Arc::new(new_cache(dir, limit)?); + + // Lancer la consolidation en arrière-plan pour nettoyer les fichiers incomplets + let cache_clone = cache.clone(); + tokio::spawn(async move { + if let Err(e) = cache_clone.consolidate().await { + tracing::warn!("Failed to consolidate cache on startup: {}", e); + } else { + tracing::info!("Cache consolidated successfully on startup"); + } + }); + + Ok(cache) +} + /// Ajoute une piste audio depuis une URL avec extraction et stockage des métadonnées /// /// Cette fonction étend `add_from_url` du cache en ajoutant : diff --git a/pmocache/src/cache.rs b/pmocache/src/cache.rs index 2021e841..095d5ed5 100755 --- a/pmocache/src/cache.rs +++ b/pmocache/src/cache.rs @@ -68,32 +68,36 @@ pub struct Cache { } impl Cache { + /// Retourne le chemin du fichier marker de complétion + fn get_completion_marker_path(&self, pk: &str) -> PathBuf { + self.get_file_path(pk).with_extension(format!("{}.complete", C::file_extension())) + } + /// Vérifie si un fichier est en cache et complet /// /// # Returns /// - /// - `Ok(true)` si le fichier est en cache et complet - /// - `Ok(false)` si le fichier n'est pas en cache ou incomplet (et supprime le fichier incomplet) + /// - `Ok(true)` si le fichier est en cache et complet (fichier .complete existe) + /// - `Ok(false)` si le fichier n'est pas en cache ou incomplet (et supprime les fichiers incomplets) /// - `Err` en cas d'erreur async fn check_cached_and_complete(&self, pk: &str) -> Result { if self.db.get(pk, false).is_ok() { let file_path = self.get_file_path(pk); + let completion_marker = self.get_completion_marker_path(pk); + if file_path.exists() { - // Vérifier si le fichier semble complet (taille >= min_prebuffer_size) - if let Ok(metadata) = std::fs::metadata(&file_path) { - let file_size = metadata.len(); - if self.min_prebuffer_size > 0 && file_size < self.min_prebuffer_size { - tracing::warn!( - "File with pk {} in cache is too small ({} bytes < {} bytes), will re-download/re-ingest", - pk, file_size, self.min_prebuffer_size - ); - // Supprimer le fichier incomplet - let _ = std::fs::remove_file(&file_path); - return Ok(false); - } else { - // Déjà en cache et complet - return Ok(true); - } + // Vérifier si le fichier marker de complétion existe + if completion_marker.exists() { + tracing::debug!("File with pk {} is complete (marker exists)", pk); + return Ok(true); + } else { + tracing::warn!( + "File with pk {} in cache has no completion marker, will re-download/re-ingest", + pk + ); + // Supprimer le fichier incomplet + let _ = std::fs::remove_file(&file_path); + return Ok(false); } } } @@ -139,12 +143,23 @@ impl Cache { tracing::debug!("Prebuffering complete for pk {} ({} bytes)", pk, self.min_prebuffer_size); } - // Lancer une tâche de nettoyage en background + // Lancer une tâche de nettoyage et marquage de complétion en background let downloads_clone = self.downloads.clone(); let pk_clone = pk.to_string(); + let completion_marker = self.get_completion_marker_path(pk); + tokio::spawn(async move { - let _ = download.wait_until_finished().await; + let result = download.wait_until_finished().await; downloads_clone.write().await.remove(&pk_clone); + + // Créer le fichier marker de complétion si le téléchargement a réussi + if result.is_ok() { + if let Err(e) = std::fs::write(&completion_marker, "") { + tracing::warn!("Failed to create completion marker for pk {}: {}", pk_clone, e); + } else { + tracing::debug!("Created completion marker for pk {}", pk_clone); + } + } }); Ok(pk.to_string()) @@ -582,15 +597,22 @@ impl Cache { } /// Consolide le cache en supprimant les orphelins et en re-téléchargeant les fichiers manquants + /// + /// Cette fonction : + /// - Supprime les entrées DB sans fichiers (ou re-télécharge si URL disponible) + /// - Supprime les fichiers sans marker de complétion et leurs entrées DB + /// - Supprime les fichiers sans entrées DB correspondantes pub async fn consolidate(&self) -> Result<()> { // Récupérer la liste des entrées à traiter let entries = self.db.get_all(false)?; - // Supprimer les entrées sans fichiers correspondants + // Supprimer les entrées sans fichiers correspondants OU sans marker de complétion for entry in entries { let file_path = self.get_file_path(&entry.pk); + let completion_marker = self.get_completion_marker_path(&entry.pk); if !file_path.exists() { + // Fichier manquant, essayer de re-télécharger match self.db.get_origin_url(&entry.pk)? { Some(url) => { if let Err(err) = self.add_from_url(&url, entry.collection.as_deref()).await @@ -607,6 +629,14 @@ impl Cache { self.db.delete(&entry.pk)?; } } + } else if !completion_marker.exists() { + // Fichier existe mais pas de marker de complétion -> fichier incomplet + tracing::warn!( + "Removing incomplete file {} (no completion marker)", + entry.pk + ); + let _ = tokio::fs::remove_file(&file_path).await; + self.db.delete(&entry.pk)?; } } @@ -616,11 +646,20 @@ impl Cache { let path = entry.path(); if path.is_file() && path != self.dir.join("cache.db") { if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) { + // Ignorer les fichiers .complete + if file_name.ends_with(".complete") { + continue; + } + // Format attendu: {pk}.{qualifier}.{EXT} // On extrait le pk (première partie avant le premier point) if let Some(pk) = file_name.split('.').next() { if self.db.get(pk, false).is_err() { - tokio::fs::remove_file(path).await?; + tracing::debug!("Removing orphan file: {}", file_name); + tokio::fs::remove_file(&path).await?; + // Supprimer aussi le marker de complétion s'il existe + let completion_marker = self.get_completion_marker_path(pk); + let _ = tokio::fs::remove_file(&completion_marker).await; } } }