diff --git a/moli-protocol-server/src/cdp_scheduler.rs b/moli-protocol-server/src/cdp_scheduler.rs index 414b28024d..476915b271 100644 --- a/moli-protocol-server/src/cdp_scheduler.rs +++ b/moli-protocol-server/src/cdp_scheduler.rs @@ -1019,7 +1019,7 @@ impl CdpScheduler { post_renderer_output_events, ), ) - .with_renderer_output_predecessor(renderer_output_predecessor), + .with_renderer_output_dependencies(renderer_output_predecessor), )) } } @@ -2681,16 +2681,9 @@ impl CdpScheduler { ) -> Result { let (prefix, suffix, boundary, predecessor) = self.materialize_renderer_owner_turn_outcome(outcome); - let mut output = ProtocolOutputSequence::empty(); - if let Some(predecessor) = predecessor { - output.append( - self.project_renderer_output_predecessor_before_devtools_result( - receivers, - &predecessor, - ) - .await?, - ); - } + let mut output = self + .project_renderer_output_dependencies_before_devtools_result(receivers, predecessor) + .await?; // Fetch can commit a Document owned by a different navigation. Its // metadata prefix must remain before that independent renderer boundary. output.append(prefix); @@ -2725,21 +2718,39 @@ impl CdpScheduler { "non-navigation owner turn must not carry a renderer insertion boundary" ); - let mut causal_output = ProtocolOutputSequence::empty(); - if let Some(predecessor) = renderer_output_predecessor { - causal_output.append( - self.project_renderer_output_predecessor_before_devtools_result( - receivers, - &predecessor, - ) - .await?, - ); - } + let mut causal_output = self + .project_renderer_output_dependencies_before_devtools_result( + receivers, + renderer_output_predecessor, + ) + .await?; causal_output.append(output); causal_output.append(post_renderer_output); Ok(causal_output) } + async fn project_renderer_output_dependencies_before_devtools_result( + &mut self, + receivers: &mut CdpSchedulerEventReceivers, + dependencies: moli_protocol::RendererOutputDependencies, + ) -> Result { + let mut output = ProtocolOutputSequence::empty(); + for predecessor in dependencies { + match self + .project_renderer_output_predecessor_before_devtools_result(receivers, &predecessor) + .await + { + Ok(events) => output.append(events), + Err(failure) => { + let (events, error) = failure.into_parts(); + output.append(events); + return Err(RendererOutputTransportFailure::new(output, error)); + } + } + } + Ok(output) + } + fn materialize_protocol_only_turn_outcome( &mut self, outcome: moli_protocol::CdpTurnOutcome, @@ -2771,7 +2782,7 @@ impl CdpScheduler { ProtocolOutputSequence, ProtocolOutputSequence, Option, - Option, + moli_protocol::RendererOutputDependencies, ) { let ( events, @@ -3170,7 +3181,7 @@ impl CdpScheduler { ProtocolOutputSequence, ProtocolOutputSequence, Option, - Option, + moli_protocol::RendererOutputDependencies, ) { let trace_started = moli_trace::cdp_runtime_trace_enabled().then(Instant::now); let outcome = self @@ -3214,7 +3225,7 @@ impl CdpScheduler { let (completion_prefix, mut suffix, renderer_output_boundary, renderer_output_predecessor) = self.drain_background_navigation_completion(completion); assert!( - renderer_output_predecessor.is_none(), + renderer_output_predecessor.is_empty(), "navigation completion must use its exact insertion boundary, not a command predecessor" ); prefix.append(completion_prefix); diff --git a/moli-protocol-server/src/cdp_scheduler/actor.rs b/moli-protocol-server/src/cdp_scheduler/actor.rs index 597af8201b..17dce4ea22 100644 --- a/moli-protocol-server/src/cdp_scheduler/actor.rs +++ b/moli-protocol-server/src/cdp_scheduler/actor.rs @@ -632,17 +632,19 @@ async fn handle_scheduler_input( let (output, post_output, boundary, predecessor) = scheduler.materialize_renderer_owner_turn_outcome(outcome); assert!(boundary.is_none() && post_output.is_empty()); - if !flush_renderer_publication_predecessor( - frontend_router, - scheduler, - scheduler_input_rx, - pending_runtime_deferred_replies, - adapter_scheduler, - predecessor.as_ref(), - ) - .await - { - return false; + for predecessor in predecessor.iter() { + if !flush_renderer_publication_predecessor( + frontend_router, + scheduler, + scheduler_input_rx, + pending_runtime_deferred_replies, + adapter_scheduler, + Some(predecessor), + ) + .await + { + return false; + } } flush_protocol_output_with_runtime_deferred_reply_routing( frontend_router, @@ -938,7 +940,7 @@ impl PendingRuntimeDeferredReplyState { impl RuntimeDeferredReplyCompletion { async fn settle_exact_outputs_before_response(mut self, scheduler: &mut CdpScheduler) -> Self { - if let Some(predecessor) = self.output.take_renderer_output_predecessor() { + for predecessor in self.output.take_renderer_output_dependencies() { assert!( scheduler .state @@ -975,7 +977,7 @@ impl RuntimeDeferredReplyCompletion { output_release_permit, ) = self.output.into_parts(); debug_assert!( - renderer_output_predecessor.is_none(), + renderer_output_predecessor.is_empty(), "deferred Runtime response consumed its renderer output predecessor" ); debug_assert!( @@ -1317,8 +1319,8 @@ fn complete_runtime_deferred_reply_state( CommandTaskStep::Complete(output) => { RuntimeDeferredReplyAdvance::Complete(Box::new(RuntimeDeferredReplyCompletion { output: (*output) - .with_renderer_output_predecessor( - pending.command_context.take_renderer_output_predecessor(), + .with_renderer_output_dependencies( + pending.command_context.take_renderer_output_dependencies(), ) .with_renderer_output_predecessor(renderer_output_predecessor) .with_output_release_permit(pending.output_release_permit), @@ -1348,8 +1350,8 @@ fn complete_runtime_deferred_reply_state( session_id.as_deref(), pending_kind, )) - .with_renderer_output_predecessor( - pending.command_context.take_renderer_output_predecessor(), + .with_renderer_output_dependencies( + pending.command_context.take_renderer_output_dependencies(), ) .with_renderer_output_predecessor(renderer_output_predecessor) .with_output_release_permit(pending.output_release_permit), @@ -1750,8 +1752,8 @@ async fn start_ready_command_dispatch( CommandTaskStep::Complete(output) => { trace_command(command, "command_dispatch_complete_sync", None, None, None); let output = (*output) - .with_renderer_output_predecessor( - command_context.take_renderer_output_predecessor(), + .with_renderer_output_dependencies( + command_context.take_renderer_output_dependencies(), ) .with_output_release_permit(output_release_permit); let ok = flush_completed_command_output( @@ -1917,22 +1919,24 @@ async fn flush_completed_command_output( metadata: &InFlightCommandMetadata, mut output: CommandTurnOutput, ) -> bool { - let renderer_output_predecessor = output.take_renderer_output_predecessor(); - if !flush_renderer_publication_predecessor( - frontend_router, - scheduler, - scheduler_input_rx, - pending_runtime_deferred_replies, - adapter_scheduler, - renderer_output_predecessor.as_ref(), - ) - .await - { - return false; + let renderer_output_dependencies = output.take_renderer_output_dependencies(); + for predecessor in renderer_output_dependencies.iter() { + if !flush_renderer_publication_predecessor( + frontend_router, + scheduler, + scheduler_input_rx, + pending_runtime_deferred_replies, + adapter_scheduler, + Some(predecessor), + ) + .await + { + return false; + } } let mut causal_prefix = ProtocolOutputSequence::empty(); - if let Some(predecessor) = renderer_output_predecessor { + for predecessor in renderer_output_dependencies { causal_prefix.append( scheduler.complete_renderer_output_predecessor_before_runtime_response(&predecessor), ); @@ -1953,7 +1957,7 @@ async fn flush_completed_command_output( output_release_permit, ) = output.into_parts(); debug_assert!( - renderer_output_predecessor.is_none(), + renderer_output_predecessor.is_empty(), "command response consumed its exact renderer output predecessor before decomposition" ); let internal_response = @@ -2365,8 +2369,8 @@ async fn handle_pending_command_completion( ); } let output = (*output) - .with_renderer_output_predecessor( - state.command_context.take_renderer_output_predecessor(), + .with_renderer_output_dependencies( + state.command_context.take_renderer_output_dependencies(), ) .with_output_release_permit(state.output_release_permit); let ok = flush_completed_command_output( @@ -2506,6 +2510,110 @@ mod tests { }) } + #[tokio::test] + async fn multi_stream_command_keeps_response_and_flush_permit_until_every_dependency() { + use moli_core::{ + PageId, RendererOutputCursor, RendererOutputItem, RendererOutputPublication, + RendererOutputRecord, RendererOutputStreamControl, RendererOutputStreamIdentity, + RendererProtocolObservation, + }; + use moli_protocol::RendererOutputDependencies; + + tokio::task::LocalSet::new().run_until(async { + for internal in [false, true] { + let router = CdpFrontendRouter::new(); + let (sink, mut writer) = crate::cdp_writer::CdpSocketSink::with_stalled_writer_for_test(8); + router.register_browser_frontend(1, "browser-session".into(), sink).unwrap(); + let command = ParsedCdpCommand::parse_str(r#"{"id":42,"method":"Target.detachFromTarget"}"#).unwrap(); + let command = if internal { + command + } else { + let Some(CdpPreparedFrontendCommand::Command(command)) = router.prepare_command(1, Ok(command)) else { + panic!("frontend command must be admitted"); + }; + command + }; + let mut metadata = command_metadata(&command); + let (internal_tx, mut internal_rx) = oneshot::channel(); + let (owner, _owner_rx) = mpsc::unbounded_channel(); + if internal { + metadata.internal_response = Some(Rc::new(RefCell::new(Some(InternalCommandResponse { + sender: internal_tx, + registration: None, + owner: owner.downgrade(), + })))); + } + let streams = [71, 72].map(|page| RendererOutputStreamIdentity::new_page_for_protocol_test(PageId::new_for_testing(page))); + let mut connection = CdpConnection::new(); + for stream in streams { + moli_protocol::test_support::bind_renderer_page_output_stream_to_retired_owner(&mut connection, stream); + } + let (mut scheduler, protocol_completions) = CdpScheduler::new(connection); + let state = scheduler.state.handle(); + let (permit, context) = state.with_state(|state| state.begin_command_response_flush_permit()).unwrap(); + let flushed = moli_protocol::test_support::command_response_flush_observer(&context); + for stream in streams { + assert!(scheduler.ingest_renderer_publication_now(RendererOutputStreamControl::Opened { stream }.into()).is_empty()); + } + let first = RendererOutputCursor::new_for_test(streams[0], 3); + let second = RendererOutputCursor::new_for_test(streams[1], 1); + let mut dependencies = RendererOutputDependencies::default(); + dependencies.extend([first, second].map(RendererOutputFence::new_for_test)); + let output = CommandTurnOutput::new( + ProtocolOutputSequence::from_background_event(BackgroundProtocolEvent::command_success(metadata.id, metadata.session_id.as_deref(), json!({}))), + Vec::new(), + ).with_renderer_output_dependencies(dependencies) + .with_output_release_permit(CommandOutputReleasePermit::new(permit, None)); + let publication = |cursor| RendererOutputTransportMessage::Publication(RendererOutputPublication::new_for_test( + cursor, + vec![RendererOutputRecord::new_for_test(RendererOutputItem::Observation(RendererProtocolObservation::RuntimeLifecycleError { + text: "historical renderer output".into(), execution_context_id: None, + }))], + )); + let (renderer_tx, renderer_rx) = moli_core::renderer_output_transport_channel(); + for sequence in 1..=3 { + renderer_tx.send(publication(RendererOutputCursor::new_for_test(streams[0], sequence))).unwrap(); + } + let mut inputs = empty_scheduler_inputs(); + inputs.state_events_ready = scheduler.state.scheduler_events_ready(); + inputs.protocol_work_completion_rx = protocol_completions; + inputs.renderer_publication_rx = renderer_rx; + let mut deferred = VecDeque::new(); + let mut adapter = ProtocolAdapterScheduler::default(); + let flush = flush_completed_command_output(&router, &mut scheduler, &mut inputs, &mut deferred, &mut adapter, &metadata, output); + tokio::pin!(flush); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + tokio::select! { + result = &mut flush => panic!("response completed with a missing Page dependency: {result}"), + _ = tokio::task::yield_now() => { + if state.with_state(|state| state.renderer_output_cursor_is_projected(first)).unwrap() { + break; + } + } + } + } + }).await.expect("first Page output is projected"); + assert!(!*flushed.borrow(), "the command still owns its flush permit"); + assert!(writer.is_empty(), "no response may reach the frontend yet"); + if internal { + assert!(matches!(internal_rx.try_recv(), Err(oneshot::error::TryRecvError::Empty))); + } + renderer_tx.send(publication(second)).unwrap(); + assert!(tokio::time::timeout(std::time::Duration::from_secs(2), &mut flush).await.unwrap()); + assert!(*flushed.borrow()); + if internal { + let result = internal_rx.try_recv().expect("internal response is delivered before flush completes").expect("internal command retains its original response"); + assert_eq!(result.response, json!({"id": 42, "result": {}})); + assert!(writer.is_empty(), "private response must not leak to the frontend"); + } else { + assert_eq!(writer.take_message(), json!({"id": 42, "result": {}})); + assert!(writer.is_empty(), "the command response is emitted once"); + } + } + }).await; + } + #[tokio::test] async fn internal_attach_registers_only_a_session_still_open_at_flush_confirmation() { for phase in ["open", "closing", "closed"] { @@ -2758,7 +2866,7 @@ mod tests { "id": 1, "method": "Target.attachToTarget", "params": {"targetId": moli_protocol::DEFAULT_CDP_PAGE_TARGET_ID, "flatten": true}, }).to_string()).await.unwrap().into_protocol_event_parts(); - assert!(predecessor.is_none()); + assert!(predecessor.is_empty()); let attached: Vec<_> = attached.into_iter().map(BackgroundProtocolEvent::into_protocol_message).collect(); let session_id = attached.iter().find(|message| message["id"] == 1).unwrap()["result"]["sessionId"].as_str().unwrap().to_owned(); let enable = json!({"id": 2, "sessionId": session_id, "method": "Fetch.enable", "params": { @@ -2769,7 +2877,7 @@ mod tests { "id": 3, "sessionId": session_id, "method": "Page.navigate", "params": {"url": "http://unused.example/held"}, }).to_string()).await.unwrap().into_protocol_event_parts(); - assert!(predecessor.is_none()); + assert!(predecessor.is_empty()); let paused: Vec<_> = paused.into_iter().map(BackgroundProtocolEvent::into_protocol_message).collect(); let request_id = paused.iter().find(|message| message["method"] == "Fetch.requestPaused").unwrap()["params"]["requestId"].clone(); let mut command = json!({ diff --git a/moli-protocol-server/src/cdp_scheduler/command_dispatch.rs b/moli-protocol-server/src/cdp_scheduler/command_dispatch.rs index c16387b4c3..f9df64e142 100644 --- a/moli-protocol-server/src/cdp_scheduler/command_dispatch.rs +++ b/moli-protocol-server/src/cdp_scheduler/command_dispatch.rs @@ -1,4 +1,4 @@ -use moli_protocol::{BackgroundProtocolEvent, CdpSchedulerEvent}; +use moli_protocol::{BackgroundProtocolEvent, CdpSchedulerEvent, RendererOutputDependencies}; use super::{CommandOutputReleasePermit, ProtocolOutputSequence}; @@ -8,7 +8,7 @@ pub(crate) struct CommandTurnOutput { renderer_output_boundary: Option, post_response_events: Vec, post_flush_scheduler_events: Vec, - renderer_output_predecessor: Option, + renderer_output_dependencies: RendererOutputDependencies, output_release_permit: Option, } @@ -49,7 +49,7 @@ impl CommandTurnOutput { renderer_output_boundary: None, post_response_events, post_flush_scheduler_events, - renderer_output_predecessor: None, + renderer_output_dependencies: RendererOutputDependencies::default(), output_release_permit, } } @@ -84,15 +84,22 @@ impl CommandTurnOutput { predecessor: Option, ) -> Self { if let Some(predecessor) = predecessor { - predecessor.merge_into_same_stream_tail(&mut self.renderer_output_predecessor); + self.renderer_output_dependencies + .add_same_stream(predecessor); } self } - pub(crate) fn take_renderer_output_predecessor( - &mut self, - ) -> Option { - self.renderer_output_predecessor.take() + pub(crate) fn with_renderer_output_dependencies( + mut self, + dependencies: RendererOutputDependencies, + ) -> Self { + self.renderer_output_dependencies.merge(dependencies); + self + } + + pub(crate) fn take_renderer_output_dependencies(&mut self) -> RendererOutputDependencies { + std::mem::take(&mut self.renderer_output_dependencies) } /// Place output projected from the command's exact renderer cause before @@ -119,7 +126,7 @@ impl CommandTurnOutput { Option, Vec, Vec, - Option, + RendererOutputDependencies, Option, ) { ( @@ -128,7 +135,7 @@ impl CommandTurnOutput { self.renderer_output_boundary, self.post_response_events, self.post_flush_scheduler_events, - self.renderer_output_predecessor, + self.renderer_output_dependencies, self.output_release_permit, ) } diff --git a/moli-protocol-server/src/cdp_writer.rs b/moli-protocol-server/src/cdp_writer.rs index c6e57076a6..02f98b61a3 100644 --- a/moli-protocol-server/src/cdp_writer.rs +++ b/moli-protocol-server/src/cdp_writer.rs @@ -369,6 +369,10 @@ pub(crate) struct StalledCdpSocketWriter { #[cfg(test)] impl StalledCdpSocketWriter { + pub(crate) fn is_empty(&self) -> bool { + self.output_rx.is_empty() + } + pub(crate) fn take_message(&mut self) -> serde_json::Value { let command = self .output_rx diff --git a/moli-protocol-server/src/protocol_server/tests/cdp_dynamic_page/browser_close.rs b/moli-protocol-server/src/protocol_server/tests/cdp_dynamic_page/browser_close.rs index 37abe5fdde..96527a8569 100644 --- a/moli-protocol-server/src/protocol_server/tests/cdp_dynamic_page/browser_close.rs +++ b/moli-protocol-server/src/protocol_server/tests/cdp_dynamic_page/browser_close.rs @@ -69,6 +69,173 @@ async fn close_instance(socket: &mut TestCdpSocket, id: u64) { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn browser_disconnect_with_fetch_on_two_pages_preserves_each_output_stream() { + for (leave_paused, abrupt_disconnect) in + [(false, false), (true, false), (false, true), (true, true)] + { + let fixture = Router::new().route( + "/page", + get(|| async { axum::response::Html("

Fetch teardown

") }), + ); + let (fixture_addr, _fixture_server) = + spawn_dedicated_fixture_server(fixture, "multi-page-fetch-disconnect"); + let (addr, task, _) = instance(None).await; + let mut root = browser(addr).await; + let targets = [ + create_dynamic_target(&mut root, 1).await, + create_dynamic_target(&mut root, 2).await, + ]; + let attached = send_cdp_command( + &mut root, + 3, + "Target.setAutoAttach", + None, + json!({"autoAttach": true, "waitForDebuggerOnStart": false, + "flatten": true, "filter": [{"type": "page"}]}), + ) + .await; + assert_eq!(response_by_id(&attached, 3)["result"], json!({})); + let mut sessions = Vec::new(); + for (index, target) in targets.iter().enumerate() { + let session = attached + .iter() + .find(|event| { + event["method"] == "Target.attachedToTarget" + && event["params"]["targetInfo"]["targetId"] == *target + }) + .expect("each Page is auto-attached")["params"]["sessionId"] + .as_str() + .unwrap() + .to_owned(); + let enabled = send_cdp_command( + &mut root, + 10, + "Fetch.enable", + Some(&session), + if index == 0 { + json!({}) + } else { + json!({"patterns": [{"urlPattern": "*"}]}) + }, + ) + .await; + assert_eq!(response_by_id(&enabled, 10)["result"], json!({})); + send_cdp_command_without_wait( + &mut root, + 12, + "Page.navigate", + Some(&session), + json!({"url": format!("http://{fixture_addr}/page?{index}")}), + ) + .await; + let paused = recv_until_match(&mut root, |event| { + event["method"] == "Fetch.requestPaused" && event["sessionId"] == session + }) + .await; + let request_id = paused.last().unwrap()["params"]["requestId"].clone(); + let continued = send_cdp_command( + &mut root, + 13, + "Fetch.continueRequest", + Some(&session), + json!({"requestId": request_id}), + ) + .await; + assert_eq!(response_by_id(&continued, 13)["result"], json!({})); + wait_for_cdp_runtime_string( + &mut root, + &session, + 100, + "document.querySelector('p')?.textContent ?? ''", + "Fetch teardown", + ) + .await; + if leave_paused { + send_cdp_command_without_wait( + &mut root, + 20, + "Runtime.evaluate", + Some(&session), + json!({"expression": "globalThis.fetchOutcome = 'pending'; fetch('/paused').then(() => fetchOutcome = 'fulfilled', () => fetchOutcome = 'rejected'); 'started'"}), + ) + .await; + recv_until_match(&mut root, |event| { + event["method"] == "Fetch.requestPaused" + && event["sessionId"] == session + && event["params"]["request"]["url"] + .as_str() + .is_some_and(|url| url.ends_with("/paused")) + }) + .await; + } + sessions.push(session); + } + + if !abrupt_disconnect { + // Playwright disconnects a CDP browser by disabling auto-attachment + // before closing its socket. Both Fetch handlers produce independent + // Page output prefixes, even when every request was continued. + let detached = send_cdp_command( + &mut root, + 30, + "Target.setAutoAttach", + None, + json!({"autoAttach": false, "waitForDebuggerOnStart": false, "flatten": true}), + ) + .await; + assert_eq!(response_by_id(&detached, 30)["result"], json!({})); + let response_index = detached.iter().position(|event| event["id"] == 30).unwrap(); + for session in &sessions { + let detach_index = detached + .iter() + .position(|event| { + event["method"] == "Target.detachedFromTarget" + && event["params"]["sessionId"] == *session + }) + .expect("each Page session detaches before the command response"); + assert!(detach_index < response_index, "{detached:#?}"); + } + } + root.close(None) + .await + .expect("disconnect Playwright frontend"); + closed(&mut root).await; + let mut observer = browser(addr).await; + if leave_paused { + // Domain disposal also disables Network observation. Verify the + // native requests actually settle through fresh sessions rather + // than requiring notifications on the removed subscriptions. + for target in &targets { + let attached = send_cdp_command( + &mut observer, + 31, + "Target.attachToTarget", + None, + json!({"targetId": target, "flatten": true}), + ) + .await; + let session = response_by_id(&attached, 31)["result"]["sessionId"] + .as_str() + .unwrap(); + wait_for_cdp_runtime_string( + &mut observer, + session, + 100, + "globalThis.fetchOutcome", + "rejected", + ) + .await; + } + } + let version = + send_cdp_command(&mut observer, 40, "Browser.getVersion", None, json!({})).await; + assert!(response_by_id(&version, 40)["result"]["protocolVersion"].is_string()); + close_instance(&mut observer, 41).await; + joined(task).await; + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn browser_close_joins_its_listener_connections_and_storage_only() { assert_instance_shutdown(false).await; diff --git a/moli-protocol/src/conn.rs b/moli-protocol/src/conn.rs index 9bc2a87549..204b6c2e93 100644 --- a/moli-protocol/src/conn.rs +++ b/moli-protocol/src/conn.rs @@ -83,6 +83,8 @@ mod page_state; pub(crate) use page_host_scope::ProtocolPageHostScope; mod popup_activation_work; mod popup_navigation_work; +mod renderer_output_dependencies; +pub use renderer_output_dependencies::RendererOutputDependencies; mod protocol_output; pub(crate) use protocol_output::PendingBackgroundNavigationProtocol; pub use protocol_output::{ @@ -430,7 +432,7 @@ pub struct CommandDispatchContext { post_renderer_output_events: Vec, renderer_output_boundary: Option, post_response_events: Vec, - renderer_output_predecessor: Option, + renderer_output_dependencies: RendererOutputDependencies, command_failure_cleanup: Rc>>, deferred_command_events: Vec, runtime_output_projection_permits: @@ -446,7 +448,7 @@ impl CommandDispatchContext { post_renderer_output_events: Vec::new(), renderer_output_boundary: None, post_response_events: Vec::new(), - renderer_output_predecessor: None, + renderer_output_dependencies: RendererOutputDependencies::default(), command_failure_cleanup: Rc::default(), deferred_command_events: Vec::new(), runtime_output_projection_permits: Rc::default(), @@ -592,18 +594,38 @@ impl CommandDispatchContext { &mut self, predecessor: moli_core::RendererOutputFence, ) { - predecessor.merge_into_same_stream_tail(&mut self.renderer_output_predecessor); + self.renderer_output_dependencies + .add_same_stream(predecessor); } pub(crate) fn renderer_output_predecessor_mut( &mut self, ) -> &mut Option { - &mut self.renderer_output_predecessor + self.renderer_output_dependencies.single_stream_mut() } #[doc(hidden)] pub fn take_renderer_output_predecessor(&mut self) -> Option { - self.renderer_output_predecessor.take() + self.take_renderer_output_dependencies() + .into_single_stream() + } + + /// Browser/Target commands explicitly compose independently owned renderer + /// streams without moving their response out of the command lifecycle. + pub(crate) fn add_renderer_output_dependency(&mut self, fence: moli_core::RendererOutputFence) { + self.renderer_output_dependencies.add(fence); + } + + pub(crate) fn extend_renderer_output_dependencies( + &mut self, + dependencies: RendererOutputDependencies, + ) { + self.renderer_output_dependencies.merge(dependencies); + } + + #[doc(hidden)] + pub fn take_renderer_output_dependencies(&mut self) -> RendererOutputDependencies { + std::mem::take(&mut self.renderer_output_dependencies) } } diff --git a/moli-protocol/src/conn/automation_command.rs b/moli-protocol/src/conn/automation_command.rs index 5f32e2278c..4fc31305e6 100644 --- a/moli-protocol/src/conn/automation_command.rs +++ b/moli-protocol/src/conn/automation_command.rs @@ -98,6 +98,7 @@ impl AutomationCommandDispatchOutcome { /// Commands with an independent renderer boundary must consume the typed /// owner turn to preserve the insertion point between their output pieces. + /// Composite renderer dependencies likewise require `into_result_and_owner_turn`. #[track_caller] pub fn into_complete_parts( self, @@ -115,7 +116,12 @@ impl AutomationCommandDispatchOutcome { ); assert!(after.is_empty()); before.extend(post); - (self.result, scheduler, before, predecessor) + ( + self.result, + scheduler, + before, + predecessor.into_single_stream(), + ) } } diff --git a/moli-protocol/src/conn/automation_command/dispatch.rs b/moli-protocol/src/conn/automation_command/dispatch.rs index 8784b4ec97..c78f90badc 100644 --- a/moli-protocol/src/conn/automation_command/dispatch.rs +++ b/moli-protocol/src/conn/automation_command/dispatch.rs @@ -213,7 +213,7 @@ impl CdpConnection { prefix, post, scheduler, ) .with_renderer_output_boundary(boundary, after) - .with_renderer_output_predecessor(predecessor); + .with_renderer_output_dependencies(predecessor); AutomationCommandTaskStep::Complete(Box::new(AutomationCommandDispatchOutcome { result, output, diff --git a/moli-protocol/src/conn/command_workflow.rs b/moli-protocol/src/conn/command_workflow.rs index de1eb110e2..fab801a5db 100644 --- a/moli-protocol/src/conn/command_workflow.rs +++ b/moli-protocol/src/conn/command_workflow.rs @@ -266,9 +266,7 @@ impl CommandWorkflowContext { for permit in local.take_runtime_output_projection_permits() { target.retain_runtime_output_projection_permit(permit); } - if let Some(predecessor) = local.take_renderer_output_predecessor() { - target.set_renderer_output_predecessor(predecessor); - } + target.extend_renderer_output_dependencies(local.take_renderer_output_dependencies()); } /// Borrow the command context only for this synchronous state operation. diff --git a/moli-protocol/src/conn/dispatch.rs b/moli-protocol/src/conn/dispatch.rs index 4e0bed40bc..ff8feee90f 100644 --- a/moli-protocol/src/conn/dispatch.rs +++ b/moli-protocol/src/conn/dispatch.rs @@ -382,9 +382,7 @@ pub(super) fn append_command_output_plan( command_id: Option, session_id: Option<&str>, ) { - if let Some(predecessor) = plan.take_renderer_output_predecessor() { - command_context.set_renderer_output_predecessor(predecessor); - } + command_context.extend_renderer_output_dependencies(plan.take_renderer_output_dependencies()); let (events, renderer_output_boundary, post_renderer_output_events, post_response_events) = plan.into_renderer_fenced_background_and_post_response_events(command_id, session_id); command_context @@ -450,7 +448,7 @@ impl CdpConnection { self.take_scheduler_events(), ) .with_renderer_output_boundary(renderer_output_boundary, post_renderer_output_events) - .with_renderer_output_predecessor(command_context.take_renderer_output_predecessor()), + .with_renderer_output_dependencies(command_context.take_renderer_output_dependencies()), ) } @@ -1648,7 +1646,7 @@ impl CdpConnection { scheduler_events, ) .with_renderer_output_boundary(renderer_output_boundary, post_renderer_output_events) - .with_renderer_output_predecessor(command_context.take_renderer_output_predecessor()), + .with_renderer_output_dependencies(command_context.take_renderer_output_dependencies()), ) } } diff --git a/moli-protocol/src/conn/dispatch_tests/mod.rs b/moli-protocol/src/conn/dispatch_tests/mod.rs index f02e290431..fe38372209 100644 --- a/moli-protocol/src/conn/dispatch_tests/mod.rs +++ b/moli-protocol/src/conn/dispatch_tests/mod.rs @@ -59,7 +59,7 @@ fn command_wait_failure_preserves_context_events_and_does_not_ack_early() { let (before, renderer_tail, boundary, after, _, predecessor) = outcome.into_renderer_owner_turn_parts(); assert!(renderer_tail.is_empty()); - assert!(boundary.is_none() && predecessor.is_none()); + assert!(boundary.is_none() && predecessor.is_empty()); let before = before .into_iter() .map(BackgroundProtocolEvent::into_protocol_message) diff --git a/moli-protocol/src/conn/document_activity_work.rs b/moli-protocol/src/conn/document_activity_work.rs index 7e52afda8b..f5ffbc45b1 100644 --- a/moli-protocol/src/conn/document_activity_work.rs +++ b/moli-protocol/src/conn/document_activity_work.rs @@ -22,6 +22,6 @@ impl CdpConnection { Err(error) => tracing::debug!(%error, "document activity owner update ended"), } CdpTurnOutcome::new_with_protocol_events(Vec::new(), self.take_scheduler_events()) - .with_renderer_output_predecessor(context.take_renderer_output_predecessor()) + .with_renderer_output_dependencies(context.take_renderer_output_dependencies()) } } diff --git a/moli-protocol/src/conn/protocol_output/workflow.rs b/moli-protocol/src/conn/protocol_output/workflow.rs index 6a290ca3f7..6ded1d5786 100644 --- a/moli-protocol/src/conn/protocol_output/workflow.rs +++ b/moli-protocol/src/conn/protocol_output/workflow.rs @@ -109,31 +109,26 @@ impl CdpConnection { ) -> Vec { let (mut events, boundary, mut after) = context.take_renderer_fenced_protocol_events(); after.extend(context.take_post_response_events()); - let predecessor = context.take_renderer_output_predecessor(); + let dependencies = context.take_renderer_output_dependencies(); if let Some(boundary) = boundary { - if let Some(predecessor) = predecessor.as_ref() - && !events.is_empty() - { - self.publish_renderer_fenced_protocol_output( + if !dependencies.is_empty() && !events.is_empty() { + self.publish_renderer_fenced_protocol_output_after( std::mem::take(&mut events), - predecessor.clone(), + dependencies.clone().into_iter().collect(), target_id.clone(), ); } // The insertion also releases independently transported Native // observations. Its receipt must survive an empty wire suffix. - let mut predecessors = Vec::new(); - predecessors.extend(predecessor); + let mut predecessors = dependencies.into_iter().collect::>(); predecessors.push(boundary); self.publish_renderer_fenced_protocol_output_after(after, predecessors, target_id); } else { events.extend(after); - if let Some(predecessor) = predecessor - && !events.is_empty() - { - self.publish_renderer_fenced_protocol_output( + if !dependencies.is_empty() && !events.is_empty() { + self.publish_renderer_fenced_protocol_output_after( std::mem::take(&mut events), - predecessor, + dependencies.into_iter().collect(), target_id, ); } diff --git a/moli-protocol/src/conn/renderer_output_dependencies.rs b/moli-protocol/src/conn/renderer_output_dependencies.rs new file mode 100644 index 0000000000..836e869ca4 --- /dev/null +++ b/moli-protocol/src/conn/renderer_output_dependencies.rs @@ -0,0 +1,204 @@ +use moli_core::RendererOutputFence; + +/// Exact renderer FIFO prefixes required before one protocol result is exposed. +/// +/// Browser-level commands can compose independent Page/Worker owners. Each +/// stream retains its own lifetime lease and latest cursor; sequence numbers +/// from different streams are never compared. The usual single-owner case +/// does not allocate. +#[derive(Clone, Debug, Default)] +pub struct RendererOutputDependencies { + first: Option, + additional: Vec, + explicitly_composed: bool, +} + +impl RendererOutputDependencies { + pub fn is_empty(&self) -> bool { + self.first.is_none() + } + + pub fn iter(&self) -> impl Iterator { + self.first.iter().chain(&self.additional) + } + + pub fn add(&mut self, fence: RendererOutputFence) { + self.explicitly_composed = true; + let Some(first) = self.first.as_mut() else { + self.first = Some(fence); + return; + }; + let stream = fence.cursor().stream(); + if let Some(tail) = std::iter::once(first) + .chain(&mut self.additional) + .find(|tail| tail.cursor().stream() == stream) + { + *tail = tail.clone().latest_in_same_stream(fence); + } else { + self.additional.push(fence); + } + } + + /// Preserve the declaring command's ownership contract across plan and + /// workflow handoffs. Ordinary single-owner plans must not silently gain + /// permission to join unrelated streams merely because they are combined. + pub fn merge(&mut self, other: Self) { + self.explicitly_composed |= other.explicitly_composed; + for fence in other { + if self.explicitly_composed { + self.add(fence); + } else { + self.add_same_stream(fence); + } + } + } + + /// Ordinary renderer-owner operations still join exactly one source FIFO. + /// Only explicit composition through `add`/`extend` permits other streams. + pub fn add_same_stream(&mut self, fence: RendererOutputFence) { + fence.merge_into_same_stream_tail(self.single_stream_mut()); + } + + pub(crate) fn single_stream_mut(&mut self) -> &mut Option { + assert!( + self.additional.is_empty(), + "a single renderer owner cannot consume composite output dependencies" + ); + &mut self.first + } + + #[track_caller] + pub fn into_single_stream(self) -> Option { + assert!( + self.additional.is_empty(), + "composite output dependencies require consuming every renderer stream" + ); + self.first + } +} + +impl From> for RendererOutputDependencies { + fn from(first: Option) -> Self { + Self { + first, + additional: Vec::new(), + explicitly_composed: false, + } + } +} + +impl Extend for RendererOutputDependencies { + fn extend>(&mut self, fences: T) { + for fence in fences { + self.add(fence); + } + } +} + +impl IntoIterator for RendererOutputDependencies { + type Item = RendererOutputFence; + type IntoIter = std::iter::Chain< + std::option::IntoIter, + std::vec::IntoIter, + >; + + fn into_iter(self) -> Self::IntoIter { + self.first.into_iter().chain(self.additional) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use moli_core::{PageId, RendererOutputCursor, RendererOutputStreamIdentity}; + + #[test] + fn dependencies_keep_the_latest_cursor_in_each_exact_stream() { + let first = + RendererOutputStreamIdentity::new_page_for_protocol_test(PageId::new_for_testing(71)); + let second = + RendererOutputStreamIdentity::new_page_for_protocol_test(PageId::new_for_testing(72)); + // Reusing a Page ID does not reuse its renderer agent or stream epoch. + let replacement = + RendererOutputStreamIdentity::new_page_for_protocol_test(PageId::new_for_testing(71)); + let mut dependencies = RendererOutputDependencies::default(); + assert!(dependencies.is_empty()); + for (stream, sequence) in [ + (first, 3), + (second, 1), + (first, 2), + (first, 5), + (replacement, 1), + ] { + dependencies.add(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(stream, sequence), + )); + } + assert_eq!( + dependencies + .iter() + .map(|fence| fence.cursor()) + .collect::>(), + [(first, 5), (second, 1), (replacement, 1)] + .map(|(stream, sequence)| RendererOutputCursor::new_for_test(stream, sequence)) + ); + } + + #[test] + #[should_panic(expected = "one command response cannot join unrelated renderer output streams")] + fn ordinary_owner_dependencies_keep_the_single_stream_constraint() { + let mut dependencies = RendererOutputDependencies::default(); + for page in [71, 72] { + let stream = RendererOutputStreamIdentity::new_page_for_protocol_test( + PageId::new_for_testing(page), + ); + dependencies.add_same_stream(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(stream, 1), + )); + } + } + + #[test] + #[should_panic(expected = "one command response cannot join unrelated renderer output streams")] + fn merging_ordinary_plans_does_not_enable_cross_stream_composition() { + let mut dependencies = RendererOutputDependencies::default(); + for page in [71, 72] { + let stream = RendererOutputStreamIdentity::new_page_for_protocol_test( + PageId::new_for_testing(page), + ); + dependencies.merge( + Some(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(stream, 1), + )) + .into(), + ); + } + } + + #[test] + fn explicit_composition_survives_plan_handoffs_with_one_stream() { + for composed_first in [false, true] { + let first = RendererOutputStreamIdentity::new_page_for_protocol_test( + PageId::new_for_testing(71), + ); + let second = RendererOutputStreamIdentity::new_page_for_protocol_test( + PageId::new_for_testing(72), + ); + let mut composed = RendererOutputDependencies::default(); + composed.add(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(first, 1), + )); + let ordinary = Some(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(second, 1), + )) + .into(); + let mut left = composed; + let mut right = ordinary; + if !composed_first { + std::mem::swap(&mut left, &mut right); + } + left.merge(right); + assert_eq!(left.iter().count(), 2); + } + } +} diff --git a/moli-protocol/src/conn/runtime_eval/helper_workflow.rs b/moli-protocol/src/conn/runtime_eval/helper_workflow.rs index 816956b068..9176387954 100644 --- a/moli-protocol/src/conn/runtime_eval/helper_workflow.rs +++ b/moli-protocol/src/conn/runtime_eval/helper_workflow.rs @@ -455,7 +455,7 @@ mod tests { let (_, _, _, _, _, fence) = output.into_renderer_owner_turn_parts(); if boundary == 2 { assert!( - fence.is_some(), + !fence.is_empty(), "accepted Native output keeps its original position even when state settlement is rejected" ); } @@ -634,7 +634,7 @@ mod tests { }; assert_eq!(value.value, json!(42)); assert!( - fence.is_some(), + !fence.is_empty(), "ready loss preserves the accepted Native cursor" ); } else { diff --git a/moli-protocol/src/conn/scheduler_state.rs b/moli-protocol/src/conn/scheduler_state.rs index ce8d255f4e..24745d18e5 100644 --- a/moli-protocol/src/conn/scheduler_state.rs +++ b/moli-protocol/src/conn/scheduler_state.rs @@ -1,6 +1,6 @@ use std::collections::VecDeque; -use super::{BackgroundProtocolEvent, NavigationBackgroundEvent}; +use super::{BackgroundProtocolEvent, NavigationBackgroundEvent, RendererOutputDependencies}; #[cfg(test)] use crate::domains::command_output::protocol_message_background_event; use serde_json::{Value, json}; @@ -29,8 +29,9 @@ pub struct CdpTurnOutcome { scheduler_events: Vec, } -/// One renderer-owner turn whose exact publication predecessor must be -/// projected before its protocol result becomes observable. +/// One turn whose exact renderer dependencies must be projected before its +/// protocol result becomes observable. Ordinary owner turns use one stream; +/// Browser/Target commands explicitly compose several owners. /// /// This wrapper is intentionally a different type from [`CdpTurnOutcome`]. A /// scheduler hook cannot pass it to a protocol-only turn consumer and rely on @@ -39,7 +40,7 @@ pub struct CdpTurnOutcome { #[derive(Debug)] pub struct CdpRendererOwnerTurnOutcome { turn: CdpTurnOutcome, - renderer_output_predecessor: Option, + renderer_output_dependencies: RendererOutputDependencies, } impl CdpTurnOutcome { @@ -120,10 +121,17 @@ impl CdpTurnOutcome { pub(crate) fn with_renderer_output_predecessor( self, predecessor: Option, + ) -> CdpRendererOwnerTurnOutcome { + self.with_renderer_output_dependencies(predecessor.into()) + } + + pub(crate) fn with_renderer_output_dependencies( + self, + dependencies: RendererOutputDependencies, ) -> CdpRendererOwnerTurnOutcome { CdpRendererOwnerTurnOutcome { turn: self, - renderer_output_predecessor: predecessor, + renderer_output_dependencies: dependencies, } } @@ -179,13 +187,15 @@ impl CdpTurnOutcome { impl CdpRendererOwnerTurnOutcome { pub fn renderer_output_predecessor(&self) -> Option { - self.renderer_output_predecessor.clone() + self.renderer_output_dependencies + .clone() + .into_single_stream() } #[cfg(test)] pub fn into_parts(self) -> (Vec, Vec) { assert!( - self.renderer_output_predecessor.is_none(), + self.renderer_output_dependencies.is_empty(), "tests must project an exact renderer predecessor instead of flattening it" ); self.turn.into_parts() @@ -196,13 +206,13 @@ impl CdpRendererOwnerTurnOutcome { ) -> ( Vec, Vec, - Option, + RendererOutputDependencies, ) { let (protocol_events, scheduler_events) = self.turn.into_protocol_event_parts(); ( protocol_events, scheduler_events, - self.renderer_output_predecessor, + self.renderer_output_dependencies, ) } @@ -214,7 +224,7 @@ impl CdpRendererOwnerTurnOutcome { Option, Vec, Vec, - Option, + RendererOutputDependencies, ) { let ( protocol_events, @@ -229,7 +239,7 @@ impl CdpRendererOwnerTurnOutcome { renderer_output_boundary, post_response_events, scheduler_events, - self.renderer_output_predecessor, + self.renderer_output_dependencies, ) } } diff --git a/moli-protocol/src/conn/state_access/auto_attach/tests.rs b/moli-protocol/src/conn/state_access/auto_attach/tests.rs index b5e185cbe2..5b78ff987a 100644 --- a/moli-protocol/src/conn/state_access/auto_attach/tests.rs +++ b/moli-protocol/src/conn/state_access/auto_attach/tests.rs @@ -469,7 +469,7 @@ async fn auto_attach_multi_context_sweep_keeps_page_streams_independent() { }; let (events, _, predecessor) = outcome.into_protocol_event_parts(); assert!( - predecessor.is_none(), + predecessor.is_empty(), "a Browser policy ACK cannot join independent Page tails" ); let messages = events diff --git a/moli-protocol/src/conn/state_access/automation_configuration.rs b/moli-protocol/src/conn/state_access/automation_configuration.rs index 7be9eee789..56e1d1d595 100644 --- a/moli-protocol/src/conn/state_access/automation_configuration.rs +++ b/moli-protocol/src/conn/state_access/automation_configuration.rs @@ -1559,7 +1559,7 @@ mod tests { ); if is_input { assert!( - predecessor.is_some(), + !predecessor.is_empty(), "the accepted Native input turn retains its exact cursor" ); } diff --git a/moli-protocol/src/conn/state_access/browser_context_disposal/tests.rs b/moli-protocol/src/conn/state_access/browser_context_disposal/tests.rs index bbdcc3b430..db082fdad7 100644 --- a/moli-protocol/src/conn/state_access/browser_context_disposal/tests.rs +++ b/moli-protocol/src/conn/state_access/browser_context_disposal/tests.rs @@ -162,7 +162,7 @@ async fn context_disposal_owns_native_close_through_cancellation_and_ready_loss( }; let (events, _, predecessor) = outcome.into_protocol_event_parts(); assert!( - predecessor.is_none(), + predecessor.is_empty(), "a Browser ACK cannot join unrelated Page streams" ); let messages = events diff --git a/moli-protocol/src/conn/state_access/command_dispatch.rs b/moli-protocol/src/conn/state_access/command_dispatch.rs index 25726db07f..7b6326f44d 100644 --- a/moli-protocol/src/conn/state_access/command_dispatch.rs +++ b/moli-protocol/src/conn/state_access/command_dispatch.rs @@ -556,9 +556,7 @@ fn retain_step( ), ); } - if let Some(predecessor) = predecessor { - output.set_renderer_output_predecessor(predecessor); - } + output.extend_renderer_output_dependencies(predecessor); // Cancellation and recovery must preserve the accepted terminal // and its exact renderer output fence. context.retain_finalized_command(output); @@ -751,9 +749,7 @@ impl ProtocolWorkflowState<'_> { } output.extend_background_events(after); output.extend_background_events(post); - if let Some(predecessor) = predecessor { - output.set_renderer_output_predecessor(predecessor); - } + output.extend_renderer_output_dependencies(predecessor); context.retain_output(output); context.context_mut().take_command_failure_cleanup(); self.connection.extend_scheduler_events(scheduler); @@ -798,9 +794,7 @@ impl ProtocolWorkflowState<'_> { } output.extend_background_events(after); output.extend_background_events(post_response); - if let Some(predecessor) = predecessor { - output.set_renderer_output_predecessor(predecessor); - } + output.extend_renderer_output_dependencies(predecessor); self.connection.extend_scheduler_events(scheduler); context.retain_finalized_command(output); context.context_mut().take_command_failure_cleanup(); diff --git a/moli-protocol/src/conn/state_access/direct/native_tests.rs b/moli-protocol/src/conn/state_access/direct/native_tests.rs index 7291264ae8..1819a92f1f 100644 --- a/moli-protocol/src/conn/state_access/direct/native_tests.rs +++ b/moli-protocol/src/conn/state_access/direct/native_tests.rs @@ -82,7 +82,7 @@ async fn native_completion_and_discarded_ready_recovery_use_direct_state() { }; let (events, _, predecessor) = outcome.into_protocol_event_parts(); assert!( - predecessor.is_some(), + !predecessor.is_empty(), "the renderer terminal retains its exact predecessor" ); assert!( @@ -502,7 +502,7 @@ async fn discarded_input_workflow_result_keeps_the_original_native_output() { }; let (events, _, predecessor) = outcome.into_protocol_event_parts(); assert!( - predecessor.is_some(), + !predecessor.is_empty(), "accepted Native output survives the discarded workflow result" ); let messages: Vec<_> = events diff --git a/moli-protocol/src/conn/state_access/direct/target_message_tests.rs b/moli-protocol/src/conn/state_access/direct/target_message_tests.rs index 2e5ee8363f..57ec272810 100644 --- a/moli-protocol/src/conn/state_access/direct/target_message_tests.rs +++ b/moli-protocol/src/conn/state_access/direct/target_message_tests.rs @@ -217,7 +217,7 @@ async fn attached_command_recovers_its_original_output(internal: bool) { terminals[0]["result"]["result"]["value"], 42, "{phase}: {messages:?}" ); - assert!(predecessor.is_some()); + assert!(!predecessor.is_empty()); } else { assert_eq!( terminals[0]["error"]["code"], -32000, @@ -249,7 +249,7 @@ async fn attached_command_recovers_its_original_output(internal: bool) { nested[0]["result"]["result"]["value"], 42, "{phase}: {messages:?}" ); - assert!(predecessor.is_some()); + assert!(!predecessor.is_empty()); } else { assert_eq!(nested[0]["error"]["code"], -32000, "{phase}: {messages:?}"); } diff --git a/moli-protocol/src/conn/state_access/direct/termination_tests.rs b/moli-protocol/src/conn/state_access/direct/termination_tests.rs index e0720138c2..2ff7963ea5 100644 --- a/moli-protocol/src/conn/state_access/direct/termination_tests.rs +++ b/moli-protocol/src/conn/state_access/direct/termination_tests.rs @@ -705,7 +705,7 @@ async fn intercepted_target_close_retains_cleanup_across_cancelled_and_discarded }; let (events, scheduler, predecessor) = outcome.into_protocol_event_parts(); assert!( - predecessor.is_some(), + !predecessor.is_empty(), "the real failed request retains its renderer predecessor" ); let messages: Vec<_> = events diff --git a/moli-protocol/src/conn/tests/navigation_commit.rs b/moli-protocol/src/conn/tests/navigation_commit.rs index 2fa3e9a57f..af2ebbb7bf 100644 --- a/moli-protocol/src/conn/tests/navigation_commit.rs +++ b/moli-protocol/src/conn/tests/navigation_commit.rs @@ -179,7 +179,7 @@ async fn background_protocol_completion_rejoins_waits_and_keeps_its_original_pub let native = completions.recv().await.unwrap(); let started = ctx.with_fixture(|conn| conn.admit_background_navigation_completion(native)) ; let (events, scheduler_events, predecessor) = started.into_protocol_event_parts(); - assert!(predecessor.is_none()); + assert!(predecessor.is_empty()); assert!(events.is_empty(), "the finite admission does not publish a source result"); let mut work = None; let mut other_events = Vec::new(); diff --git a/moli-protocol/src/domains/command_output.rs b/moli-protocol/src/domains/command_output.rs index ccbc5d6e95..a99d6b6bc7 100644 --- a/moli-protocol/src/domains/command_output.rs +++ b/moli-protocol/src/domains/command_output.rs @@ -24,14 +24,14 @@ use moli_cookie_jar::{ #[cfg(test)] use crate::conn::build_command_success_response; -use crate::conn::{BackgroundProtocolEvent, build_event}; +use crate::conn::{BackgroundProtocolEvent, RendererOutputDependencies, build_event}; use crate::domains::observable_output::log_entry_event; #[derive(Debug, Default)] pub(crate) struct CommandOutputPlan { outputs: Vec, post_response_events: Vec, - renderer_output_predecessor: Option, + renderer_output_dependencies: RendererOutputDependencies, renderer_output_boundary: Option>, } @@ -161,7 +161,7 @@ impl CommandOutputPlan { let CommandOutputPlan { outputs, post_response_events, - renderer_output_predecessor, + renderer_output_dependencies, renderer_output_boundary, } = other; if let Some(boundary) = renderer_output_boundary { @@ -178,9 +178,8 @@ impl CommandOutputPlan { } self.outputs.extend(outputs); self.post_response_events.extend(post_response_events); - if let Some(predecessor) = renderer_output_predecessor { - self.set_renderer_output_predecessor(predecessor); - } + self.renderer_output_dependencies + .merge(renderer_output_dependencies); } /// Converts a successfully completed nested command into the side effects @@ -207,13 +206,26 @@ impl CommandOutputPlan { &mut self, predecessor: moli_core::RendererOutputFence, ) { - predecessor.merge_into_same_stream_tail(&mut self.renderer_output_predecessor); + self.renderer_output_dependencies + .add_same_stream(predecessor); } pub(crate) fn take_renderer_output_predecessor( &mut self, ) -> Option { - self.renderer_output_predecessor.take() + self.take_renderer_output_dependencies() + .into_single_stream() + } + + pub(crate) fn extend_renderer_output_dependencies( + &mut self, + dependencies: RendererOutputDependencies, + ) { + self.renderer_output_dependencies.merge(dependencies); + } + + pub(crate) fn take_renderer_output_dependencies(&mut self) -> RendererOutputDependencies { + std::mem::take(&mut self.renderer_output_dependencies) } /// Inserts one exact renderer publication at the current protocol-output @@ -466,7 +478,7 @@ impl CommandOutputPlan { }) .collect(), post_response_events: self.post_response_events, - renderer_output_predecessor: self.renderer_output_predecessor, + renderer_output_dependencies: self.renderer_output_dependencies, renderer_output_boundary: self.renderer_output_boundary, } } diff --git a/moli-protocol/src/domains/log.rs b/moli-protocol/src/domains/log.rs index 7122e8b0bb..f9f6bc4499 100644 --- a/moli-protocol/src/domains/log.rs +++ b/moli-protocol/src/domains/log.rs @@ -352,7 +352,7 @@ mod tests { .expect("live protocol owner") }; let (events, _, renderer_output_predecessor) = outcome.into_protocol_event_parts(); - assert!(renderer_output_predecessor.is_none()); + assert!(renderer_output_predecessor.is_empty()); assert_eq!(events.len(), 2, "expected response and replay event"); let (log_message, log_sidecar) = events[0].clone().into_parts(); diff --git a/moli-protocol/src/domains/page/tests/lifecycle.rs b/moli-protocol/src/domains/page/tests/lifecycle.rs index 6c7849ba64..0b8207adb1 100644 --- a/moli-protocol/src/domains/page/tests/lifecycle.rs +++ b/moli-protocol/src/domains/page/tests/lifecycle.rs @@ -3597,7 +3597,7 @@ async fn close_command_background_events_keep_target_detached_sidecar() { ) = outcome.into_renderer_owner_turn_parts(); assert!(renderer_output_boundary.is_none()); assert!(post_renderer_output_events.is_empty()); - assert!(renderer_output_predecessor.is_none()); + assert!(renderer_output_predecessor.is_empty()); events.extend(post_response_events); let [CdpSchedulerEvent::ProtocolWorkPublished { work }] = <[_; 1]>::try_from(scheduler_events) .expect("Page.close should publish exactly one target-termination owner action") diff --git a/moli-protocol/src/domains/runtime/tests/evaluate/contexts.rs b/moli-protocol/src/domains/runtime/tests/evaluate/contexts.rs index f1e3d1c9ae..23681ad9e5 100644 --- a/moli-protocol/src/domains/runtime/tests/evaluate/contexts.rs +++ b/moli-protocol/src/domains/runtime/tests/evaluate/contexts.rs @@ -182,7 +182,9 @@ async fn runtime_enable_retired_owner_error_keeps_journal_order_and_single_termi assert!(scheduler_events.is_empty()); let sent_start = ctx.sent.len(); ctx.route_direct_command_renderer_predecessor_for_test( - predecessor.expect("the adapter must acknowledge its published terminal"), + predecessor + .into_single_stream() + .expect("the adapter must acknowledge its published terminal"), ) .await; assert!( @@ -247,7 +249,9 @@ async fn runtime_disable_retired_owner_keeps_committed_success_and_single_termin let (messages, scheduler_events, predecessor) = outcome.into_protocol_event_parts(); assert!(messages.is_empty() && scheduler_events.is_empty()); ctx.route_direct_command_renderer_predecessor_for_test( - predecessor.expect("the adapter must acknowledge the committed disable"), + predecessor + .into_single_stream() + .expect("the adapter must acknowledge the committed disable"), ) .await; // wait_for_scheduler_message removed the original response above. diff --git a/moli-protocol/src/domains/target/attachment/detachment.rs b/moli-protocol/src/domains/target/attachment/detachment.rs index eedf1408d3..a58df2b15e 100644 --- a/moli-protocol/src/domains/target/attachment/detachment.rs +++ b/moli-protocol/src/domains/target/attachment/detachment.rs @@ -243,7 +243,7 @@ impl TargetDetachmentOutput { let cleanup_error = match session.result { Ok(predecessor) => { if let Some(predecessor) = predecessor { - command_context.set_renderer_output_predecessor(predecessor); + command_context.add_renderer_output_dependency(predecessor); } None } @@ -335,9 +335,171 @@ fn route_target_id(route: &CdpSessionRoute) -> Option<&str> { #[cfg(test)] mod tests { use super::*; - use crate::conn::BrowserContext; + use crate::conn::{BrowserContext, RendererPageResidenceIdentity}; use crate::testing::TestContext; + #[tokio::test] + async fn multi_page_detachment_waits_for_every_original_output_prefix() { + use moli_core::{ + PageId, RendererOutputCursor, RendererOutputFence, RendererOutputItem, + RendererOutputPublication, RendererOutputRecord, RendererOutputStreamControl, + RendererOutputStreamIdentity, RendererOutputTransportMessage, + RendererProtocolObservation, + }; + + let mut browser = BrowserContext::new("BID-detach-prefixes".into()); + browser.set_active_target_id("TID-first"); + browser.attach_active_session("SID-first"); + browser.stage_background_target( + "TID-second".into(), + Some("SID-second".into()), + "about:blank".into(), + None, + None, + ); + let mut context = TestContext::new(); + context.with_fixture(|conn| conn.install_browser_context_fixture_for_test(browser)); + let prefixes = [ + ("TID-first", "SID-first", 71, 3), + ("TID-second", "SID-second", 72, 1), + ] + .map(|(target, session, page, sequence)| { + let stream = RendererOutputStreamIdentity::new_page_for_protocol_test( + PageId::new_for_testing(page), + ); + context.with_fixture(|conn| { + let residence = + RendererPageResidenceIdentity::from_residence(stream.residence()).unwrap(); + conn.bind_renderer_page_output_owner( + residence, + crate::conn::TargetPageResidenceIdentity::new_for_test( + "BID-detach-prefixes".into(), + Some(target.into()), + page, + ), + ); + conn.apply_renderer_output_stream_control(RendererOutputStreamControl::Opened { + stream, + }); + }); + (target, session, stream, sequence) + }); + let (output, dependencies) = context.with_fixture(|conn| { + let sessions = prefixes + .iter() + .map(|&(target, session, stream, sequence)| { + let route = conn.session_route(Some(session)).unwrap(); + let plan = SessionDisposalPlan::for_session_route(session, &route).unwrap(); + let identity = SessionDisposalIdentity::capture(conn, plan.clone()); + conn.mark_session_closing(&plan); + SessionDetachmentOutput { + identity, + commit: SessionDetachmentCommit::Target( + TargetSessionDetachCleanupPlan::new(target, session, None, None), + ), + child: true, + response_before_event: false, + background_events: Vec::new(), + protocol_events: Vec::new(), + result: Ok(Some(RendererOutputFence::new_for_test( + RendererOutputCursor::new_for_test(stream, sequence), + ))), + } + }) + .collect(); + let mut command_context = CommandDispatchContext::default(); + let output = TargetDetachmentOutput { sessions }.finish(conn, &mut command_context); + assert!( + !output.is_empty(), + "the original response must remain in the command output plan" + ); + assert!(command_context.take_protocol_events().is_empty()); + assert!( + conn.take_scheduler_events() + .into_iter() + .all(|event| !matches!( + event, + crate::conn::CdpSchedulerEvent::ProtocolWorkPublished { .. } + )), + "Target cleanup must not publish the command response as background work" + ); + (output, command_context.take_renderer_output_dependencies()) + }); + assert_eq!(dependencies.iter().count(), 2); + for (index, &(_, _, stream, sequence)) in prefixes.iter().enumerate() { + context.with_fixture(|conn| { + assert!( + !dependencies + .iter() + .all(|fence| conn.renderer_output_cursor_is_projected(fence.cursor())) + ) + }); + for sequence in 1..=sequence { + let outcome = context + .handle() + .with_state(|state| { + state.admit_renderer_output(RendererOutputTransportMessage::Publication( + RendererOutputPublication::new_for_test( + RendererOutputCursor::new_for_test(stream, sequence), + vec![RendererOutputRecord::new_for_test( + RendererOutputItem::Observation( + RendererProtocolObservation::RuntimeLifecycleError { + text: "historical renderer output".into(), + execution_context_id: None, + }, + ), + )], + ), + )) + }) + .unwrap(); + let (events, scheduler) = outcome.into_protocol_event_parts(); + assert!(events.is_empty()); + for event in scheduler { + let crate::conn::CdpSchedulerEvent::RendererOutputProjectionPrepared { + pending, + } = event + else { + panic!("expected the exact renderer projection"); + }; + let completed = context.drive_owned_wait(pending.wait()).await; + let outcome = context + .handle() + .with_state(|state| state.complete_owned_protocol_scheduler_work(completed)) + .unwrap(); + assert!(outcome.into_protocol_event_parts().0.is_empty()); + } + } + context.with_fixture(|conn| { + assert_eq!( + dependencies + .iter() + .all(|fence| conn.renderer_output_cursor_is_projected(fence.cursor())), + index == 1 + ); + }); + } + let events = output + .into_background_events(Some(91), Some("SID-browser-command")) + .into_iter() + .map(BackgroundProtocolEvent::into_protocol_message) + .collect::>(); + assert_eq!( + events.len(), + 3, + "each detach and the response appear once: {events:?}" + ); + assert!( + events[..2] + .iter() + .all(|event| event["method"] == "Target.detachedFromTarget") + ); + assert_eq!( + events[2], + json!({"id": 91, "sessionId": "SID-browser-command", "result": {}}) + ); + } + #[tokio::test] async fn rebound_session_keeps_new_scripts_and_binding_after_old_cleanup_or_recovery() { for cancelled in [false, true] { diff --git a/moli-protocol/src/domains/target/attachment/message.rs b/moli-protocol/src/domains/target/attachment/message.rs index 393d0f738a..909eff514f 100644 --- a/moli-protocol/src/domains/target/attachment/message.rs +++ b/moli-protocol/src/domains/target/attachment/message.rs @@ -227,8 +227,6 @@ pub(crate) fn retain_target_message_output( BackgroundProtocolEvent::target_received_message_from_target(&session_id, event, strip) })); } - if let Some(predecessor) = predecessor { - response.set_renderer_output_predecessor(predecessor); - } + response.extend_renderer_output_dependencies(predecessor); response } diff --git a/moli-protocol/src/domains/target/tests/tests_cdp_chromium_imports/p1_target_multipage.rs b/moli-protocol/src/domains/target/tests/tests_cdp_chromium_imports/p1_target_multipage.rs index 0c62be8f88..0bd86dbfa9 100644 --- a/moli-protocol/src/domains/target/tests/tests_cdp_chromium_imports/p1_target_multipage.rs +++ b/moli-protocol/src/domains/target/tests/tests_cdp_chromium_imports/p1_target_multipage.rs @@ -328,7 +328,7 @@ async fn internal_create_isolated_world_restart_does_not_inherit_the_stale_rende }; let (events, after, boundary, post, scheduler, predecessor) = outcome.into_renderer_owner_turn_parts(); - if let Some(predecessor) = &predecessor { + for predecessor in predecessor.iter() { assert!( matches!(predecessor.cursor().stream().residence(), moli_core::RendererOutputResidenceIdentity::Page { page_id, .. } if page_id == replacement_page_id), "an abandoned renderer stream cannot precede the replacement response" @@ -338,7 +338,7 @@ async fn internal_create_isolated_world_restart_does_not_inherit_the_stale_rende events, post, scheduler, ) .with_renderer_output_boundary(boundary, after) - .with_renderer_output_predecessor(predecessor); + .with_renderer_output_dependencies(predecessor); let (messages, _) = ctx.route_completed_command_outcome_for_test(outcome).await; assert!( response(&messages, 2_600_072)["result"]["executionContextId"] diff --git a/moli-protocol/src/lib.rs b/moli-protocol/src/lib.rs index ebfba50953..623454fb1a 100644 --- a/moli-protocol/src/lib.rs +++ b/moli-protocol/src/lib.rs @@ -35,7 +35,8 @@ pub use conn::{ OwnedProtocolWaitTask, ParsedCdpCommand, PendingCdpCommandDispatch, PendingDeferredMainDocumentLoadCompletion, PendingProtocolSchedulerWork, PendingRuntimeProtocolMessageDispatch, ProtocolOwnerClosed, ProtocolStateHandle, - ProtocolStateOwner, ProtocolWorkflowState, WebDriverNodeReference, spawn_owned_protocol_wait, + ProtocolStateOwner, ProtocolWorkflowState, RendererOutputDependencies, WebDriverNodeReference, + spawn_owned_protocol_wait, }; pub use conn::{ AutomationCommandTaskStep, CompletedAutomationCommandDispatch, PendingAutomationCommandDispatch, diff --git a/moli-protocol/src/test_support.rs b/moli-protocol/src/test_support.rs index c65efdb2d3..f8aa2c7463 100644 --- a/moli-protocol/src/test_support.rs +++ b/moli-protocol/src/test_support.rs @@ -6,7 +6,10 @@ //! callers must obtain these values from their owning connection, navigation, //! lifecycle observer, or publication boundary. -use moli_core::{RendererDocumentLifecycleIdentity, RendererOutputResidenceIdentity}; +use moli_core::{ + RendererDocumentLifecycleIdentity, RendererOutputResidenceIdentity, + RendererOutputStreamIdentity, +}; use crate::{ DeferredMainDocumentLoadCompletionOutputInterest, DeferredMainDocumentLoadObservationId, @@ -14,6 +17,30 @@ use crate::{ conn::{CdpConnection, DocumentNavigationToken, RendererPageResidenceIdentity}, }; +pub fn command_response_flush_observer( + context: &crate::CommandResponseFlushContext, +) -> tokio::sync::watch::Receiver { + context + .receiver() + .expect("test command owns a flush permit") +} + +/// Binds a synthetic Page stream to a retired protocol Page owner, allowing +/// scheduler tests to project historical output without a live Page. +pub fn bind_renderer_page_output_stream_to_retired_owner( + conn: &mut CdpConnection, + stream: RendererOutputStreamIdentity, +) { + let renderer_page = RendererPageResidenceIdentity::from_residence(stream.residence()) + .expect("a synthetic Page output stream requires a Page residence"); + let page_owner = crate::conn::TargetPageResidenceIdentity::new( + "BID-retired-scheduler-fixture".into(), + Some("TID-retired-scheduler-fixture".into()), + crate::conn::TargetPageAttachmentId::allocate(), + ); + conn.bind_renderer_page_output_owner(renderer_page, page_owner); +} + /// Opaque exact-token fixture for scheduler tests that need a real /// target-owned background navigation request. pub struct BackgroundNavigationRequestFixture { diff --git a/moli-protocol/src/testing.rs b/moli-protocol/src/testing.rs index bfe2c1e153..ce62b1e09f 100644 --- a/moli-protocol/src/testing.rs +++ b/moli-protocol/src/testing.rs @@ -1477,13 +1477,9 @@ impl TestContext { mut new_scheduler_events, mut renderer_output_predecessor, ) = outcome.into_renderer_owner_turn_parts(); - if let Some(command_predecessor) = - command_context.take_renderer_output_predecessor() - { - command_predecessor - .merge_into_same_stream_tail(&mut renderer_output_predecessor); - } - if let Some(predecessor) = renderer_output_predecessor { + renderer_output_predecessor + .merge(command_context.take_renderer_output_dependencies()); + for predecessor in renderer_output_predecessor { Box::pin( self.route_renderer_output_predecessor_before_command_response( predecessor, @@ -1718,7 +1714,7 @@ impl TestContext { before, post, scheduler, ) .with_renderer_output_boundary(boundary, after) - .with_renderer_output_predecessor(predecessor); + .with_renderer_output_dependencies(predecessor); let (messages, scheduler) = self.route_completed_command_outcome_for_test(output).await; self.sent.extend(messages); self.route_direct_command_output_for_test(Vec::new(), scheduler) @@ -1746,7 +1742,7 @@ impl TestContext { scheduler_events, renderer_output_predecessor, ) = outcome.into_renderer_owner_turn_parts(); - if let Some(predecessor) = renderer_output_predecessor { + for predecessor in renderer_output_predecessor { Box::pin(self.route_renderer_output_predecessor_before_command_response(predecessor)) .await; } @@ -2013,7 +2009,7 @@ impl TestContext { renderer_output_predecessor, ) = outcome.into_renderer_owner_turn_parts(); assert!( - renderer_output_predecessor.is_none(), + renderer_output_predecessor.is_empty(), "background navigation completion must use an insertion boundary" ); if !completion_prefix.is_empty() {