Fix lock

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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumi_tui"
-version = "0.2.6"
+version = "0.2.8"
 edition = "2024"
 rust-version = "1.97"
 description = "A federated P2P player for personal music libraries"
@@ -22,17 +22,19 @@ 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"] }
 rodio = { version = "0.22.2", default-features = false, features = ["playback", "mp3", "flac", "vorbis", "wav", "symphonia-aac", "symphonia-isomp4", "symphonia-alac"] }
+rusty-opus = "0.9.1"
 rustfft = "6.4.1"
 rusqlite = { version = "0.32", features = ["bundled", "functions"] }
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.150"
 sha2 = "0.10.9"
 souvlaki = { version = "0.8.3", default-features = false, features = ["use_zbus"] }
+symphonia = { version = "0.5.5", default-features = false, features = ["ogg"] }
 thiserror = "2.0.18"
 tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "time", "fs", "io-util"] }
 toml = "1.1.2"
@@ -54,3 +56,6 @@ windows-sys = { version = "0.61.2", features = ["Win32_Foundation", "Win32_UI_Wi
 
 [target."cfg(unix)".dependencies]
 libc = "0.2.186"
+
+[dev-dependencies]
+base64 = "0.22.1"

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());

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::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
-use std::time::Duration;
+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<String>,
+    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<String>,
     pub known_contacts: usize,
     pub stored_dht_records: Option<usize>,
     pub stored_dht_bytes: Option<u64>,
@@ -419,6 +430,10 @@ pub struct Federation {
     metadata_cache: std::sync::Mutex<std::collections::HashMap<String, CachedTrackMetadata>>,
     settings: std::sync::Mutex<FedSettings>,
     running: tokio::sync::Mutex<Option<Running>>,
+    supervisor_task: std::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
+    supervisor_shutdown: tokio::sync::Notify,
+    shutting_down: AtomicBool,
+    recovery_count: AtomicU64,
     last_sync: std::sync::Mutex<Option<String>>,
     last_error: std::sync::Mutex<Option<String>>,
     transport_stats: Arc<TransportStats>,
@@ -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<Self>) {
+        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<Self>,
+        network_id: NetworkId,
+        network_name: String,
+        recovery_only: bool,
+    ) -> Result<bool> {
         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<Self>) {
+        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<Self>) -> Result<bool> {
+        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<Arc<MusicDhtService>> {
         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 =

src/player/mod.rs

@@ -5,13 +5,14 @@
 //! the UI or app state.
 
 mod analyzer;
+mod opus;
 
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
 use std::time::Duration;
 
-use rodio::{Decoder, DeviceSinkBuilder, Player, stream::MixerDeviceSink};
+use rodio::{Decoder, DeviceSinkBuilder, Player, Source, stream::MixerDeviceSink};
 
 pub use analyzer::AudioAnalysisSnapshot;
 
@@ -226,23 +227,13 @@ fn handle(
             }
             let out = output.as_ref().expect("output opened above");
 
-            let mut builder = Decoder::builder()
+            match decode_source(reader, byte_len, mime_type.as_deref(), seekable) {
-                .with_data(reader)
+                Ok(source) => {
-                .with_seekable(seekable)
-                .with_gapless(true);
-            if let Some(len) = byte_len {
-                builder = builder.with_byte_len(len);
-            }
-            if let Some(mime_type) = mime_type.as_deref() {
-                builder = builder.with_mime_type(mime_type);
-            }
-            match builder.build() {
-                Ok(decoder) => {
                     shared.analysis.clear();
                     out.player.stop();
                     out.player.set_volume(volume);
                     out.player.append(analyzer::AnalyzedSource::new(
-                        decoder,
+                        source,
                         Arc::clone(&shared.analysis),
                     ));
                     out.player.play();
@@ -258,16 +249,9 @@ fn handle(
             let Some(out) = output.as_ref() else {
                 return;
             };
-            let mut builder = Decoder::builder()
+            match decode_source(reader, byte_len, None, true) {
-                .with_data(reader)
+                Ok(source) => out.player.append(analyzer::AnalyzedSource::new(
-                .with_seekable(true)
+                    source,
-                .with_gapless(true);
-            if let Some(len) = byte_len {
-                builder = builder.with_byte_len(len);
-            }
-            match builder.build() {
-                Ok(decoder) => out.player.append(analyzer::AnalyzedSource::new(
-                    decoder,
                     Arc::clone(&shared.analysis),
                 )),
                 Err(err) => {
@@ -309,3 +293,38 @@ fn handle(
         }
     }
 }
+
+pub(crate) type DecodedSource = Box<dyn Source + Send>;
+
+pub(crate) fn decode_source(
+    mut reader: TrackReader,
+    byte_len: Option<u64>,
+    mime_type: Option<&str>,
+    seekable: bool,
+) -> Result<DecodedSource, String> {
+    let mime_is_opus = mime_type.is_some_and(|mime| {
+        let mime = mime.to_ascii_lowercase();
+        mime == "audio/opus" || mime.contains("codecs=opus") || mime.contains("codecs=\"opus\"")
+    });
+    let ogg_is_opus = opus::is_ogg_opus(&mut reader)
+        .map_err(|error| format!("cannot inspect audio stream: {error}"))?;
+    if mime_is_opus || ogg_is_opus {
+        return opus::OggOpusSource::new(reader, byte_len, seekable)
+            .map(|source| Box::new(source) as DecodedSource);
+    }
+
+    let mut builder = Decoder::builder()
+        .with_data(reader)
+        .with_seekable(seekable)
+        .with_gapless(true);
+    if let Some(len) = byte_len {
+        builder = builder.with_byte_len(len);
+    }
+    if let Some(mime_type) = mime_type {
+        builder = builder.with_mime_type(mime_type);
+    }
+    builder
+        .build()
+        .map(|decoder| Box::new(decoder) as DecodedSource)
+        .map_err(|error| error.to_string())
+}

src/player/opus.rs

@@ -0,0 +1,433 @@
+//! Ogg/Opus decoding for rodio.
+//!
+//! Rodio 0.22 can demux Ogg and decode Vorbis, but its Symphonia version does
+//! not include an Opus decoder.  This source keeps Symphonia's mature Ogg
+//! demuxing and feeds the packets into the pure-Rust `rusty-opus` decoder.
+
+use std::io::{self, Read, Seek, SeekFrom};
+use std::sync::Arc;
+use std::time::Duration;
+
+use rodio::source::SeekError;
+use rodio::{ChannelCount, SampleRate, Source};
+use rusty_opus::OpusDecoder;
+use symphonia::core::codecs::{CODEC_TYPE_OPUS, CodecParameters};
+use symphonia::core::errors::Error as SymphoniaError;
+use symphonia::core::formats::{FormatOptions, FormatReader, SeekMode, SeekTo};
+use symphonia::core::io::{MediaSource, MediaSourceStream};
+use symphonia::core::meta::MetadataOptions;
+use symphonia::core::probe::Hint;
+
+use super::TrackReader;
+
+const OPUS_SAMPLE_RATE: u32 = 48_000;
+const OPUS_SEEK_PREROLL_FRAMES: u64 = OPUS_SAMPLE_RATE as u64 * 80 / 1_000;
+const SNIFF_BYTES: usize = 512;
+
+/// Inspect the Ogg identification page without changing the reader position.
+pub(super) fn is_ogg_opus(reader: &mut TrackReader) -> io::Result<bool> {
+    let position = reader.stream_position()?;
+    let mut header = [0; SNIFF_BYTES];
+    let read_result = reader.read(&mut header);
+    let rewind_result = reader.seek(SeekFrom::Start(position));
+
+    let read = read_result?;
+    rewind_result?;
+    Ok(header[..read].starts_with(b"OggS")
+        && header[..read]
+            .windows(b"OpusHead".len())
+            .any(|window| window == b"OpusHead"))
+}
+
+struct ReaderMediaSource {
+    reader: TrackReader,
+    byte_len: Option<u64>,
+    seekable: bool,
+}
+
+impl Read for ReaderMediaSource {
+    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
+        self.reader.read(buffer)
+    }
+}
+
+impl Seek for ReaderMediaSource {
+    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
+        self.reader.seek(position)
+    }
+}
+
+impl MediaSource for ReaderMediaSource {
+    fn is_seekable(&self) -> bool {
+        self.seekable
+    }
+
+    fn byte_len(&self) -> Option<u64> {
+        self.byte_len
+    }
+}
+
+pub(super) struct OggOpusSource {
+    format: Box<dyn FormatReader>,
+    decoder: OpusDecoder,
+    track_id: u32,
+    channels: u16,
+    pre_skip: u64,
+    playable_frames: Option<u64>,
+    output_position_frames: u64,
+    discard_frames: u64,
+    output_gain: f32,
+    buffer: Vec<f32>,
+    scratch: Vec<f32>,
+    buffer_position: usize,
+    seekable: bool,
+    done: bool,
+}
+
+impl OggOpusSource {
+    pub(super) fn new(
+        reader: TrackReader,
+        byte_len: Option<u64>,
+        seekable: bool,
+    ) -> Result<Self, String> {
+        let source = ReaderMediaSource {
+            reader,
+            byte_len,
+            seekable,
+        };
+        let stream = MediaSourceStream::new(Box::new(source), Default::default());
+        let mut hint = Hint::new();
+        hint.with_extension("ogg");
+        hint.mime_type("audio/ogg");
+        let format_options = FormatOptions {
+            enable_gapless: false,
+            ..Default::default()
+        };
+        let probed = symphonia::default::get_probe()
+            .format(&hint, stream, &format_options, &MetadataOptions::default())
+            .map_err(|error| format!("cannot read Ogg container: {error}"))?;
+        let format = probed.format;
+
+        let (track_id, channels, pre_skip, playable_frames, output_gain) = {
+            let track = format
+                .default_track()
+                .ok_or_else(|| "Ogg container has no audio track".to_string())?;
+            if track.codec_params.codec != CODEC_TYPE_OPUS {
+                return Err("Ogg track is not Opus".to_string());
+            }
+            let channels = track
+                .codec_params
+                .channels
+                .map(|channels| channels.count() as u16)
+                .ok_or_else(|| "Opus track has no channel layout".to_string())?;
+            if !(1..=2).contains(&channels) {
+                return Err(format!(
+                    "Opus track has {channels} channels; only mono and stereo are supported"
+                ));
+            }
+            let extra_data = track.codec_params.extra_data.as_deref();
+            let pre_skip = opus_pre_skip(extra_data);
+            let playable_frames = opus_playable_frames(&track.codec_params, pre_skip);
+            let output_gain = opus_output_gain(extra_data);
+            (track.id, channels, pre_skip, playable_frames, output_gain)
+        };
+        let decoder = OpusDecoder::new(OPUS_SAMPLE_RATE as i32, usize::from(channels))
+            .map_err(|error| format!("cannot initialize Opus decoder: {error}"))?;
+
+        let mut source = Self {
+            format,
+            decoder,
+            track_id,
+            channels,
+            pre_skip,
+            playable_frames,
+            output_position_frames: 0,
+            discard_frames: pre_skip,
+            output_gain,
+            buffer: Vec::new(),
+            scratch: Vec::new(),
+            buffer_position: 0,
+            seekable,
+            done: false,
+        };
+        source.fill_buffer();
+        if source.buffer.is_empty() {
+            return Err("Opus track contains no decodable audio".to_string());
+        }
+        Ok(source)
+    }
+
+    fn fill_buffer(&mut self) {
+        self.buffer.clear();
+        self.buffer_position = 0;
+
+        while self.buffer.is_empty() && !self.done {
+            let packet = match self.format.next_packet() {
+                Ok(packet) => packet,
+                Err(SymphoniaError::IoError(error))
+                    if error.kind() == io::ErrorKind::UnexpectedEof =>
+                {
+                    self.done = true;
+                    break;
+                }
+                Err(error) => {
+                    tracing::warn!(%error, "Ogg/Opus demux failed");
+                    self.done = true;
+                    break;
+                }
+            };
+            if packet.track_id() != self.track_id {
+                continue;
+            }
+
+            let Ok(decoded_frames) = usize::try_from(packet.dur) else {
+                tracing::warn!("Ogg/Opus packet duration is too large");
+                self.done = true;
+                break;
+            };
+            if decoded_frames == 0 {
+                continue;
+            }
+
+            let sample_count = match decoded_frames.checked_mul(usize::from(self.channels)) {
+                Some(sample_count) => sample_count,
+                None => {
+                    tracing::warn!("Ogg/Opus packet sample count overflow");
+                    self.done = true;
+                    break;
+                }
+            };
+            self.scratch.resize(sample_count, 0.0);
+            let frames = match self
+                .decoder
+                .decode(&packet.data, decoded_frames, &mut self.scratch)
+            {
+                Ok(frames) => frames,
+                Err(error) => {
+                    tracing::warn!(%error, "Opus packet decode failed");
+                    continue;
+                }
+            };
+
+            let discarded = self.discard_frames.min(frames as u64) as usize;
+            self.discard_frames -= discarded as u64;
+            let start_frame = discarded;
+            let remaining = self
+                .playable_frames
+                .map(|total| total.saturating_sub(self.output_position_frames))
+                .unwrap_or(u64::MAX);
+            let end_frame = frames
+                .min(start_frame.saturating_add(usize::try_from(remaining).unwrap_or(usize::MAX)));
+            if start_frame >= end_frame {
+                if self
+                    .playable_frames
+                    .is_some_and(|total| self.output_position_frames >= total)
+                {
+                    self.done = true;
+                }
+                continue;
+            }
+
+            let channels = usize::from(self.channels);
+            let start = start_frame * channels;
+            let end = end_frame.saturating_mul(channels).min(self.scratch.len());
+            self.output_position_frames = self
+                .output_position_frames
+                .saturating_add((end_frame - start_frame) as u64);
+            self.buffer.extend(
+                self.scratch[start..end]
+                    .iter()
+                    .map(|sample| (sample * self.output_gain).clamp(-1.0, 1.0)),
+            );
+            if self
+                .playable_frames
+                .is_some_and(|total| self.output_position_frames >= total)
+            {
+                self.done = true;
+            }
+        }
+    }
+
+    fn reset_decoder(&mut self) -> Result<(), SeekError> {
+        self.decoder = OpusDecoder::new(OPUS_SAMPLE_RATE as i32, usize::from(self.channels))
+            .map_err(|error| seek_error(format!("cannot reset Opus decoder: {error}")))?;
+        Ok(())
+    }
+}
+
+impl Iterator for OggOpusSource {
+    type Item = f32;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            if self.buffer_position < self.buffer.len() {
+                let sample = self.buffer[self.buffer_position];
+                self.buffer_position += 1;
+                return Some(sample);
+            }
+            if self.done {
+                return None;
+            }
+            self.fill_buffer();
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.buffer.len().saturating_sub(self.buffer_position), None)
+    }
+}
+
+impl Source for OggOpusSource {
+    fn current_span_len(&self) -> Option<usize> {
+        Some(self.buffer.len())
+    }
+
+    fn channels(&self) -> ChannelCount {
+        ChannelCount::new(self.channels).expect("Opus channel count was validated")
+    }
+
+    fn sample_rate(&self) -> SampleRate {
+        SampleRate::new(OPUS_SAMPLE_RATE).expect("Opus sample rate is non-zero")
+    }
+
+    fn total_duration(&self) -> Option<Duration> {
+        self.playable_frames
+            .map(|frames| Duration::from_secs_f64(frames as f64 / f64::from(OPUS_SAMPLE_RATE)))
+    }
+
+    fn try_seek(&mut self, position: Duration) -> Result<(), SeekError> {
+        if !self.seekable {
+            return Err(SeekError::NotSupported {
+                underlying_source: std::any::type_name::<Self>(),
+            });
+        }
+
+        let requested = duration_frames(position);
+        let target = self
+            .playable_frames
+            .map_or(requested, |total| requested.min(total));
+        let raw_target = target.saturating_add(self.pre_skip);
+        let preroll = raw_target.saturating_sub(OPUS_SEEK_PREROLL_FRAMES);
+        let seeked = self
+            .format
+            .seek(
+                SeekMode::Accurate,
+                SeekTo::TimeStamp {
+                    ts: preroll,
+                    track_id: self.track_id,
+                },
+            )
+            .map_err(|error| seek_error(format!("Ogg seek failed: {error}")))?;
+        self.reset_decoder()?;
+        self.buffer.clear();
+        self.buffer_position = 0;
+        self.output_position_frames = target;
+        self.discard_frames = raw_target.saturating_sub(seeked.actual_ts);
+        self.done = false;
+        self.fill_buffer();
+        Ok(())
+    }
+}
+
+fn duration_frames(duration: Duration) -> u64 {
+    let frames = duration.as_secs_f64() * f64::from(OPUS_SAMPLE_RATE);
+    frames.round().clamp(0.0, u64::MAX as f64) as u64
+}
+
+fn opus_pre_skip(extra_data: Option<&[u8]>) -> u64 {
+    extra_data
+        .filter(|header| header.len() >= 12)
+        .map(|header| u64::from(u16::from_le_bytes([header[10], header[11]])))
+        .unwrap_or(0)
+}
+
+fn opus_playable_frames(params: &CodecParameters, pre_skip: u64) -> Option<u64> {
+    let encoded_frames = params.n_frames?;
+    let padding = u64::from(params.padding.unwrap_or(0));
+
+    // Symphonia normally leaves the OpusHead pre-skip in `delay`. For a very
+    // short stream whose first audio page is also its last page, it instead
+    // reports the page's trailing padding there. Preserve that information so
+    // both one-page and ordinary Ogg/Opus streams end on the correct sample.
+    let one_page_padding = params
+        .delay
+        .map(u64::from)
+        .filter(|delay| *delay != pre_skip)
+        .unwrap_or(0);
+
+    Some(
+        encoded_frames
+            .saturating_sub(pre_skip)
+            .saturating_sub(padding)
+            .saturating_sub(one_page_padding),
+    )
+}
+
+fn opus_output_gain(extra_data: Option<&[u8]>) -> f32 {
+    let Some(header) = extra_data.filter(|header| header.len() >= 18) else {
+        return 1.0;
+    };
+    let gain_q8 = i16::from_le_bytes([header[16], header[17]]);
+    10.0_f32.powf(f32::from(gain_q8) / (20.0 * 256.0))
+}
+
+fn seek_error(message: String) -> SeekError {
+    SeekError::Other(Arc::new(io::Error::other(message)))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::io::Cursor;
+
+    use base64::Engine as _;
+
+    use super::*;
+
+    // 30 ms mono Ogg/Opus sine, encoded by FFmpeg/libopus. Keeping a real
+    // interoperable stream here catches container and codec regressions.
+    const OGG_OPUS: &str = "T2dnUwACAAAAAAAAAABkiBtVAAAAAFZb1toBE09wdXNIZWFkAQE4AYC7AAAAAABPZ2dTAAAAAAAAAAAAAGSIG1UBAAAAq+b2jQE+T3B1c1RhZ3MNAAAATGF2ZjYyLjEyLjEwMgEAAAAdAAAAZW5jb2Rlcj1MYXZjNjIuMjguMTAyIGxpYm9wdXNPZ2dTAATYBgAAAAAAAGSIG1UCAAAAEkNnrgT/NP8x+HJJRycQ5MhbeCeLhvfY79vodePsNuYR8hAn2iaItTjKzvL0UULTvai7BFd4FJ2BZdtZQn4K2vCS9rjadnLmp+u4QMCAJiL3MXbVJBVHjtDhATQq5rg5ZlscpBXWk/NqnQE/QaT/nhMi/VqnLKwRCFcpFnH/NqPx7RhjpvzicAu/blC/sRygALOf6blR6HYkOb7BLn/Vn3ijqdTVwCzwtcpWU2hbCXUOWnTEuXUOWaOmQKy8zmqYgzM7aOueo/2siHPpomLeiHPp4NH0cpD63qkSv4/XxX8HWdBtg9IvLpQ42ch0PrqI7BR4ZhvJoDmg45q/177KfOUxlfoG/8GpJWKTCpaN1u64it+1vkpE7sfaLIuSVNbz6BDOIREzM5pPkdIyBzqThkbRkGJr8KDGTyQyJfi0JXhIR5fGM2RptXd5mTRUZHmCz9dCMyKay/nzCir7oSDyEbWNlfaV7qzrxPXTNJRpJ/6GgfI3Ht2Y9jKPV8WkGzGO/Pyz+hznJTgJdGOpxRmzAWXmcwcSdI0TjPVLIs8SIBtipLhr0R+yDe2ar7ZQR3hxQzoIn5ydLO7mk8QZvr+4gAAAAAAAAAAAAAAAAAAAAAAAAE+e11+HmY5uncnz1m153WYm1urhAnMTa3qa3G/MIezSH+urKDn4eZAnEgz8PMnPgAOL22PlLS7647fYrYbJfnfYvLgwLpXSlHeh7VHf+GQd99vAzM/086lHsksCvwXkd12GFX2BTQbtW/3wPMmi4zL6VsKJUH/Q5ZQ00cvMtFDgSw5CLIg8nHONgH2/XVeZC+VwngKBeuzHHK4=";
+
+    fn fixture() -> Vec<u8> {
+        base64::engine::general_purpose::STANDARD
+            .decode(OGG_OPUS)
+            .unwrap()
+    }
+
+    #[test]
+    fn detects_and_decodes_ogg_opus() {
+        let bytes = fixture();
+        let len = bytes.len() as u64;
+        let mut reader: TrackReader = Box::new(Cursor::new(bytes));
+        assert!(is_ogg_opus(&mut reader).unwrap());
+
+        let source = OggOpusSource::new(reader, Some(len), true).unwrap();
+        assert_eq!(source.channels().get(), 1);
+        assert_eq!(source.sample_rate().get(), OPUS_SAMPLE_RATE);
+        let samples: Vec<_> = source.collect();
+        assert_eq!(samples.len(), 1_440);
+        assert!(samples.iter().any(|sample| sample.abs() > 0.001));
+    }
+
+    #[test]
+    fn seeks_within_ogg_opus() {
+        let bytes = fixture();
+        let len = bytes.len() as u64;
+        let reader: TrackReader = Box::new(Cursor::new(bytes));
+        let mut source = OggOpusSource::new(reader, Some(len), true).unwrap();
+
+        assert_eq!(source.by_ref().count(), 1_440);
+        source.try_seek(Duration::from_millis(15)).unwrap();
+        let samples: Vec<_> = source.collect();
+        assert_eq!(samples.len(), 720);
+        assert!(samples.iter().any(|sample| sample.abs() > 0.001));
+    }
+
+    #[test]
+    fn ogg_vorbis_header_is_not_misclassified_as_opus() {
+        let mut bytes =
+            b"OggS\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\x01\x1e\x01vorbis".to_vec();
+        bytes.resize(SNIFF_BYTES, 0);
+        let mut reader: TrackReader = Box::new(Cursor::new(bytes));
+        assert!(!is_ogg_opus(&mut reader).unwrap());
+    }
+}

src/similarity.rs

@@ -13,7 +13,7 @@ use std::time::Instant;
 
 use anyhow::{Context as _, Result};
 use futures_util::StreamExt as _;
-use rodio::{Decoder, Source as _};
+use rodio::Source as _;
 use rustfft::FftPlanner;
 use rustfft::num_complex::Complex;
 use sha2::{Digest as _, Sha256};
@@ -847,8 +847,15 @@ fn decode_mono_window(
     length_seconds: Option<f64>,
 ) -> Result<Vec<f32>> {
     let file = File::open(path).with_context(|| format!("opening {}", path.display()))?;
-    let mut decoder =
+    let byte_len = file.metadata().ok().map(|metadata| metadata.len());
-        Decoder::try_from(file).with_context(|| format!("decoding {}", path.display()))?;
+    let mut decoder = crate::player::decode_source(
+        Box::new(std::io::BufReader::new(file)),
+        byte_len,
+        None,
+        true,
+    )
+    .map_err(anyhow::Error::msg)
+    .with_context(|| format!("decoding {}", path.display()))?;
     let channels = decoder.channels().get() as usize;
     let source_rate = decoder.sample_rate().get() as usize;
     if start_seconds > 0.0 {

src/streaming.rs

@@ -101,11 +101,30 @@ impl Read for GrowingFileReader {
 }
 
 impl Seek for GrowingFileReader {
-    fn seek(&mut self, _: SeekFrom) -> io::Result<u64> {
+    fn seek(&mut self, position: SeekFrom) -> io::Result<u64> {
-        Err(io::Error::new(
+        // Random access beyond the downloaded prefix is intentionally not
-            io::ErrorKind::Unsupported,
+        // exposed, but decoders may inspect and rewind the available header.
-            "streaming playback is not seekable yet",
+        let available = self
-        ))
+            .shared
+            .state
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .available;
+        let target = match position {
+            SeekFrom::Start(position) => i128::from(position),
+            SeekFrom::Current(offset) => i128::from(self.pos) + i128::from(offset),
+            SeekFrom::End(offset) => i128::from(available) + i128::from(offset),
+        };
+        if target < 0 || target > i128::from(available) {
+            return Err(io::Error::new(
+                io::ErrorKind::Unsupported,
+                "cannot seek outside the downloaded audio prefix",
+            ));
+        }
+        let target = target as u64;
+        self.file.seek(SeekFrom::Start(target))?;
+        self.pos = target;
+        Ok(target)
     }
 }
 
@@ -129,6 +148,12 @@ mod tests {
         reader.read_exact(&mut first).unwrap();
         assert_eq!(&first, b"fur");
 
+        reader.seek(SeekFrom::Start(0)).unwrap();
+        let mut rewind = [0u8; 3];
+        reader.read_exact(&mut rewind).unwrap();
+        assert_eq!(&rewind, b"fur");
+        assert!(reader.seek(SeekFrom::Start(4)).is_err());
+
         output.write_all(b"umi").unwrap();
         writer.add_available(3);
         writer.finish();

src/ui/federation.rs

@@ -961,7 +961,10 @@ fn node_summary_lines(state: &AppState) -> Vec<Line<'static>> {
             ]
         }
         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<Lin
                     status.known_contacts
                 ),
             ));
+            lines.push(status_line(
+                "Discovery",
+                format!(
+                    "{} · {} DHT / {} dial failures · {} rebuilds · {} full recoveries",
+                    status.network_health,
+                    status.rendezvous_failures,
+                    status.peer_dial_failures,
+                    status.rendezvous_restarts,
+                    status.recovery_count
+                ),
+            ));
             if !status.connected_peers.is_empty() {
                 let mut peers: Vec<String> = status
                     .connected_peers
@@ -1260,6 +1274,9 @@ fn status_detail_status_lines(state: &AppState, status_cursor: usize) -> Vec<Lin
             if let Some(error) = &status.last_error {
                 lines.push(status_line("Error", first_line(error)));
             }
+            if let Some(error) = &status.last_rendezvous_error {
+                lines.push(status_line("Discovery error", first_line(error)));
+            }
             push_transport_summary_status(&mut lines, state, status);
         }
     }