Desktop: fix quit hang when a stream fetch is cancelled

crates/hocket-core/src/audio/native/http.rs

@@ -475,12 +475,33 @@ fn is_closed(shared: &Shared) -> bool {
     shared.state.lock().closed
 }
 
+/// Marks the state closed and wakes readers when the downloader task ends,
+/// however it ends: completion, an error return, task cancellation (engine
+/// stop) or runtime shutdown. Without this, a decoder blocked in `read`
+/// would wait forever once the downloader is gone.
+struct CloseGuard(Arc<Shared>);
+
+impl Drop for CloseGuard {
+    fn drop(&mut self) {
+        self.0.state.lock().closed = true;
+        self.0.data_ready.notify_all();
+    }
+}
+
+/// Errors that can never succeed on retry: the fetch was cancelled because
+/// the downloader or runtime is being torn down. Waiting through them is
+/// pointless — the downloader is never coming back.
+fn is_final_fetch_error(e: &FetchError) -> bool {
+    matches!(e, FetchError::Cancelled)
+}
+
 async fn downloader(
     shared: Arc<Shared>,
     fetcher: Arc<dyn RangeFetcher>,
     url: String,
     headers: HashMap<String, String>,
 ) {
+    let _close_on_exit = CloseGuard(shared.clone());
     loop {
         let start = match plan_next(&shared) {
             Next::Fetch(s) => s,
@@ -490,7 +511,11 @@ async fn downloader(
             Ok(r) => r,
             Err(e) => {
                 tracing::warn!(target: "hocket::audio::http", error = %e, "fetch failed");
-                if record_fetch_error(&shared, e) {
+                let final_error = is_final_fetch_error(&e);
+                if record_fetch_error(&shared, e) || final_error {
+                    // Either `open` reports the error itself, or retrying is
+                    // pointless: the exit guard marks the state closed and
+                    // wakes any reader waiting on us.
                     return;
                 }
                 // Mid-stream failure: back off and retry from the same gap.
@@ -522,6 +547,10 @@ async fn downloader(
                     pos += bytes.len() as u64;
                 }
                 Some(Err(e)) => {
+                    // Not final even when cancelled: the stream reader also
+                    // cancels handles it evicts as idle (a long pause), and
+                    // a fresh fetch recovers from that. After a real stop the
+                    // refetch itself fails as cancelled and ends us.
                     tracing::warn!(target: "hocket::audio::http", error = %e, "body stream error");
                     break;
                 }
@@ -553,6 +582,13 @@ impl Read for HttpSource {
                     return Ok(0);
                 }
             }
+            if let Some(e) = &st.error {
+                if is_final_fetch_error(e) {
+                    // The fetch was cancelled (engine stop, runtime shutdown)
+                    // and the downloader is gone: surface instead of waiting.
+                    return Err(io::Error::other(e.to_string()));
+                }
+            }
             if st.closed {
                 return Err(io::Error::new(io::ErrorKind::BrokenPipe, "source closed"));
             }
@@ -733,6 +769,139 @@ mod tests {
         assert!(err.to_string().contains("403"), "{err}");
     }
 
+    /// Serves one chunk, then the body fails as cancelled and every further
+    /// fetch returns `Cancelled` too — as when the stream reader stops.
+    struct CancelFetcher {
+        data: Arc<Vec<u8>>,
+        chunk: usize,
+        requests: Arc<Mutex<Vec<u64>>>,
+    }
+
+    impl RangeFetcher for CancelFetcher {
+        fn fetch(
+            &self,
+            _url: String,
+            _headers: HashMap<String, String>,
+            start: u64,
+        ) -> FetchFuture {
+            let data = self.data.clone();
+            let chunk = self.chunk;
+            let requests = self.requests.clone();
+            self.requests.lock().push(start);
+            Box::pin(async move {
+                if requests.lock().len() > 1 {
+                    return Err(FetchError::Cancelled);
+                }
+                let end = (start as usize + chunk).min(data.len());
+                let first = Bytes::copy_from_slice(&data[start as usize..end]);
+                let stream = futures::stream::once(async move { Ok(first) })
+                    .chain(futures::stream::once(async { Err(FetchError::Cancelled) }));
+                Ok(FetchResponse {
+                    range_start: start,
+                    total_len: Some(data.len() as u64),
+                    supports_ranges: true,
+                    body: Box::pin(stream),
+                })
+            })
+        }
+    }
+
+    #[test]
+    fn cancelled_fetch_surfaces_from_read_instead_of_waiting() {
+        let rt = runtime();
+        let bytes = data(100_000);
+        let fetcher = CancelFetcher {
+            data: Arc::new(bytes),
+            chunk: 4096,
+            requests: Arc::new(Mutex::new(Vec::new())),
+        };
+        let requests = fetcher.requests.clone();
+        let mut src = HttpSource::open(
+            Arc::new(fetcher),
+            rt.handle().clone(),
+            "http://x/a".into(),
+            HashMap::new(),
+            Duration::from_secs(5),
+        )
+        .unwrap();
+        let mut first = vec![0u8; 4096];
+        src.read_exact(&mut first).unwrap();
+        // The body was cancelled after the first chunk: this must not hang.
+        let mut next = vec![0u8; 16];
+        let err = src.read(&mut next).unwrap_err();
+        assert!(err.to_string().contains("cancelled"), "{err}");
+        // A cancelled body is retried once (an idle-evicted handle recovers
+        // that way); the refetch failing as cancelled ends the downloader.
+        let made = requests.lock().clone();
+        assert_eq!(made.len(), 2, "{made:?}");
+    }
+
+    /// Serves the first chunk, then the body stalls forever.
+    struct StallFetcher {
+        data: Arc<Vec<u8>>,
+        chunk: usize,
+    }
+
+    impl RangeFetcher for StallFetcher {
+        fn fetch(
+            &self,
+            _url: String,
+            _headers: HashMap<String, String>,
+            start: u64,
+        ) -> FetchFuture {
+            let data = self.data.clone();
+            let chunk = self.chunk;
+            Box::pin(async move {
+                let end = (start as usize + chunk).min(data.len());
+                let first = Bytes::copy_from_slice(&data[start as usize..end]);
+                let stream = futures::stream::once(async move { Ok(first) })
+                    .chain(futures::stream::pending());
+                Ok(FetchResponse {
+                    range_start: start,
+                    total_len: Some(data.len() as u64),
+                    supports_ranges: true,
+                    body: Box::pin(stream),
+                })
+            })
+        }
+    }
+
+    #[test]
+    fn runtime_shutdown_wakes_a_blocked_decoder() {
+        let rt = runtime();
+        let bytes = data(100_000);
+        let fetcher = StallFetcher {
+            data: Arc::new(bytes),
+            chunk: 4096,
+        };
+        let mut src = HttpSource::open(
+            Arc::new(fetcher),
+            rt.handle().clone(),
+            "http://x/a".into(),
+            HashMap::new(),
+            Duration::from_secs(5),
+        )
+        .unwrap();
+        let (tx, rx) = std::sync::mpsc::channel();
+        std::thread::spawn(move || {
+            let mut out = Vec::new();
+            let res = src.read_to_end(&mut out).map(|_| out.len());
+            let _ = tx.send(res);
+        });
+        // Let the decoder consume the first chunk and block on the stall.
+        std::thread::sleep(Duration::from_millis(200));
+        // Engine stop / runtime shutdown cancels the downloader mid-fetch.
+        drop(rt);
+        let res = rx
+            .recv_timeout(Duration::from_secs(5))
+            .expect("decoder thread did not exit");
+        let err = res.unwrap_err();
+        assert!(
+            err.to_string().contains("source closed"),
+            "unexpected error: {err}"
+        );
+    }
+
     #[test]
     fn content_range_parsing() {
         assert_eq!(