diff --git a/src/app/mod.rs b/src/app/mod.rs index 7276b59..8e57e3f 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -44,6 +44,10 @@ pub struct Runtime { pub similarity: Arc, /// When the last Federation-tab status snapshot was requested. pub fed_status_at: Option, + /// Keeps the last successful local-data snapshot visible while a newer + /// one is calculated and collapses bursts of library-change events. + pub local_library_stats_refreshing: Arc, + pub local_library_stats_refresh_requested: Arc, pub library_network_refresh_at: Option, pub library_network_refreshing: Arc, pub library_network_cursors: @@ -102,11 +106,35 @@ fn refresh_local_content_ids(runtime: &Runtime) { } fn refresh_local_library_stats(runtime: &Runtime) { + runtime + .local_library_stats_refresh_requested + .store(true, std::sync::atomic::Ordering::Release); + if runtime + .local_library_stats_refreshing + .swap(true, std::sync::atomic::Ordering::AcqRel) + { + return; + } let library = Arc::clone(&runtime.library); let tx = runtime.event_tx.clone(); + let refreshing = Arc::clone(&runtime.local_library_stats_refreshing); + let requested = Arc::clone(&runtime.local_library_stats_refresh_requested); tokio::task::spawn_blocking(move || { - let result = library.local_stats().map_err(err_string); - let _ = tx.send(AppEvent::LocalLibraryStatsLoaded(result)); + loop { + requested.store(false, std::sync::atomic::Ordering::Release); + let result = library.local_stats().map_err(err_string); + let _ = tx.send(AppEvent::LocalLibraryStatsLoaded(result)); + if requested.load(std::sync::atomic::Ordering::Acquire) { + continue; + } + refreshing.store(false, std::sync::atomic::Ordering::Release); + if requested.swap(false, std::sync::atomic::Ordering::AcqRel) + && !refreshing.swap(true, std::sync::atomic::Ordering::AcqRel) + { + continue; + } + break; + } }); } @@ -303,6 +331,8 @@ pub async fn run( federation, similarity, fed_status_at: None, + local_library_stats_refreshing: Arc::new(std::sync::atomic::AtomicBool::new(false)), + local_library_stats_refresh_requested: Arc::new(std::sync::atomic::AtomicBool::new(false)), library_network_refresh_at: None, library_network_refreshing: Arc::new(std::sync::atomic::AtomicBool::new(false)), library_network_cursors: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())), @@ -2746,7 +2776,12 @@ fn on_library_changed(state: &mut AppState, runtime: &mut Runtime) { // until then. state.likes_loaded = false; state.local_content_ids_loaded = false; - state.local_library_stats = None; + // Refresh in place: status cards keep the last successful snapshot + // instead of flashing `loading` for every background library event. + if state.local_library_stats.is_none() { + state.local_library_stats = Some(state::Loadable::Loading); + } + refresh_local_library_stats(runtime); // Fresh copies of whatever sits in the queue. Federated placeholders // and ephemeral tracks (negative ids) are not library rows and keep @@ -3792,15 +3827,17 @@ fn handle_app_event(state: &mut AppState, runtime: &mut Runtime, event: AppEvent tracing::warn!(%message, "local content id load failed"); } }, - AppEvent::LocalLibraryStatsLoaded(result) => { - state.local_library_stats = Some(match result { - Ok(stats) => state::Loadable::Ready(stats), - Err(message) => { - tracing::warn!(%message, "local library stats load failed"); - state::Loadable::Failed(message) + AppEvent::LocalLibraryStatsLoaded(result) => match result { + Ok(stats) => { + state.local_library_stats = Some(state::Loadable::Ready(stats)); + } + Err(message) => { + tracing::warn!(%message, "local library stats load failed"); + if !matches!(state.local_library_stats, Some(state::Loadable::Ready(_))) { + state.local_library_stats = Some(state::Loadable::Failed(message)); } - }); - } + } + }, AppEvent::LocalContentAvailable { content_id } => { if let Some(content_id) = music_dht::normalize_content_id(&content_id) { state.local_content_ids.insert(content_id); diff --git a/src/similarity.rs b/src/similarity.rs index 4a90370..4c732a8 100644 --- a/src/similarity.rs +++ b/src/similarity.rs @@ -7,7 +7,7 @@ use std::collections::{HashMap, VecDeque}; use std::fs::File; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use std::time::Instant; @@ -101,7 +101,7 @@ impl Phase { } } -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct SimilarityStatus { pub phase: Phase, pub active_profile: Option, @@ -144,6 +144,8 @@ pub struct Manager { settings: Mutex, workers: AtomicUsize, generation: AtomicU64, + pipeline_running: AtomicBool, + rescan_requested: AtomicBool, status: Mutex, index: RwLock, model: Mutex>, @@ -169,13 +171,20 @@ impl Manager { Err(err) => tracing::warn!(%err, "similarity index restore failed"), } } + let target_profile = model_by_id(&settings.model) + .filter(|_| profile_by_id(&settings.profile).is_some()) + .map(|model| profile_fingerprint(model, &settings.profile)); + let restored_profile_is_current = index.profile_id == target_profile; let status = SimilarityStatus { - phase: if settings.enabled { - Phase::Loading - } else { + phase: if !settings.enabled { Phase::Disabled + } else if restored_profile_is_current { + Phase::Ready + } else { + Phase::Loading }, active_profile: index.profile_id.clone(), + target_profile, model: settings.model.clone(), ..SimilarityStatus::default() }; @@ -184,6 +193,8 @@ impl Manager { event_tx, workers: AtomicUsize::new(settings.workers.clamp(1, 16)), generation: AtomicU64::new(0), + pipeline_running: AtomicBool::new(false), + rescan_requested: AtomicBool::new(false), settings: Mutex::new(settings), status: Mutex::new(status), index: RwLock::new(index), @@ -223,23 +234,51 @@ impl Manager { || previous.model != settings.model || previous.profile != settings.profile { + self.generation.fetch_add(1, Ordering::AcqRel); self.start(); } } + /// Requests a scan without cancelling useful work already in progress. + /// Bursts of library-change notifications collapse into one follow-up + /// pass, so metadata refreshes cannot repeatedly restart the model. pub fn start(self: &Arc) { - let generation = self.generation.fetch_add(1, Ordering::AcqRel) + 1; + self.rescan_requested.store(true, Ordering::Release); + if self.pipeline_running.swap(true, Ordering::AcqRel) { + return; + } let this = Arc::clone(self); tokio::spawn(async move { - if let Err(err) = this.run_pipeline(generation).await - && this.generation.load(Ordering::Acquire) == generation - { - tracing::error!(%err, "similarity pipeline failed"); - this.update_status(|status| { - status.phase = Phase::Error; - status.current_track = None; - status.last_error = Some(format!("{err:#}")); - }); + loop { + // This pass covers every notification received before it + // starts. A notification during the pass requests one more. + this.rescan_requested.store(false, Ordering::Release); + let generation = this.generation.load(Ordering::Acquire); + if let Err(err) = this.run_pipeline(generation).await + && this.generation.load(Ordering::Acquire) == generation + { + tracing::error!(%err, "similarity pipeline failed"); + this.update_status(|status| { + status.phase = Phase::Error; + status.current_track = None; + status.last_error = Some(format!("{err:#}")); + }); + } + + if this.rescan_requested.load(Ordering::Acquire) { + continue; + } + + this.pipeline_running.store(false, Ordering::Release); + // Close the small race between checking the request flag and + // releasing ownership of the worker. If another worker has + // already claimed it, that worker owns the pending pass. + if this.rescan_requested.swap(false, Ordering::AcqRel) + && !this.pipeline_running.swap(true, Ordering::AcqRel) + { + continue; + } + break; } }); } @@ -431,7 +470,6 @@ impl Manager { )?; let stats = self.library.similarity_storage_stats(&profile_id)?; self.update_status(|status| { - status.phase = Phase::Downloading; status.target_profile = Some(profile_id.clone()); status.model = spec.id.to_string(); status.total_tracks = stats.total_tracks; @@ -442,21 +480,18 @@ impl Manager { status.current_track = None; status.last_error = None; }); - let model_path = self.ensure_model(spec, generation).await?; - self.ensure_generation(generation)?; - self.update_status(|status| status.phase = Phase::Loading); - let model = self.load_model(&profile_id, &model_path).await?; - self.ensure_generation(generation)?; let mut pending: VecDeque<_> = self.library.pending_similarity_tracks(&profile_id)?.into(); - let pending_total = pending.len(); - self.update_status(|status| { - status.phase = if pending_total == 0 { - Phase::Loading - } else { - Phase::Processing - }; - }); + if pending.is_empty() { + self.ensure_generation(generation)?; + return self.activate_profile(profile_id); + } + + let model_path = self.ensure_model(spec, generation).await?; + self.ensure_generation(generation)?; + let model = self.load_model(&profile_id, &model_path).await?; + self.ensure_generation(generation)?; + self.update_status(|status| status.phase = Phase::Processing); let mut jobs = tokio::task::JoinSet::new(); while !pending.is_empty() || !jobs.is_empty() { @@ -506,6 +541,10 @@ impl Manager { } } self.ensure_generation(generation)?; + self.activate_profile(profile_id) + } + + fn activate_profile(&self, profile_id: String) -> Result<()> { let entries = self.library.load_similarity_index(&profile_id)?; let total_tracks = self .library @@ -519,7 +558,12 @@ impl Manager { profile_id: Some(profile_id.clone()), entries, }; - lock(&self.settings).active_profile = Some(profile_id.clone()); + let profile_changed = { + let mut settings = lock(&self.settings); + let changed = settings.active_profile.as_deref() != Some(&profile_id); + settings.active_profile = Some(profile_id.clone()); + changed + }; let stats = self.library.similarity_storage_stats(&profile_id)?; self.update_status(|status| { status.phase = Phase::Ready; @@ -531,9 +575,11 @@ impl Manager { status.stored_bytes = stats.stored_bytes; status.current_track = None; }); - let _ = self - .event_tx - .send(AppEvent::SimilarityProfileActivated(Some(profile_id))); + if profile_changed { + let _ = self + .event_tx + .send(AppEvent::SimilarityProfileActivated(Some(profile_id))); + } Ok(()) } @@ -561,6 +607,7 @@ impl Manager { tokio::fs::remove_file(&path).await?; } + self.update_status(|status| status.phase = Phase::Downloading); let response = reqwest::get(spec.url).await?.error_for_status()?; let tmp = path.with_extension(format!("part-{}-{generation}", std::process::id())); let mut file = tokio::fs::File::create(&tmp).await?; @@ -603,6 +650,7 @@ impl Manager { { return Ok(Arc::clone(model)); } + self.update_status(|status| status.phase = Phase::Loading); let path = path.to_path_buf(); let model = tokio::task::spawn_blocking(move || load_onnx(&path)) .await @@ -614,10 +662,13 @@ impl Manager { fn update_status(&self, update: impl FnOnce(&mut SimilarityStatus)) { let snapshot = { let mut status = lock(&self.status); + let previous = status.clone(); update(&mut status); - status.clone() + (*status != previous).then(|| status.clone()) }; - let _ = self.event_tx.send(AppEvent::SimilarityStatus(snapshot)); + if let Some(snapshot) = snapshot { + let _ = self.event_tx.send(AppEvent::SimilarityStatus(snapshot)); + } } } @@ -977,6 +1028,14 @@ fn write(lock: &RwLock) -> std::sync::RwLockWriteGuard<'_, T> { mod tests { use super::*; + fn unique_test_dir(label: &str) -> PathBuf { + let unique = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("furumi-{label}-{}-{unique}", std::process::id())) + } + #[test] fn profile_fingerprint_changes_with_contract() { let model = &MODELS[0]; @@ -1010,6 +1069,43 @@ mod tests { assert!(!is_near_duplicate(&distinct, &[&query])); } + #[tokio::test] + async fn repeated_rescans_keep_an_up_to_date_profile_ready() { + let directory = unique_test_dir("similarity-stable-status"); + let library = Arc::new(Library::open(&directory.join("library.db")).unwrap()); + let profile_id = profile_fingerprint(&MODELS[0], DEFAULT_PROFILE_ID); + let settings = SimilaritySettings { + enabled: true, + active_profile: Some(profile_id), + ..SimilaritySettings::default() + }; + let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel(); + let manager = Manager::new(Arc::clone(&library), event_tx, settings); + + assert_eq!(manager.status().phase, Phase::Ready); + for _ in 0..32 { + manager.start(); + } + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while manager.pipeline_running.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + + assert_eq!(manager.status().phase, Phase::Ready); + while let Ok(event) = event_rx.try_recv() { + if let AppEvent::SimilarityStatus(status) = event { + assert_eq!(status.phase, Phase::Ready); + } + } + + drop(manager); + drop(library); + std::fs::remove_dir_all(directory).unwrap(); + } + #[test] fn resampling_keeps_a_constant_signal() { let output = resample_sinc(&vec![0.25; 441], 44_100, 16_000); diff --git a/src/ui/federation.rs b/src/ui/federation.rs index bd76676..29ade78 100644 --- a/src/ui/federation.rs +++ b/src/ui/federation.rs @@ -541,21 +541,20 @@ fn short_id(id: &str) -> String { } fn draw_status_column(frame: &mut Frame, area: Rect, state: &AppState) { - if area.height < 12 { - draw_status(frame, area, state); - return; - } - let [similarity_area, _, federation_area] = Layout::vertical([ - Constraint::Length(8), - Constraint::Length(1), - Constraint::Min(0), - ]) - .areas(area); - draw_similarity_status(frame, similarity_area, state); - draw_status(frame, federation_area, state); + draw_status(frame, area, state); } fn draw_similarity_status(frame: &mut Frame, area: Rect, state: &AppState) { + draw_summary_card( + frame, + area, + state, + " Similarity Processing ", + similarity_summary_lines(state), + ); +} + +fn similarity_summary_lines(state: &AppState) -> Vec> { let status = &state.similarity.status; let progress = if status.total_tracks == 0 { "0 / 0".to_string() @@ -572,34 +571,28 @@ fn draw_similarity_status(frame: &mut Frame, area: Rect, state: &AppState) { .as_deref() .map(short_id) .unwrap_or_else(|| "—".to_string()); - draw_summary_card( - frame, - area, - state, - " Similarity Processing ", - vec![ - status_line("State", status.phase.label().to_string()), - status_line("Progress", progress), - status_line("Active", active), - status_line("Processing", target), - status_line( - "Stored", - format!( - "{} vectors / {}", - status.stored_vectors, - short_bytes_label(status.stored_bytes) - ), + vec![ + summary_line("State", status.phase.label().to_string()), + summary_line("Progress", progress), + summary_line("Active", active), + summary_line("Processing", target), + summary_line( + "Stored", + format!( + "{} vectors / {}", + status.stored_vectors, + short_bytes_label(status.stored_bytes) ), - status_line( - "Current / errors", - status - .current_track - .clone() - .or_else(|| status.last_error.clone()) - .unwrap_or_else(|| format!("{} errors", status.failed_tracks)), - ), - ], - ); + ), + summary_line( + "Current", + status + .current_track + .clone() + .or_else(|| status.last_error.clone()) + .unwrap_or_else(|| format!("{} errors", status.failed_tracks)), + ), + ] } fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { @@ -616,69 +609,81 @@ fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { return; } - if area.width >= 60 && area.height >= 20 { - let protocols_height = - protocol_card_height(state, area.width.saturating_sub(2), area.height); - let [top_area, _, bottom_area, _, protocols_area, _] = Layout::vertical([ - Constraint::Length(7), - Constraint::Length(1), - Constraint::Length(7), - Constraint::Length(1), - Constraint::Length(protocols_height), - Constraint::Min(0), - ]) - .areas(area); - let [status_area, _, local_area] = Layout::horizontal([ - Constraint::Percentage(50), - Constraint::Length(1), - Constraint::Percentage(50), - ]) - .areas(top_area); - let [transport_area, _, devices_area] = Layout::horizontal([ - Constraint::Percentage(50), - Constraint::Length(1), - Constraint::Percentage(50), - ]) - .areas(bottom_area); - draw_summary_card( - frame, - status_area, - state, - " Status ", - node_summary_lines(state), - ); - draw_summary_card( - frame, - local_area, - state, - " Local Data ", - local_data_summary_lines(state), - ); - draw_summary_card( - frame, - transport_area, - state, - " Iroh Transport ", - transport_summary_lines(state), - ); - draw_summary_card( - frame, - devices_area, - state, - " Connected Devices ", - device_summary_lines(state), - ); - draw_summary_card( - frame, - protocols_area, - state, - " Protocol Versions ", - protocol_summary_lines(state, protocols_area.width.saturating_sub(2)), - ); - return; + if area.width >= 60 { + let paired_width = area.width.saturating_sub(1) / 2; + let paired_height = + protocol_card_height(state, paired_width.saturating_sub(2), area.height).max(8); + if area.height >= 16 + paired_height { + let [top_area, _, middle_area, _, paired_area, _] = Layout::vertical([ + Constraint::Length(7), + Constraint::Length(1), + Constraint::Length(7), + Constraint::Length(1), + Constraint::Length(paired_height), + Constraint::Min(0), + ]) + .areas(area); + let [status_area, _, local_area] = Layout::horizontal([ + Constraint::Percentage(50), + Constraint::Length(1), + Constraint::Percentage(50), + ]) + .areas(top_area); + let [transport_area, _, devices_area] = Layout::horizontal([ + Constraint::Percentage(50), + Constraint::Length(1), + Constraint::Percentage(50), + ]) + .areas(middle_area); + let [similarity_area, _, protocols_area] = Layout::horizontal([ + Constraint::Percentage(50), + Constraint::Length(1), + Constraint::Percentage(50), + ]) + .areas(paired_area); + draw_summary_card( + frame, + status_area, + state, + " Status ", + node_summary_lines(state), + ); + draw_summary_card( + frame, + local_area, + state, + " Local Data ", + local_data_summary_lines(state), + ); + draw_summary_card( + frame, + transport_area, + state, + " Iroh Transport ", + transport_summary_lines(state), + ); + draw_summary_card( + frame, + devices_area, + state, + " Connected Devices ", + device_summary_lines(state), + ); + draw_similarity_status(frame, similarity_area, state); + draw_summary_card( + frame, + protocols_area, + state, + " Protocol Versions ", + protocol_summary_lines(state, protocols_area.width.saturating_sub(2)), + ); + return; + } } - if area.height < 39 { + let protocols_height = protocol_card_height(state, area.width.saturating_sub(2), area.height); + let required_vertical_height = 41u16.saturating_add(protocols_height); + if area.height < required_vertical_height { frame.render_widget( Paragraph::new(compact_status_lines(state)) .wrap(ratatui::widgets::Wrap { trim: false }), @@ -696,6 +701,8 @@ fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { _, local_area, _, + similarity_area, + _, protocols_area, _, ] = Layout::vertical([ @@ -707,11 +714,9 @@ fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { Constraint::Length(1), Constraint::Length(7), Constraint::Length(1), - Constraint::Length(protocol_card_height( - state, - area.width.saturating_sub(2), - area.height, - )), + Constraint::Length(8), + Constraint::Length(1), + Constraint::Length(protocols_height), Constraint::Min(0), ]) .areas(area); @@ -744,6 +749,7 @@ fn draw_status(frame: &mut Frame, area: Rect, state: &AppState) { " Local Data ", local_data_summary_lines(state), ); + draw_similarity_status(frame, similarity_area, state); draw_summary_card( frame, protocols_area, @@ -767,6 +773,12 @@ fn compact_status_lines(state: &AppState) -> Vec> { lines.push(Line::styled("Connected Devices", theme::header_for(state))); lines.extend(device_summary_lines(state).into_iter().take(2)); lines.push(Line::default()); + lines.push(Line::styled( + "Similarity Processing", + theme::header_for(state), + )); + lines.extend(similarity_summary_lines(state).into_iter().take(3)); + lines.push(Line::default()); lines.push(Line::styled("Protocol Versions", theme::header_for(state))); lines.extend(protocol_summary_lines(state, 0).into_iter().take(3)); lines