From 0cf1cc09a9c4ecabbdd62bc951e8f59526ccef32 Mon Sep 17 00:00:00 2001 From: ldm0 Date: Sun, 18 Oct 2026 23:46:49 +0800 Subject: [PATCH] fix(fetch): recover HTML response channels for cached redirects Redirects and retries retain start/body senders in the reused easy handle. Return those channels to the HTML job before publishing a final cache hit, and use the retained easy when an attempt fails before reconfiguration. Cover network-to-cache redirects with discarded interim/body observations, and preserve the original error instead of reporting a closed start channel. --- moli-fetch/src/runtime.rs | 16 +++- .../src/runtime/response_timing_tests.rs | 86 +++++++++++++++++++ 2 files changed, 99 insertions(+), 3 deletions(-) diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index a07e04ff7c..d2d1e4e6ec 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -2978,10 +2978,19 @@ fn complete_streaming_html_job(job: StreamingRuntimeJob, response: Response) { } fn complete_cached_streaming_html_job( - job: StreamingRuntimeJob, + mut job: StreamingRuntimeJob, cached: CachedStreamingResponseLookup, request_cookie_report: Option, ) { + // Redirects and retries keep the response channels in the reused easy. + // Return them to the job before publishing a final cache hit. + if let Some(mut easy) = job.easy.take() { + (job.started_tx, job.body_tx) = easy + .get_mut() + .streaming_mut() + .expect("cached HTML response should use streaming collector") + .take_response_channels(); + } let redirected = !job.request.redirect_chain.is_empty(); let redirect_chain = job.request.redirect_chain.clone(); let CachedStreamingResponseLookup { @@ -3705,10 +3714,11 @@ fn fail_streaming_job(job: StreamingRuntimeJob, error: anyhow::Error) { } fn fail_streaming_job_with_easy( - job: StreamingRuntimeJob, - mut easy: Option>, + mut job: StreamingRuntimeJob, + easy: Option>, error: anyhow::Error, ) { + let mut easy = easy.or_else(|| job.easy.take()); let error = network_fetch_failure_for_request( &job.request, &job.current_url, diff --git a/moli-fetch/src/runtime/response_timing_tests.rs b/moli-fetch/src/runtime/response_timing_tests.rs index 4e60d3e0b5..dd72c1d130 100644 --- a/moli-fetch/src/runtime/response_timing_tests.rs +++ b/moli-fetch/src/runtime/response_timing_tests.rs @@ -403,3 +403,89 @@ fn transfer_progress_preserves_cancellation_and_body_observations_on_reuse() { cancel.cancel(); assert!(!handler.progress(4.0, 4.0, 0.0, 0.0)); } + +#[tokio::test] +async fn html_redirect_to_a_cached_response_preserves_channels_and_timing() -> Result<()> { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = format!("http://{}", listener.local_addr()?); + let cache_dir = std::env::temp_dir().join(format!( + "moli-html-redirect-timing-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH)? + .as_nanos(), + )); + let server = thread::spawn(move || -> std::io::Result<()> { + for response in [ + b"HTTP/1.1 200 OK\r\nCache-Control: max-age=600\r\nContent-Length: 4\r\nConnection: close\r\n\r\nbody".as_slice(), + b"HTTP/1.1 103 Early Hints\r\n\r\nHTTP/1.1 302 Found\r\nLocation: /cache\r\nContent-Length: 4\r\nConnection: close\r\n\r\njunk".as_slice(), + ] { + let (mut stream, _) = listener.accept()?; + read_request(&mut stream)?; + stream.write_all(response)?; + } + Ok(()) + }); + let mut config = FetchConfig::default(); + config.set_http_cache_dir(Some(cache_dir.display().to_string())); + let (owner, _, pending) = + submit_html_request(&config, Request::get(&format!("{address}/cache"))?); + pending.into_response().await?.finish().await?; + + let pending = owner + .handle() + .submit_html_stream(Request::get(&format!("{address}/redirect"))?)?; + let cancel = pending.cancel_handle.clone(); + let mut response = pending.into_response().await?; + assert!(response.from_cache); + assert!(response.redirected); + assert_eq!(response.final_url.path(), "/cache"); + assert_eq!(response.redirect_chain.len(), 1); + assert!(!response.redirect_chain[0].from_cache); + let mut body = String::new(); + while let Some(chunk) = response.next_chunk().await { + body.push_str(&chunk); + } + response.finish().await?; + assert_eq!(body, "body"); + let timing = cancel.response_timing(); + assert!(timing.request_started_at().is_none()); + assert!(timing.first_interim_response_started_at().is_none()); + assert!(!timing.body_bytes_received()); + assert!(timing.response_started_at().unwrap() <= timing.response_end_at().unwrap()); + assert!(!cancel.cancel_if_response_pending()); + server.join().expect("server thread")?; + drop(owner); + std::fs::remove_dir_all(cache_dir)?; + Ok(()) +} + +#[tokio::test] +async fn html_failure_before_reusing_an_easy_preserves_the_original_error() -> Result<()> { + let cancel = FetchCancelHandle::new(); + let (started_tx, started_rx) = oneshot::channel(); + let (body_tx, mut body_rx) = mpsc::unbounded_channel(); + let (completion_tx, completion_rx) = oneshot::channel(); + let mut job = StreamingRuntimeJob::new( + Request::get("http://example.test/cache")?, + started_tx, + body_tx, + completion_tx, + cancel.clone(), + ); + job.easy = Some(Easy2::new(FetchTransferHandler::new_streaming( + StreamingResponseCollector::new( + new_shared_browser_cookie_store(), + job.started_tx.take().unwrap(), + job.body_tx.take().unwrap(), + cancel.clone(), + ), + ))); + fail_streaming_job_with_easy(job, None, anyhow!("cached redirect rejected")); + let error = started_rx.await?.unwrap_err(); + assert!(format!("{error:#}").contains("cached redirect rejected")); + assert!(body_rx.recv().await.is_none()); + assert!(completion_rx.await?.is_err()); + assert!(cancel.response_timing().response_end_at().is_some()); + Ok(()) +}