diff --git a/CHANGELOG.md b/CHANGELOG.md index d705db5..3145a1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.2.8] - 2026-09-02 + ### Added - Optional offline music-similarity search for local tracks, backed by @@ -27,7 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - Similarity wire types, bounds, validation, and stream framing now come from - the shared `music-dht 0.4.0` API so native, web, and future clients can + the shared `music-dht 0.4` API so native, web, and future clients can interoperate without sharing an embedding implementation. - Existing SQLite embeddings are backfilled once with compact 256-bit routing signatures; new embeddings store them immediately without changing exact @@ -41,6 +43,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Federation now recovers automatically after sleep, prolonged idle, or a + degraded rendezvous transport while preserving local playback and state. - Current-track information (`Shift+I`) now uses the enriched queue entry, so it shows the same complete metadata and similarity action as `I` on that track in the queue. @@ -74,5 +78,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Music-directory validation and migration now reject overlapping changes, resolve canonical paths, and produce Windows-portable managed filenames. -[Unreleased]: https://gt.hexor.cy/ab/furumi_tui/compare/v0.2.5...HEAD +[Unreleased]: https://gt.hexor.cy/ab/furumi_tui/compare/v0.2.8...HEAD +[0.2.8]: https://gt.hexor.cy/ab/furumi_tui/compare/v0.2.7...v0.2.8 [0.2.5]: https://gt.hexor.cy/ab/furumi_tui/compare/v0.2.4...v0.2.5 diff --git a/Cargo.lock b/Cargo.lock index e66645a..efd10a4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1520,9 +1520,9 @@ dependencies = [ [[package]] name = "federation-net" -version = "0.3.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3e690b370c505d153bef214b21a8f2aa55d667367ac1e16bde8bc0de88963c2" +checksum = "01e5c2e6e5af020dc202d4924125ef57719aa21f8d2567e2d943914386bc8980" dependencies = [ "blake3", "data-encoding", @@ -1666,7 +1666,7 @@ dependencies = [ [[package]] name = "furumi_tui" -version = "0.2.7" +version = "0.2.8" dependencies = [ "anyhow", "base64", @@ -3161,9 +3161,9 @@ dependencies = [ [[package]] name = "music-dht" -version = "0.4.0" +version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c5b429b90a8f1b0980b3a35a6fa5445d7a275c737eb04db18db4d7f14c81478" +checksum = "78799c049bac981b3b84fbe8c9965865ab061d7e1c6e40ed2c5297749569922d" dependencies = [ "async-trait", "blake3", diff --git a/Cargo.toml b/Cargo.toml index 7de9c0a..f136c37 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumi_tui" -version = "0.2.7" +version = "0.2.8" edition = "2024" rust-version = "1.97" description = "A federated P2P player for personal music libraries" @@ -22,7 +22,7 @@ image = { version = "0.25.10", default-features = false, features = ["jpeg", "pn lofty = "0.22" # P2P federation: library index in a shared DHT + audio streaming between # peers (same protocol as furumi-fd). -music-dht = "0.4.0" +music-dht = "0.4.1" ratatui = "0.30.1" reqwest = { version = "0.12.28", default-features = false, features = ["rustls-tls", "stream"] } rhai = { version = "1", features = ["sync"] } diff --git a/src/app/mod.rs b/src/app/mod.rs index 68f3a65..6b482d0 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -312,6 +312,7 @@ pub async fn run( Arc::clone(&similarity), settings.music_dir.clone(), ); + federation.start_supervisor(); state.music_dir = federation.media_dir(); state.federation.settings = federation.settings(); state.federation.devices = Some(devices.status()); diff --git a/src/federation/mod.rs b/src/federation/mod.rs index 7a0eba9..4a8e189 100644 --- a/src/federation/mod.rs +++ b/src/federation/mod.rs @@ -21,8 +21,8 @@ use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; use std::str::FromStr; use std::sync::Arc; -use std::sync::atomic::{AtomicI64, Ordering}; -use std::time::Duration; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; +use std::time::{Duration, Instant}; use anyhow::{Context, Result}; use music_dht::similarity_dht::SimilarityDht; @@ -46,6 +46,11 @@ pub use similarity::SIMILARITY_ALPN; /// How often the published library is re-synchronized with the local index. const SYNC_INTERVAL: Duration = Duration::from_secs(60); +/// How often the application checks whether the shared transport needs a +/// full restart so all application-owned protocol acceptors are recreated. +const SUPERVISOR_INTERVAL: Duration = Duration::from_secs(15); +/// Prevents repeated restarts during a prolonged external network outage. +const RECOVERY_COOLDOWN: Duration = Duration::from_secs(5 * 60); /// How many times a share-link content lookup is retried before the label /// fallback kicks in. @@ -370,6 +375,12 @@ pub struct FedStatus { pub endpoint_id: String, pub dht_node_id: String, pub connected_peers: Vec, + pub network_health: String, + pub rendezvous_failures: u32, + pub peer_dial_failures: u32, + pub rendezvous_restarts: u64, + pub recovery_count: u64, + pub last_rendezvous_error: Option, pub known_contacts: usize, pub stored_dht_records: Option, pub stored_dht_bytes: Option, @@ -419,6 +430,10 @@ pub struct Federation { metadata_cache: std::sync::Mutex>, settings: std::sync::Mutex, running: tokio::sync::Mutex>, + supervisor_task: std::sync::Mutex>>, + supervisor_shutdown: tokio::sync::Notify, + shutting_down: AtomicBool, + recovery_count: AtomicU64, last_sync: std::sync::Mutex>, last_error: std::sync::Mutex>, transport_stats: Arc, @@ -549,6 +564,10 @@ impl Federation { metadata_cache: std::sync::Mutex::new(Default::default()), settings: std::sync::Mutex::new(load_settings()), running: tokio::sync::Mutex::new(None), + supervisor_task: std::sync::Mutex::new(None), + supervisor_shutdown: tokio::sync::Notify::new(), + shutting_down: AtomicBool::new(false), + recovery_count: AtomicU64::new(0), last_sync: std::sync::Mutex::new(None), last_error: std::sync::Mutex::new(initial_error), transport_stats: Arc::new(TransportStats::default()), @@ -560,6 +579,18 @@ impl Federation { lock(&self.settings).clone() } + /// Starts the application-level federation supervisor once. + pub fn start_supervisor(self: &Arc) { + let mut task = lock(&self.supervisor_task); + if task.is_some() || self.shutting_down.load(Ordering::SeqCst) { + return; + } + let federation = Arc::clone(self); + *task = Some(tokio::spawn(async move { + federation.supervisor_loop().await; + })); + } + pub fn media_dir(&self) -> PathBuf { lock(&self.media_dir).clone() } @@ -665,12 +696,43 @@ impl Federation { network_id: NetworkId, network_name: String, ) -> Result<()> { + self.start_with_network_id_mode(network_id, network_name, false) + .await + .map(|_| ()) + } + + async fn start_with_network_id_mode( + self: &Arc, + network_id: NetworkId, + network_name: String, + recovery_only: bool, + ) -> Result { let mut guard = self.running.lock().await; + if recovery_only { + let current = self.settings(); + if !current.enabled + || current.network_id.trim() != network_name + || NetworkId::from_name(current.network_id.trim()) != network_id + { + return Ok(false); + } + } if let Some(running) = guard.as_ref() { if running.network_id == network_id { - return Ok(()); + if !recovery_only || !running.service.network_health().restart_recommended { + return Ok(false); + } + tracing::warn!( + network = %network_name, + health = %running.service.network_health().state, + "restarting degraded federation service" + ); + } else if recovery_only { + return Ok(false); } stop_running(guard.take()).await; + } else if recovery_only { + tracing::warn!(network = %network_name, "retrying stopped federation service"); } std::fs::create_dir_all(&self.data_dir) .with_context(|| format!("creating {}", self.data_dir.display()))?; @@ -828,7 +890,7 @@ impl Federation { ], }); self.set_error(None); - Ok(()) + Ok(true) } async fn stop(&self) { @@ -837,9 +899,57 @@ impl Federation { } pub async fn shutdown(&self) { + self.shutting_down.store(true, Ordering::SeqCst); + self.supervisor_shutdown.notify_one(); + let supervisor = lock(&self.supervisor_task).take(); + if let Some(supervisor) = supervisor { + let _ = supervisor.await; + } self.stop().await; } + async fn supervisor_loop(self: Arc) { + let mut interval = tokio::time::interval(SUPERVISOR_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + let mut last_attempt = None; + loop { + tokio::select! { + _ = self.supervisor_shutdown.notified() => return, + _ = interval.tick() => {} + } + if self.shutting_down.load(Ordering::SeqCst) { + return; + } + if last_attempt.is_some_and(|attempt: Instant| attempt.elapsed() < RECOVERY_COOLDOWN) { + continue; + } + match self.recover_if_needed().await { + Ok(false) => {} + Ok(true) => { + last_attempt = Some(Instant::now()); + self.recovery_count.fetch_add(1, Ordering::Relaxed); + self.spawn_sync_soon().await; + } + Err(err) => { + last_attempt = Some(Instant::now()); + tracing::error!(error = %err, "federation recovery failed"); + self.set_error(Some(format!("network recovery failed: {err}"))); + } + } + } + } + + async fn recover_if_needed(self: &Arc) -> Result { + let settings = self.settings(); + if !settings.enabled || settings.network_id.trim().is_empty() { + return Ok(false); + } + let network_name = settings.network_id.trim().to_string(); + self.start_with_network_id_mode(NetworkId::from_name(&network_name), network_name, true) + .await + } + async fn service(&self) -> Result> { self.running .lock() @@ -975,6 +1085,7 @@ impl Federation { let guard = self.running.lock().await; let mut status = FedStatus { network: settings.network_id, + recovery_count: self.recovery_count.load(Ordering::Relaxed), last_sync: lock(&self.last_sync).clone(), last_error: lock(&self.last_error).clone(), protocols: ProtocolVersions::snapshot(&self.observed_protocols), @@ -982,6 +1093,7 @@ impl Federation { }; if let Some(running) = guard.as_ref() { let service = &running.service; + let health = service.network_health(); status.running = true; status.network = running.network_name.clone(); status.endpoint_id = service.endpoint_id().to_string(); @@ -991,6 +1103,11 @@ impl Federation { .iter() .map(|p| p.to_string()) .collect(); + status.network_health = health.state.to_string(); + status.rendezvous_failures = health.consecutive_rendezvous_failures; + status.peer_dial_failures = health.consecutive_peer_dial_failures; + status.rendezvous_restarts = health.rendezvous_restarts; + status.last_rendezvous_error = health.last_rendezvous_error; status.known_contacts = service.known_peers().len(); status.stored_dht_records = service.dht_record_count().await.ok(); status.stored_dht_bytes = diff --git a/src/ui/federation.rs b/src/ui/federation.rs index 90edaa8..0d68083 100644 --- a/src/ui/federation.rs +++ b/src/ui/federation.rs @@ -961,7 +961,10 @@ fn node_summary_lines(state: &AppState) -> Vec> { ] } Some(status) => vec![ - summary_line("Node", format!("running on {}", status.network)), + summary_line( + "Node", + format!("running on {} · {}", status.network, status.network_health), + ), summary_line( "Peers", format!( @@ -1215,6 +1218,17 @@ fn status_detail_status_lines(state: &AppState, status_cursor: usize) -> Vec = status .connected_peers @@ -1260,6 +1274,9 @@ fn status_detail_status_lines(state: &AppState, status_cursor: usize) -> Vec