From a8364ec0e10dcf8eeb5aa38de8e27b3a21a70875 Mon Sep 17 00:00:00 2001 From: Ingo Wolf Date: Tue, 10 Nov 2026 22:27:25 +1100 Subject: [PATCH 1/3] audio: stop decoder hanging on quit when a stream fetch is cancelled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #214. When the core stopped, the downloader task was cancelled but nothing set `closed`, and the resulting `cancelled` fetch error went through `record_fetch_error` as if transient — so a decoder blocked in `HttpSource::read` waited through it forever and shutdown hung. - Add a drop guard in `downloader` that sets `closed` and notifies `data_ready` whenever the task ends, however it ends (error return, task cancellation, runtime shutdown). - Treat `FetchError::Cancelled` as final: the downloader stops instead of retrying, and `read` surfaces the error instead of waiting through it. A cancelled body stream is final too. - Tests: a cancelled fetch surfaces from `read` promptly, and dropping the runtime wakes a decoder blocked on a stalled fetch. Generated by GLM-5.3-Flash in [OpenCode](https://opencode.ai) Co-authored-by: GLM-5.3-Flash --- crates/hocket-core/src/audio/native/http.rs | 162 +++++++++++++++++++- 1 file changed, 161 insertions(+), 1 deletion(-) diff --git a/crates/hocket-core/src/audio/native/http.rs b/crates/hocket-core/src/audio/native/http.rs index 2361f21c..1fcb413f 100644 --- a/crates/hocket-core/src/audio/native/http.rs +++ b/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); + +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, fetcher: Arc, url: String, headers: HashMap, ) { + 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. @@ -523,6 +548,10 @@ async fn downloader( } Some(Err(e)) => { tracing::warn!(target: "hocket::audio::http", error = %e, "body stream error"); + if is_final_fetch_error(&e) { + record_fetch_error(&shared, e); + return; + } break; } None => { @@ -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")); } @@ -615,6 +651,7 @@ impl MediaSource for HttpSource { #[cfg(test)] mod tests { use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; fn runtime() -> tokio::runtime::Runtime { tokio::runtime::Builder::new_multi_thread() @@ -733,6 +770,129 @@ 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 downloader is torn down. + struct CancelFetcher { + data: Arc>, + chunk: usize, + requests: AtomicUsize, + } + + impl RangeFetcher for CancelFetcher { + fn fetch( + &self, + _url: String, + _headers: HashMap, + start: u64, + ) -> FetchFuture { + let n = self.requests.fetch_add(1, Ordering::SeqCst); + let data = self.data.clone(); + let chunk = self.chunk; + Box::pin(async move { + if n > 0 { + 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: AtomicUsize::new(0), + }; + 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}"); + } + + /// Serves the first chunk, then the body stalls forever. + struct StallFetcher { + data: Arc>, + chunk: usize, + } + + impl RangeFetcher for StallFetcher { + fn fetch( + &self, + _url: String, + _headers: HashMap, + 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); + let _ = tx.send((res, out.len())); + }); + // 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, len) = rx + .recv_timeout(Duration::from_secs(5)) + .expect("decoder thread did not exit"); + assert!(res.is_err(), "read should fail, got {len} bytes"); + } + #[test] fn content_range_parsing() { assert_eq!( From b83ff0297916555f0972cf5ef7d8416b77fe244c Mon Sep 17 00:00:00 2001 From: Ingo Wolf Date: Tue, 10 Nov 2026 22:50:41 +1100 Subject: [PATCH 2/3] audio: strengthen cancellation test assertions from review - CancelFetcher reuses the Arc>> request log and the test pins that the downloader stops after the first cancelled body (exactly one request, no retries). - The runtime-shutdown test asserts the read error is "source closed" rather than just any error. Generated by GLM-5.3-Flash in [OpenCode](https://opencode.ai) Co-authored-by: GLM-5.3-Flash --- crates/hocket-core/src/audio/native/http.rs | 26 ++++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/crates/hocket-core/src/audio/native/http.rs b/crates/hocket-core/src/audio/native/http.rs index 1fcb413f..40abdacd 100644 --- a/crates/hocket-core/src/audio/native/http.rs +++ b/crates/hocket-core/src/audio/native/http.rs @@ -651,7 +651,6 @@ impl MediaSource for HttpSource { #[cfg(test)] mod tests { use super::*; - use std::sync::atomic::{AtomicUsize, Ordering}; fn runtime() -> tokio::runtime::Runtime { tokio::runtime::Builder::new_multi_thread() @@ -775,7 +774,7 @@ mod tests { struct CancelFetcher { data: Arc>, chunk: usize, - requests: AtomicUsize, + requests: Arc>>, } impl RangeFetcher for CancelFetcher { @@ -785,11 +784,12 @@ mod tests { _headers: HashMap, start: u64, ) -> FetchFuture { - let n = self.requests.fetch_add(1, Ordering::SeqCst); let data = self.data.clone(); let chunk = self.chunk; + let requests = self.requests.clone(); + self.requests.lock().push(start); Box::pin(async move { - if n > 0 { + if requests.lock().len() > 1 { return Err(FetchError::Cancelled); } let end = (start as usize + chunk).min(data.len()); @@ -813,8 +813,9 @@ mod tests { let fetcher = CancelFetcher { data: Arc::new(bytes), chunk: 4096, - requests: AtomicUsize::new(0), + requests: Arc::new(Mutex::new(Vec::new())), }; + let requests = fetcher.requests.clone(); let mut src = HttpSource::open( Arc::new(fetcher), rt.handle().clone(), @@ -829,6 +830,9 @@ mod tests { let mut next = vec![0u8; 16]; let err = src.read(&mut next).unwrap_err(); assert!(err.to_string().contains("cancelled"), "{err}"); + // And the downloader stopped instead of retrying: only the first + // fetch (whose body was cancelled) was made. + assert_eq!(requests.lock().len(), 1, "{:?}", requests.lock()); } /// Serves the first chunk, then the body stalls forever. @@ -880,17 +884,21 @@ mod tests { let (tx, rx) = std::sync::mpsc::channel(); std::thread::spawn(move || { let mut out = Vec::new(); - let res = src.read_to_end(&mut out); - let _ = tx.send((res, out.len())); + 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, len) = rx + let res = rx .recv_timeout(Duration::from_secs(5)) .expect("decoder thread did not exit"); - assert!(res.is_err(), "read should fail, got {len} bytes"); + let err = res.unwrap_err(); + assert!( + err.to_string().contains("source closed"), + "unexpected error: {err}" + ); } #[test] From 310c30ef19f0fc0171051a2fcd960bbef432dca9 Mon Sep 17 00:00:00 2001 From: Ingo Wolf Date: Wed, 11 Nov 2026 11:15:16 +1100 Subject: [PATCH 3/3] http: keep a cancelled body retryable The stream reader also cancels handles it evicts as idle (after a long pause), and refetching from the gap recovers from that. A cancelled fetch stays final, so a stopped reader still ends the downloader after one refetch. Co-Authored-By: Claude Opus 5.5 --- crates/hocket-core/src/audio/native/http.rs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/crates/hocket-core/src/audio/native/http.rs b/crates/hocket-core/src/audio/native/http.rs index 40abdacd..31715570 100644 --- a/crates/hocket-core/src/audio/native/http.rs +++ b/crates/hocket-core/src/audio/native/http.rs @@ -547,11 +547,11 @@ 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"); - if is_final_fetch_error(&e) { - record_fetch_error(&shared, e); - return; - } break; } None => { @@ -770,7 +770,7 @@ mod tests { } /// Serves one chunk, then the body fails as cancelled and every further - /// fetch returns `Cancelled` too — as when the downloader is torn down. + /// fetch returns `Cancelled` too — as when the stream reader stops. struct CancelFetcher { data: Arc>, chunk: usize, @@ -830,9 +830,10 @@ mod tests { let mut next = vec![0u8; 16]; let err = src.read(&mut next).unwrap_err(); assert!(err.to_string().contains("cancelled"), "{err}"); - // And the downloader stopped instead of retrying: only the first - // fetch (whose body was cancelled) was made. - assert_eq!(requests.lock().len(), 1, "{:?}", requests.lock()); + // 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.