From 0d428d0b220f45f2a24ad71785378411de064cfa Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sun, 26 Oct 2025 07:03:56 +0100 Subject: [PATCH] evite les doubles download de block --- pmoparadise/src/paradise/worker.rs | 198 +++++++++++++++++++++++++++-- 1 file changed, 188 insertions(+), 10 deletions(-) diff --git a/pmoparadise/src/paradise/worker.rs b/pmoparadise/src/paradise/worker.rs index 1d94d463..7923520a 100644 --- a/pmoparadise/src/paradise/worker.rs +++ b/pmoparadise/src/paradise/worker.rs @@ -15,7 +15,6 @@ use bytes::Bytes; use chrono::Utc; use futures::stream; use pmosource::{SourceCacheManager, TrackMetadata}; -use sha2::{Digest, Sha256}; use std::collections::{HashSet, VecDeque}; use std::pin::Pin; use std::sync::Arc; @@ -149,6 +148,7 @@ struct WorkerState { active_clients: usize, status: ChannelLifecycle, processed_blocks: HashSet, + processing_blocks: HashSet, recent_blocks: VecDeque, next_block_hint: Option, scheduled_task: Option, @@ -175,6 +175,7 @@ impl WorkerState { active_clients: 0, status: ChannelLifecycle::Idle, processed_blocks: HashSet::new(), + processing_blocks: HashSet::new(), recent_blocks: VecDeque::new(), next_block_hint: None, scheduled_task: None, @@ -308,16 +309,57 @@ impl WorkerState { } async fn process_block(&mut self, block: Block) -> Result<()> { + // Check if we just processed this block (songs are already in playlist) if self.is_recent_block(block.event) { debug!( channel = self.descriptor.slug, event = block.event, - "Skipping already processed block" + "Skipping already processed block (songs already in playlist)" ); self.next_block_hint = Some(block.end_event); return Ok(()); } + // Check if this block is currently being processed by another task + // This prevents race conditions when the same block is requested multiple times + if self.processing_blocks.contains(&block.event) { + warn!( + channel = self.descriptor.slug, + event = block.event, + "Block is already being processed, skipping duplicate request" + ); + return Ok(()); + } + + // Check if all songs from this block are in cache + // If yes, restore from cache instead of downloading + if self.check_all_songs_cached(&block).await { + info!( + channel = self.descriptor.slug, + event = block.event, + "Block found in cache, restoring without download" + ); + self.restore_from_cache(&block).await?; + self.record_processed_block(block.event); + self.next_block_hint = Some(block.end_event); + self.backoff.reset(); + return Ok(()); + } + + // Mark block as being processed + self.processing_blocks.insert(block.event); + let event = block.event; // Save for cleanup + + // Process the block and ensure cleanup even on error + let result = self.process_block_inner(block).await; + + // Always remove from processing set, whether success or error + self.processing_blocks.remove(&event); + + result + } + + async fn process_block_inner(&mut self, block: Block) -> Result<()> { info!( channel = self.descriptor.slug, event = block.event, @@ -534,7 +576,7 @@ impl WorkerState { .await .context("Failed to encode song to FLAC")?; - let track_id = self.compute_track_id(&flac_bytes); + let track_id = self.compute_track_id(block, *song_index); let placeholder_uri = format!("{}#{}", block.url, song_index); let mut metadata = TrackMetadata { @@ -616,7 +658,7 @@ impl WorkerState { .await .context("Failed to encode song to FLAC")?; - let track_id = self.compute_track_id(&flac_bytes); + let track_id = self.compute_track_id(block, song_index); let placeholder_uri = format!("{}#{}", block.url, song_index); let mut metadata = TrackMetadata { @@ -679,12 +721,10 @@ impl WorkerState { Ok(None) } - fn compute_track_id(&self, flac_bytes: &[u8]) -> String { - let slice_len = flac_bytes.len().min(self.config.cache.track_id_hash_bytes); - let mut hasher = Sha256::new(); - hasher.update(&flac_bytes[..slice_len]); - let hash = hasher.finalize(); - format!("rp:{}:{}", self.descriptor.id, hex::encode(hash)) + fn compute_track_id(&self, block: &Block, song_index: usize) -> String { + // Use deterministic ID based on block event and song index + // This allows checking if a song is cached before downloading the block + format!("rp:{}:event_{}_song_{}", self.descriptor.id, block.event, song_index) } async fn maybe_schedule_poll(&mut self) { @@ -734,6 +774,144 @@ impl WorkerState { fn is_recent_block(&self, event: u64) -> bool { self.processed_blocks.contains(&event) } + + /// Check if all songs from a block are already cached + async fn check_all_songs_cached(&self, block: &Block) -> bool { + let ordered_songs = block.songs_ordered(); + + for (song_index, _song) in &ordered_songs { + let track_id = self.compute_track_id(block, *song_index); + + // Check if metadata exists + let metadata = match self.cache_manager.get_metadata(&track_id).await { + Some(m) => m, + None => { + debug!( + channel = self.descriptor.slug, + event = block.event, + song_index = *song_index, + "Song not in cache: no metadata" + ); + return false; + } + }; + + // Check if audio is cached + let audio_pk = match metadata.cached_audio_pk { + Some(pk) => pk, + None => { + debug!( + channel = self.descriptor.slug, + event = block.event, + song_index = *song_index, + "Song not in cache: no audio_pk" + ); + return false; + } + }; + + // Check if file exists + if self.cache_manager.audio_file_path(&audio_pk).await.is_none() { + debug!( + channel = self.descriptor.slug, + event = block.event, + song_index = *song_index, + "Song not in cache: file not found" + ); + return false; + } + } + + debug!( + channel = self.descriptor.slug, + event = block.event, + "All {} songs are cached", + ordered_songs.len() + ); + true + } + + /// Restore songs from cache and add them to the playlist + async fn restore_from_cache(&mut self, block: &Block) -> Result<()> { + info!( + channel = self.descriptor.slug, + event = block.event, + "Restoring block from cache (no download needed)" + ); + + let ordered_songs = block.songs_ordered(); + + for (song_index, song) in &ordered_songs { + let track_id = self.compute_track_id(block, *song_index); + + // Get metadata (we already checked it exists in check_all_songs_cached) + let metadata = self.cache_manager.get_metadata(&track_id).await + .ok_or_else(|| anyhow!("Metadata disappeared for track_id: {}", track_id))?; + + let audio_pk = metadata.cached_audio_pk.clone() + .ok_or_else(|| anyhow!("Audio PK disappeared for track_id: {}", track_id))?; + + // Get cover PK if available + let cover_pk = if let Some(ref cover_path) = song.cover { + if let Some(cover_url) = block.cover_url(cover_path) { + match self.cache_manager.cache_cover(&cover_url).await { + Ok(pk) => Some(pk), + Err(err) => { + warn!(channel = self.descriptor.slug, "Cover cache error: {err}"); + None + } + } + } else { + None + } + } else { + None + }; + + // Update metadata with cover if we just cached it + if cover_pk.is_some() && metadata.cached_cover_pk.is_none() { + let updated_metadata = TrackMetadata { + cached_cover_pk: cover_pk, + ..metadata.clone() + }; + self.cache_manager.update_metadata(track_id.clone(), updated_metadata).await; + } + + let file_path = self.cache_manager.audio_file_path(&audio_pk).await + .ok_or_else(|| anyhow!("File disappeared for audio_pk: {}", audio_pk))?; + + let duration_ms = song.duration; + + let entry = Arc::new(PlaylistEntry::new( + track_id, + self.descriptor.id, + Arc::new((*song).clone()), + Utc::now(), + duration_ms, + Some(audio_pk), + Some(file_path), + self.active_clients, + )); + + self.playlist.push_active(entry).await; + + info!( + channel = self.descriptor.slug, + song_index = *song_index, + "🎵 Restored '{}' from cache", + song.title + ); + } + + info!( + channel = self.descriptor.slug, + event = block.event, + "Block restored from cache: {} songs", + ordered_songs.len() + ); + + Ok(()) + } } struct ScheduledTask {