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

fix(cdp): preserve per-page output fences during Fetch teardown

Browse Source 
Fix MOL-554 by keeping renderer output fences for different Pages as
independent predecessors of the existing protocol output work item.
Preserve command correlation when publishing cleanup and detach events
after every original renderer prefix has crossed protocol ingress.

Cover continued and outstanding Fetch requests across two Pages, and prove
that projecting one Page's cursor cannot release another Page's output.

Validation: cargo fmt --all; workspace all-target/all-feature Clippy;
Nextest 21,987 passed, 19 skipped; Playwright reproduction passes 3/3.
This commit is contained in:
ldm0
2026-11-03 10:48:27 +08:00
committed by Donough Liu
parent 6043f02345
commit ea1076e2db
34 changed files with 888 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
moli-protocol-server/src/cdp_scheduler.rs
+35 -24
View File
@@ -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<ProtocolOutputSequence, RendererOutputTransportFailure> {
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<ProtocolOutputSequence, RendererOutputTransportFailure> {
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<moli_core::RendererOutputFence>,
Option<moli_core::RendererOutputFence>,
moli_protocol::RendererOutputDependencies,
) {
let (
events,
@@ -3170,7 +3181,7 @@ impl CdpScheduler {
ProtocolOutputSequence,
ProtocolOutputSequence,
Option<moli_core::RendererOutputFence>,
Option<moli_core::RendererOutputFence>,
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);
moli-protocol-server/src/cdp_scheduler/actor.rs
+145 -37
View File
@@ -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!({
moli-protocol-server/src/cdp_scheduler/command_dispatch.rs
+17 -10
View File
@@ -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<moli_core::RendererOutputFence>,
post_response_events: Vec<BackgroundProtocolEvent>,
post_flush_scheduler_events: Vec<CdpSchedulerEvent>,
renderer_output_predecessor: Option<moli_core::RendererOutputFence>,
renderer_output_dependencies: RendererOutputDependencies,
output_release_permit: Option<CommandOutputReleasePermit>,
}
@@ -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<moli_core::RendererOutputFence>,
) -> 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<moli_core::RendererOutputFence> {
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<moli_core::RendererOutputFence>,
Vec<BackgroundProtocolEvent>,
Vec<CdpSchedulerEvent>,
Option<moli_core::RendererOutputFence>,
RendererOutputDependencies,
Option<CommandOutputReleasePermit>,
) {
(
@@ -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,
)
}
moli-protocol-server/src/cdp_writer.rs
+4
View File
@@ -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
moli-protocol-server/src/protocol_server/tests/cdp_dynamic_page/browser_close.rs
+167
View File
@@ -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("<p>Fetch teardown</p>") }),
);
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;
moli-protocol/src/conn.rs
+27 -5
View File
@@ -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<BackgroundProtocolEvent>,
renderer_output_boundary: Option<moli_core::RendererOutputFence>,
post_response_events: Vec<BackgroundProtocolEvent>,
renderer_output_predecessor: Option<moli_core::RendererOutputFence>,
renderer_output_dependencies: RendererOutputDependencies,
command_failure_cleanup: Rc<RefCell<Option<CommandFailureCleanup>>>,
deferred_command_events: Vec<BackgroundProtocolEvent>,
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<moli_core::RendererOutputFence> {
&mut self.renderer_output_predecessor
self.renderer_output_dependencies.single_stream_mut()
}
#[doc(hidden)]
pub fn take_renderer_output_predecessor(&mut self) -> Option<moli_core::RendererOutputFence> {
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)
}
}
moli-protocol/src/conn/automation_command.rs
+7 -1
View File
@@ -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(),
)
}
}
moli-protocol/src/conn/automation_command/dispatch.rs
+1 -1
View File
@@ -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,
moli-protocol/src/conn/command_workflow.rs
+1 -3
View File
@@ -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.
moli-protocol/src/conn/dispatch.rs
+3 -5
View File
@@ -382,9 +382,7 @@ pub(super) fn append_command_output_plan(
command_id: Option<u64>,
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()),
)
}
}
moli-protocol/src/conn/dispatch_tests/mod.rs
+1 -1
View File
@@ -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)
moli-protocol/src/conn/document_activity_work.rs
+1 -1
View File
@@ -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())
}
}
moli-protocol/src/conn/protocol_output/workflow.rs
+8 -13
View File
@@ -109,31 +109,26 @@ impl CdpConnection {
) -> Vec<crate::conn::BackgroundProtocolEvent> {
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::<Vec<_>>();
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,
);
}
moli-protocol/src/conn/renderer_output_dependencies.rs
+204
View File
@@ -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<RendererOutputFence>,
additional: Vec<RendererOutputFence>,
explicitly_composed: bool,
}
impl RendererOutputDependencies {
pub fn is_empty(&self) -> bool {
self.first.is_none()
}
pub fn iter(&self) -> impl Iterator<Item = &RendererOutputFence> {
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<RendererOutputFence> {
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<RendererOutputFence> {
assert!(
self.additional.is_empty(),
"composite output dependencies require consuming every renderer stream"
);
self.first
}
}
impl From<Option<RendererOutputFence>> for RendererOutputDependencies {
fn from(first: Option<RendererOutputFence>) -> Self {
Self {
first,
additional: Vec::new(),
explicitly_composed: false,
}
}
}
impl Extend<RendererOutputFence> for RendererOutputDependencies {
fn extend<T: IntoIterator<Item = RendererOutputFence>>(&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<RendererOutputFence>,
std::vec::IntoIter<RendererOutputFence>,
>;
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::<Vec<_>>(),
[(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);
}
}
}
moli-protocol/src/conn/runtime_eval/helper_workflow.rs
+2 -2
View File
@@ -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 {
moli-protocol/src/conn/scheduler_state.rs
+21 -11
View File
@@ -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<CdpSchedulerEvent>,
}
/// 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<moli_core::RendererOutputFence>,
renderer_output_dependencies: RendererOutputDependencies,
}
impl CdpTurnOutcome {
@@ -120,10 +121,17 @@ impl CdpTurnOutcome {
pub(crate) fn with_renderer_output_predecessor(
self,
predecessor: Option<moli_core::RendererOutputFence>,
) -> 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<moli_core::RendererOutputFence> {
self.renderer_output_predecessor.clone()
self.renderer_output_dependencies
.clone()
.into_single_stream()
}
#[cfg(test)]
pub fn into_parts(self) -> (Vec<serde_json::Value>, Vec<CdpSchedulerEvent>) {
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<BackgroundProtocolEvent>,
Vec<CdpSchedulerEvent>,
Option<moli_core::RendererOutputFence>,
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<moli_core::RendererOutputFence>,
Vec<BackgroundProtocolEvent>,
Vec<CdpSchedulerEvent>,
Option<moli_core::RendererOutputFence>,
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,
)
}
}
moli-protocol/src/conn/state_access/auto_attach/tests.rs
+1 -1
View File
@@ -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
moli-protocol/src/conn/state_access/automation_configuration.rs
+1 -1
View File
@@ -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"
);
}
moli-protocol/src/conn/state_access/browser_context_disposal/tests.rs
+1 -1
View File
@@ -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
moli-protocol/src/conn/state_access/command_dispatch.rs
+3 -9
View File
@@ -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();
moli-protocol/src/conn/state_access/direct/native_tests.rs
+2 -2
View File
@@ -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
moli-protocol/src/conn/state_access/direct/target_message_tests.rs
+2 -2
View File
@@ -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:?}");
}
moli-protocol/src/conn/state_access/direct/termination_tests.rs
+1 -1
View File
@@ -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
moli-protocol/src/conn/tests/navigation_commit.rs
+1 -1
View File
@@ -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();
moli-protocol/src/domains/command_output.rs
+21 -9
View File
@@ -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<CommandOutput>,
post_response_events: Vec<BackgroundProtocolEvent>,
renderer_output_predecessor: Option<moli_core::RendererOutputFence>,
renderer_output_dependencies: RendererOutputDependencies,
renderer_output_boundary: Option<Box<RendererOutputBoundary>>,
}
@@ -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<moli_core::RendererOutputFence> {
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,
}
}
moli-protocol/src/domains/log.rs
+1 -1
View File
@@ -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();
moli-protocol/src/domains/page/tests/lifecycle.rs
+1 -1
View File
@@ -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")
moli-protocol/src/domains/runtime/tests/evaluate/contexts.rs
+6 -2
View File
@@ -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.
moli-protocol/src/domains/target/attachment/detachment.rs
+164 -2
View File
@@ -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::<Vec<_>>();
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] {
moli-protocol/src/domains/target/attachment/message.rs
+1 -3
View File
@@ -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
}
moli-protocol/src/domains/target/tests/tests_cdp_chromium_imports/p1_target_multipage.rs
+2 -2
View File
@@ -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"]
moli-protocol/src/lib.rs
+2 -1
View File
@@ -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,
moli-protocol/src/test_support.rs
+28 -1
View File
@@ -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<bool> {
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 {
moli-protocol/src/testing.rs
+6 -10
View File
@@ -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() {