Merge pull request #560 from InstaZDLL/feat/remote-stream-cache

feat(remote): cache a remote stream from the bytes playback already reads

Author: InstaZDLL

docs/rfcs/RFC-005-remote-source-and-sync-v2.md

@@ -428,6 +428,34 @@ negative sentinel id at load time, per position rather than per track, and the
 rows are moved rather than re-projected — so the handle survives an optimistic
 reorder even on a playlist holding one track twice.
 
+**A remote stream is cached on disk, and it is filled by playback itself.**
+Until now the audio was re-fetched in full on every play — the projection
+caches metadata and the cover cache caches covers, but the bytes that actually
+cost bandwidth were not kept. They are now, under
+`profile_remote_stream_dir`, keyed by `(track id, format, bitrate)`: the triple
+that determines the bytes, and deliberately not the URL, which carries a
+single-use ticket and differs on every play.
+
+The cache is **not** a downloader. Nothing extra is fetched and nothing is
+delayed: every block the decoder reads is written at its **absolute offset**
+into a sparse working file, so the first play sounds exactly as it did and the
+second reads from disk. Writing by offset rather than by append is what makes
+this survive symphonia, which seeks while probing and again on a scrub — an
+append-only tee would have to give up at the first seek, which for most formats
+arrives within the first few kilobytes.
+
+An entry is published only when the covered ranges merge into one span over the
+whole body, by a single atomic rename out of `.part`. A partial file is worse
+than an absent one: it decodes for a while and then stops, which reads as a
+broken track rather than as a cold cache. A body whose length the server did
+not declare is never cached, because completeness could not be decided. On a
+hit the track loads through the existing `LoadRemoteFileAndPlay` path with a
+freshly-minted ticket as its `fallback_url` — a small JSON round-trip, not the
+body the cache just saved, which buys back the decoder's repair path so a
+cached file that will not decode falls back to the server once instead of
+failing that track forever. Offline, the ticket is simply not minted and the
+cached file plays alone, which is the point of having it.
+
 **The queue panel** switches to a dedicated `RemoteQueueView` while a remote
 session plays (keyed on `isRemoteTrack`), reading `remote_get_play_queue` (an
 in-memory snapshot) and jumping with `remote_queue_jump` — the local

src-tauri/crates/app/src/audio/decoder.rs

@@ -434,6 +434,7 @@ fn decoder_loop(
                 artist,
                 artwork_url,
                 fallback_url,
+                discard_on_failure,
                 replay_gain,
             } => {
                 shared.clear_ab_loop();
@@ -507,6 +508,7 @@ fn decoder_loop(
                             has_server_fallback = fallback_url.is_some(),
                             "remote local source open failed"
                         );
+                        discard_unplayable(&path, discard_on_failure);
                         if let Some(url) = fallback_url.filter(|_| !crate::offline::is_offline()) {
                             pending_cmd = Some(AudioCmd::LoadUrlAndPlay {
                                 url,
@@ -519,6 +521,12 @@ fn decoder_loop(
                                 // server instead of the local file must
                                 // not change its level.
                                 replay_gain,
+                                // Not cached: this is the repair path for a
+                                // file that would not open, and the target
+                                // that named it is two commands upstream.
+                                // Caching here would also risk re-writing the
+                                // very entry that just failed to decode.
+                                cache: None,
                             });
                             continue;
                         }
@@ -544,6 +552,7 @@ fn decoder_loop(
                             path = %path.display(),
                             "remote local decode failed; falling back to server"
                         );
+                        discard_unplayable(&path, discard_on_failure);
                         pending_cmd = Some(AudioCmd::LoadUrlAndPlay {
                             url,
                             ext_hint: None,
@@ -552,6 +561,8 @@ fn decoder_loop(
                             title,
                             artist,
                             artwork_url,
+                            // See above: a repair path does not repopulate.
+                            cache: None,
                         });
                         continue;
                     }
@@ -572,6 +583,7 @@ fn decoder_loop(
                 artist,
                 artwork_url,
                 replay_gain,
+                cache,
             } => {
                 tracing::info!(
                     track_id,
@@ -685,7 +697,15 @@ fn decoder_loop(
                 // with the live `StreamTitle` while keeping the station's
                 // cover + name; it stays forward-only.
                 let opened = if is_remote {
-                    super::http_source::HttpMediaSource::open_seekable(&url)
+                    // A cache target only ever accompanies a finite
+                    // remote-queue track, so the seekable open is the only
+                    // one that can honour it.
+                    match cache {
+                        Some(target) => {
+                            super::http_source::HttpMediaSource::open_seekable_caching(&url, target)
+                        }
+                        None => super::http_source::HttpMediaSource::open_seekable(&url),
+                    }
                 } else {
                     let icy_ctx = super::http_source::IcyContext {
                         app: app.clone(),
@@ -2423,3 +2443,21 @@ mod tests {
         approx(ro, 2.0 + K * (0.4 + 0.8 + 0.2 + 0.07));
     }
 }
+
+/// Drop a reproducible copy that would not play.
+///
+/// Only ever called with `allowed` set for a stream-cache entry: the server
+/// still holds those bytes, so a file that fails to open or decode is worth
+/// losing to make the next play refetch it. A reconciled file from the user's
+/// own library reaches the same failure path and must survive it — it is
+/// theirs, and a decoder that deletes a listener's music because a codec
+/// tripped would be a far worse bug than a cache miss.
+fn discard_unplayable(path: &std::path::Path, allowed: bool) {
+    if !allowed {
+        return;
+    }
+    match std::fs::remove_file(path) {
+        Ok(()) => tracing::info!(path = %path.display(), "dropped unplayable cached stream"),
+        Err(err) => tracing::warn!(?err, path = %path.display(), "could not drop cached stream"),
+    }
+}

src-tauri/crates/app/src/audio/engine.rs

@@ -59,6 +59,15 @@ pub enum AudioCmd {
         artist: Option<String>,
         artwork_url: Option<String>,
         fallback_url: Option<String>,
+        /// Whether `path` is a reproducible copy that should be discarded if
+        /// it will not open or decode.
+        ///
+        /// True for a stream-cache entry, which can always be fetched again;
+        /// false for a reconciled file from the user's own library, which is
+        /// theirs and is never ours to delete. Without the distinction a bad
+        /// cache entry would fall back to the server on every single play
+        /// instead of once.
+        discard_on_failure: bool,
         replay_gain: TrackGain,
     },
     Pause,
@@ -118,6 +127,10 @@ pub enum AudioCmd {
         title: Option<String>,
         artist: Option<String>,
         artwork_url: Option<String>,
+        /// Where to cache the bytes this stream reads, when it is a finite
+        /// remote-queue track worth keeping. `None` for radio, which is
+        /// endless and therefore never complete.
+        cache: Option<crate::audio::stream_cache::CacheTarget>,
         /// Loudness metadata for the stream. `TrackGain::default()`
         /// for a live radio station — nothing knows anything about it
         /// — but a library track that fell back to streaming from the
@@ -404,13 +417,18 @@ impl RadioResumeState {
                 artist: self.artist,
                 artwork_url: self.artwork_url,
                 replay_gain,
+                // Resuming radio after a device change: endless, so never a
+                // complete body to cache.
+                cache: None,
             },
             RadioResumeSource::RemoteFile {
                 path,
                 duration_ms,
                 fallback_url,
                 replay_gain,
             } => AudioCmd::LoadRemoteFileAndPlay {
+                // Radio resume never restores a cache entry.
+                discard_on_failure: false,
                 path,
                 start_ms: position_ms,
                 track_id: self.track_id,
@@ -1499,6 +1517,10 @@ fn apply_radio_resume_update(snapshot: &Mutex<Option<RadioResumeState>>, cmd: &A
             artist,
             artwork_url,
             replay_gain,
+            // The resume snapshot exists to restart radio after a device
+            // change; a cache target belongs to one open response and does
+            // not survive into a new one.
+            cache: _,
         } => {
             if let Ok(mut guard) = snapshot.lock() {
                 *guard = Some(RadioResumeState {
@@ -1515,6 +1537,7 @@ fn apply_radio_resume_update(snapshot: &Mutex<Option<RadioResumeState>>, cmd: &A
             }
         }
         AudioCmd::LoadRemoteFileAndPlay {
+            discard_on_failure: false,
             path,
             duration_ms,
             fallback_url,
@@ -1699,6 +1722,7 @@ mod radio_resume_tests {
             title: Some("Test stream".to_string()),
             artist: Some("Test artist".to_string()),
             artwork_url: Some("https://example.invalid/art.jpg".to_string()),
+            cache: None,
             replay_gain: TrackGain::default(),
         }
     }
@@ -1725,6 +1749,7 @@ mod radio_resume_tests {
             artist: Some("Remote artist".to_string()),
             artwork_url: None,
             fallback_url: fallback_url.map(str::to_string),
+            discard_on_failure: false,
             replay_gain: TrackGain {
                 gain_db: Some(-4.0),
                 peak: None,

src-tauri/crates/app/src/audio/http_source.rs

@@ -174,6 +174,11 @@ pub struct HttpMediaSource {
     /// (radio servers resend the same title every interval) doesn't
     /// re-fire the event.
     last_title: Option<String>,
+    /// Writes what this source reads into the on-disk stream cache, when the
+    /// caller asked for it and the body declared a length. `None` for radio
+    /// (endless, so never complete) and whenever the cache is unusable — it
+    /// is an optimisation, and its absence is never a playback failure.
+    cache: Option<crate::audio::stream_cache::CacheWriter>,
 }
 
 impl HttpMediaSource {
@@ -182,15 +187,15 @@ impl HttpMediaSource {
     /// 404 / 502 surfaces as `Err` instead of producing a `MediaSource`
     /// that would only fail at the first `probe()` call.
     pub fn open(url: &str) -> Result<Self, String> {
-        Self::open_inner(url, None, false)
+        Self::open_inner(url, None, false, None)
     }
 
     /// Open a streaming HTTP GET that also requests + de-interleaves ICY
     /// metadata, re-emitting `player:radio-metadata` through `icy.app`
     /// whenever the live `StreamTitle` changes. Falls back transparently
     /// to passthrough when the server ignores `Icy-MetaData: 1`.
     pub fn open_with_icy(url: &str, icy: IcyContext) -> Result<Self, String> {
-        Self::open_inner(url, Some(icy), false)
+        Self::open_inner(url, Some(icy), false, None)
     }
 
     /// Open a finite, range-capable file for **seekable** playback — a
@@ -200,10 +205,29 @@ impl HttpMediaSource {
     /// drive a real `format.seek`. Degrades to forward-only if the server
     /// doesn't answer ranges — playback still works, only scrubbing won't.
     pub fn open_seekable(url: &str) -> Result<Self, String> {
-        Self::open_inner(url, None, true)
+        Self::open_inner(url, None, true, None)
     }
 
-    fn open_inner(url: &str, icy: Option<IcyContext>, want_seek: bool) -> Result<Self, String> {
+    /// [`Self::open_seekable`], additionally filling the on-disk stream cache
+    /// from the bytes playback reads.
+    ///
+    /// Nothing extra is fetched and nothing is delayed: the cache is written
+    /// from the blocks the decoder was going to read anyway, at their absolute
+    /// offsets, so a seek leaves a hole rather than corrupting the file and
+    /// the entry is simply never published.
+    pub fn open_seekable_caching(
+        url: &str,
+        target: crate::audio::stream_cache::CacheTarget,
+    ) -> Result<Self, String> {
+        Self::open_inner(url, None, true, Some(target))
+    }
+
+    fn open_inner(
+        url: &str,
+        icy: Option<IcyContext>,
+        want_seek: bool,
+        cache_target: Option<crate::audio::stream_cache::CacheTarget>,
+    ) -> Result<Self, String> {
         // Offline short-circuit at the HTTP boundary itself. The decoder
         // already gates `LoadUrlAndPlay` on this before reaching here, but
         // guarding the source too makes it self-honouring for any future
@@ -269,6 +293,12 @@ impl HttpMediaSource {
         // `byte_len() == None` exactly as before, even on the rare stream
         // that sends a Content-Length, so symphonia never treats it as
         // finite.
+        // Two different questions were sharing one variable. Seeking needs a
+        // length AND ranges; caching needs only a length, because it is filled
+        // by reading forward. Collapsing them meant a server that sends
+        // `Content-Length` without `Accept-Ranges` — perfectly cacheable —
+        // was never cached at all.
+        let declared_len = len;
         let len = if seekable { len } else { None };
 
         Ok(Self {
@@ -285,6 +315,14 @@ impl HttpMediaSource {
             // actually parse blocks).
             icy: if metaint > 0 { icy } else { None },
             last_title: None,
+            // Only a body that declared its length can be cached: without one
+            // there is no way to decide that every byte has been covered, and
+            // a file we cannot call complete must never be offered as one.
+            cache: cache_target.and_then(|target| {
+                declared_len.and_then(|len| {
+                    crate::audio::stream_cache::CacheWriter::create(&target.dir, &target.name, len)
+                })
+            }),
         })
     }
 
@@ -406,7 +444,17 @@ impl Read for HttpMediaSource {
         // byte cursor here for `SeekFrom::Current`.
         if self.metaint == 0 {
             let n = guard.read(buf)?;
+            drop(guard);
+            let start = self.pos;
             self.pos += n as u64;
+            // Record where these bytes live, not merely that they arrived: a
+            // seek reopens the body at another offset, so an append would
+            // interleave two regions into one corrupt file.
+            if n > 0 {
+                if let Some(cache) = self.cache.as_mut() {
+                    cache.write_at(start, &buf[..n]);
+                }
+            }
             return Ok(n);
         }
         if buf.is_empty() {

src-tauri/crates/app/src/audio/mod.rs

@@ -36,6 +36,7 @@ pub mod replay_gain;
 pub mod resampler;
 pub mod spectrum;
 pub mod state;
+pub mod stream_cache;
 #[cfg(target_os = "windows")]
 pub mod wasapi_exclusive;