feat(audio): streamed music decode — OGG/MP3 no longer fully decoded into RAM

A 5-minute 44.1kHz stereo track was held as ~57MB of f32 PCM; it's now
~5MB of compressed bytes (shared Arc) plus ~1.5s of decoded ring.

audio/stream.rs: per-playback decode worker (lewton/minimp3 are both
incremental) feeding 32k-sample chunks over the existing lock-free SPSC
ring with sleep-based backpressure; looping restarts the decoder inside
the worker so loop seams can't underrun; a stop flag (set by
StreamConsumer::drop — i.e. stop/replace/teardown) kills the worker.

The render-side MusicVoice grows a Stream source alongside Full:
chunk-pulling mix loop, silence on underrun (resumes next callback),
shared playing/position atomics maintained as before. WAV stays
full-decode (already PCM); wasm32 falls back to full decode (no
threads) — both via the same load_music_bytes entry the FFI loader now
calls.

Non-goal kept explicit: no resampling yet — non-44.1kHz tracks play at
the same wrong pitch they did before (tracked).

97 shared tests green incl. a stop-flag/backpressure test.

Author: proggeramlug

native/shared/src/audio/mod.rs

@@ -29,6 +29,7 @@
 mod decode;
 mod render;
 mod spsc;
+pub mod stream;
 
 pub use decode::{decode_audio, parse_ogg, parse_wav};
 #[cfg(feature = "mp3")]
@@ -53,8 +54,20 @@ pub struct MusicShared {
     pub position: AtomicUsize,
 }
 
+enum MusicSource {
+    /// Fully decoded PCM (WAV always; everything on wasm32).
+    Full(Arc<SoundData>),
+    /// Compressed bytes decoded by a background worker at play time.
+    #[cfg(not(target_arch = "wasm32"))]
+    Streamed {
+        kind: stream::StreamKind,
+        bytes: Arc<Vec<u8>>,
+        channels: u16,
+    },
+}
+
 struct MusicEntry {
-    data: Arc<SoundData>,
+    source: MusicSource,
     shared: Arc<MusicShared>,
     volume: f32,
     looping: bool,
@@ -174,8 +187,50 @@ impl AudioMixer {
     // ------------------------------------------------------------ music
 
     pub fn load_music(&mut self, data: SoundData) -> f64 {
+        self.alloc_music(MusicSource::Full(Arc::new(data)))
+    }
+
+    /// Load music from raw file bytes, streaming the decode when the
+    /// format supports it (OGG/MP3 on native — keeps ~5 MB of compressed
+    /// bytes resident instead of ~57 MB of PCM for a 5-minute track).
+    /// WAV — and everything on wasm32, which has no threads — falls back
+    /// to full decode. Returns 0 on undecodable data.
+    pub fn load_music_bytes(&mut self, path: &str, data: Vec<u8>) -> f64 {
+        #[cfg(not(target_arch = "wasm32"))]
+        {
+            let lower = path.to_ascii_lowercase();
+            let kind = if lower.ends_with(".ogg") {
+                Some(stream::StreamKind::Ogg)
+            } else {
+                #[cfg(feature = "mp3")]
+                if lower.ends_with(".mp3") {
+                    Some(stream::StreamKind::Mp3)
+                } else {
+                    None
+                }
+                #[cfg(not(feature = "mp3"))]
+                None
+            };
+            if let Some(kind) = kind {
+                if let Some((_rate, channels)) = stream::probe(kind, &data) {
+                    return self.alloc_music(MusicSource::Streamed {
+                        kind,
+                        bytes: Arc::new(data),
+                        channels,
+                    });
+                }
+                // Mislabelled file — fall through to sniffing decode.
+            }
+        }
+        match decode_audio(path, &data) {
+            Some(s) => self.load_music(s),
+            None => 0.0,
+        }
+    }
+
+    fn alloc_music(&mut self, source: MusicSource) -> f64 {
         self.music.alloc(MusicEntry {
-            data: Arc::new(data),
+            source,
             shared: Arc::new(MusicShared {
                 playing: AtomicBool::new(false),
                 position: AtomicUsize::new(0),
@@ -191,9 +246,17 @@ impl AudioMixer {
         // moment play_music returns (the render thread confirms on its
         // next callback).
         m.shared.playing.store(true, Ordering::Relaxed);
+        let payload = match &m.source {
+            MusicSource::Full(data) => render::MusicPayload::Full(data.clone()),
+            #[cfg(not(target_arch = "wasm32"))]
+            MusicSource::Streamed { kind, bytes, channels } => render::MusicPayload::Stream {
+                consumer: stream::start(*kind, bytes.clone(), m.looping),
+                channels: *channels,
+            },
+        };
         let cmd = Cmd::PlayMusic {
             music_id: handle.to_bits(),
-            data: m.data.clone(),
+            payload,
             shared: m.shared.clone(),
             volume: m.volume,
             looping: m.looping,

native/shared/src/audio/render.rs

@@ -11,6 +11,8 @@
 //! voice playing it finishes, then the Arc drops on this thread.
 
 use super::spsc::Consumer;
+#[cfg(not(target_arch = "wasm32"))]
+use super::stream::{StreamConsumer, StreamMsg};
 use super::{MusicShared, SoundData};
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
@@ -30,7 +32,7 @@ pub enum Cmd {
     SetSoundVolume { sound_id: u64, volume: f32 },
     PlayMusic {
         music_id: u64,
-        data: Arc<SoundData>,
+        payload: MusicPayload,
         shared: Arc<MusicShared>,
         volume: f32,
         looping: bool,
@@ -49,13 +51,35 @@ struct Voice {
     spatial: Option<[f32; 3]>,
 }
 
+/// How a music voice gets its samples.
+pub enum MusicPayload {
+    Full(Arc<SoundData>),
+    /// Chunks arrive from a background decode worker; the worker handles
+    /// looping internally (End only arrives for finished non-loop tracks).
+    #[cfg(not(target_arch = "wasm32"))]
+    Stream { consumer: StreamConsumer, channels: u16 },
+}
+
+enum MusicSamples {
+    Full { data: Arc<SoundData>, position: usize },
+    #[cfg(not(target_arch = "wasm32"))]
+    Stream {
+        consumer: StreamConsumer,
+        channels: u16,
+        current: Vec<f32>,
+        offset: usize,
+        ended: bool,
+    },
+}
+
 struct MusicVoice {
     music_id: u64,
-    data: Arc<SoundData>,
+    samples: MusicSamples,
     shared: Arc<MusicShared>,
-    position: usize,
     volume: f32,
     looping: bool,
+    /// Total samples consumed (drives the shared position mirror).
+    consumed: usize,
 }
 
 pub struct AudioRenderer {
@@ -94,12 +118,23 @@ impl AudioRenderer {
                     }
                 }
             }
-            Cmd::PlayMusic { music_id, data, shared, volume, looping } => {
+            Cmd::PlayMusic { music_id, payload, shared, volume, looping } => {
                 // Restart-from-zero semantics (matches the old mixer).
                 self.music.retain(|m| m.music_id != music_id);
                 shared.playing.store(true, Ordering::Relaxed);
                 shared.position.store(0, Ordering::Relaxed);
-                self.music.push(MusicVoice { music_id, data, shared, position: 0, volume, looping });
+                let samples = match payload {
+                    MusicPayload::Full(data) => MusicSamples::Full { data, position: 0 },
+                    #[cfg(not(target_arch = "wasm32"))]
+                    MusicPayload::Stream { consumer, channels } => MusicSamples::Stream {
+                        consumer,
+                        channels,
+                        current: Vec::new(),
+                        offset: 0,
+                        ended: false,
+                    },
+                };
+                self.music.push(MusicVoice { music_id, samples, shared, volume, looping, consumed: 0 });
             }
             Cmd::StopMusic { music_id } => {
                 if let Some(m) = self.music.iter().position(|m| m.music_id == music_id) {
@@ -192,31 +227,81 @@ impl AudioRenderer {
         self.music.retain_mut(|m| {
             let vol = m.volume * master;
             let mut i = 0;
-            while i < output.len() && m.position < m.data.samples.len() {
-                if m.data.channels == 1 {
-                    let sample = m.data.samples[m.position] * vol;
-                    output[i] += sample;
-                    if i + 1 < output.len() {
-                        output[i + 1] += sample;
+            let mut finished = false;
+            match &mut m.samples {
+                MusicSamples::Full { data, position } => {
+                    while i < output.len() && *position < data.samples.len() {
+                        if data.channels == 1 {
+                            let sample = data.samples[*position] * vol;
+                            output[i] += sample;
+                            if i + 1 < output.len() {
+                                output[i + 1] += sample;
+                            }
+                            *position += 1;
+                            i += 2;
+                        } else {
+                            output[i] += data.samples[*position] * vol;
+                            *position += 1;
+                            i += 1;
+                        }
                     }
-                    m.position += 1;
-                    i += 2;
-                } else {
-                    output[i] += m.data.samples[m.position] * vol;
-                    m.position += 1;
-                    i += 1;
+                    if *position >= data.samples.len() {
+                        if m.looping {
+                            *position = 0;
+                        } else {
+                            finished = true;
+                        }
+                    }
+                    m.consumed = *position;
                 }
-            }
-            if m.position >= m.data.samples.len() {
-                if m.looping {
-                    m.position = 0;
-                } else {
-                    m.shared.playing.store(false, Ordering::Relaxed);
-                    m.shared.position.store(0, Ordering::Relaxed);
-                    return false;
+                #[cfg(not(target_arch = "wasm32"))]
+                MusicSamples::Stream { consumer, channels, current, offset, ended } => {
+                    let mono = *channels == 1;
+                    while i < output.len() {
+                        if *offset >= current.len() {
+                            // Refill from the decode worker's ring.
+                            match consumer.rx.pop() {
+                                Some(StreamMsg::Chunk(c)) => {
+                                    *current = c;
+                                    *offset = 0;
+                                }
+                                Some(StreamMsg::End) => {
+                                    *ended = true;
+                                    break;
+                                }
+                                // Underrun: worker is behind (cold start
+                                // or scheduling hiccup) — emit silence for
+                                // the rest of this callback, resume next.
+                                None => break,
+                            }
+                        }
+                        if mono {
+                            let sample = current[*offset] * vol;
+                            output[i] += sample;
+                            if i + 1 < output.len() {
+                                output[i + 1] += sample;
+                            }
+                            *offset += 1;
+                            m.consumed += 1;
+                            i += 2;
+                        } else {
+                            output[i] += current[*offset] * vol;
+                            *offset += 1;
+                            m.consumed += 1;
+                            i += 1;
+                        }
+                    }
+                    if *ended && *offset >= current.len() {
+                        finished = true;
+                    }
                 }
             }
-            m.shared.position.store(m.position, Ordering::Relaxed);
+            if finished {
+                m.shared.playing.store(false, Ordering::Relaxed);
+                m.shared.position.store(0, Ordering::Relaxed);
+                return false;
+            }
+            m.shared.position.store(m.consumed, Ordering::Relaxed);
             true
         });
     }