Audio: decode Ogg-Opus via libopus so opus streams play

rodio decodes through symphonia 0.5, which ships no Opus decoder, so raw
.opus sources (audiobookshelf files, and opus from any provider) failed
Decoder::build(). Add OpusSource, a rodio Source that demuxes Ogg with
symphonia own Ogg reader and decodes with libopus (via
symphonia-adapter-libopus, registered into an explicit codec registry), and
route to it by content-sniffing OggS+OpusHead in the player -- the abs stream
URL has no file extension, so the extension hint is not enough. The bundled
libopus builds with cmake/ninja (added to devenv).

Verified end-to-end with ffmpeg mono/stereo opus fixtures including seeking;
11 audio-player tests pass, clippy/fmt clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -388,6 +388,8 @@ dependencies = [
  "reqwest 0.13.1",
  "rodio",
  "stream-download",
+ "symphonia",
+ "symphonia-adapter-libopus",
  "thiserror 2.0.19",
  "tokio",
  "tracing",
@@ -3215,6 +3217,15 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
 
+[[package]]
+name = "opusic-sys"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2804e694ef0de3b4cbb254de565053b7cb48d3398df7fd60c6c62bed40c5372a"
+dependencies = [
+ "cmake",
+]
+
 [[package]]
 name = "or_poisoned"
 version = "0.1.0"
@@ -4869,6 +4880,17 @@ dependencies = [
  "symphonia-metadata",
 ]
 
+[[package]]
+name = "symphonia-adapter-libopus"
+version = "0.2.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2bfc8e95f95c23ed1b5328eb66920ad28d9968c797f9c7aa755d4b45a5f47a41"
+dependencies = [
+ "log",
+ "opusic-sys",
+ "symphonia-core",
+]
+
 [[package]]
 name = "symphonia-bundle-flac"
 version = "0.5.5"

Cargo.toml

@@ -67,6 +67,11 @@ rustypipe = { version = "0.11", default-features = false, features = [
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_urlencoded = "0.7"
+# Opus decoding: symphonia has no Opus decoder, so we demux Ogg-Opus with
+# symphonia's Ogg reader and decode via the libopus adapter (matches
+# rodio's symphonia 0.5). The adapter bundles libopus (needs cmake).
+symphonia = { version = "0.5", default-features = false, features = ["ogg"] }
+symphonia-adapter-libopus = "0.2"
 stream-download = { version = "0.24", default-features = false, features = [
   "reqwest",
   "reqwest-rustls",

audio-player/Cargo.toml

@@ -11,6 +11,8 @@ futures.workspace = true
 reqwest.workspace = true
 rodio.workspace = true
 stream-download.workspace = true
+symphonia.workspace = true
+symphonia-adapter-libopus.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
 tracing.workspace = true

audio-player/src/lib.rs

@@ -1,3 +1,4 @@
+mod opus_source;
 mod player;
 mod player_engine;
 mod spectrum_tap;

audio-player/src/opus_source.rs

@@ -0,0 +1,342 @@
+//! Ogg-Opus decoding for the rodio pipeline.
+//!
+//! rodio decodes through symphonia, and symphonia 0.5 ships no Opus
+//! decoder — so raw `.opus` streams (notably audiobookshelf files, but any
+//! provider serving Opus) fail rodio's `Decoder::build()`. This module fills
+//! that gap: it demuxes the container with symphonia's own Ogg reader and
+//! decodes the Opus packets with libopus via [`symphonia_adapter_libopus`],
+//! then exposes the result as a rodio [`Source`] so the rest of the player
+//! (sink, spectrum tap, seeking) is unchanged.
+//!
+//! Only Ogg-*Opus* is handled here; every other format (including Ogg-Vorbis
+//! and Ogg-FLAC, which symphonia decodes natively) stays on rodio's decoder.
+//! [`is_ogg_opus`] does the sniffing and the player routes accordingly.
+//!
+//! The decode loop, buffering, and seek refinement mirror rodio 0.22's own
+//! `SymphoniaDecoder`; the only substantive difference is that we build the
+//! format reader and the codec from an explicit registry that includes the
+//! libopus adapter, instead of symphonia's default `get_codecs()`.
+
+use std::io::{Read, Result as IoResult, Seek, SeekFrom};
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{anyhow, Context, Result};
+use symphonia::core::audio::{AudioBufferRef, SampleBuffer, SignalSpec};
+use symphonia::core::codecs::{
+    CodecRegistry, Decoder as SymphoniaDecoderTrait, DecoderOptions, CODEC_TYPE_OPUS,
+};
+use symphonia::core::errors::Error as SymphoniaError;
+use symphonia::core::formats::{FormatOptions, FormatReader, SeekMode, SeekTo, SeekedTo};
+use symphonia::core::io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions};
+use symphonia::core::units;
+use symphonia::default::formats::OggReader;
+use symphonia_adapter_libopus::OpusDecoder;
+
+use rodio::source::SeekError;
+use rodio::{ChannelCount, Sample, SampleRate, Source};
+
+/// The Ogg page capture pattern (every page starts with it).
+const OGG_MAGIC: &[u8] = b"OggS";
+/// The magic signature of the Opus identification header packet, which is
+/// the first packet of an Ogg-Opus stream (RFC 7845 §5.1).
+const OPUS_HEAD_MAGIC: &[u8] = b"OpusHead";
+
+/// Returns true if `header` looks like the start of an Ogg-Opus stream: an
+/// Ogg page whose first packet is the Opus identification header. `header`
+/// should be at least the first Ogg page (~64 bytes is plenty; `OpusHead`
+/// sits at offset 28 in a well-formed stream). Ogg-Vorbis/FLAC deliberately
+/// do *not* match, so they keep flowing through rodio's native decoder.
+pub fn is_ogg_opus(header: &[u8]) -> bool {
+    header.starts_with(OGG_MAGIC) && contains(header, OPUS_HEAD_MAGIC)
+}
+
+fn contains(haystack: &[u8], needle: &[u8]) -> bool {
+    haystack
+        .windows(needle.len())
+        .any(|window| window == needle)
+}
+
+/// A rodio [`Source`] that decodes an Ogg-Opus stream to interleaved `f32`
+/// samples at libopus's native 48 kHz.
+pub struct OpusSource {
+    decoder: Box<dyn SymphoniaDecoderTrait>,
+    format: Box<dyn FormatReader>,
+    track_id: u32,
+    /// Interleaved samples of the most recently decoded packet.
+    buffer: SampleBuffer<Sample>,
+    /// Read cursor into `buffer`, in samples (not frames).
+    span_offset: usize,
+    spec: SignalSpec,
+    total_duration: Option<Duration>,
+}
+
+impl OpusSource {
+    /// Builds an Opus source from a seekable byte stream. `byte_len` (the
+    /// content length) and `is_seekable` mirror the values the player already
+    /// computes for rodio; supplying them lets symphonia's Ogg reader learn
+    /// the final granule position (hence duration) and seek by time.
+    pub fn new<R>(reader: R, byte_len: Option<u64>, is_seekable: bool) -> Result<Self>
+    where
+        R: Read + Seek + Send + Sync + 'static,
+    {
+        let source = Box::new(ReadSeekSource {
+            inner: reader,
+            byte_len,
+            is_seekable,
+        });
+        let mss = MediaSourceStream::new(source, MediaSourceStreamOptions::default());
+
+        let format = OggReader::try_new(mss, &FormatOptions::default())
+            .context("failed to open Ogg-Opus container")?;
+        let format: Box<dyn FormatReader> = Box::new(format);
+
+        let track = format
+            .tracks()
+            .iter()
+            .find(|t| t.codec_params.codec == CODEC_TYPE_OPUS)
+            .context("Ogg stream has no Opus track")?;
+        let track_id = track.id;
+
+        // A registry holding only the libopus adapter — the container is
+        // Opus-only, so no other codec can appear.
+        let mut registry = CodecRegistry::new();
+        registry.register_all::<OpusDecoder>();
+        let decoder = registry
+            .make(&track.codec_params, &DecoderOptions::default())
+            .context("failed to create libopus decoder")?;
+
+        let total_duration = track
+            .codec_params
+            .time_base
+            .zip(track.codec_params.n_frames)
+            .map(|(base, frames)| Duration::from(base.calc_time(frames)))
+            .filter(|d| !d.is_zero());
+
+        // Decode the first packet up front so the struct always holds a real
+        // buffer and spec. symphonia's `SampleBuffer::new` divides by the
+        // channel count, so a zero-channel placeholder would panic; this also
+        // makes channel/rate queries valid before the first `next()` (rodio
+        // inspects the source right after construction).
+        let mut format = format;
+        let mut decoder = decoder;
+        let (spec, buffer) = decode_next_buffer(&mut format, &mut decoder, track_id)?
+            .context("Opus stream produced no audio")?;
+
+        Ok(Self {
+            decoder,
+            format,
+            track_id,
+            buffer,
+            span_offset: 0,
+            spec,
+            total_duration,
+        })
+    }
+}
+
+/// Decodes packets — skipping ones from other logical streams and benign
+/// decode errors — until one yields audio frames, returning that frame's
+/// interleaved samples and signal spec. `Ok(None)` is a clean end of stream.
+fn decode_next_buffer(
+    format: &mut Box<dyn FormatReader>,
+    decoder: &mut Box<dyn SymphoniaDecoderTrait>,
+    track_id: u32,
+) -> Result<Option<(SignalSpec, SampleBuffer<Sample>)>> {
+    loop {
+        let packet = match format.next_packet() {
+            Ok(packet) => packet,
+            // An IO error at a page boundary is how symphonia's Ogg reader
+            // signals end-of-stream for a bounded file.
+            Err(SymphoniaError::IoError(_)) => return Ok(None),
+            Err(e) => return Err(anyhow!(e)).context("reading Opus packet"),
+        };
+        if packet.track_id() != track_id {
+            continue;
+        }
+        match decoder.decode(&packet) {
+            Ok(decoded) if decoded.frames() > 0 => {
+                let spec = *decoded.spec();
+                return Ok(Some((spec, copy_to_buffer(decoded, &spec))));
+            }
+            // A metadata-only packet (0 frames) or a recoverable decode error:
+            // skip it and keep going, matching rodio's behaviour.
+            Ok(_) => continue,
+            Err(SymphoniaError::DecodeError(_)) => continue,
+            Err(e) => return Err(anyhow!(e)).context("decoding Opus packet"),
+        }
+    }
+}
+
+/// Copies a decoded symphonia buffer into an interleaved `f32` sample buffer.
+fn copy_to_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer<Sample> {
+    let capacity = units::Duration::from(decoded.capacity() as u64);
+    let mut buffer = SampleBuffer::<Sample>::new(capacity, *spec);
+    buffer.copy_interleaved_ref(decoded);
+    buffer
+}
+
+impl Iterator for OpusSource {
+    type Item = Sample;
+
+    fn next(&mut self) -> Option<Sample> {
+        if self.span_offset >= self.buffer.len() {
+            match decode_next_buffer(&mut self.format, &mut self.decoder, self.track_id) {
+                Ok(Some((spec, buffer))) => {
+                    self.spec = spec;
+                    self.buffer = buffer;
+                    self.span_offset = 0;
+                }
+                // Clean EOS or a fatal error both end the source.
+                _ => return None,
+            }
+        }
+        let sample = *self.buffer.samples().get(self.span_offset)?;
+        self.span_offset += 1;
+        Some(sample)
+    }
+}
+
+impl Source for OpusSource {
+    fn current_span_len(&self) -> Option<usize> {
+        Some(self.buffer.len())
+    }
+
+    fn channels(&self) -> ChannelCount {
+        let count = u16::try_from(self.spec.channels.count().max(1)).unwrap_or(2);
+        ChannelCount::new(count).unwrap_or(ChannelCount::new(2).expect("2 is nonzero"))
+    }
+
+    fn sample_rate(&self) -> SampleRate {
+        SampleRate::new(self.spec.rate)
+            .unwrap_or(SampleRate::new(48_000).expect("48000 is nonzero"))
+    }
+
+    fn total_duration(&self) -> Option<Duration> {
+        self.total_duration
+    }
+
+    fn try_seek(&mut self, pos: Duration) -> Result<(), SeekError> {
+        // Clamp beyond-end seeks to the end (saturating), like rodio.
+        let mut target = pos;
+        if let Some(total) = self.total_duration {
+            if target > total {
+                target = total;
+            }
+        }
+
+        // Preserve the active channel so we resume on a frame boundary.
+        let channels = self.channels().get() as usize;
+        let active_channel = self.span_offset % channels;
+
+        let seeked = self
+            .format
+            .seek(
+                SeekMode::Accurate,
+                SeekTo::Time {
+                    time: target.into(),
+                    track_id: Some(self.track_id),
+                },
+            )
+            .map_err(|e| SeekError::Other(Arc::new(e)))?;
+
+        // The demuxer moved without telling the decoder; reset it and force
+        // the next `next()` to refill from the new position (`span_offset`
+        // past the current buffer's end triggers a decode in `next()`).
+        self.decoder.reset();
+        self.span_offset = usize::MAX;
+
+        // Ogg seeks land on a page boundary before the target; fast-forward
+        // the residual so playback resumes at the requested instant.
+        self.refine_position(seeked);
+
+        // Re-align to the channel we were on before seeking.
+        for _ in 0..active_channel {
+            self.next();
+        }
+        Ok(())
+    }
+}
+
+impl OpusSource {
+    /// Skips the samples between the keyframe symphonia seeked to and the
+    /// exact requested timestamp. No-op when the demuxer hit the target.
+    fn refine_position(&mut self, seeked: SeekedTo) {
+        let Some(base) = self.decoder.codec_params().time_base else {
+            return;
+        };
+        let residual = seeked.required_ts.saturating_sub(seeked.actual_ts);
+        if residual == 0 {
+            return;
+        }
+        let seconds = Duration::from(base.calc_time(residual)).as_secs_f64();
+        let channels = self.channels().get() as f64;
+        let mut samples = (seconds * self.sample_rate().get() as f64 * channels).ceil() as usize;
+        samples -= samples % self.channels().get() as usize;
+        for _ in 0..samples {
+            if self.next().is_none() {
+                break;
+            }
+        }
+    }
+}
+
+/// Adapts a `Read + Seek` byte stream into symphonia's [`MediaSource`]
+/// (equivalent to rodio's private `ReadSeekSource`).
+struct ReadSeekSource<R> {
+    inner: R,
+    byte_len: Option<u64>,
+    is_seekable: bool,
+}
+
+impl<R: Read + Seek + Send + Sync> MediaSource for ReadSeekSource<R> {
+    fn is_seekable(&self) -> bool {
+        self.is_seekable
+    }
+    fn byte_len(&self) -> Option<u64> {
+        self.byte_len
+    }
+}
+
+impl<R: Read> Read for ReadSeekSource<R> {
+    fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
+        self.inner.read(buf)
+    }
+}
+
+impl<R: Seek> Seek for ReadSeekSource<R> {
+    fn seek(&mut self, pos: SeekFrom) -> IoResult<u64> {
+        self.inner.seek(pos)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn detects_ogg_opus_header() {
+        // "OggS" page header with an "OpusHead" first packet.
+        let mut header = Vec::new();
+        header.extend_from_slice(b"OggS");
+        header.extend_from_slice(&[0u8; 24]); // rest of the 28-byte page header
+        header.extend_from_slice(b"OpusHead");
+        assert!(is_ogg_opus(&header));
+    }
+
+    #[test]
+    fn rejects_ogg_vorbis() {
+        let mut header = Vec::new();
+        header.extend_from_slice(b"OggS");
+        header.extend_from_slice(&[0u8; 24]);
+        header.extend_from_slice(b"\x01vorbis");
+        assert!(!is_ogg_opus(&header));
+    }
+
+    #[test]
+    fn rejects_non_ogg() {
+        assert!(!is_ogg_opus(b"ID3\x04OpusHead")); // mp3 with a coincidental needle
+        assert!(!is_ogg_opus(b"fLaC"));
+        assert!(!is_ogg_opus(b""));
+    }
+}

audio-player/src/player_engine.rs

@@ -1,5 +1,5 @@
 use std::fs::File;
-use std::io::BufReader;
+use std::io::{BufReader, Read, Seek, SeekFrom};
 use std::path::Path;
 use std::time::Duration;
 
@@ -11,6 +11,7 @@ use rodio::{Decoder, Source};
 use stream_download::storage::temp::TempStorageProvider;
 use stream_download::{Settings, StreamDownload};
 
+use crate::opus_source::{is_ogg_opus, OpusSource};
 use crate::spectrum_tap::{SpectrumTap, TappingSource};
 use crate::windowed_http::{WindowedHttpParams, WindowedHttpStream};
 use std::sync::Arc;
@@ -325,26 +326,17 @@ impl PlayerEngine {
                     .map_err(|_| anyhow!("timed out opening stream after {STREAM_OPEN_TIMEOUT:?}"))?
                     .context("failed to open http stream")
                 })?;
-                // Symphonia probes the container length during init; without a
-                // known byte length it seeks from the end, which rodio 0.22 turns
-                // into an `unreachable!` panic on a streamed source. Handing it the
-                // content length up front avoids that seek entirely.
                 let byte_len = reader.content_length();
-                let mut builder = Decoder::builder().with_data(reader).with_seekable(true);
+                let hint = Path::new(url.path())
-                if let Some(len) = byte_len {
+                    .extension()
-                    builder = builder.with_byte_len(len);
+                    .and_then(|e| e.to_str())
-                }
+                    .map(str::to_owned);
-                if let Some(extension) = Path::new(url.path()).extension().and_then(|e| e.to_str())
+                self.build_source(
-                {
+                    reader,
-                    builder = builder.with_hint(extension);
+                    byte_len,
-                }
+                    hint.as_deref(),
-                let decoder = builder.build().context("failed to decode http stream")?;
+                    "failed to decode http stream",
-                let duration = decoder.total_duration();
+                )
-                // Mirror the played audio into the spectrum tap (it only
-                // observes; playback is unaffected).
-                let source: Box<dyn Source + Send> =
-                    Box::new(TappingSource::new(decoder, self.spectrum.clone()));
-                Ok((source, duration))
             }
             Ok(url) => Err(anyhow!("Not a valid URL scheme: {}", url.scheme())),
             Err(_) => {
@@ -352,23 +344,67 @@ impl PlayerEngine {
                 let file = File::open(source_str)
                     .with_context(|| format!("failed to open file {source_str}"))?;
                 let byte_len = file.metadata().ok().map(|m| m.len());
-                let mut builder = Decoder::builder()
+                let hint = Path::new(source_str)
-                    .with_data(BufReader::new(file))
+                    .extension()
-                    .with_seekable(true);
+                    .and_then(|e| e.to_str())
+                    .map(str::to_owned);
+                self.build_source(
+                    BufReader::new(file),
+                    byte_len,
+                    hint.as_deref(),
+                    "failed to decode file",
+                )
+            }
+        }
+    }
+
+    /// Sniffs the stream's header and decodes it. Ogg-Opus — which symphonia
+    /// (hence rodio) cannot decode — is routed to [`OpusSource`]; every other
+    /// format goes through rodio's decoder. The reader is rewound after
+    /// sniffing so the chosen decoder sees the whole stream, and both paths
+    /// are wrapped in the spectrum tap (it only observes; playback is
+    /// unaffected).
+    fn build_source<R>(
+        &self,
+        mut reader: R,
+        byte_len: Option<u64>,
+        hint: Option<&str>,
+        decode_err: &'static str,
+    ) -> Result<(Box<dyn Source + Send>, Option<Duration>)>
+    where
+        R: Read + Seek + Send + Sync + 'static,
+    {
+        let mut header = [0u8; 64];
+        let n = read_header(&mut reader, &mut header)?;
+        reader
+            .seek(SeekFrom::Start(0))
+            .context("failed to rewind after sniffing the header")?;
+
+        if is_ogg_opus(&header[..n]) {
+            debug!("decoding Ogg-Opus via libopus");
+            let source = OpusSource::new(reader, byte_len, true)?;
+            let duration = source.total_duration();
+            let tapped: Box<dyn Source + Send> =
+                Box::new(TappingSource::new(source, self.spectrum.clone()));
+            return Ok((tapped, duration));
+        }
+
+        // Symphonia probes the container length during init; without a known
+        // byte length it seeks from the end, which rodio 0.22 turns into an
+        // `unreachable!` panic on a streamed source. Handing it the content
+        // length up front avoids that seek entirely.
+        let mut builder = Decoder::builder().with_data(reader).with_seekable(true);
         if let Some(len) = byte_len {
             builder = builder.with_byte_len(len);
         }
-                if let Some(extension) = Path::new(source_str).extension().and_then(|e| e.to_str())
+        if let Some(extension) = hint {
-                {
             builder = builder.with_hint(extension);
         }
-                let decoder = builder.build().context("failed to decode file")?;
+        let decoder = builder.build().context(decode_err)?;
         let duration = decoder.total_duration();
-                let source: Box<dyn Source + Send> =
+        let tapped: Box<dyn Source + Send> =
             Box::new(TappingSource::new(decoder, self.spectrum.clone()));
-                Ok((source, duration))
+        Ok((tapped, duration))
-            }
-        }
     }
 
     /// Appends an already-decoded source to the freshly-reset sink, followed
@@ -538,6 +574,22 @@ fn send_reply<T>(tx: Sender<T>, value: T) {
     }
 }
 
+/// Reads up to `buf.len()` bytes for format sniffing, tolerating short reads
+/// and `Interrupted`. Returns how many bytes were read (fewer than the buffer
+/// only at end-of-stream), so a stream shorter than the buffer is not an error.
+fn read_header<R: Read>(reader: &mut R, buf: &mut [u8]) -> Result<usize> {
+    let mut filled = 0;
+    while filled < buf.len() {
+        match reader.read(&mut buf[filled..]) {
+            Ok(0) => break,
+            Ok(n) => filled += n,
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(e) => return Err(e).context("failed to read stream header"),
+        }
+    }
+    Ok(filled)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

devenv.nix

@@ -15,6 +15,10 @@ let
     pkg-config
     protobuf
     cargo-cross
+    # opusic-sys (via symphonia-adapter-libopus) compiles libopus from
+    # source with CMake; ninja is picked up automatically if present.
+    cmake
+    ninja
     # Stream-URL sidecar for the ytdy provider: YouTube caps tokenless
     # stream URLs at ~1 MiB and yt-dlp is the only maintained cipher
     # solver (architecture/youtube-rustypipe.md, D2-revised).