starred/moli
1
0
Fork 0
mirror of https://github.com/lexmount/moli.git synced 2026-09-23 08:01:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(navigation): preserve errors through document loading

Browse Source
This commit is contained in:
ldm0
2026-09-20 19:27:31 +08:00
committed by Donough Liu
parent 5be3c7b644
commit 81663c6650
17 changed files with 491 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
moli-protocol/src/conn/browser_context/fetch_owner.rs
+3 -3
View File
@@ -63,7 +63,7 @@ impl SessionFetchBodyStreamOwner<'_> {
fn open_pending_fetch_response_body_stream(
&mut self,
request_id: &str,
) -> Result<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
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<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
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<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
let Some(mut owner) = self.fetch_body_stream_owner_mut() else {
return Ok(None);
};
moli-protocol/src/conn/fetch_support.rs
+44 -39
View File
@@ -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<u8>, 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<u8>, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, String)>>,
completed: Result<
(Vec<u8>, bool, PausedDocumentTransfer),
Box<(PausedDocumentTransfer, anyhow::Error)>,
>,
}
impl PendingFetchResponseBodyStreamReadDispatch {
@@ -766,7 +769,7 @@ impl CompletedFetchResponseBodyStreamReadDispatch {
pub(crate) fn into_completed(
self,
) -> Result<(Vec<u8>, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, String)>>
) -> Result<(Vec<u8>, bool, PausedDocumentTransfer), Box<(PausedDocumentTransfer, anyhow::Error)>>
{
self.completed
}
@@ -777,7 +780,7 @@ pub(crate) enum OpenBodyStreamError {
NotOpenable(Box<PausedDocumentTransfer>),
Failed {
transfer: Box<PausedDocumentTransfer>,
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<usize>,
) -> Result<(Vec<u8>, bool, Self), (Self, String)> {
) -> Result<(Vec<u8>, 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<Vec<u8>>, Self), (String, Self)> {
) -> Result<(Option<Vec<u8>>, Self), (anyhow::Error, Self)> {
let Self {
fetch_request_id,
state,
@@ -1161,7 +1162,7 @@ impl PausedDocumentTransfer {
(
Option<DocumentNavigationToken>,
NavigationDispatchState,
Result<NavigationLoadOutcome, String>,
anyhow::Result<NavigationLoadOutcome>,
),
Self,
> {
@@ -1193,7 +1194,7 @@ impl PausedDocumentTransfer {
) -> (
Option<DocumentNavigationToken>,
NavigationDispatchState,
Result<NavigationLoadOutcome, String>,
anyhow::Result<NavigationLoadOutcome>,
) {
match self.state {
PausedDocumentTransferState::Pending {
@@ -1232,7 +1233,7 @@ impl PausedDocumentTransfer {
) -> (
Option<DocumentNavigationToken>,
NavigationDispatchState,
Result<NavigationLoadOutcome, String>,
anyhow::Result<NavigationLoadOutcome>,
) {
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<usize>) -> Result<(Vec<u8>, bool), String> {
async fn read_async(&mut self, size: Option<usize>) -> anyhow::Result<(Vec<u8>, 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<DocumentBodySource, String> {
fn finish_pending_body_source(&mut self) -> anyhow::Result<DocumentBodySource> {
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<DocumentNavigationToken>,
NavigationDispatchState,
Result<NavigationLoadOutcome, String>,
anyhow::Result<NavigationLoadOutcome>,
) {
let navigation_state = self.navigation.clone();
let final_url = self.response.final_url.clone();
@@ -1352,12 +1357,12 @@ impl ActiveDocumentBodyStreamState {
) -> (
Option<DocumentNavigationToken>,
NavigationDispatchState,
Result<NavigationLoadOutcome, String>,
anyhow::Result<NavigationLoadOutcome>,
) {
(
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<usize>,
) -> Result<(Vec<u8>, bool), String> {
) -> anyhow::Result<(Vec<u8>, 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<u8>,
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<u16>,
response_headers: Vec<(String, String)>,
) -> Result<NavigationLoadOutcome, String> {
) -> anyhow::Result<NavigationLoadOutcome> {
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<u8>, Self), (String, Self)> {
) -> Result<(Vec<u8>, 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))
}
moli-protocol/src/conn/page_state/fetch_state.rs
+1 -1
View File
@@ -76,7 +76,7 @@ impl BrowserContext {
pub(crate) fn open_pending_fetch_response_body_stream(
&mut self,
request_id: &str,
) -> Result<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
let handle = self
.active_page_target_mut()
.runtime_slot
moli-protocol/src/conn/resource_runtime_support.rs
+5 -4
View File
@@ -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<ResourceRequestClient, String> {
) -> anyhow::Result<ResourceRequestClient> {
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(
moli-protocol/src/conn/runtime_load.rs
+258 -174
View File
File diff suppressed because it is too large Load Diff
moli-protocol/src/conn/state/fetch.rs
+4 -4
View File
@@ -581,7 +581,7 @@ impl TargetFetchState {
runtime_slot: &mut TargetRuntimeSlot,
request_id: &str,
handle: String,
) -> Result<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
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<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
self.pending
.open_pending_fetch_response_body_stream(runtime_slot, request_id, handle)
}
moli-protocol/src/conn/tests/navigation_error.rs
+96
View File
@@ -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::<std::io::Error>()
.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::<std::io::Error>()
.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());
moli-protocol/src/domains/fetch.rs
+3 -3
View File
@@ -202,7 +202,7 @@ enum CompletedFetchCommandOperation {
result: Box<
Result<
(Option<Vec<u8>>, 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,
moli-protocol/src/domains/fetch/auth.rs
+1 -1
View File
@@ -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,
moli-protocol/src/domains/fetch/body_stream.rs
+4 -4
View File
@@ -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<Option<String>, String> {
) -> anyhow::Result<Option<String>> {
conn.open_pending_fetch_response_body_stream_for_session_owner(session_id, request_id)
}
moli-protocol/src/domains/fetch/commands.rs
+1 -1
View File
@@ -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,
moli-protocol/src/domains/fetch/navigation.rs
+7 -21
View File
@@ -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;
moli-protocol/src/domains/io.rs
+2 -2
View File
@@ -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:#}"))
}
}
}
moli-protocol/src/domains/network/main_document_progress/mod.rs
+8 -3
View File
@@ -364,15 +364,20 @@ fn materialize_navigation_load_outcome(
pub(crate) fn materialize_navigation_load_result(
conn: &mut CdpConnection,
state: &NavigationDispatchState,
navigation: Result<NavigationLoadOutcome, String>,
navigation: anyhow::Result<NavigationLoadOutcome>,
) -> 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,
))
moli-protocol/src/domains/page/navigation.rs
+15 -11
View File
@@ -60,7 +60,7 @@ pub(super) struct CompletedNavigateLoadCommand {
prefix_events: Vec<BackgroundProtocolEvent>,
token: DocumentNavigationToken,
state: NavigationDispatchState,
navigation: Result<NavigationLoadOutcome, String>,
navigation: anyhow::Result<NavigationLoadOutcome>,
}
pub(super) struct PendingChildFrameNavigateCommand {
@@ -436,7 +436,7 @@ impl MaterializedNavigationCompletion {
pub struct BackgroundMainDocumentBodyCompletion {
token: DocumentNavigationToken,
state: NavigationDispatchState,
body: Result<CapturedBody, String>,
body: anyhow::Result<CapturedBody>,
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<CapturedBody, String>,
body: anyhow::Result<CapturedBody>,
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<NavigationLoadOutcome, String>,
navigation: anyhow::Result<NavigationLoadOutcome>,
ready_at: std::time::Instant,
}
@@ -539,7 +543,7 @@ impl BackgroundNavigationLifecycleCompletion {
pub(crate) fn new(
token: DocumentNavigationToken,
state: NavigationDispatchState,
navigation: Result<NavigationLoadOutcome, String>,
navigation: anyhow::Result<NavigationLoadOutcome>,
) -> Self {
Self {
token,
@@ -577,7 +581,7 @@ impl BackgroundNavigationCompletion {
pub(crate) fn new(
token: DocumentNavigationToken,
state: NavigationDispatchState,
navigation: Result<NavigationLoadOutcome, String>,
navigation: anyhow::Result<NavigationLoadOutcome>,
) -> 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<CapturedBody, String>,
body: anyhow::Result<CapturedBody>,
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<String>,
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 {
moli-renderer-v8/src/runtime/phase_one/streaming_input.rs
+37 -2
View File
@@ -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<Vec<u8>>),
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::<std::io::Error>()
.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();
moli-renderer-v8/src/runtime/phase_one/streaming_residence.rs
+2 -2
View File
@@ -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() {