fix(cast): serve arriving bytes, one thread per request, close the receiver app on disconnect

MediaServer gains a RangeSource entry whose bytes may still be arriving
(known final size, blocking reads) and serves every request on its own
thread so a slow media body no longer stalls the HEAD probe or a seek's
second Range request. Chromecast: play/pause/stop never launch the
Default Media Receiver (an app launched with nothing loaded sat IDLE and
read as "track ended"); disconnect stops the receiver app instead of
leaving it on screen with the last item.

cc

crates/qbz-cast/src/chromecast/device.rs

@@ -91,7 +91,14 @@ impl CastDeviceConnection {
 
     /// Disconnect from device
     pub fn disconnect(&mut self) -> Result<(), CastError> {
+        // CLOSE the media receiver app, not just our transport to it. A bare
+        // transport disconnect leaves the Default Media Receiver running on
+        // the device with our last item on its screen (and, if it was
+        // playing, still playing) — the "remote queue never clears" symptom.
+        // Tearing the app down is what the official senders do on "stop
+        // casting".
         if let Some(session) = &self.session {
+            let _ = self.device.receiver.stop_app(session.session_id.as_str());
             let _ = self
                 .device
                 .connection
@@ -285,8 +292,15 @@ impl CastDeviceConnection {
         self.launch_media_app().map(|_| ())
     }
 
+    /// Transport commands NEVER launch the receiver app: `play`/`pause`/
+    /// `stop` on a receiver with nothing loaded used to launch the Default
+    /// Media Receiver just to ask it for a media session it could not have,
+    /// which then sat there IDLE — and an IDLE report is what the poll reads
+    /// as "track ended". Only `load_media` launches the app.
     fn ensure_media_session(&mut self) -> Result<(String, i32), CastError> {
-        self.ensure_session()?;
+        if self.session.is_none() {
+            return Err(CastError::NoMediaSession);
+        }
 
         let destination = self
             .session

crates/qbz-cast/src/lib.rs

@@ -22,7 +22,7 @@ pub mod media_server;
 pub use errors::{CastError, DlnaError};
 
 // Re-export media server
-pub use media_server::MediaServer;
+pub use media_server::{MediaServer, RangeSource};
 
 // Re-export Chromecast types
 pub use chromecast::{

crates/qbz-cast/src/media_server.rs

@@ -53,6 +53,20 @@ enum MediaSource {
     // that is still being served (#550).
     Data(std::sync::Arc<Vec<u8>>),
     File(PathBuf),
+    /// Bytes that are still ARRIVING while they are served: a Qobuz track
+    /// being downloaded, or a Plex/Jellyfin/Subsonic stream proxied from its
+    /// server. The total size is known up front (advertised as
+    /// Content-Length); reads past the buffered edge block until the bytes
+    /// exist. This is what turns "download 100 MB, then play" into "play
+    /// after the first segment".
+    Reader(Arc<dyn RangeSource>),
+}
+
+/// A range-addressable byte source whose bytes may still be arriving.
+/// `open` returns a reader positioned at `start` that yields at most `len`
+/// bytes and BLOCKS (never errors) while a requested byte is not there yet.
+pub trait RangeSource: Send + Sync {
+    fn open(&self, start: u64, len: u64) -> std::io::Result<Box<dyn Read + Send>>;
 }
 
 /// Simple HTTP server for audio streaming to cast devices
@@ -112,14 +126,25 @@ impl MediaServer {
             while !shutdown_clone.load(Ordering::SeqCst) {
                 match server.recv_timeout(Duration::from_millis(250)) {
                     Ok(Some(request)) => {
+                        // ONE THREAD PER REQUEST. A media body is written at
+                        // the renderer's pace (a 100 MB track takes the whole
+                        // song to drain, and a progressive source blocks on
+                        // bytes that have not arrived), so serving inline
+                        // stalled every other request — the HEAD probe strict
+                        // renderers send before Play, a second Range request
+                        // for a seek — for the duration of the track.
+                        let entries = entries_clone.clone();
+                        let token = token_clone.clone();
+                        thread::spawn(move || {
                             let response = handle_request(
                                 request.method(),
                                 request.url(),
                                 &request,
-                            &entries_clone,
-                            &token_clone,
+                                &entries,
+                                &token,
                             );
                             let _ = request.respond(response);
+                        });
                     }
                     Ok(None) => {}
                     Err(e) => {
@@ -188,6 +213,29 @@ impl MediaServer {
         self.audio_path(id)
     }
 
+    /// Register a range-addressable source whose bytes may still be arriving
+    /// (see [`RangeSource`]). `size` is the FINAL byte length — it is what the
+    /// renderer is told in Content-Length, so it must be exact.
+    pub fn register_reader(
+        &mut self,
+        id: u64,
+        size: u64,
+        content_type: &str,
+        source: Arc<dyn RangeSource>,
+    ) -> String {
+        let entry = MediaEntry {
+            content_type: content_type.to_string(),
+            size,
+            source: MediaSource::Reader(source),
+        };
+        if let Ok(mut entries) = self.entries.lock() {
+            // Same eviction rule as register_audio (#550).
+            entries.retain(|k, _| *k == id);
+            entries.insert(id, entry);
+        }
+        self.audio_path(id)
+    }
+
     /// Register a local file to serve, inferring the content type from the
     /// extension via [`content_type_for_path`].
     pub fn register_file(&mut self, id: u64, file_path: &str) -> Result<String, CastError> {
@@ -444,6 +492,13 @@ fn open_range(
                 content_range,
             ))
         }
+        MediaSource::Reader(source) => {
+            let end = end.min(entry.size.saturating_sub(1));
+            let start = start.min(end);
+            let length = end - start + 1;
+            let reader = source.open(start, length)?;
+            Ok((reader, length as usize, status_code, content_range))
+        }
     }
 }