diff --git a/moli-protocol/src/conn/browser_context/fetch_owner.rs b/moli-protocol/src/conn/browser_context/fetch_owner.rs index a778fa8a2..998dac422 100644 --- a/moli-protocol/src/conn/browser_context/fetch_owner.rs +++ b/moli-protocol/src/conn/browser_context/fetch_owner.rs @@ -63,7 +63,7 @@ impl SessionFetchBodyStreamOwner<'_> { fn open_pending_fetch_response_body_stream( &mut self, request_id: &str, - ) -> Result, String> { + ) -> anyhow::Result> { let handle = target_scoped_stream_handle( &self.owner_key, self.runtime_slot.allocate_io_stream_handle(), @@ -611,7 +611,7 @@ impl CdpConnection { &mut self, session_id: Option<&str>, request_id: &str, - ) -> Result, String> { + ) -> anyhow::Result> { let Some(mut owner) = self.target_session_owner_mut(session_id) else { return Ok(None); }; @@ -1516,7 +1516,7 @@ impl TargetSessionOwnerMut<'_> { fn open_pending_fetch_response_body_stream( &mut self, request_id: &str, - ) -> Result, String> { + ) -> anyhow::Result> { let Some(mut owner) = self.fetch_body_stream_owner_mut() else { return Ok(None); }; diff --git a/moli-protocol/src/conn/fetch_support.rs b/moli-protocol/src/conn/fetch_support.rs index b2f7c4143..584f2c478 100644 --- a/moli-protocol/src/conn/fetch_support.rs +++ b/moli-protocol/src/conn/fetch_support.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use std::collections::HashMap; use std::str::FromStr; use std::sync::Arc; @@ -692,11 +693,11 @@ pub struct PendingFetchResponseOpenedBodyStream { pub transfer: PausedDocumentTransfer, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug)] pub(crate) enum PendingFetchResponseBodyStreamRead { NotFound, Read { bytes: Vec, eof: bool }, - Failed(String), + Failed(anyhow::Error), } #[derive(Debug)] @@ -718,8 +719,10 @@ pub(crate) struct PendingFetchResponseBodyStreamReadDispatch { pub(crate) struct CompletedFetchResponseBodyStreamReadDispatch { request_id: String, handle: String, - completed: - Result<(Vec, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, String)>>, + completed: Result< + (Vec, bool, PausedDocumentTransfer), + Box<(PausedDocumentTransfer, anyhow::Error)>, + >, } impl PendingFetchResponseBodyStreamReadDispatch { @@ -766,7 +769,7 @@ impl CompletedFetchResponseBodyStreamReadDispatch { pub(crate) fn into_completed( self, - ) -> Result<(Vec, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, String)>> + ) -> Result<(Vec, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, anyhow::Error)>> { self.completed } @@ -777,7 +780,7 @@ pub(crate) enum OpenBodyStreamError { NotOpenable(Box), Failed { transfer: Box, - message: String, + error: anyhow::Error, }, } @@ -953,9 +956,7 @@ impl PausedDocumentTransfer { }, }, }), - message: format!( - "failed to materialize captured response body: {error}" - ), + error: error.context("failed to materialize captured response body"), }); } }; @@ -1049,7 +1050,7 @@ impl PausedDocumentTransfer { pub(crate) async fn read_body_stream_async( self, size: Option, - ) -> Result<(Vec, bool, Self), (Self, String)> { + ) -> Result<(Vec, bool, Self), (Self, anyhow::Error)> { let Self { fetch_request_id, state, @@ -1060,7 +1061,7 @@ impl PausedDocumentTransfer { fetch_request_id, state, }, - "StreamHandleNotFound".to_owned(), + anyhow::anyhow!("StreamHandleNotFound"), )); }; match stream.read_async(size).await { @@ -1111,7 +1112,7 @@ impl PausedDocumentTransfer { pub(crate) async fn materialize_body_limited_async( self, limit: usize, - ) -> Result<(Option>, Self), (String, Self)> { + ) -> Result<(Option>, Self), (anyhow::Error, Self)> { let Self { fetch_request_id, state, @@ -1161,7 +1162,7 @@ impl PausedDocumentTransfer { ( Option, NavigationDispatchState, - Result, + anyhow::Result, ), Self, > { @@ -1193,7 +1194,7 @@ impl PausedDocumentTransfer { ) -> ( Option, NavigationDispatchState, - Result, + anyhow::Result, ) { match self.state { PausedDocumentTransferState::Pending { @@ -1232,7 +1233,7 @@ impl PausedDocumentTransfer { ) -> ( Option, NavigationDispatchState, - Result, + anyhow::Result, ) { match self.state { PausedDocumentTransferState::Pending { @@ -1241,7 +1242,11 @@ impl PausedDocumentTransfer { body, } => { let _ = body; - (document_navigation_token, navigation, Err(error_text)) + ( + document_navigation_token, + navigation, + Err(anyhow::Error::msg(error_text)), + ) } PausedDocumentTransferState::ActiveBodyStream { stream, .. } => stream.fail(error_text), } @@ -1279,7 +1284,7 @@ impl ActiveDocumentBodyStreamState { self.offset } - async fn read_async(&mut self, size: Option) -> Result<(Vec, bool), String> { + async fn read_async(&mut self, size: Option) -> anyhow::Result<(Vec, bool)> { read_active_body_stream_async( &mut self.response, &mut self.captured_body, @@ -1291,7 +1296,7 @@ impl ActiveDocumentBodyStreamState { .await } - fn finish_pending_body_source(&mut self) -> Result { + fn finish_pending_body_source(&mut self) -> anyhow::Result { let head = ResponseHead { final_url: self.response.final_url.clone(), status: self.response.status, @@ -1306,7 +1311,7 @@ impl ActiveDocumentBodyStreamState { let body = self .captured_body .finish_in_place() - .map_err(|error| format!("failed to finish captured response body: {error}"))?; + .context("failed to finish captured response body")?; Ok(DocumentBodySource::CapturedRaw { requested_url: self.requested_url.clone(), request_method: self.request_method.clone(), @@ -1327,7 +1332,7 @@ impl ActiveDocumentBodyStreamState { ) -> ( Option, NavigationDispatchState, - Result, + anyhow::Result, ) { let navigation_state = self.navigation.clone(); let final_url = self.response.final_url.clone(); @@ -1352,12 +1357,12 @@ impl ActiveDocumentBodyStreamState { ) -> ( Option, NavigationDispatchState, - Result, + anyhow::Result, ) { ( self.document_navigation_token, self.navigation, - Err(error_text), + Err(anyhow::Error::msg(error_text)), ) } } @@ -1398,7 +1403,7 @@ async fn read_active_body_stream_async( offset: &mut usize, finished: &mut bool, size: Option, -) -> Result<(Vec, bool), String> { +) -> anyhow::Result<(Vec, bool)> { let mut bytes = Vec::new(); match size { Some(limit) => { @@ -1455,21 +1460,21 @@ async fn read_next_active_body_stream_chunk_async( captured_body: &mut CapturedBodyWriter, unread_body: &mut Vec, finished: &mut bool, -) -> Result<(), String> { +) -> anyhow::Result<()> { if *finished { return Ok(()); } if let Some(chunk) = response.next_chunk().await { captured_body .append(&chunk) - .map_err(|error| format!("failed to capture response body stream: {error}"))?; + .context("failed to capture response body stream")?; unread_body.extend(chunk); return Ok(()); } response .finish() .await - .map_err(|error| format!("failed to read page body from stream: {error}"))?; + .context("failed to read page body from stream")?; *finished = true; Ok(()) } @@ -1511,7 +1516,7 @@ impl DocumentBodySource { navigation: &NavigationDispatchState, response_code: Option, response_headers: Vec<(String, String)>, - ) -> Result { + ) -> anyhow::Result { let has_response_override = response_code.is_some() || !response_headers.is_empty(); match self { Self::BufferedRaw { @@ -1621,7 +1626,7 @@ impl DocumentBodySource { pub(crate) async fn materialize_body_limited_async( self, limit: usize, - ) -> Result<(Vec, Self), (String, Self)> { + ) -> Result<(Vec, Self), (anyhow::Error, Self)> { match self { Self::BufferedRaw { requested_url, @@ -1632,7 +1637,7 @@ impl DocumentBodySource { } => { if let Err(error) = ensure_materialize_limit(response.body_bytes().len(), limit) { return Err(( - error.to_string(), + error, Self::BufferedRaw { requested_url, request_method, @@ -1681,9 +1686,9 @@ impl DocumentBodySource { )); } }; - let result = body.materialize_bytes_limited(limit).map_err(|error| { - format!("failed to materialize captured response body: {error}") - }); + let result = body + .materialize_bytes_limited(limit) + .context("failed to materialize captured response body"); let source = Self::CapturedRaw { requested_url, request_method, @@ -1707,9 +1712,9 @@ impl DocumentBodySource { network_observation_journal, body_progress_source, } => { - let result = body.materialize_bytes_limited(limit).map_err(|error| { - format!("failed to materialize captured response body: {error}") - }); + let result = body + .materialize_bytes_limited(limit) + .context("failed to materialize captured response body"); let source = Self::CapturedRaw { requested_url, request_method, @@ -1730,20 +1735,20 @@ impl DocumentBodySource { async fn capture_streaming_raw_response( mut response: StreamingRawResponse, -) -> Result<(ResponseHead, CapturedBody), String> { +) -> anyhow::Result<(ResponseHead, CapturedBody)> { let head = response.head(); let mut body = CapturedBodyWriter::default(); while let Some(chunk) = response.next_chunk().await { body.append(&chunk) - .map_err(|error| format!("failed to capture response body stream: {error}"))?; + .context("failed to capture response body stream")?; } response .finish() .await - .map_err(|error| format!("failed to read page body from stream: {error}"))?; + .context("failed to read page body from stream")?; let body = body .finish() - .map_err(|error| format!("failed to finish captured response body: {error}"))?; + .context("failed to finish captured response body")?; Ok((head, body)) } diff --git a/moli-protocol/src/conn/page_state/fetch_state.rs b/moli-protocol/src/conn/page_state/fetch_state.rs index 4c70c5507..a228496ae 100644 --- a/moli-protocol/src/conn/page_state/fetch_state.rs +++ b/moli-protocol/src/conn/page_state/fetch_state.rs @@ -76,7 +76,7 @@ impl BrowserContext { pub(crate) fn open_pending_fetch_response_body_stream( &mut self, request_id: &str, - ) -> Result, String> { + ) -> anyhow::Result> { let handle = self .active_page_target_mut() .runtime_slot diff --git a/moli-protocol/src/conn/resource_runtime_support.rs b/moli-protocol/src/conn/resource_runtime_support.rs index cc9a1381c..6cc0ddcc4 100644 --- a/moli-protocol/src/conn/resource_runtime_support.rs +++ b/moli-protocol/src/conn/resource_runtime_support.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use moli_core::network::{ BrowserResourceRuntime, BrowserResourceRuntimeOwner, ResourceRequestClient, }; @@ -44,17 +45,17 @@ impl CdpConnection { pub(crate) fn ensure_resource_request_client_for_navigation_load_inputs( &mut self, load_inputs: &TargetNavigationLoadInputs, - ) -> Result { + ) -> anyhow::Result { let storage = load_inputs.resource_storage_handles(); let engine = self .configured_navigation_engine_for_load_inputs_mut(load_inputs) - .ok_or_else(|| "navigation Page engine unavailable".to_owned())?; + .context("navigation Page engine unavailable")?; engine .ensure_resource_runtime_ready_for_navigation_storage(storage.into_navigation_storage()) - .map_err(|error| format!("failed to initialize resource runtime: {error}"))?; + .context("failed to initialize resource runtime")?; engine .resource_request_client() - .ok_or_else(|| "resource request client unavailable".to_owned()) + .context("resource request client unavailable") } pub(super) fn configured_navigation_engine_for_load_inputs_mut( diff --git a/moli-protocol/src/conn/runtime_load.rs b/moli-protocol/src/conn/runtime_load.rs index e2a98f55a..61039a4d0 100644 --- a/moli-protocol/src/conn/runtime_load.rs +++ b/moli-protocol/src/conn/runtime_load.rs @@ -95,7 +95,7 @@ async fn prepare_browser_owned_error_page_navigation_with_engine_async( error_text: String, body: CapturedBody, reply_boundary: RendererReplyBoundary, -) -> Result { +) -> anyhow::Result { let error_page_url = Url::parse(NETWORK_ERROR_PAGE_URL) .expect("the browser-owned network error page URL must be valid"); let error_page = NetworkErrorPageNavigation::new(error_text, unreachable_url.clone()); @@ -141,7 +141,7 @@ async fn prepare_network_error_page_navigation_with_engine_async( request_headers: Vec<(String, String)>, error_text: String, reply_boundary: RendererReplyBoundary, -) -> Result { +) -> anyhow::Result { let body = CapturedBody::from_string(network_error_page_html(&unreachable_url, &error_text)); prepare_browser_owned_error_page_navigation_with_engine_async( engine, @@ -309,7 +309,7 @@ pub struct ResponseCommitReady { } enum ResponseCommitBodyCapture { - Pending(tokio::task::JoinHandle>), + Pending(tokio::task::JoinHandle>), Ready(CapturedBody), } @@ -320,11 +320,11 @@ impl ResponseCommitBodyCapture { } } - async fn resolve(self) -> Result { + async fn resolve(self) -> anyhow::Result { match self { Self::Pending(task) => task .await - .map_err(|error| format!("main document body capture task failed: {error}"))?, + .context("main document body capture task failed")?, Self::Ready(body) => Ok(body), } } @@ -375,15 +375,15 @@ impl ResponseCommitReady { pub(crate) async fn update_commit_configuration( &self, configuration: PreparedDocumentPageCommitConfiguration, - ) -> Result<(), String> { + ) -> anyhow::Result<()> { self.prepared_page .as_ref() .expect("response commit-ready value must retain its prepared Page") .update_commit_configuration(configuration) .await - .map_err(|error| { + .with_context(|| { format!( - "failed to attach commit-time target configuration for page `{}`: {error:#}", + "failed to attach commit-time target configuration for page `{}`", self.requested_url ) }) @@ -399,7 +399,7 @@ impl ResponseCommitReady { pub(crate) async fn commit( mut self, permit: PreparedDocumentPageCommitPermit, - ) -> Result { + ) -> anyhow::Result { let prepared_page = self .prepared_page .take() @@ -410,10 +410,10 @@ impl ResponseCommitReady { if let Some(body_capture) = self.body_capture.take() { body_capture.abort(); } - return Err(format!( - "failed to execute scripts for page `{}`: {error:#}", + return Err(error.context(format!( + "failed to execute scripts for page `{}`", self.requested_url - )); + ))); } }; let body_capture = self @@ -454,9 +454,9 @@ impl ResponseCommitReady { }); CompletedDocumentProgressTransfer::new_pending_body(body_network_progress_state) } else { - let captured_body = body_capture.resolve().await.map_err(|error| { + let captured_body = body_capture.resolve().await.with_context(|| { format!( - "failed to execute scripts for page `{}`: {error}", + "failed to execute scripts for page `{}`", self.requested_url ) })?; @@ -588,7 +588,7 @@ impl PausedResponsePreparedDocument { async fn first_nonempty_response_body_chunk( response: &mut StreamingRawResponse, -) -> Result>, String> { +) -> anyhow::Result>> { loop { match response.next_chunk().await { Some(chunk) if chunk.is_empty() => continue, @@ -597,7 +597,7 @@ async fn first_nonempty_response_body_chunk( response .finish() .await - .map_err(|error| format!("failed to read page body from stream: {error:#}"))?; + .context("failed to read page body from stream")?; return Ok(None); } } @@ -609,40 +609,37 @@ fn spawn_streaming_body_capture( initial_chunk: Option>, body_tx: mpsc::Sender>, completion_tx: oneshot::Sender>, -) -> tokio::task::JoinHandle> { +) -> tokio::task::JoinHandle> { tokio::spawn(async move { - let mut body = CapturedBodyWriter::default(); + // Publish completion before closing the parser's body channel, as the + // transport-backed capture did before sharing errors with both owners. let mut renderer_body_tx = Some(body_tx); - if let Some(chunk) = initial_chunk { - body.append(&chunk) - .map_err(|error| format!("failed to capture page body: {error}"))?; - if let Some(body_tx) = renderer_body_tx.as_ref() - && body_tx.send(chunk).await.is_err() - { - renderer_body_tx = None; + let result = async { + let mut body = CapturedBodyWriter::default(); + if let Some(chunk) = initial_chunk { + body.append(&chunk).context("failed to capture page body")?; + if let Some(body_tx) = renderer_body_tx.as_ref() + && body_tx.send(chunk).await.is_err() + { + renderer_body_tx = None; + } } - } - while let Some(chunk) = response.next_chunk().await { - body.append(&chunk) - .map_err(|error| format!("failed to capture page body: {error}"))?; - if let Some(body_tx) = renderer_body_tx.as_ref() - && body_tx.send(chunk).await.is_err() - { - renderer_body_tx = None; + while let Some(chunk) = response.next_chunk().await { + body.append(&chunk).context("failed to capture page body")?; + if let Some(body_tx) = renderer_body_tx.as_ref() + && body_tx.send(chunk).await.is_err() + { + renderer_body_tx = None; + } } + response + .finish() + .await + .context("failed to read page body from stream")?; + body.finish().context("failed to finish captured page body") } - let finish_result = response - .finish() - .await - .map_err(|error| format!("failed to read page body from stream: {error:#}")); - let completion_result = finish_result - .as_ref() - .map(|_| ()) - .map_err(|error| anyhow::anyhow!(error.clone())); - let _ = completion_tx.send(completion_result); - finish_result?; - body.finish() - .map_err(|error| format!("failed to finish captured page body: {error}")) + .await; + complete_body_capture(result, completion_tx) }) } @@ -650,33 +647,64 @@ fn spawn_captured_body_replay( body: CapturedBody, body_tx: mpsc::Sender>, completion_tx: oneshot::Sender>, -) -> tokio::task::JoinHandle> { +) -> tokio::task::JoinHandle> { tokio::spawn(async move { let replay_result = async { let mut reader = body .chunk_reader(CAPTURED_RAW_REPLAY_CHUNK_SIZE) - .map_err(|error| error.to_string())?; + .context("failed to open captured page body")?; let mut renderer_body_tx = Some(body_tx); - while let Some(chunk) = reader.next_chunk().map_err(|error| error.to_string())? { + while let Some(chunk) = reader + .next_chunk() + .context("failed to replay captured page body")? + { if let Some(body_tx) = renderer_body_tx.as_ref() && body_tx.send(chunk).await.is_err() { renderer_body_tx = None; } } - Ok::<(), String>(()) + Ok(body) } .await; - let completion_result = replay_result - .as_ref() - .map(|_| ()) - .map_err(|error| anyhow::anyhow!(error.clone())); - let _ = completion_tx.send(completion_result); - replay_result?; - Ok(body) + complete_body_capture(replay_result, completion_tx) }) } +/// Both the renderer and the navigation task must observe the same failure. +/// Keep its source chain alive instead of cloning a formatted diagnostic. +#[derive(Clone, Debug)] +struct SharedBodyCaptureError(Arc); + +impl std::fmt::Display for SharedBodyCaptureError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("main document body transfer failed") + } +} + +impl std::error::Error for SharedBodyCaptureError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.0.as_ref().as_ref()) + } +} + +fn complete_body_capture( + result: anyhow::Result, + completion_tx: oneshot::Sender>, +) -> anyhow::Result { + match result { + Ok(body) => { + let _ = completion_tx.send(Ok(())); + Ok(body) + } + Err(error) => { + let error = SharedBodyCaptureError(Arc::new(error)); + let _ = completion_tx.send(Err(error.clone().into())); + Err(error.into()) + } + } +} + pub(crate) struct BackgroundNavigationLoadJob { engine: NavigationEngine, page_reservation: RendererPageReservationToken, @@ -742,7 +770,7 @@ impl BackgroundNavigationBodyCompletionSink { fn send( self, - body: Result, + body: anyhow::Result, body_progress_source: MainDocumentBodyProgressSource, final_url: Url, response_headers: Vec<(String, String)>, @@ -793,7 +821,7 @@ impl BackgroundNavigationEarlyResult { impl BackgroundNavigationLoadJob { fn emit_early_result_for_successful_document( early_result: &mut Option, - navigation: &Result, + navigation: &anyhow::Result, ) -> bool { let is_successful_document = match navigation { Ok(NavigationLoadOutcome::ResponseCommitReady(navigation)) => { @@ -815,7 +843,7 @@ impl BackgroundNavigationLoadJob { pub(crate) async fn run( mut self, body_completion_sink: Option, - ) -> (Result, bool) { + ) -> (anyhow::Result, bool) { let timing_started = moli_trace::cdp_nav_timing_enabled().then(std::time::Instant::now); let timing_url = self.raw_url.clone(); let mut engine = self.engine; @@ -823,7 +851,7 @@ impl BackgroundNavigationLoadJob { if let Some(resource_runtime) = self.shared_resource_runtime.take() && let Err(error) = engine.adopt_registered_resource_runtime(resource_runtime) { - return (Err(error.to_string()), false); + return (Err(error), false); } let mut early_result_sent = false; let navigation = async { @@ -866,9 +894,8 @@ impl BackgroundNavigationLoadJob { // browser-owned error page so Page.navigate resolves with an // `errorText` and the target stays responsive (Chromium // reports net::ERR_INTERNET_DISCONNECTED here). - let requested_url = Url::parse(&self.raw_url).map_err(|error| { - format!("failed to parse request url `{}`: {error}", self.raw_url) - })?; + let requested_url = Url::parse(&self.raw_url) + .with_context(|| format!("failed to parse request url `{}`", self.raw_url))?; tracing::debug!( url = %self.raw_url, network_error_text = NET_ERR_INTERNET_DISCONNECTED_ERROR_TEXT, @@ -887,9 +914,8 @@ impl BackgroundNavigationLoadJob { .await; } - let requested_url = Url::parse(&self.raw_url).map_err(|error| { - format!("failed to parse request url `{}`: {error}", self.raw_url) - })?; + let requested_url = Url::parse(&self.raw_url) + .with_context(|| format!("failed to parse request url `{}`", self.raw_url))?; let resource_storage = self.load_inputs.resource_storage_handles(); let navigation_response = engine .fetch_navigation_streaming_raw_response_bytes_with_storage_async( @@ -947,7 +973,7 @@ impl BackgroundNavigationLoadJob { ) .await; } - return Err(format!("failed to fetch page `{}`: {error}", self.raw_url)); + return Err(error.context(format!("failed to fetch page `{}`", self.raw_url))); } }; let (response, network_observation_journal) = navigation_response @@ -1004,12 +1030,12 @@ impl BackgroundStreamingResponseNavigationLoadJob { pub(crate) async fn run( mut self, body_completion_sink: Option, - ) -> Result { + ) -> anyhow::Result { let mut engine = self.engine; if let Some(resource_runtime) = self.shared_resource_runtime.take() && let Err(error) = engine.adopt_registered_resource_runtime(resource_runtime) { - return Err(error.to_string()); + return Err(error); } build_navigation_from_streaming_raw_response_with_engine_async( &mut engine, @@ -1033,7 +1059,7 @@ impl BackgroundStreamingResponseNavigationLoadJob { } #[cfg(test)] -pub(crate) fn decode_data_url_body(raw_url: &str) -> Option, String>> { +pub(crate) fn decode_data_url_body(raw_url: &str) -> Option>> { decode_data_url_response(raw_url).map(|result| result.map(|response| response.body)) } @@ -1055,24 +1081,23 @@ struct InlineHtmlNavigationSource { pub(crate) fn decode_data_url_response( raw_url: &str, -) -> Option> { +) -> Option> { let data_url = DataUrl::process(raw_url).ok()?; let content_type = data_url.mime_type().to_string(); Some( data_url .decode_to_vec() .map(|(body, _fragment)| DecodedDataUrlResponse { content_type, body }) - .map_err(|_| "failed to decode data url body".to_owned()), + .context("failed to decode data url body"), ) } fn decoded_data_url_navigation_response( raw_url: &str, -) -> Option> { +) -> Option> { let decoded = decode_data_url_response(raw_url)?; Some(decoded.and_then(|decoded| { - let requested_url = - Url::parse(raw_url).map_err(|error| format!("failed to parse data url: {error}"))?; + let requested_url = Url::parse(raw_url).context("failed to parse data url")?; let response = RawResponse::from_head_and_body( ResponseHead { final_url: requested_url.clone(), @@ -1094,7 +1119,7 @@ fn decoded_data_url_navigation_response( })) } -pub(crate) fn decode_text_html_data_url(raw_url: &str) -> Option> { +pub(crate) fn decode_text_html_data_url(raw_url: &str) -> Option> { if let Some(payload) = raw_url.strip_prefix("data:text/html,") && payload.contains('#') { @@ -1110,16 +1135,16 @@ pub(crate) fn decode_text_html_data_url(raw_url: &str) -> Option Option> { +) -> Option> { if raw_url == "about:blank" { return Some( Url::parse(raw_url) @@ -1128,7 +1153,7 @@ fn inline_html_navigation_source( html: ABOUT_BLANK_DOCUMENT_HTML.to_owned(), response_headers: vec![("content-type".into(), "text/html".into())], }) - .map_err(|error| format!("failed to parse about:blank url: {error}")), + .context("failed to parse about:blank url"), ); } @@ -1144,7 +1169,7 @@ fn inline_html_navigation_source( html, response_headers: vec![("Content-Type".into(), content_type)], }) - .map_err(|error| format!("failed to parse data url: {error}")) + .context("failed to parse data url") })) } @@ -1156,7 +1181,7 @@ async fn load_inline_html_navigation_with_engine_async( raw_url: &str, request_headers: Vec<(String, String)>, reply_boundary: RendererReplyBoundary, -) -> Option> { +) -> Option> { let source = inline_html_navigation_source(raw_url)?; Some( async { @@ -1206,7 +1231,7 @@ async fn load_data_url_navigation_with_engine_async( raw_url: &str, request_headers: Vec<(String, String)>, reply_boundary: RendererReplyBoundary, -) -> Option> { +) -> Option> { let source = decoded_data_url_navigation_response(raw_url)?; Some( async { @@ -1254,7 +1279,7 @@ async fn build_navigation_from_streaming_raw_response_with_engine_async( reserved_service_worker_client: Option, resource_source: CommittedDocumentResourceSource, reply_boundary: RendererReplyBoundary, -) -> Result { +) -> anyhow::Result { let timing_enabled = moli_trace::cdp_nav_timing_enabled(); let timing_started = std::time::Instant::now(); let network_extra_info_available = !network_observation_journal.is_empty(); @@ -1362,24 +1387,24 @@ async fn build_navigation_from_streaming_raw_response_with_engine_async( if let Some(chunk) = initial_body_chunk.take() { body_writer .append(&chunk) - .map_err(|error| format!("failed to capture XML page body: {error}"))?; + .context("failed to capture XML page body")?; } while let Some(chunk) = response.next_chunk().await { body_writer .append(&chunk) - .map_err(|error| format!("failed to capture XML page body: {error}"))?; + .context("failed to capture XML page body")?; } response .finish() .await - .map_err(|error| format!("failed to read XML page body from stream: {error}"))?; + .context("failed to read XML page body from stream")?; let captured_body = body_writer .finish() - .map_err(|error| format!("failed to finish captured XML page body: {error}"))?; + .context("failed to finish captured XML page body")?; let response_text = captured_body .materialize_bytes() .map(|bytes| String::from_utf8_lossy(&bytes).into_owned()) - .map_err(|error| format!("failed to materialize XML page body: {error}"))?; + .context("failed to materialize XML page body")?; let page_storage = load_inputs.page_storage_handles(); let main_document_commit = load_inputs .main_document_commit_for_final_url(&final_url, None) @@ -1415,7 +1440,7 @@ async fn build_navigation_from_streaming_raw_response_with_engine_async( main_document_commit.as_deref().cloned(), ) .await - .map_err(|error| format!("failed to prepare XML page `{}`: {error}", requested_url))?; + .with_context(|| format!("failed to prepare XML page `{}`", requested_url))?; if timing_enabled { tracing::info!( target: "moli_cdp_nav_timing", @@ -1489,7 +1514,7 @@ async fn build_navigation_from_streaming_raw_response_with_engine_async( let prepared_future = async { prepared_future .await - .map_err(|error| format!("failed to prepare streaming raw page: {error:#}")) + .context("failed to prepare streaming raw page") }; let body_capture_task = spawn_streaming_body_capture(response, initial_body_chunk, body_tx, completion_tx); @@ -1498,10 +1523,7 @@ async fn build_navigation_from_streaming_raw_response_with_engine_async( Ok(prepared_page) => prepared_page, Err(error) => { body_capture_task.abort(); - return Err(format!( - "failed to prepare page `{}`: {error}", - requested_url - )); + return Err(error.context(format!("failed to prepare page `{}`", requested_url))); } }; if timing_enabled { @@ -1551,7 +1573,7 @@ impl CdpConnection { &mut self, owner: &CommandOwnerScope, final_url: &Url, - ) -> Result { + ) -> anyhow::Result { let idle_override = self.idle_override_for_navigation(owner, final_url); let load_inputs = self.navigation_load_inputs_for_owner(owner); // The renderer runtime is shared by the BrowserContext, but each Page @@ -1619,7 +1641,7 @@ impl CdpConnection { response: &StreamingRawResponse, network_observation_journal: &NetworkObservationJournal, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result, String> { + ) -> anyhow::Result> { if super::downloads::response_headers_indicate_download(&response.headers) || response_headers_indicate_xml_document(&response.headers) || response_status_may_use_http_error_page(response.status) @@ -1676,7 +1698,7 @@ impl CdpConnection { if let Some(resource_runtime) = shared_resource_runtime { engine .adopt_registered_resource_runtime(resource_runtime) - .map_err(|error| error.to_string())?; + .context("failed to update navigation runtime policy")?; } let (fetch_subresource_interception_enabled, fetch_subresource_interception_resource_type) = load_inputs.fetch_subresource_interception; @@ -1720,12 +1742,10 @@ impl CdpConnection { main_document_commit.as_deref().cloned(), ) .await - .map_err(|error| { - format!( - "failed to prepare response-stage page `{}`: {error:#}", + .with_context(|| format!( + "failed to prepare response-stage page `{}`", requested_url - ) - })?; + ))?; if let Some(started) = timing_started { tracing::info!( target: "moli_cdp_nav_timing", @@ -2144,6 +2164,7 @@ impl CdpConnection { } apply_navigation_load_input_overrides_async(&mut page, &load_inputs, override_mode) .await + .map_err(|error| format!("{error:#}")) .inspect_err(|message| { self.fail_initial_document_page_build_for_owner(&owner, message.clone()); })?; @@ -2164,7 +2185,7 @@ impl CdpConnection { pub async fn load_navigation_via_runtime_async( &mut self, raw_url: &str, - ) -> Result { + ) -> anyhow::Result { let owner = CommandOwnerScope::capture(self, None); let load_inputs = self.navigation_load_inputs_for_owner(&owner); self.load_navigation_via_runtime_with_load_inputs_async(&owner, raw_url, load_inputs) @@ -2183,7 +2204,7 @@ impl CdpConnection { &mut self, session_id: Option<&str>, raw_url: &str, - ) -> Result { + ) -> anyhow::Result { let owner = CommandOwnerScope::capture(self, session_id); let load_inputs = self.navigation_fixture_load_inputs_for_session_owner(session_id)?; self.load_navigation_via_runtime_with_load_inputs_async(&owner, raw_url, load_inputs) @@ -2195,7 +2216,7 @@ impl CdpConnection { owner: &CommandOwnerScope, raw_url: &str, load_inputs: TargetNavigationLoadInputs, - ) -> Result { + ) -> anyhow::Result { let request_headers = load_inputs.extra_http_headers.clone(); let navigation = self .load_navigation_request_via_runtime_with_network_events_and_load_inputs_async( @@ -2207,8 +2228,7 @@ impl CdpConnection { request_headers, MainDocumentBodyProgressSource::default(), ) - .await - .map_err(|error| format!("{error:#}"))?; + .await?; self.commit_navigation_load_outcome_for_owner_async(owner, navigation) .await } @@ -2217,7 +2237,7 @@ impl CdpConnection { &mut self, owner: &CommandOwnerScope, navigation: NavigationLoadOutcome, - ) -> Result { + ) -> anyhow::Result { match navigation { NavigationLoadOutcome::ResponseCommitReady(navigation) => { let navigation = *navigation; @@ -2233,9 +2253,11 @@ impl CdpConnection { } NavigationLoadOutcome::Loaded(navigation) => Ok(*navigation), NavigationLoadOutcome::Download(_) => { - Err("navigation resolved to a download".to_owned()) + Err(anyhow::anyhow!("navigation resolved to a download")) + } + NavigationLoadOutcome::NetworkFailure(error_text) => { + Err(anyhow::Error::msg(error_text)) } - NavigationLoadOutcome::NetworkFailure(error_text) => Err(error_text), } } @@ -2327,7 +2349,6 @@ impl CdpConnection { RendererReplyBoundary::Stage, ) .await - .map_err(anyhow::Error::msg) .context("failed to prepare network error document") } @@ -2357,7 +2378,7 @@ impl CdpConnection { ) .await { - return navigation.map_err(anyhow::Error::msg); + return navigation; } if let Some(navigation) = load_data_url_navigation_with_engine_async( self.standalone_navigation_engine.ensure_mut(), @@ -2370,7 +2391,7 @@ impl CdpConnection { ) .await { - return navigation.map_err(anyhow::Error::msg); + return navigation; } } else { let mut inline_engine = self.navigation_engine_handle_for_load_inputs(&load_inputs); @@ -2388,7 +2409,7 @@ impl CdpConnection { ) .await { - return navigation.map_err(anyhow::Error::msg); + return navigation; } if let Some(navigation) = load_data_url_navigation_with_engine_async( &mut inline_engine, @@ -2401,7 +2422,7 @@ impl CdpConnection { ) .await { - return navigation.map_err(anyhow::Error::msg); + return navigation; } } @@ -2430,7 +2451,6 @@ impl CdpConnection { body_progress_source, ) .await - .map_err(anyhow::Error::msg) } pub(crate) fn navigation_load_job_for_navigation( @@ -2667,7 +2687,7 @@ impl CdpConnection { fetch_config } - pub async fn load_page_via_runtime_async(&mut self, raw_url: &str) -> Result { + pub async fn load_page_via_runtime_async(&mut self, raw_url: &str) -> anyhow::Result { let navigation = self.load_navigation_via_runtime_async(raw_url).await?; Ok(navigation.page) } @@ -2679,7 +2699,7 @@ impl CdpConnection { method: &str, raw_url: &str, request_headers: Vec<(String, String)>, - ) -> Option> { + ) -> Option> { load_inline_html_navigation_with_engine_async( self.standalone_navigation_engine.ensure_mut(), page_reservation, @@ -2701,7 +2721,7 @@ impl CdpConnection { raw_url: &str, request_headers: Vec<(String, String)>, reply_boundary: RendererReplyBoundary, - ) -> Option> { + ) -> Option> { load_inline_html_navigation_with_engine_async( engine, page_reservation, @@ -2728,7 +2748,7 @@ impl CdpConnection { response_status: u16, response_headers: Vec<(String, String)>, response_body: String, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_session_owner(None); let initial_request_cookie_report = load_inputs.request_cookie_report_for_navigation(&requested_url, &request_method, true); @@ -2756,7 +2776,7 @@ impl CdpConnection { response_status: u16, response_headers: Vec<(String, String)>, response_body: String, - ) -> Result { + ) -> anyhow::Result { let owner = CommandOwnerScope::capture(self, session_id); let load_inputs = self.navigation_fixture_load_inputs_for_session_owner(session_id)?; let initial_request_cookie_report = @@ -2785,11 +2805,12 @@ impl CdpConnection { fn navigation_fixture_load_inputs_for_session_owner( &self, session_id: Option<&str>, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_session_owner(session_id); - let frame_id = load_inputs.root_frame_id.clone().ok_or_else(|| { - "navigation fixture requires an installed target root frame".to_owned() - })?; + let frame_id = load_inputs + .root_frame_id + .clone() + .context("navigation fixture requires an installed target root frame")?; Ok(load_inputs.with_main_document_commit_seed( RendererMainDocumentCommitSeed::from_navigation_fixture( frame_id, @@ -2816,7 +2837,7 @@ impl CdpConnection { response_headers: Vec<(String, String)>, response_body: String, initial_request_cookie_report: Option, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_session_owner(None); self.build_loaded_navigation_from_buffered_response_with_request_cookie_report_async( &load_inputs, @@ -2842,7 +2863,7 @@ impl CdpConnection { initial_request_cookie_report: Option, network_observation_journal: NetworkObservationJournal, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_navigation(navigation); self.build_navigation_from_buffered_body_source_with_load_inputs_async( &navigation.owner, @@ -2876,7 +2897,7 @@ impl CdpConnection { initial_request_cookie_report: Option, network_observation_journal: NetworkObservationJournal, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let response_cookie_reports = load_inputs.store_response_cookie_reports(&final_url, &response_headers); let head = ResponseHead { @@ -2915,7 +2936,7 @@ impl CdpConnection { response_body: String, captured_response_body: Option, initial_request_cookie_report: Option, - ) -> Result { + ) -> anyhow::Result { let response_cookie_reports = load_inputs.store_response_cookie_reports(&requested_url, &response_headers); let (fetch_subresource_interception_enabled, fetch_subresource_interception_resource_type) = @@ -2926,7 +2947,7 @@ impl CdpConnection { .map(Arc::new); let built = self .navigation_engine_for_load_inputs_mut(load_inputs) - .ok_or_else(|| "navigation Page engine unavailable".to_owned())? + .context("navigation Page engine unavailable")? .build_html_page_from_response_with_storage_and_inspector_session_restores_async( page_storage.into_navigation_storage(), requested_url.clone(), @@ -2955,9 +2976,9 @@ impl CdpConnection { main_document_commit.as_deref().cloned(), ) .await - .map_err(|error| { + .with_context(|| { format!( - "failed to execute scripts for synthetic response `{}`: {error}", + "failed to execute scripts for synthetic response `{}`", requested_url ) })?; @@ -3080,8 +3101,7 @@ impl CdpConnection { request.set_auth(Some(auth.into())); let loader = self - .ensure_resource_request_client_for_navigation_load_inputs(&load_inputs) - .map_err(anyhow::Error::msg)? + .ensure_resource_request_client_for_navigation_load_inputs(&load_inputs)? .clone(); loader .fetch_raw_with_network_metadata(request) @@ -3170,8 +3190,7 @@ impl CdpConnection { request.set_auth(auth.map(Into::into)); let loader = self - .ensure_resource_request_client_for_navigation_load_inputs(load_inputs) - .map_err(anyhow::Error::msg)? + .ensure_resource_request_client_for_navigation_load_inputs(load_inputs)? .clone(); loader .fetch_raw_stream_with_cancel_and_network_metadata(request, FetchCancelHandle::new()) @@ -3185,7 +3204,7 @@ impl CdpConnection { request_method: String, request_headers: Vec<(String, String)>, response: NetworkFetchResult, - ) -> Result { + ) -> anyhow::Result { self.build_navigation_from_network_response_for_session_owner_async( None, requested_url, @@ -3203,7 +3222,7 @@ impl CdpConnection { request_method: String, request_headers: Vec<(String, String)>, response: NetworkFetchResult, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_session_owner(session_id); let (response, network_observation_journal) = response.into_parts_with_observation_journal(); @@ -3232,7 +3251,7 @@ impl CdpConnection { .map(Arc::new); let built = self .navigation_engine_for_load_inputs_mut(&load_inputs) - .ok_or_else(|| "navigation Page engine unavailable".to_owned())? + .context("navigation Page engine unavailable")? .build_html_page_from_response_with_storage_and_inspector_session_restores_async( page_storage.into_navigation_storage(), requested_url.clone(), @@ -3261,12 +3280,7 @@ impl CdpConnection { main_document_commit.as_deref().cloned(), ) .await - .map_err(|error| { - format!( - "failed to execute scripts for page `{}`: {error}", - requested_url - ) - })?; + .with_context(|| format!("failed to execute scripts for page `{}`", requested_url))?; let diagnostics = loaded_page_creation_diagnostics_parts(built.page_creation_diagnostics); let mut page = built.page; apply_navigation_load_input_overrides_async( @@ -3325,7 +3339,7 @@ impl CdpConnection { request_method: String, request_headers: Vec<(String, String)>, response: RawResponse, - ) -> Result { + ) -> anyhow::Result { self.build_navigation_from_buffered_raw_response_for_session_owner_async( None, requested_url, @@ -3343,7 +3357,7 @@ impl CdpConnection { request_method: String, request_headers: Vec<(String, String)>, response: NetworkFetchResult, - ) -> Result { + ) -> anyhow::Result { let owner = CommandOwnerScope::capture(self, session_id); let load_inputs = self.navigation_load_inputs_for_owner(&owner); self.build_navigation_from_buffered_raw_response_with_load_inputs_async( @@ -3361,7 +3375,7 @@ impl CdpConnection { &mut self, navigation: &NavigationDispatchState, response: NetworkFetchResult, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_navigation(navigation); self.build_navigation_from_buffered_raw_response_with_load_inputs_async( &navigation.owner, @@ -3382,7 +3396,7 @@ impl CdpConnection { request_method: String, request_headers: Vec<(String, String)>, response: NetworkFetchResult, - ) -> Result { + ) -> anyhow::Result { let (response, network_observation_journal) = response.into_parts_with_observation_journal(); if super::downloads::response_headers_indicate_download(&response.headers) { @@ -3419,7 +3433,7 @@ impl CdpConnection { body: CapturedBody, network_observation_journal: NetworkObservationJournal, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_navigation(navigation); self.build_navigation_from_captured_raw_response_with_load_inputs_async( &navigation.owner, @@ -3447,11 +3461,11 @@ impl CdpConnection { body: CapturedBody, network_observation_journal: NetworkObservationJournal, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { if super::downloads::response_headers_indicate_download(&head.headers) { - let body_bytes = body.materialize_bytes().map_err(|error| { - format!("failed to materialize captured download body: {error}") - })?; + let body_bytes = body + .materialize_bytes() + .context("failed to materialize captured download body")?; return Ok(NavigationLoadOutcome::download( self.build_download_from_raw_response( request_method, @@ -3514,7 +3528,7 @@ impl CdpConnection { request_headers: Vec<(String, String)>, response: StreamingRawResponse, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { self.build_navigation_from_streaming_raw_response_for_session_owner_async( None, requested_url, @@ -3534,7 +3548,7 @@ impl CdpConnection { request_headers: Vec<(String, String)>, response: NetworkFetchResult, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let owner = CommandOwnerScope::capture(self, session_id); let load_inputs = self.navigation_load_inputs_for_owner(&owner); self.build_navigation_from_streaming_raw_response_with_load_inputs_async( @@ -3556,7 +3570,7 @@ impl CdpConnection { navigation: &NavigationDispatchState, response: NetworkFetchResult, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_navigation(navigation); self.build_navigation_from_streaming_raw_response_with_load_inputs_async( &navigation.owner, @@ -3579,7 +3593,7 @@ impl CdpConnection { response_code: Option, response_headers_override: Vec<(String, String)>, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let load_inputs = self.navigation_load_inputs_for_navigation(navigation); self.build_navigation_from_streaming_raw_response_with_load_inputs_async( &navigation.owner, @@ -3607,7 +3621,7 @@ impl CdpConnection { response_code: Option, response_headers_override: Vec<(String, String)>, body_progress_source: MainDocumentBodyProgressSource, - ) -> Result { + ) -> anyhow::Result { let (response, network_observation_journal) = response.into_parts_with_observation_journal(); if load_inputs.browser_context_id.is_none() { @@ -3738,7 +3752,7 @@ async fn prepare_navigation_from_captured_raw_response_with_engine_async( network_error_page: Option, synthetic_body: bool, reply_boundary: RendererReplyBoundary, -) -> Result { +) -> anyhow::Result { let network_extra_info_available = !network_observation_journal.is_empty(); if network_error_page.is_none() && response_status_may_use_http_error_page(head.status) @@ -3807,7 +3821,7 @@ async fn prepare_captured_document_response_with_engine_async( network_error_page: Option, synthetic_body: bool, reply_boundary: RendererReplyBoundary, -) -> Result { +) -> anyhow::Result { let network_extra_info_available = !network_observation_journal.is_empty(); body_progress_source.emit_response_metadata( &request_method, @@ -3900,10 +3914,10 @@ async fn prepare_captured_document_response_with_engine_async( Ok(prepared_page) => prepared_page, Err(error) => { body_capture_task.abort(); - return Err(format!( - "failed to prepare captured page `{}`: {error:#}", + return Err(error.context(format!( + "failed to prepare captured page `{}`", requested_url - )); + ))); } }; @@ -3933,7 +3947,7 @@ fn validate_navigation_network_request( raw_url: &str, request_headers: &[(String, String)], ) -> anyhow::Result<()> { - ensure_url_not_blocked_for_load_inputs(load_inputs, raw_url).map_err(anyhow::Error::msg)?; + ensure_url_not_blocked_for_load_inputs(load_inputs, raw_url)?; if load_inputs.network_offline { let requested_url = Url::parse(raw_url) .with_context(|| format!("failed to parse request url `{raw_url}`"))?; @@ -3951,13 +3965,13 @@ fn validate_navigation_network_request( fn ensure_url_not_blocked_for_load_inputs( load_inputs: &TargetNavigationLoadInputs, raw_url: &str, -) -> Result<(), String> { +) -> anyhow::Result<()> { if load_inputs .blocked_url_patterns .iter() .any(|pattern| url_pattern_matches(pattern, raw_url)) { - Err(BLOCKED_BY_CLIENT_ERROR_TEXT.to_owned()) + Err(anyhow::anyhow!(BLOCKED_BY_CLIENT_ERROR_TEXT)) } else { Ok(()) } @@ -3973,13 +3987,13 @@ async fn apply_navigation_load_input_overrides_async( page: &mut moli_core::page::Page, load_inputs: &TargetNavigationLoadInputs, mode: NavigationLoadInputOverrideMode, -) -> Result<(), String> { +) -> anyhow::Result<()> { if !load_inputs.permission_overrides.is_empty() || mode == NavigationLoadInputOverrideMode::ExistingPage { page.set_permission_overrides_async(&load_inputs.permission_overrides) .await - .map_err(|error| format!("failed to apply page permission overrides: {error}"))?; + .context("failed to apply page permission overrides")?; } if mode == NavigationLoadInputOverrideMode::FreshlyBuiltPage { return Ok(()); @@ -3988,16 +4002,16 @@ async fn apply_navigation_load_input_overrides_async( // the new Document. Native isolates inherit the process default at entry. page.set_script_execution_disabled_async(load_inputs.script_execution_disabled) .await - .map_err(|error| format!("failed to apply page script execution override: {error}"))?; + .context("failed to apply page script execution override")?; page.set_bypass_content_security_policy_async(load_inputs.bypass_content_security_policy) .await - .map_err(|error| format!("failed to apply page CSP bypass override: {error}"))?; + .context("failed to apply page CSP bypass override")?; page.set_emulated_media_async(&load_inputs.emulated_media) .await - .map_err(|error| format!("failed to apply page emulated media: {error}"))?; + .context("failed to apply page emulated media")?; page.set_viewport_surface_async(load_inputs.viewport_surface) .await - .map_err(|error| format!("failed to apply page viewport surface: {error}"))?; + .context("failed to apply page viewport surface")?; Ok(()) } @@ -4010,6 +4024,76 @@ mod tests { }; use serde_json::json; + #[tokio::test] + async fn streaming_body_failure_preserves_source_for_renderer_and_navigation() { + let (chunks_tx, chunks_rx) = tokio::sync::mpsc::unbounded_channel(); + chunks_tx.send(b"partial body".to_vec()).unwrap(); + drop(chunks_tx); + let (finish_tx, finish_rx) = tokio::sync::oneshot::channel(); + finish_tx + .send(Err(anyhow::Error::new(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + "typed body transport failure", + )) + .context("transport completion failed"))) + .unwrap(); + let response = moli_fetch::StreamingRawResponse::new( + url::Url::parse("https://example.test/document").unwrap(), + 200, + Vec::new(), + None, + Vec::new(), + false, + Vec::new(), + chunks_rx, + moli_fetch::FetchCancelHandle::new(), + finish_rx, + ); + let (body_tx, mut body_rx) = tokio::sync::mpsc::channel(1); + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); + let task = super::spawn_streaming_body_capture(response, None, body_tx, completion_tx); + assert_eq!(body_rx.recv().await.unwrap(), b"partial body"); + assert!(body_rx.recv().await.is_none()); + let navigation_error = super::ResponseCommitBodyCapture::Pending(task) + .resolve() + .await + .expect_err("partial response must fail navigation capture"); + let renderer_error = completion_rx + .await + .unwrap() + .expect_err("renderer must receive the transport failure"); + for error in [&navigation_error, &renderer_error] { + let cause = error + .root_cause() + .downcast_ref::() + .expect("shared body failure must retain the typed I/O cause"); + assert_eq!(cause.kind(), std::io::ErrorKind::ConnectionReset); + assert!(format!("{error:#}").contains("transport completion failed")); + assert!(format!("{error:#}").contains("failed to read page body from stream")); + } + assert!(std::ptr::eq( + navigation_error.root_cause(), + renderer_error.root_cause() + )); + } + + #[tokio::test] + async fn cancelled_body_capture_retains_join_error() { + let task = tokio::spawn(std::future::pending::>()); + task.abort(); + let error = super::ResponseCommitBodyCapture::Pending(task) + .resolve() + .await + .expect_err("cancelled capture must fail"); + assert!( + error + .downcast_ref::() + .expect("capture must preserve the task failure type") + .is_cancelled() + ); + assert!(format!("{error:#}").contains("main document body capture task failed")); + } + #[test] fn decode_text_html_data_url_uses_data_url_processor() { assert_eq!( diff --git a/moli-protocol/src/conn/state/fetch.rs b/moli-protocol/src/conn/state/fetch.rs index 4bbe4a457..8705ce57e 100644 --- a/moli-protocol/src/conn/state/fetch.rs +++ b/moli-protocol/src/conn/state/fetch.rs @@ -581,7 +581,7 @@ impl TargetFetchState { runtime_slot: &mut TargetRuntimeSlot, request_id: &str, handle: String, - ) -> Result, String> { + ) -> anyhow::Result> { let Some(transfer) = self.take_pending_fetch_response_transfer(request_id) else { return Ok(None); }; @@ -591,9 +591,9 @@ impl TargetFetchState { self.register_pending_fetch_response_transfer(request_id.to_owned(), *transfer); return Ok(None); } - Err(OpenBodyStreamError::Failed { transfer, message }) => { + Err(OpenBodyStreamError::Failed { transfer, error }) => { self.register_pending_fetch_response_transfer(request_id.to_owned(), *transfer); - return Err(message); + return Err(error); } }; @@ -2077,7 +2077,7 @@ impl TargetFetchOwner { runtime_slot: &mut TargetRuntimeSlot, request_id: &str, handle: String, - ) -> Result, String> { + ) -> anyhow::Result> { self.pending .open_pending_fetch_response_body_stream(runtime_slot, request_id, handle) } diff --git a/moli-protocol/src/conn/tests/navigation_error.rs b/moli-protocol/src/conn/tests/navigation_error.rs index c3a29dc4e..631422463 100644 --- a/moli-protocol/src/conn/tests/navigation_error.rs +++ b/moli-protocol/src/conn/tests/navigation_error.rs @@ -6,6 +6,102 @@ use moli_core::page::{SubresourceAuthCredentials, SubresourceAuthScheme, Subreso const OFFLINE_ERROR_TEXT: &str = "net::ERR_INTERNET_DISCONNECTED"; +fn failing_streamed_document( + navigation: &NavigationDispatchState, +) -> crate::conn::DocumentBodySource { + let (chunks_tx, chunks_rx) = tokio::sync::mpsc::unbounded_channel(); + chunks_tx.send(b"partial document".to_vec()).unwrap(); + drop(chunks_tx); + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); + completion_tx + .send(Err(anyhow::Error::new(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "typed response body failure", + )) + .context("response transport failed"))) + .unwrap(); + let response = moli_fetch::StreamingRawResponse::new( + navigation.requested_url.clone(), + 200, + Vec::new(), + None, + Vec::new(), + false, + Vec::new(), + chunks_rx, + moli_fetch::FetchCancelHandle::new(), + completion_rx, + ); + crate::conn::DocumentBodySource::StreamingRaw { + requested_url: navigation.requested_url.clone(), + request_method: navigation.request_method.clone(), + request_headers: navigation.request_headers.clone(), + response, + network_observation_journal: Default::default(), + body_progress_source: Default::default(), + prepared_document: None, + } +} + +#[tokio::test] +async fn fetch_body_materialization_preserves_error_type_and_request_identity() { + let (_ctx, navigation) = navigation_fixture(); + let (error, body) = failing_streamed_document(&navigation) + .materialize_body_limited_async(1024) + .await + .expect_err("partial transport failure must fail body materialization"); + assert_eq!( + error + .downcast_ref::() + .expect("materialization must preserve the I/O cause") + .kind(), + std::io::ErrorKind::UnexpectedEof + ); + assert!(format!("{error:#}").contains("response transport failed")); + assert!(format!("{error:#}").contains("failed to read page body from stream")); + let crate::conn::DocumentBodySource::CapturedRaw { + requested_url, + request_method, + request_headers, + .. + } = body + else { + panic!("failed streamed materialization must return its captured source"); + }; + assert_eq!(requested_url, navigation.requested_url); + assert_eq!(request_method, navigation.request_method); + assert_eq!(request_headers, navigation.request_headers); +} + +#[tokio::test] +async fn fetch_body_stream_read_preserves_error_type_and_paused_transfer() { + let (_ctx, navigation) = navigation_fixture(); + let body = failing_streamed_document(&navigation); + let transfer = crate::conn::PausedDocumentTransfer::pending( + "fetch-typed-error".to_owned(), + None, + navigation, + body, + ) + .open_body_stream("stream-typed-error".to_owned()) + .expect("streamed response should open") + .transfer; + let (transfer, error) = transfer + .read_body_stream_async(None) + .await + .expect_err("partial transport failure must fail the stream read"); + assert_eq!(transfer.fetch_request_id(), "fetch-typed-error"); + assert_eq!( + error + .downcast_ref::() + .expect("stream read must preserve the I/O cause") + .kind(), + std::io::ErrorKind::UnexpectedEof + ); + assert!(format!("{error:#}").contains("response transport failed")); + assert!(format!("{error:#}").contains("failed to read page body from stream")); +} + fn navigation_fixture() -> (TestContext, NavigationDispatchState) { let mut ctx = TestContext::new(); let mut browser_context = BrowserContext::new("BID-1".to_owned()); diff --git a/moli-protocol/src/domains/fetch.rs b/moli-protocol/src/domains/fetch.rs index 5855abecf..b866f03a0 100644 --- a/moli-protocol/src/domains/fetch.rs +++ b/moli-protocol/src/domains/fetch.rs @@ -202,7 +202,7 @@ enum CompletedFetchCommandOperation { result: Box< Result< (Option>, crate::conn::PausedDocumentTransfer), - (String, crate::conn::PausedDocumentTransfer), + (anyhow::Error, crate::conn::PausedDocumentTransfer), >, >, }, @@ -978,7 +978,7 @@ async fn complete_disable_command_async( let navigation = network::materialize_navigation_load_result( conn, &navigation_state, - Err("Fetch interception disabled".to_owned()), + Err(anyhow::anyhow!("Fetch interception disabled")), ); navigation::complete_tokened_materialized_navigation_as_background_events_async( conn, @@ -995,7 +995,7 @@ async fn complete_disable_command_async( let navigation = network::materialize_navigation_load_result( conn, &navigation_state, - Err("Fetch interception disabled".to_owned()), + Err(anyhow::anyhow!("Fetch interception disabled")), ); navigation::complete_tokened_materialized_navigation_as_background_events_async( conn, diff --git a/moli-protocol/src/domains/fetch/auth.rs b/moli-protocol/src/domains/fetch/auth.rs index c2b3d6c15..634f11bfb 100644 --- a/moli-protocol/src/domains/fetch/auth.rs +++ b/moli-protocol/src/domains/fetch/auth.rs @@ -551,7 +551,7 @@ pub(super) async fn complete_continue_with_auth_command_async( let navigation = network::materialize_navigation_load_result( conn, &navigation_state, - Err("Fetch auth challenge aborted".to_owned()), + Err(anyhow::anyhow!("Fetch auth challenge aborted")), ); complete_tokened_materialized_navigation_as_background_events_async( conn, diff --git a/moli-protocol/src/domains/fetch/body_stream.rs b/moli-protocol/src/domains/fetch/body_stream.rs index d262a4f16..144c13302 100644 --- a/moli-protocol/src/domains/fetch/body_stream.rs +++ b/moli-protocol/src/domains/fetch/body_stream.rs @@ -125,7 +125,7 @@ pub(super) fn complete_get_response_body_from_transfer( } Err((message, transfer)) => { conn.register_pending_fetch_response_transfer_for_owner(owner, request_id, transfer); - CommandOutputPlan::error(-32000, message) + CommandOutputPlan::error(-32000, format!("{message:#}")) } }; out.extend_plan_as_command_response(plan); @@ -148,7 +148,7 @@ pub(super) fn take_response_body_as_stream_command( } Ok(None) => {} Err(message) => { - return CommandOutputPlan::error(-32000, message); + return CommandOutputPlan::error(-32000, format!("{message:#}")); } } if let Some(pending) = pending_subresource_response_request_for_action_session( @@ -176,7 +176,7 @@ pub(super) fn take_response_body_as_stream_command( return CommandOutputPlan::result(json!({ "stream": handle })); } Err(message) => { - return CommandOutputPlan::error(-32000, message); + return CommandOutputPlan::error(-32000, format!("{message:#}")); } } } @@ -187,7 +187,7 @@ fn open_pending_response_navigation_body_stream( conn: &mut CdpConnection, session_id: Option<&str>, request_id: &str, -) -> Result, String> { +) -> anyhow::Result> { conn.open_pending_fetch_response_body_stream_for_session_owner(session_id, request_id) } diff --git a/moli-protocol/src/domains/fetch/commands.rs b/moli-protocol/src/domains/fetch/commands.rs index 66cbc1998..07619212b 100644 --- a/moli-protocol/src/domains/fetch/commands.rs +++ b/moli-protocol/src/domains/fetch/commands.rs @@ -672,7 +672,7 @@ pub(super) async fn complete_fail_request_command_async( let navigation = network::materialize_navigation_load_result( conn, &navigation_state, - Err(error_text), + Err(anyhow::Error::msg(error_text)), ); complete_tokened_materialized_navigation_as_background_events_async( conn, diff --git a/moli-protocol/src/domains/fetch/navigation.rs b/moli-protocol/src/domains/fetch/navigation.rs index 3585af28b..13d59fc59 100644 --- a/moli-protocol/src/domains/fetch/navigation.rs +++ b/moli-protocol/src/domains/fetch/navigation.rs @@ -195,8 +195,7 @@ pub(crate) async fn load_or_pause_navigation_for_auth_into_buffer_async( &pending.navigation, response, ) - .await - .map_err(anyhow::Error::msg); + .await; complete_or_pause_response_stage_into_buffer_async( conn, out, pending, navigation, ) @@ -358,8 +357,7 @@ pub(super) async fn cancel_navigation_auth_as_background_events_async( &pending.navigation, response, ) - .await - .map_err(anyhow::Error::msg); + .await; complete_or_pause_response_stage_into_buffer_async(conn, &mut output, pending, navigation) .await; } @@ -449,18 +447,8 @@ async fn complete_pending_fetch_navigation_result_into_buffer_async( } navigation => navigation, }; - let navigation = network::materialize_navigation_load_result( - conn, - &navigation_state, - navigation.map_err(|error| { - tracing::debug!( - error = ?error, - session_id = navigation_state.owner.session_id(), - "intercepted navigation failed" - ); - error.root_cause().to_string() - }), - ); + let navigation = + network::materialize_navigation_load_result(conn, &navigation_state, navigation); complete_tokened_materialized_navigation_into_buffer_async( conn, out, @@ -528,8 +516,7 @@ async fn handle_streaming_response_head_for_navigation_into_buffer_async( response, network::MainDocumentBodyProgressSource::default(), ) - .await - .map_err(anyhow::Error::msg); + .await; complete_pending_fetch_navigation_result_into_buffer_async(conn, out, pending, navigation) .await; return; @@ -542,8 +529,7 @@ async fn handle_streaming_response_head_for_navigation_into_buffer_async( response, network::MainDocumentBodyProgressSource::default(), ) - .await - .map_err(anyhow::Error::msg); + .await; complete_pending_fetch_navigation_result_into_buffer_async(conn, out, pending, navigation) .await; return; @@ -589,7 +575,7 @@ async fn handle_streaming_response_head_for_navigation_into_buffer_async( conn, out, pending, - Err(anyhow::Error::msg(error)), + Err(error), ) .await; return; diff --git a/moli-protocol/src/domains/io.rs b/moli-protocol/src/domains/io.rs index 4a8954235..ef1c45489 100644 --- a/moli-protocol/src/domains/io.rs +++ b/moli-protocol/src/domains/io.rs @@ -285,8 +285,8 @@ fn read_fetch_response_body_stream_output_plan( CommandOutputPlan::error(-32000, "StreamHandleNotFound") } PendingFetchResponseBodyStreamRead::Read { bytes, eof } => read_output_plan(&bytes, eof), - PendingFetchResponseBodyStreamRead::Failed(message) => { - CommandOutputPlan::error(-32000, message) + PendingFetchResponseBodyStreamRead::Failed(error) => { + CommandOutputPlan::error(-32000, format!("{error:#}")) } } } diff --git a/moli-protocol/src/domains/network/main_document_progress/mod.rs b/moli-protocol/src/domains/network/main_document_progress/mod.rs index fd9014f0b..0b194101f 100644 --- a/moli-protocol/src/domains/network/main_document_progress/mod.rs +++ b/moli-protocol/src/domains/network/main_document_progress/mod.rs @@ -364,15 +364,20 @@ fn materialize_navigation_load_outcome( pub(crate) fn materialize_navigation_load_result( conn: &mut CdpConnection, state: &NavigationDispatchState, - navigation: Result, + navigation: anyhow::Result, ) -> MaterializedNavigationLoadOutcome { match navigation { Ok(navigation) => materialize_navigation_load_outcome(conn, state, navigation), - Err(error_text) => { + Err(error) => { + tracing::debug!( + error = ?error, + session_id = state.owner.session_id(), + "navigation load failed" + ); MaterializedNavigationLoadOutcome::Failed(materialize_failed_navigation_progress( conn, state, - error_text, + error.root_cause().to_string(), FailedNavigationDocumentPolicy::InvalidateCommittedDocument, FailedNavigationResponseMode::ProtocolError, )) diff --git a/moli-protocol/src/domains/page/navigation.rs b/moli-protocol/src/domains/page/navigation.rs index 73be7206e..f66c7ba03 100644 --- a/moli-protocol/src/domains/page/navigation.rs +++ b/moli-protocol/src/domains/page/navigation.rs @@ -60,7 +60,7 @@ pub(super) struct CompletedNavigateLoadCommand { prefix_events: Vec, token: DocumentNavigationToken, state: NavigationDispatchState, - navigation: Result, + navigation: anyhow::Result, } pub(super) struct PendingChildFrameNavigateCommand { @@ -436,7 +436,7 @@ impl MaterializedNavigationCompletion { pub struct BackgroundMainDocumentBodyCompletion { token: DocumentNavigationToken, state: NavigationDispatchState, - body: Result, + body: anyhow::Result, synthetic: bool, body_progress_source: network::MainDocumentBodyProgressSource, final_url: Url, @@ -448,7 +448,7 @@ impl BackgroundMainDocumentBodyCompletion { pub(crate) fn new( token: DocumentNavigationToken, state: NavigationDispatchState, - body: Result, + body: anyhow::Result, synthetic: bool, body_progress_source: network::MainDocumentBodyProgressSource, final_url: Url, @@ -501,7 +501,11 @@ impl BackgroundMainDocumentBodyCompletion { ?error, "background main document body capture failed after lifecycle commit" ); - network::record_failed_main_document_response_body(conn, &self.state, error); + network::record_failed_main_document_response_body( + conn, + &self.state, + format!("{error:#}"), + ); } } } @@ -531,7 +535,7 @@ impl BackgroundNavigationCompletion { pub struct BackgroundNavigationLifecycleCompletion { token: DocumentNavigationToken, state: NavigationDispatchState, - navigation: Result, + navigation: anyhow::Result, ready_at: std::time::Instant, } @@ -539,7 +543,7 @@ impl BackgroundNavigationLifecycleCompletion { pub(crate) fn new( token: DocumentNavigationToken, state: NavigationDispatchState, - navigation: Result, + navigation: anyhow::Result, ) -> Self { Self { token, @@ -577,7 +581,7 @@ impl BackgroundNavigationCompletion { pub(crate) fn new( token: DocumentNavigationToken, state: NavigationDispatchState, - navigation: Result, + navigation: anyhow::Result, ) -> Self { Self::Lifecycle(Box::new(BackgroundNavigationLifecycleCompletion::new( token, state, navigation, @@ -587,7 +591,7 @@ impl BackgroundNavigationCompletion { pub(crate) fn main_document_body( token: DocumentNavigationToken, state: NavigationDispatchState, - body: Result, + body: anyhow::Result, synthetic: bool, body_progress_source: network::MainDocumentBodyProgressSource, final_url: Url, @@ -1261,7 +1265,7 @@ fn direct_navigation_result_from_completed_load( } Err(message) => Err(DevToolsError::new( DevToolsErrorKind::Internal, - message.clone(), + format!("{message:#}"), )), } } @@ -3404,9 +3408,9 @@ async fn complete_materialized_navigation_into_buffer_inner_async( fn push_navigation_commit_error( out: &mut CommandOutputBuffer, state: &NavigationDispatchState, - error: impl Into, + error: impl std::fmt::Display, ) { - let error = error.into(); + let error = format!("{error:#}"); if state.navigate_id.is_some() { out.push_error_after_messages(-32000, error); } else { diff --git a/moli-renderer-v8/src/runtime/phase_one/streaming_input.rs b/moli-renderer-v8/src/runtime/phase_one/streaming_input.rs index 59cba20e8..839b26dbe 100644 --- a/moli-renderer-v8/src/runtime/phase_one/streaming_input.rs +++ b/moli-renderer-v8/src/runtime/phase_one/streaming_input.rs @@ -13,7 +13,7 @@ const STREAMING_DOCUMENT_INPUT_BUFFERED_EVENTS: usize = 1; /// continuation runnable and are consumed only by that continuation. pub(super) enum StreamingDocumentInputEvent { Chunks(Vec>), - Finished(std::result::Result<(), String>), + Finished(Result<()>), } #[derive(Clone)] @@ -76,7 +76,7 @@ impl StreamingDocumentInputSource { } let terminal = tokio::select! { _ = sender.receiver_closed() => return, - terminal = raw_body.finish() => terminal.map_err(|error| error.to_string()), + terminal = raw_body.finish() => terminal, }; let _ = sender .send(StreamingDocumentInputEvent::Finished(terminal)) @@ -173,6 +173,41 @@ mod tests { (networking, continuation, wake_rx) } + #[tokio::test] + async fn body_terminal_preserves_transport_error_and_context() { + let (body_tx, completion_tx, _cancel_handle, raw_body) = pending_fetch_body(); + let (_networking, continuation, mut wake_rx) = continuation_fixture(96); + let mut source = StreamingDocumentInputSource::bridge( + raw_body, + continuation, + crate::network::RendererResourceTaskRunner::from_current_tokio().unwrap(), + ); + drop(body_tx); + completion_tx + .send(Err(anyhow::Error::new(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "typed parser input failure", + )) + .context("main document transport failed"))) + .unwrap(); + wake_rx + .recv() + .await + .expect("failed terminal must wake the parser owner"); + let Some(StreamingDocumentInputEvent::Finished(Err(error))) = source.try_next().unwrap() + else { + panic!("the parser must receive its failed body terminal"); + }; + assert_eq!( + error + .downcast_ref::() + .expect("parser input must preserve the transport error type") + .kind(), + std::io::ErrorKind::UnexpectedEof + ); + assert!(format!("{error:#}").contains("main document transport failed")); + } + #[tokio::test] async fn body_payloads_are_resident_before_their_owner_wakes() { let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); diff --git a/moli-renderer-v8/src/runtime/phase_one/streaming_residence.rs b/moli-renderer-v8/src/runtime/phase_one/streaming_residence.rs index 714542f0f..9404fde33 100644 --- a/moli-renderer-v8/src/runtime/phase_one/streaming_residence.rs +++ b/moli-renderer-v8/src/runtime/phase_one/streaming_residence.rs @@ -102,8 +102,8 @@ impl PendingStreamingPhaseOneContinuation { } } StreamingDocumentInputEvent::Finished(result) => { - if let Err(message) = result { - deferred_main_resource_failure = Some(anyhow!(message)); + if let Err(error) = result { + deferred_main_resource_failure = Some(error); break; } if let Some(tail) = decoder.finish() {