audio: http streaming, opus/mkv support and stream recovery

- read http(s) urls through a MediaSourceStream instead of failing
- register the libopus decoder and enable mkv/aac/isomp4 in symphonia
- resample to the device rate with linear interpolation
- rebuild the output stream when the backend dies instead of leaving
  playback silent while the daemon still reports "Playing"

A dead pcm makes cpal call the error callback hundreds of thousands of
times per second, so errors are only counted there and reported once when
the stream is dropped. The first failure of a healthy stream is retried
immediately, consecutive failures back off, and a stream that played long
enough clears the failure budget.

3 files changed, 761 insertions, 243 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 9c69f54..7a4524a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -22,6 +22,7 @@ dependencies = [
  "serde",
  "serde_json",
  "symphonia",
+ "symphonia-adapter-libopus",
  "toml",
  "zbus",
 ]
@@ -1654,6 +1655,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
 
 [[package]]
+name = "opusic-sys"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c9d1ecdf206421bc74343ab3bb2f30ad2abbfee41fa341f7181fecbaf957769a"
+dependencies = [
+ "cmake",
+]
+
+[[package]]
 name = "ordered-stream"
 version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2325,14 +2335,28 @@ dependencies = [
  "lazy_static",
  "symphonia-bundle-flac",
  "symphonia-bundle-mp3",
+ "symphonia-codec-aac",
  "symphonia-codec-vorbis",
  "symphonia-core",
+ "symphonia-format-isomp4",
+ "symphonia-format-mkv",
  "symphonia-format-ogg",
  "symphonia-format-riff",
  "symphonia-metadata",
 ]
 
 [[package]]
+name = "symphonia-adapter-libopus"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6febe6f88f9a9483db7e5b72a2dad916d8f8eb588d18905a39c861319fc7fa1"
+dependencies = [
+ "log",
+ "opusic-sys",
+ "symphonia-core",
+]
+
+[[package]]
 name = "symphonia-bundle-flac"
 version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2356,6 +2380,18 @@ dependencies = [
 ]
 
 [[package]]
+name = "symphonia-codec-aac"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1979c515a76371b186aad2feff5f23e21cbec775bf95de08bf1e3af92a2ad76"
+dependencies = [
+ "lazy_static",
+ "log",
+ "symphonia-common",
+ "symphonia-core",
+]
+
+[[package]]
 name = "symphonia-codec-vorbis"
 version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2392,6 +2428,30 @@ dependencies = [
 ]
 
 [[package]]
+name = "symphonia-format-isomp4"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d179a01305b3505940135a9f0180d6ef4b487912748fe97554756f120fbd05e"
+dependencies = [
+ "log",
+ "symphonia-common",
+ "symphonia-core",
+ "symphonia-metadata",
+]
+
+[[package]]
+name = "symphonia-format-mkv"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb17713e134f5ad316c2690fa3104590ccc85842cdbcf82c3cd1a845cb08aa74"
+dependencies = [
+ "lazy_static",
+ "log",
+ "symphonia-common",
+ "symphonia-core",
+]
+
+[[package]]
 name = "symphonia-format-ogg"
 version = "0.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index e3567d7..51eebbb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -24,7 +24,8 @@ directories = "6.0"
 serde = { version = "1.0.228", features = ["derive"] }
 toml = "1.1.2"
 reqwest = { version = "0.13.3", features = ["json", "form", "blocking"] }
-symphonia = { version = "0.6", default-features = false, features = ["mp3", "flac", "vorbis", "wav", "ogg",  "id3v1", "id3v2"] }
+symphonia = { version = "0.6", default-features = false, features = ["mp3", "flac", "vorbis", "wav", "ogg", "mkv", "aac", "isomp4", "id3v1", "id3v2"] }
+symphonia-adapter-libopus = "0.3"
 cpal = "0.17.3"
 serde_json = "1.0.149"
 dirs = "6.0"
diff --git a/src/audio.rs b/src/audio.rs
index f8eba04..8bac26f 100644
--- a/src/audio.rs
+++ b/src/audio.rs
@@ -8,10 +8,20 @@
 use crate::notification;
 use std::io::{Read, Seek, SeekFrom};
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::mpsc::{self, Receiver, Sender};
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, LazyLock, Mutex};
 use std::thread;
+use std::time::{Duration, Instant};
+use symphonia::core::codecs::registry::CodecRegistry;
+use symphonia_adapter_libopus::OpusDecoder;
+
+static CODEC_REGISTRY: LazyLock<CodecRegistry> = LazyLock::new(|| {
+    let mut registry = CodecRegistry::new();
+    symphonia::default::register_enabled_codecs(&mut registry);
+    registry.register_audio_decoder::<OpusDecoder>();
+    registry
+});
 
 struct HttpMediaSource {
     reader: Mutex<Box<dyn Read + Send>>,
@@ -249,6 +259,353 @@ fn resample_linear(input: &[f32], ratio: f64, channels: usize) -> Vec<f32> {
     output
 }
 
+const WATCHDOG_INTERVAL: Duration = Duration::from_millis(250);
+/// A playing stream must fill a buffer every few milliseconds, so a gap this
+/// long means the backend stopped driving the callback entirely.
+const STALL_TIMEOUT: Duration = Duration::from_secs(2);
+/// A stream that stayed up at least this long is considered healthy: its
+/// failure does not count against the retry budget.
+const HEALTHY_STREAM: Duration = Duration::from_secs(15);
+const MAX_RECOVERIES: usize = 3;
+const RECOVERY_WINDOW: Duration = Duration::from_secs(60);
+/// Delay before rebuilding after a stream died almost immediately. Doubled per
+/// consecutive failure, so a device that is really gone is not hammered.
+const RETRY_BACKOFF: Duration = Duration::from_millis(250);
+const MAX_BACKOFF: Duration = Duration::from_secs(2);
+
+struct StreamContext {
+    audio_data: Arc<Mutex<Option<AudioData>>>,
+    volume: Arc<Mutex<f32>>,
+    track_ended: Arc<AtomicBool>,
+    is_playing: Arc<AtomicBool>,
+    position: Arc<Mutex<f64>>,
+    failed: Arc<AtomicBool>,
+    callbacks: Arc<AtomicU64>,
+    /// Error callbacks seen on the current stream. A dead PCM fires the error
+    /// callback hundreds of thousands of times per second, so it is only
+    /// counted; the tally is reported once, when the stream is torn down.
+    stream_errors: Arc<AtomicU64>,
+    /// Message of the first error on the current stream, formatted once.
+    first_error: Arc<Mutex<Option<String>>>,
+}
+
+fn make_data_callback(
+    ctx: &StreamContext,
+    device_sample_rate: u32,
+) -> impl FnMut(&mut [f32], &cpal::OutputCallbackInfo) + Send + 'static {
+    let audio_data = Arc::clone(&ctx.audio_data);
+    let volume = Arc::clone(&ctx.volume);
+    let track_ended = Arc::clone(&ctx.track_ended);
+    let is_playing = Arc::clone(&ctx.is_playing);
+    let position = Arc::clone(&ctx.position);
+    let callbacks = Arc::clone(&ctx.callbacks);
+
+    move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
+        // Liveness marker for the watchdog: a relaxed increment is enough and
+        // avoids a clock read in the realtime callback.
+        callbacks.fetch_add(1, Ordering::Relaxed);
+
+        let vol = volume.lock().map(|v| *v).unwrap_or(1.0);
+
+        if let Ok(mut audio_data) = audio_data.lock() {
+            let mut should_clear = false;
+
+            if let Some(ref mut state) = *audio_data {
+                let channels = state.channels as usize;
+                let mut offset = 0usize;
+
+                // Keep decoding until we fill the buffer or run out of data
+                while offset < data.len() {
+                    // First, copy any buffered samples
+                    if state.buffer_pos < state.sample_buffer.len() {
+                        let remaining = &state.sample_buffer[state.buffer_pos..];
+                        let to_copy = remaining.len().min(data.len() - offset);
+                        data[offset..offset + to_copy].copy_from_slice(&remaining[..to_copy]);
+                        state.buffer_pos += to_copy;
+                        offset += to_copy;
+
+                        // If we've consumed all buffered samples, clear the buffer
+                        if state.buffer_pos >= state.sample_buffer.len() {
+                            state.sample_buffer.clear();
+                            state.buffer_pos = 0;
+                        }
+                        continue; // Try to fill more of the buffer
+                    }
+
+                    // Decode more data
+                    match state.format.next_packet() {
+                        Ok(Some(packet)) => {
+                            match state.decoder.decode(&packet) {
+                                Ok(decoded) => {
+                                    let num_samples = decoded.samples_interleaved();
+                                    let mut samples = vec![0.0f32; num_samples];
+                                    decoded.copy_to_slice_interleaved(&mut samples);
+
+                                    // Resample if needed
+                                    let resampled = if let Some(ratio) = state.resample_ratio {
+                                        resample_linear(&samples, ratio, channels)
+                                    } else {
+                                        samples
+                                    };
+
+                                    // Copy what we can to output
+                                    let to_copy = resampled.len().min(data.len() - offset);
+                                    data[offset..offset + to_copy]
+                                        .copy_from_slice(&resampled[..to_copy]);
+                                    offset += to_copy;
+
+                                    // Buffer the rest
+                                    if to_copy < resampled.len() {
+                                        state.sample_buffer.clear();
+                                        state
+                                            .sample_buffer
+                                            .extend_from_slice(&resampled[to_copy..]);
+                                        state.buffer_pos = 0;
+                                    }
+                                }
+                                Err(_) => break,
+                            }
+                        }
+                        _ => break,
+                    }
+                }
+
+                // Apply volume
+                for sample in data.iter_mut().take(offset) {
+                    *sample *= vol;
+                }
+
+                // Fill remaining with silence
+                for sample in data.iter_mut().skip(offset) {
+                    *sample = 0.0;
+                }
+
+                // Track playback position
+                if offset > 0
+                    && let Ok(mut pos) = position.lock()
+                {
+                    *pos += offset as f64 / (device_sample_rate as f64 * channels as f64);
+                }
+
+                // If no more data and offset is 0, track ended
+                if offset == 0 && state.sample_buffer.is_empty() {
+                    track_ended.store(true, Ordering::SeqCst);
+                    is_playing.store(false, Ordering::SeqCst);
+                    should_clear = true;
+                }
+            } else {
+                for sample in data.iter_mut() {
+                    *sample = 0.0;
+                }
+            }
+
+            if should_clear {
+                *audio_data = None;
+            }
+        }
+    }
+}
+
+fn build_output_stream(
+    device: &cpal::Device,
+    config: cpal::SupportedStreamConfig,
+    ctx: &StreamContext,
+) -> Result<cpal::Stream, String> {
+    use cpal::traits::{DeviceTrait, StreamTrait};
+
+    let device_sample_rate = config.sample_rate();
+    let stream_config: cpal::StreamConfig = config.into();
+    let failed = Arc::clone(&ctx.failed);
+    let errors = Arc::clone(&ctx.stream_errors);
+    let first_error = Arc::clone(&ctx.first_error);
+
+    // Every stream gets a fresh tally, so a report only covers this one.
+    errors.store(0, Ordering::Relaxed);
+    if let Ok(mut slot) = first_error.lock() {
+        *slot = None;
+    }
+    let error_seen = AtomicBool::new(false);
+
+    let stream = device
+        .build_output_stream(
+            &stream_config,
+            make_data_callback(ctx, device_sample_rate),
+            move |err: cpal::StreamError| {
+                // Buffer underruns are reported but already recovered by the
+                // backend, so they must not tear down a working stream.
+                if matches!(err, cpal::StreamError::BufferUnderrun) {
+                    if errors.fetch_add(1, Ordering::Relaxed) == 0 {
+                        eprintln!("Audio buffer underrun");
+                    }
+                    return;
+                }
+                failed.store(true, Ordering::SeqCst);
+                // A broken PCM spins on `poll()` and reports the same error
+                // again immediately: keep this path free of allocation and
+                // logging, and format the message only for the first one.
+                errors.fetch_add(1, Ordering::Relaxed);
+                if !error_seen.swap(true, Ordering::Relaxed)
+                    && let Ok(mut slot) = first_error.lock()
+                {
+                    *slot = Some(err.to_string());
+                }
+            },
+            None,
+        )
+        .map_err(|e| e.to_string())?;
+
+    stream.play().map_err(|e| e.to_string())?;
+
+    Ok(stream)
+}
+
+/// Reports the error that tore the current stream down.
+///
+/// The error callback only counts, because a dead backend calls it hundreds of
+/// thousands of times per second; the message and the tally are printed here,
+/// once, from the watchdog thread.
+fn log_stream_error(ctx: &StreamContext) {
+    let count = ctx.stream_errors.swap(0, Ordering::Relaxed);
+    let Ok(mut slot) = ctx.first_error.lock() else {
+        return;
+    };
+    let Some(message) = slot.take() else {
+        return;
+    };
+    if count > 1 {
+        eprintln!("Stream error: {} ({} callbacks)", message, count);
+    } else {
+        eprintln!("Stream error: {}", message);
+    }
+}
+
+/// Stops playback of the current track after the output device stopped working,
+/// so the queue can move on instead of reporting "Playing" in silence.
+fn abandon_playback(ctx: &StreamContext) {
+    if let Ok(mut d) = ctx.audio_data.lock() {
+        *d = None;
+    }
+    ctx.is_playing.store(false, Ordering::SeqCst);
+    ctx.track_ended.store(true, Ordering::SeqCst);
+    notification::notify_error("Audio output unavailable");
+}
+
+/// Stops trying once the retry budget is spent.
+fn give_up(recovery: &mut RecoveryState, ctx: &StreamContext) {
+    eprintln!(
+        "Audio stream failed {} times within {}s, giving up",
+        MAX_RECOVERIES,
+        RECOVERY_WINDOW.as_secs()
+    );
+    recovery.retry_at = None;
+    abandon_playback(ctx);
+}
+
+/// Decides how the audio thread recovers from a stream the backend broke.
+///
+/// cpal's ALSA backend cannot revive a dead PCM: `poll()` keeps returning
+/// `POLLERR`, the data callback is never called again, and the worker thread
+/// spins at full speed. The only way out is to drop the stream and build a new
+/// one; the decoded audio lives in [`StreamContext::audio_data`], so playback
+/// continues from where it stopped.
+///
+/// Rebuilding is cheap, but a device that is really gone would be hammered in a
+/// loop, so failures are budgeted:
+///
+/// * the first failure of a healthy stream is retried immediately,
+/// * consecutive failures back off further each time,
+/// * a stream that played for [`HEALTHY_STREAM`] clears the budget, and
+/// * [`MAX_RECOVERIES`] failures inside [`RECOVERY_WINDOW`] give up.
+#[derive(Default)]
+struct RecoveryState {
+    /// When the current stream was built.
+    stream_started: Option<Instant>,
+    /// Failures inside [`RECOVERY_WINDOW`].
+    failures: Vec<Instant>,
+    /// Failures since the last stream that played long enough to count.
+    consecutive: usize,
+    /// Earliest instant at which the next rebuild may start.
+    retry_at: Option<Instant>,
+}
+
+impl RecoveryState {
+    /// Forgets the whole budget, for a new track or an explicit stop.
+    fn clear(&mut self) {
+        *self = Self::default();
+    }
+
+    fn stream_built(&mut self, now: Instant) {
+        self.stream_started = Some(now);
+    }
+
+    /// Records a failure and returns how long to wait before rebuilding, or
+    /// `None` when the backend has failed too often to keep trying.
+    fn failed(&mut self, now: Instant) -> Option<Duration> {
+        let uptime = self
+            .stream_started
+            .map_or(Duration::ZERO, |t| now.saturating_duration_since(t));
+        self.failures
+            .retain(|t| now.saturating_duration_since(*t) < RECOVERY_WINDOW);
+        if uptime >= HEALTHY_STREAM {
+            self.failures.clear();
+            self.consecutive = 0;
+        }
+        if self.failures.len() >= MAX_RECOVERIES {
+            return None;
+        }
+        self.failures.push(now);
+        // The first retry is immediate: the outage is already audible, and a
+        // healthy device comes back without a gap. Everything after that backs
+        // off, because retrying straight away clearly did not help.
+        let delay = if self.consecutive == 0 {
+            Duration::ZERO
+        } else {
+            let shift = (self.consecutive as u32 - 1).min(4);
+            (RETRY_BACKOFF * 2u32.pow(shift)).min(MAX_BACKOFF)
+        };
+        self.consecutive += 1;
+        self.retry_at = Some(now + delay);
+        Some(delay)
+    }
+}
+
+/// Rebuilds the output stream after the backend broke it.
+///
+/// The default output can change while the daemon runs (a re-plugged DAC, a
+/// sink switch, a rate change), which leaves the cached device handle pointing
+/// at a PCM that no longer exists, so it is looked up again.
+fn rebuild_stream(
+    host: &cpal::Host,
+    device: &mut cpal::Device,
+    ctx: &StreamContext,
+) -> Result<cpal::Stream, String> {
+    use cpal::traits::{DeviceTrait, HostTrait};
+
+    if let Some(fresh) = host.default_output_device() {
+        *device = fresh;
+    }
+
+    let config = device
+        .default_output_config()
+        .map_err(|e| format!("Failed to get output config while rebuilding: {}", e))?;
+    let device_sample_rate = config.sample_rate();
+
+    // The device may report a different rate now than when the track was
+    // opened, so re-target the resampler before resuming.
+    if let Ok(mut d) = ctx.audio_data.lock()
+        && let Some(state) = d.as_mut()
+        && state.sample_rate != device_sample_rate
+    {
+        crate::debug_eprintln(format_args!(
+            "Retargeting resampler from {} Hz to {} Hz after recovery",
+            state.sample_rate, device_sample_rate
+        ));
+        state.resample_ratio = Some(device_sample_rate as f64 / state.sample_rate as f64);
+    }
+
+    build_output_stream(device, config, ctx)
+}
+
 fn audio_thread(
     rx: Receiver<AudioCommand>,
     is_playing: Arc<AtomicBool>,
@@ -260,7 +617,7 @@ fn audio_thread(
     use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 
     let host = cpal::default_host();
-    let device = match host.default_output_device() {
+    let mut device = match host.default_output_device() {
         Some(d) => d,
         None => {
             eprintln!("No output device available");
@@ -268,270 +625,240 @@ fn audio_thread(
         }
     };
 
-    let config = match device.default_output_config() {
-        Ok(c) => c,
-        Err(e) => {
-            eprintln!("Failed to get output config: {}", e);
-            return;
-        }
-    };
-
-    let device_sample_rate = config.sample_rate();
-    let channels = config.channels();
-    crate::debug_eprintln(format_args!(
-        "Default output: {} Hz, {} channels",
-        device_sample_rate, channels
-    ));
-
     let mut current_stream: Option<cpal::Stream> = None;
-    let audio_data: Arc<Mutex<Option<AudioData>>> = Arc::new(Mutex::new(None));
-
-    while let Ok(cmd) = rx.recv() {
-        match cmd {
-            AudioCommand::Play(path) => {
-                is_playing.store(true, Ordering::SeqCst);
-                if let Ok(mut track) = current_track.lock() {
-                    *track = Some(path.clone());
-                }
-                if let Ok(mut pos) = position.lock() {
-                    *pos = 0.0;
-                }
-
-                // Drop the current stream before creating a new one
-                current_stream = None;
+    let ctx = StreamContext {
+        audio_data: Arc::new(Mutex::new(None)),
+        volume: Arc::clone(&volume),
+        track_ended: Arc::clone(&track_ended),
+        is_playing: Arc::clone(&is_playing),
+        position: Arc::clone(&position),
+        failed: Arc::new(AtomicBool::new(false)),
+        callbacks: Arc::new(AtomicU64::new(0)),
+        stream_errors: Arc::new(AtomicU64::new(0)),
+        first_error: Arc::new(Mutex::new(None)),
+    };
 
-                match open_audio_file(&path) {
-                    Ok(mut data) => {
-                        crate::debug_eprintln(format_args!(
-                            "File: {} Hz, {} channels",
-                            data.sample_rate, data.channels
-                        ));
+    let mut recovery = RecoveryState::default();
+    let mut last_callbacks = 0u64;
+    let mut last_progress = Instant::now();
+
+    loop {
+        // Wake up early when a rebuild is due, so the backoff does not add
+        // jitter on top of the watchdog tick.
+        let wait = if current_stream.is_none() && is_playing.load(Ordering::SeqCst) {
+            match recovery.retry_at {
+                Some(at) => WATCHDOG_INTERVAL.min(at.saturating_duration_since(Instant::now())),
+                None => WATCHDOG_INTERVAL,
+            }
+        } else {
+            WATCHDOG_INTERVAL
+        };
+
+        let cmd = match rx.recv_timeout(wait) {
+            Ok(cmd) => Some(cmd),
+            Err(mpsc::RecvTimeoutError::Timeout) => None,
+            Err(mpsc::RecvTimeoutError::Disconnected) => break,
+        };
+
+        if let Some(cmd) = cmd {
+            match cmd {
+                AudioCommand::Play(path) => {
+                    is_playing.store(true, Ordering::SeqCst);
+                    if let Ok(mut track) = current_track.lock() {
+                        *track = Some(path.clone());
+                    }
+                    if let Ok(mut pos) = position.lock() {
+                        *pos = 0.0;
+                    }
 
-                        if data.sample_rate != device_sample_rate {
+                    // Drop the current stream before creating a new one
+                    current_stream = None;
+                    recovery.clear();
+                    ctx.failed.store(false, Ordering::SeqCst);
+
+                    // Query the config every time instead of caching it at
+                    // startup: the default device or its rate can change while
+                    // the daemon runs.
+                    let config = match device.default_output_config() {
+                        Ok(c) => c,
+                        Err(e) => {
+                            eprintln!("Failed to get output config: {}", e);
+                            notification::notify_error("Failed to query audio output");
+                            is_playing.store(false, Ordering::SeqCst);
+                            continue;
+                        }
+                    };
+                    let device_sample_rate = config.sample_rate();
+                    crate::debug_eprintln(format_args!(
+                        "Default output: {} Hz, {} channels",
+                        device_sample_rate,
+                        config.channels()
+                    ));
+
+                    match open_audio_file(&path) {
+                        Ok(mut data) => {
                             crate::debug_eprintln(format_args!(
-                                "Resampling from {} Hz to {} Hz",
-                                data.sample_rate, device_sample_rate
+                                "File: {} Hz, {} channels",
+                                data.sample_rate, data.channels
                             ));
-                            data.resample_ratio =
-                                Some(device_sample_rate as f64 / data.sample_rate as f64);
-                        }
 
-                        // Pre-decode some samples before starting playback to prevent buffer underrun
-                        let channels = data.channels as usize;
-                        for _ in 0..100 {
-                            match data.format.next_packet() {
-                                Ok(Some(packet)) => match data.decoder.decode(&packet) {
-                                    Ok(decoded) => {
-                                        let num_samples = decoded.samples_interleaved();
-                                        let mut samples = vec![0.0f32; num_samples];
-                                        decoded.copy_to_slice_interleaved(&mut samples);
-
-                                        let resampled = if let Some(ratio) = data.resample_ratio {
-                                            resample_linear(&samples, ratio, channels)
-                                        } else {
-                                            samples
-                                        };
-
-                                        data.sample_buffer.extend(resampled);
-                                    }
-                                    Err(_) => break,
-                                },
-                                _ => break,
+                            if data.sample_rate != device_sample_rate {
+                                crate::debug_eprintln(format_args!(
+                                    "Resampling from {} Hz to {} Hz",
+                                    data.sample_rate, device_sample_rate
+                                ));
+                                data.resample_ratio =
+                                    Some(device_sample_rate as f64 / data.sample_rate as f64);
                             }
-                        }
-                        data.buffer_pos = 0;
-
-                        let stream_config: cpal::StreamConfig =
-                            device.default_output_config().unwrap().into();
-
-                        let audio_data_clone = Arc::clone(&audio_data);
-                        let vol = Arc::clone(&volume);
-                        let ended = Arc::clone(&track_ended);
-                        let is_playing_clone = Arc::clone(&is_playing);
-                        let position_clone = Arc::clone(&position);
-
-                        if let Ok(mut d) = audio_data.lock() {
-                            *d = Some(data);
-                        }
-
-                        match device.build_output_stream(
-                            &stream_config,
-                            move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
-                                let vol = vol.lock().map(|v| *v).unwrap_or(1.0);
-
-                                if let Ok(mut audio_data) = audio_data_clone.lock() {
-                                    let mut should_clear = false;
-
-                                    if let Some(ref mut state) = *audio_data {
-                                        let channels = state.channels as usize;
-                                        let mut offset = 0usize;
-
-                                        // Keep decoding until we fill the buffer or run out of data
-                                        while offset < data.len() {
-                                            // First, copy any buffered samples
-                                            if state.buffer_pos < state.sample_buffer.len() {
-                                                let remaining =
-                                                    &state.sample_buffer[state.buffer_pos..];
-                                                let to_copy =
-                                                    remaining.len().min(data.len() - offset);
-                                                data[offset..offset + to_copy]
-                                                    .copy_from_slice(&remaining[..to_copy]);
-                                                state.buffer_pos += to_copy;
-                                                offset += to_copy;
-
-                                                // If we've consumed all buffered samples, clear the buffer
-                                                if state.buffer_pos >= state.sample_buffer.len() {
-                                                    state.sample_buffer.clear();
-                                                    state.buffer_pos = 0;
-                                                }
-                                                continue; // Try to fill more of the buffer
-                                            }
-
-                                            // Decode more data
-                                            match state.format.next_packet() {
-                                                Ok(Some(packet)) => {
-                                                    match state.decoder.decode(&packet) {
-                                                        Ok(decoded) => {
-                                                            let num_samples =
-                                                                decoded.samples_interleaved();
-                                                            let mut samples =
-                                                                vec![0.0f32; num_samples];
-                                                            decoded.copy_to_slice_interleaved(
-                                                                &mut samples,
-                                                            );
-
-                                                            // Resample if needed
-                                                            let resampled = if let Some(ratio) =
-                                                                state.resample_ratio
-                                                            {
-                                                                resample_linear(
-                                                                    &samples, ratio, channels,
-                                                                )
-                                                            } else {
-                                                                samples
-                                                            };
-
-                                                            // Copy what we can to output
-                                                            let to_copy = resampled
-                                                                .len()
-                                                                .min(data.len() - offset);
-                                                            data[offset..offset + to_copy]
-                                                                .copy_from_slice(
-                                                                    &resampled[..to_copy],
-                                                                );
-                                                            offset += to_copy;
-
-                                                            // Buffer the rest
-                                                            if to_copy < resampled.len() {
-                                                                state.sample_buffer.clear();
-                                                                state
-                                                                    .sample_buffer
-                                                                    .extend_from_slice(
-                                                                        &resampled[to_copy..],
-                                                                    );
-                                                                state.buffer_pos = 0;
-                                                            }
-                                                        }
-                                                        Err(_) => break,
-                                                    }
-                                                }
-                                                _ => break,
-                                            }
-                                        }
 
-                                        // Apply volume
-                                        for sample in data.iter_mut().take(offset) {
-                                            *sample *= vol;
-                                        }
-
-                                        // Fill remaining with silence
-                                        for sample in data.iter_mut().skip(offset) {
-                                            *sample = 0.0;
-                                        }
-
-                                        // Track playback position
-                                        if offset > 0
-                                            && let Ok(mut pos) = position_clone.lock()
-                                        {
-                                            *pos += offset as f64
-                                                / (device_sample_rate as f64 * channels as f64);
+                            // Pre-decode some samples before starting playback to prevent buffer underrun
+                            let channels = data.channels as usize;
+                            for _ in 0..100 {
+                                match data.format.next_packet() {
+                                    Ok(Some(packet)) => match data.decoder.decode(&packet) {
+                                        Ok(decoded) => {
+                                            let num_samples = decoded.samples_interleaved();
+                                            let mut samples = vec![0.0f32; num_samples];
+                                            decoded.copy_to_slice_interleaved(&mut samples);
+
+                                            let resampled = if let Some(ratio) = data.resample_ratio
+                                            {
+                                                resample_linear(&samples, ratio, channels)
+                                            } else {
+                                                samples
+                                            };
+
+                                            data.sample_buffer.extend(resampled);
                                         }
+                                        Err(_) => break,
+                                    },
+                                    _ => break,
+                                }
+                            }
+                            data.buffer_pos = 0;
 
-                                        // If no more data and offset is 0, track ended
-                                        if offset == 0 && state.sample_buffer.is_empty() {
-                                            ended.store(true, Ordering::SeqCst);
-                                            is_playing_clone.store(false, Ordering::SeqCst);
-                                            should_clear = true;
-                                        }
-                                    } else {
-                                        for sample in data.iter_mut() {
-                                            *sample = 0.0;
-                                        }
-                                    }
+                            if let Ok(mut d) = ctx.audio_data.lock() {
+                                *d = Some(data);
+                            }
 
-                                    if should_clear {
-                                        *audio_data = None;
-                                    }
+                            match build_output_stream(&device, config, &ctx) {
+                                Ok(stream) => {
+                                    current_stream = Some(stream);
+                                    recovery.stream_built(Instant::now());
                                 }
-                            },
-                            |err| {
-                                eprintln!("Stream error: {}", err);
-                            },
-                            None,
-                        ) {
-                            Ok(stream) => {
-                                if let Err(e) = stream.play() {
-                                    eprintln!("Failed to start stream: {}", e);
-                                    notification::notify_error("Failed to start audio stream");
+                                Err(e) => {
+                                    eprintln!("Failed to create stream: {}", e);
+                                    notification::notify_error("Failed to create audio stream");
                                     is_playing.store(false, Ordering::SeqCst);
-                                } else {
-                                    current_stream = Some(stream);
                                 }
                             }
-                            Err(e) => {
-                                eprintln!("Failed to create stream: {}", e);
-                                notification::notify_error("Failed to create audio stream");
-                                is_playing.store(false, Ordering::SeqCst);
-                            }
+                        }
+                        Err(e) => {
+                            eprintln!("Failed to open audio: {}", e);
+                            notification::notify_error("Failed to open audio file");
+                            is_playing.store(false, Ordering::SeqCst);
                         }
                     }
-                    Err(e) => {
-                        eprintln!("Failed to open audio: {}", e);
-                        notification::notify_error("Failed to open audio file");
+
+                    // Opening a file (or an http stream) can take a while, so
+                    // restart the stall window once the stream is up.
+                    last_progress = Instant::now();
+                }
+                AudioCommand::Pause => {
+                    if let Some(ref stream) = current_stream {
+                        if let Err(e) = stream.pause() {
+                            eprintln!("Failed to pause stream: {}", e);
+                        }
                         is_playing.store(false, Ordering::SeqCst);
                     }
                 }
-            }
-            AudioCommand::Pause => {
-                if let Some(ref stream) = current_stream {
-                    if let Err(e) = stream.pause() {
-                        eprintln!("Failed to pause stream: {}", e);
+                AudioCommand::Resume => {
+                    if let Some(ref stream) = current_stream {
+                        if let Err(e) = stream.play() {
+                            eprintln!("Failed to resume stream: {}", e);
+                        } else {
+                            is_playing.store(true, Ordering::SeqCst);
+                            last_progress = Instant::now();
+                        }
+                    } else if ctx.audio_data.lock().is_ok_and(|d| d.is_some()) {
+                        // The stream was dropped while paused because the
+                        // backend broke it; let the recovery path rebuild it
+                        // instead of resuming into a dead stream.
+                        is_playing.store(true, Ordering::SeqCst);
+                        recovery.retry_at = Some(Instant::now());
+                    }
+                }
+                AudioCommand::Stop => {
+                    current_stream = None;
+                    recovery.clear();
+                    if let Ok(mut d) = ctx.audio_data.lock() {
+                        *d = None;
                     }
                     is_playing.store(false, Ordering::SeqCst);
+                    if let Ok(mut track) = current_track.lock() {
+                        *track = None;
+                    }
                 }
-            }
-            AudioCommand::Resume => {
-                if let Some(ref stream) = current_stream {
-                    if let Err(e) = stream.play() {
-                        eprintln!("Failed to resume stream: {}", e);
-                    } else {
-                        is_playing.store(true, Ordering::SeqCst);
+                AudioCommand::SetVolume(vol) => {
+                    if let Ok(mut volume_ref) = volume.lock() {
+                        *volume_ref = vol;
                     }
                 }
             }
-            AudioCommand::Stop => {
-                current_stream = None;
-                if let Ok(mut d) = audio_data.lock() {
-                    *d = None;
-                }
-                is_playing.store(false, Ordering::SeqCst);
-                if let Ok(mut track) = current_track.lock() {
-                    *track = None;
-                }
+        }
+
+        // Watchdog: the backend never brings a dead stream back on its own.
+        // It either reports a fatal error and then stops feeding the callback
+        // (`POLLERR` and friends), or - when the audio server is restarted or
+        // changed underneath it - silently stops scheduling the callback.
+        // Both look the same from the outside: playback is dead while the
+        // daemon still reports "Playing", so react to either one at once.
+        let failed = ctx.failed.swap(false, Ordering::SeqCst);
+        let callbacks = ctx.callbacks.load(Ordering::Relaxed);
+        let advanced = callbacks != last_callbacks;
+        if advanced {
+            last_callbacks = callbacks;
+            last_progress = Instant::now();
+        }
+        let stalled = current_stream.is_some()
+            && !advanced
+            && is_playing.load(Ordering::SeqCst)
+            && last_progress.elapsed() > STALL_TIMEOUT;
+
+        if current_stream.is_some() && (failed || stalled) {
+            if stalled {
+                eprintln!("Audio stream stalled, rebuilding it");
+            }
+            // Dropping the stream is what stops the backend spinning on the
+            // dead PCM, so do that before deciding whether to retry.
+            current_stream = None;
+            ctx.failed.store(false, Ordering::SeqCst);
+            last_progress = Instant::now();
+            log_stream_error(&ctx);
+
+            if recovery.failed(Instant::now()).is_none() {
+                give_up(&mut recovery, &ctx);
             }
-            AudioCommand::SetVolume(vol) => {
-                if let Ok(mut volume_ref) = volume.lock() {
-                    *volume_ref = vol;
+        }
+
+        // Rebuild once the backoff has elapsed. Playback has to be wanted and
+        // the stream gone, otherwise a paused daemon would start audio.
+        let retry_due = current_stream.is_none()
+            && is_playing.load(Ordering::SeqCst)
+            && recovery.retry_at.is_some_and(|at| at <= Instant::now());
+        if retry_due {
+            recovery.retry_at = None;
+            match rebuild_stream(&host, &mut device, &ctx) {
+                Ok(stream) => {
+                    current_stream = Some(stream);
+                    recovery.stream_built(Instant::now());
+                    last_progress = Instant::now();
+                }
+                Err(e) => {
+                    eprintln!("Failed to rebuild audio stream: {}", e);
+                    if recovery.failed(Instant::now()).is_none() {
+                        give_up(&mut recovery, &ctx);
+                    }
                 }
             }
         }
@@ -609,8 +936,8 @@ fn open_audio_file(path: &PathBuf) -> Result<AudioData, Box<dyn std::error::Erro
         .unwrap_or(CHANNEL_LAYOUT_STEREO)
         .count() as u16;
 
-    let decoder = symphonia::default::get_codecs()
-        .make_audio_decoder(codec_params, &AudioDecoderOptions::default())?;
+    let decoder =
+        CODEC_REGISTRY.make_audio_decoder(codec_params, &AudioDecoderOptions::default())?;
 
     Ok(AudioData {
         format,
@@ -622,3 +949,133 @@ fn open_audio_file(path: &PathBuf) -> Result<AudioData, Box<dyn std::error::Erro
         resample_ratio: None,
     })
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_context() -> StreamContext {
+        StreamContext {
+            audio_data: Arc::new(Mutex::new(None)),
+            volume: Arc::new(Mutex::new(1.0)),
+            track_ended: Arc::new(AtomicBool::new(false)),
+            is_playing: Arc::new(AtomicBool::new(true)),
+            position: Arc::new(Mutex::new(0.0)),
+            failed: Arc::new(AtomicBool::new(false)),
+            callbacks: Arc::new(AtomicU64::new(0)),
+            stream_errors: Arc::new(AtomicU64::new(0)),
+            first_error: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    #[test]
+    fn first_failure_of_a_healthy_stream_retries_immediately() {
+        let mut state = RecoveryState::default();
+        let start = Instant::now();
+        state.stream_built(start);
+
+        let delay = state.failed(start + Duration::from_secs(30));
+
+        assert_eq!(delay, Some(Duration::ZERO), "retry without adding silence");
+        assert_eq!(state.retry_at, Some(start + Duration::from_secs(30)));
+    }
+
+    #[test]
+    fn fast_failures_back_off_then_give_up() {
+        let mut state = RecoveryState::default();
+        let start = Instant::now();
+        let mut delays = Vec::new();
+
+        for i in 0..4 {
+            let at = start + Duration::from_millis(100 * i);
+            state.stream_built(at);
+            delays.push(state.failed(at + Duration::from_millis(50)));
+        }
+
+        assert_eq!(
+            delays,
+            vec![
+                Some(Duration::ZERO),
+                Some(RETRY_BACKOFF),
+                Some(RETRY_BACKOFF * 2),
+                None,
+            ],
+            "consecutive fast failures back off, then stop trying"
+        );
+    }
+
+    #[test]
+    fn healthy_stream_clears_the_budget() {
+        let mut state = RecoveryState::default();
+        let start = Instant::now();
+
+        for i in 0..3 {
+            state.stream_built(start);
+            assert!(state.failed(start + Duration::from_secs(1 + i)).is_some());
+        }
+
+        // A stream that survived long enough must not inherit the strikes.
+        let healthy = start + Duration::from_secs(10);
+        state.stream_built(healthy);
+        let delay = state.failed(healthy + HEALTHY_STREAM + Duration::from_secs(1));
+
+        assert_eq!(delay, Some(Duration::ZERO));
+        assert_eq!(state.failures.len(), 1);
+        assert_eq!(state.consecutive, 1);
+    }
+
+    #[test]
+    fn stale_failures_do_not_count() {
+        let mut state = RecoveryState::default();
+        let start = Instant::now();
+        state.stream_built(start);
+        state.failures = vec![
+            start - RECOVERY_WINDOW - Duration::from_secs(5),
+            start - RECOVERY_WINDOW - Duration::from_secs(4),
+            start - RECOVERY_WINDOW - Duration::from_secs(3),
+        ];
+
+        assert!(state.failed(start).is_some());
+        assert_eq!(state.failures.len(), 1);
+    }
+
+    #[test]
+    fn gives_up_once_the_budget_is_spent() {
+        let ctx = test_context();
+        ctx.stream_errors.store(42, Ordering::Relaxed);
+        if let Ok(mut slot) = ctx.first_error.lock() {
+            *slot = Some("boom".to_string());
+        }
+
+        abandon_playback(&ctx);
+
+        assert!(!ctx.is_playing.load(Ordering::SeqCst));
+        assert!(ctx.track_ended.load(Ordering::SeqCst));
+        assert!(ctx.audio_data.lock().expect("lock").is_none());
+
+        // The tally is reported exactly once and does not leak into the next
+        // stream.
+        log_stream_error(&ctx);
+        assert_eq!(ctx.stream_errors.load(Ordering::Relaxed), 0);
+        assert!(ctx.first_error.lock().expect("lock").is_none());
+    }
+
+    #[test]
+    fn rebuilds_stream_after_error() {
+        use cpal::traits::{HostTrait, StreamTrait};
+
+        let host = cpal::default_host();
+        let Some(mut device) = host.default_output_device() else {
+            eprintln!("skipped: no output device available");
+            return;
+        };
+
+        let ctx = test_context();
+        let Ok(stream) = rebuild_stream(&host, &mut device, &ctx) else {
+            eprintln!("skipped: cannot build output stream");
+            return;
+        };
+
+        stream.pause().expect("rebuilt stream is usable");
+    }
+}