diff --git a/moli-curl/src/lib.rs b/moli-curl/src/lib.rs index 40bda94748..4c56672434 100644 --- a/moli-curl/src/lib.rs +++ b/moli-curl/src/lib.rs @@ -7,7 +7,7 @@ pub mod websocket; pub use dns_adapter::CurlDnsResolution; pub use runtime::{ - CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey, - CurlSubmitError, CurlTransferId, + CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, + CurlOriginKey, CurlSubmitError, CurlTransferId, }; pub use tls::CurlTlsConfig; diff --git a/moli-curl/src/runtime.rs b/moli-curl/src/runtime.rs index bf94dd83af..056331d7e6 100644 --- a/moli-curl/src/runtime.rs +++ b/moli-curl/src/runtime.rs @@ -106,31 +106,34 @@ impl fmt::Debug for CurlSubmitError { } } -/// Cloneable handle for a single libcurl multi owner thread. +/// Owns one native thread. Request handles cannot extend its lifetime or join it. #[derive(Debug)] pub struct CurlMultiRuntime { - inner: Arc>, -} - -impl Clone for CurlMultiRuntime { - fn clone(&self) -> Self { - Self { - inner: Arc::clone(&self.inner), - } - } -} - -#[derive(Debug)] -struct CurlMultiRuntimeInner { - command_tx: Sender>, - owner_waker: MultiWaker, - shutdown_requested: Arc, + http: CurlHttpSender, websocket_connector: CurlWebSocketConnector, #[cfg(test)] owner_started: Arc, owner_handle: Mutex>>, } +/// Submits HTTP work to one owner. Clones carry no shutdown or join authority. +#[derive(Debug)] +pub struct CurlHttpSender { + command_tx: Sender>, + owner_waker: MultiWaker, + shutdown_requested: Arc, +} + +impl Clone for CurlHttpSender { + fn clone(&self) -> Self { + Self { + command_tx: self.command_tx.clone(), + owner_waker: self.owner_waker.clone(), + shutdown_requested: self.shutdown_requested.clone(), + } + } +} + impl CurlMultiRuntime { pub fn new( config: CurlMultiRuntimeConfig, @@ -162,28 +165,61 @@ impl CurlMultiRuntime { .recv() .context("curl multi runtime owner did not publish a waker")?; let runtime = Self { - inner: Arc::new(CurlMultiRuntimeInner { + websocket_connector: CurlWebSocketConnector::new( + websocket_tx, + owner_waker.clone(), + shutdown_requested.clone(), + ), + http: CurlHttpSender { command_tx, - websocket_connector: CurlWebSocketConnector::new( - websocket_tx, - owner_waker.clone(), - shutdown_requested.clone(), - ), owner_waker, shutdown_requested, - #[cfg(test)] - owner_started, - owner_handle: Mutex::new(Some(owner_handle)), - }), + }, + #[cfg(test)] + owner_started, + owner_handle: Mutex::new(Some(owner_handle)), }; Ok((runtime, completion_rx)) } + pub fn http_sender(&self) -> CurlHttpSender { + self.http.clone() + } + + pub fn shutdown(&self) { + if !self.http.shutdown_requested.swap(true, Ordering::SeqCst) { + let _ = self.http.command_tx.send(CurlRuntimeCommand::Shutdown); + let _ = self.http.owner_waker.wakeup(); + } + if let Some(owner_handle) = self.owner_handle.lock().take() { + let _ = owner_handle.join(); + } + } + + /// Creates WebSockets on this runtime's owner and Multi. The capability + /// cannot keep the runtime alive or shut down unrelated HTTP transfers. + pub fn websocket_connector(&self) -> CurlWebSocketConnector { + self.websocket_connector.clone() + } + + #[cfg(test)] + pub fn owner_count_for_testing(&self) -> usize { + usize::from(self.owner_started.load(Ordering::SeqCst)) + } +} + +impl Drop for CurlMultiRuntime { + fn drop(&mut self) { + self.shutdown(); + } +} + +impl CurlHttpSender { pub fn submit( &self, job: CurlMultiJob, ) -> std::result::Result> { - if self.inner.shutdown_requested.load(Ordering::SeqCst) { + if self.shutdown_requested.load(Ordering::SeqCst) { return Err(CurlSubmitError { job, error: anyhow!("curl multi runtime is shutting down"), @@ -194,12 +230,11 @@ impl CurlMultiRuntime { Err(error) => return Err(CurlSubmitError { job, error }), }; match self - .inner .command_tx .send(CurlRuntimeCommand::Request { transfer_id, job }) { Ok(()) => { - let _ = self.inner.owner_waker.wakeup(); + let _ = self.owner_waker.wakeup(); Ok(transfer_id) } Err(error) => { @@ -213,41 +248,6 @@ impl CurlMultiRuntime { } } } - - pub fn shutdown(&self) { - self.inner.shutdown(); - } - - /// Creates WebSockets on this runtime's owner and Multi. The capability - /// cannot keep the runtime alive or shut down unrelated HTTP transfers. - pub fn websocket_connector(&self) -> CurlWebSocketConnector { - self.inner.websocket_connector.clone() - } - - #[cfg(test)] - pub fn owner_count_for_testing(&self) -> usize { - usize::from(self.inner.owner_started.load(Ordering::SeqCst)) - } -} - -impl Drop for CurlMultiRuntimeInner { - fn drop(&mut self) { - self.shutdown(); - } -} - -impl CurlMultiRuntimeInner { - fn shutdown(&self) { - if self.shutdown_requested.swap(true, Ordering::SeqCst) { - return; - } - let _ = self.command_tx.send(CurlRuntimeCommand::Shutdown); - let _ = self.owner_waker.wakeup(); - let Some(owner_handle) = self.owner_handle.lock().take() else { - return; - }; - let _ = owner_handle.join(); - } } #[cfg(test)] @@ -297,6 +297,7 @@ mod tests { easy.url(&format!("http://{address}/identity")) .expect("test curl URL should be valid"); let transfer_id = runtime + .http_sender() .submit(CurlMultiJob { easy, context: "matching-context".to_owned(), @@ -321,4 +322,35 @@ mod tests { runtime.shutdown(); server.join().expect("test HTTP server should finish"); } + + #[test] + fn http_sender_does_not_keep_owner_alive_and_returns_rejected_job() { + let (runtime, completed) = + CurlMultiRuntime::>::new(Default::default()).unwrap(); + let sender = runtime.http_sender(); + let retained = sender.clone(); + drop(runtime); + for sender in [sender, retained] { + let mut easy = Easy2::new(TestHandler); + easy.url("http://127.0.0.1:1/must-not-connect").unwrap(); + let error = sender + .submit(CurlMultiJob { + easy, + context: vec![7; 1024], + origin: None, + deadline: None, + dns_resolution: CurlDnsResolution::curl_managed(), + priority: 1, + label: "closed".into(), + }) + .unwrap_err(); + assert!(error.error.to_string().contains("shutting down")); + assert_eq!(error.job.context, vec![7; 1024]); + assert_eq!(error.job.label, "closed"); + } + assert!(matches!( + completed.try_recv(), + Err(crossbeam_channel::TryRecvError::Disconnected) + )); + } } diff --git a/moli-curl/src/websocket.rs b/moli-curl/src/websocket.rs index 5d37643992..c314db42db 100644 --- a/moli-curl/src/websocket.rs +++ b/moli-curl/src/websocket.rs @@ -377,7 +377,7 @@ impl CurlWebSocketConnector { /// Standalone owner for callers without an existing HTTP runtime. Uses the /// same Multi driver as HTTP, with no HTTP submissions. -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct CurlWebSocketRuntime { runtime: CurlMultiRuntime, } diff --git a/moli-curl/src/websocket/tests/shared.rs b/moli-curl/src/websocket/tests/shared.rs index 95c0b9e366..ae6de810eb 100644 --- a/moli-curl/src/websocket/tests/shared.rs +++ b/moli-curl/src/websocket/tests/shared.rs @@ -136,6 +136,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() { connections.push(connection); } runtime + .http_sender() .submit(request( &(base.clone() + "/ok"), HttpCapture::default(), @@ -167,6 +168,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() { } // Retiring WS does not retire the shared owner or its HTTP pool. runtime + .http_sender() .submit(request( &(base.clone() + "/after"), HttpCapture::default(), @@ -200,6 +202,7 @@ async fn shared_owner_preserves_http_connection_caps_with_live_websocket() { opened(&mut connection).await; let (headers, ready) = tokio::sync::oneshot::channel(); let first = runtime + .http_sender() .submit(request( &(base.clone() + "/held"), HttpCapture { @@ -213,6 +216,7 @@ async fn shared_owner_preserves_http_connection_caps_with_live_websocket() { // Both HTTP jobs fit max_active=2. Only the connection cap keeps the // second waiting until its own deadline, while the first is held open. let second = runtime + .http_sender() .submit(request( &(base.clone() + "/queued"), HttpCapture::default(), diff --git a/moli-fetch/src/runtime.rs b/moli-fetch/src/runtime.rs index 06f7b78139..27d74bf925 100644 --- a/moli-fetch/src/runtime.rs +++ b/moli-fetch/src/runtime.rs @@ -25,7 +25,8 @@ use moli_cookie_jar::{ advance_cookie_request_context, }; use moli_curl::{ - CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey, + CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, + CurlOriginKey, }; use moli_url_policy::ensure_http_network_transport_url; use parking_lot::Mutex; @@ -390,12 +391,14 @@ impl FetchRuntimeOwner { let (curl_runtime, curl_completion_rx) = CurlMultiRuntime::new(curl_runtime_config(config)) .expect("failed to start fetch curl multi runtime"); let websocket_connector = curl_runtime.websocket_connector(); + let curl_http = curl_runtime.http_sender(); let owner = RuntimeOwner { config: config.clone(), cookie_store, client_hint_preferences, request_rx, curl_runtime, + curl_http, curl_completion_rx, shutdown_requested: Arc::clone(&shutdown_requested), #[cfg(test)] @@ -704,6 +707,7 @@ struct RuntimeOwner { client_hint_preferences: SharedClientHintPreferences, request_rx: Receiver, curl_runtime: CurlMultiRuntime, + curl_http: CurlHttpSender, curl_completion_rx: Receiver, shutdown_requested: Arc, #[cfg(test)] @@ -942,7 +946,7 @@ impl RuntimeOwner { label, context: ActiveTransferContext::Buffered(Box::new(context)), }; - match self.curl_runtime.submit(curl_job) { + match self.curl_http.submit(curl_job) { Ok(_) => Ok(JobOutcome::Submitted), Err(error) => Err(( error @@ -1128,7 +1132,7 @@ impl RuntimeOwner { label, context: ActiveTransferContext::Streaming(Box::new(context)), }; - match self.curl_runtime.submit(curl_job) { + match self.curl_http.submit(curl_job) { Ok(_) => Ok(StreamingJobOutcome::Submitted), Err(error) => { let context = error @@ -1327,7 +1331,7 @@ impl RuntimeOwner { label, context: ActiveTransferContext::StreamingRaw(Box::new(context)), }; - match self.curl_runtime.submit(curl_job) { + match self.curl_http.submit(curl_job) { Ok(_) => Ok(StreamingJobOutcome::Submitted), Err(error) => { let context = error diff --git a/moli-renderer-v8/src/native_bridge/context_host/websockets/registration.rs b/moli-renderer-v8/src/native_bridge/context_host/websockets/registration.rs index 1db64936e8..b2b53f717a 100644 --- a/moli-renderer-v8/src/native_bridge/context_host/websockets/registration.rs +++ b/moli-renderer-v8/src/native_bridge/context_host/websockets/registration.rs @@ -9,7 +9,8 @@ use crate::types::{ }; use moli_websocket::{ ConnectOptions as WebSocketConnectOptions, spawn_connection, - spawn_connection_with_handshake_pause, spawn_failed_connection, spawn_synthetic_connection, + spawn_connection_with_handshake_pause, spawn_failed_connection, spawn_standalone_connection, + spawn_standalone_connection_with_handshake_pause, spawn_synthetic_connection, websocket_cookie_url, }; use url::Url; @@ -49,8 +50,8 @@ impl JsContextHost { .as_ref() .ok() .and_then(|header| header.clone()); + let connector = loader.map(|loader| loader.websocket_connector()); let context = WebSocketConnectOptions { - connector: loader.map(|loader| loader.websocket_connector()), origin: moli_url::origin_ascii_serialization(&document_url), user_agent: loader .map(|loader| loader.user_agent().to_owned()) @@ -91,6 +92,7 @@ impl JsContextHost { v8::Global::new(scope, scope.get_current_context()), dispatch_scope, PendingWebSocketConnection { + connector, socket_id, protocols, connect_options: context, @@ -113,13 +115,23 @@ impl JsContextHost { (None, Some(internal_id)) } else { let connection = match cookie_header { - Ok(_) => spawn_connection( - socket_id, - url.to_string(), - protocols, - context, - self.page_websocket_sender().event_sender(), - ), + Ok(_) => match connector { + Some(connector) => spawn_connection( + connector, + socket_id, + url.to_string(), + protocols, + context, + self.page_websocket_sender().event_sender(), + ), + None => spawn_standalone_connection( + socket_id, + url.to_string(), + protocols, + context, + self.page_websocket_sender().event_sender(), + ), + }, Err(error) => spawn_failed_connection( socket_id, format!("failed to build WebSocket cookie header: {error}"), @@ -166,25 +178,43 @@ impl JsContextHost { context.cookie_header = None; } let (connection, handshake_controller) = if intercept_response { - let (connection, controller) = spawn_connection_with_handshake_pause( - pending.socket_id, - url.to_string(), - pending.protocols, - context, - event_sender, - ); - (connection, Some(controller)) - } else { - ( - spawn_connection( + let (connection, controller) = match pending.connector { + Some(connector) => spawn_connection_with_handshake_pause( + connector, pending.socket_id, url.to_string(), pending.protocols, context, event_sender, ), - None, - ) + None => spawn_standalone_connection_with_handshake_pause( + pending.socket_id, + url.to_string(), + pending.protocols, + context, + event_sender, + ), + }; + (connection, Some(controller)) + } else { + let connection = match pending.connector { + Some(connector) => spawn_connection( + connector, + pending.socket_id, + url.to_string(), + pending.protocols, + context, + event_sender, + ), + None => spawn_standalone_connection( + pending.socket_id, + url.to_string(), + pending.protocols, + context, + event_sender, + ), + }; + (connection, None) }; state.handshake_controller = handshake_controller; state.url = url; diff --git a/moli-renderer-v8/src/network/request_client/tests/websocket.rs b/moli-renderer-v8/src/network/request_client/tests/websocket.rs index e70702ded4..a0de064d95 100644 --- a/moli-renderer-v8/src/network/request_client/tests/websocket.rs +++ b/moli-renderer-v8/src/network/request_client/tests/websocket.rs @@ -11,10 +11,10 @@ async fn page_and_worker_websocket_connectors_follow_resource_runtime_owner() -> let mut connections = Vec::new(); let mut peers = Vec::new(); for (index, client) in [&page, &worker].into_iter().enumerate() { - let mut context = test_websocket_context(); - context.connector = Some(client.websocket_connector()); + let context = test_websocket_context(); let (events, incoming) = tokio::sync::mpsc::channel(8); let connection = spawn_connection( + client.websocket_connector(), 3000 + index as u64, url.clone(), Vec::new(), diff --git a/moli-renderer-v8/src/types.rs b/moli-renderer-v8/src/types.rs index 0f5a05be37..d84e2c503e 100644 --- a/moli-renderer-v8/src/types.rs +++ b/moli-renderer-v8/src/types.rs @@ -307,6 +307,7 @@ impl ImageRequestKey { } } pub(super) struct PendingWebSocketConnection { + pub(super) connector: Option, pub(super) socket_id: u64, pub(super) protocols: Vec, pub(super) connect_options: moli_websocket::ConnectOptions, diff --git a/moli-renderer-v8/src/worker/global_scope/fetch.rs b/moli-renderer-v8/src/worker/global_scope/fetch.rs index 26cc34b1fc..4463317337 100644 --- a/moli-renderer-v8/src/worker/global_scope/fetch.rs +++ b/moli-renderer-v8/src/worker/global_scope/fetch.rs @@ -1276,7 +1276,6 @@ pub(crate) fn register_worker_websocket<'s>( .ok() .and_then(|header| header.clone()); let context = WebSocketConnectOptions { - connector: Some(loader.request_client().websocket_connector()), origin: moli_url::origin_ascii_serialization(&document_url), user_agent: loader.request_client().user_agent().to_owned(), extra_headers: extra_http_headers, @@ -1327,6 +1326,7 @@ pub(crate) fn register_worker_websocket<'s>( None, )?; let connection = spawn_connection( + loader.request_client().websocket_connector(), socket_id, url.to_string(), protocols, diff --git a/moli-websocket/src/connection.rs b/moli-websocket/src/connection.rs index 96602dc6eb..4aadddd5e9 100644 --- a/moli-websocket/src/connection.rs +++ b/moli-websocket/src/connection.rs @@ -10,6 +10,7 @@ use crate::{ }; pub(crate) async fn run_websocket_connection( + connector: moli_curl::websocket::CurlWebSocketConnector, socket_id: u64, url: String, protocols: Vec, @@ -51,7 +52,7 @@ pub(crate) async fn run_websocket_connection( // The handshake future owns the pending transport. End its scope before // waiting for terminal delivery, including Close received while opening. let opened = { - let handshake = open_websocket_connection(request, &context); + let handshake = open_websocket_connection(&connector, request, &context); tokio::pin!(handshake); loop { tokio::select! { diff --git a/moli-websocket/src/lib.rs b/moli-websocket/src/lib.rs index 323fa95441..ab9ca654a8 100644 --- a/moli-websocket/src/lib.rs +++ b/moli-websocket/src/lib.rs @@ -28,6 +28,7 @@ pub use protocol::{ }; pub use runtime::{ spawn_connection, spawn_connection_with_handshake_pause, spawn_failed_connection, + spawn_standalone_connection, spawn_standalone_connection_with_handshake_pause, spawn_synthetic_connection, }; pub(crate) use types::Command; diff --git a/moli-websocket/src/native_tests.rs b/moli-websocket/src/native_tests.rs index baaf231ec4..eb52d368a8 100644 --- a/moli-websocket/src/native_tests.rs +++ b/moli-websocket/src/native_tests.rs @@ -1,4 +1,4 @@ -use crate::{Event, FrameOpcode, spawn_connection, test_support::*}; +use crate::{Event, FrameOpcode, spawn_standalone_connection, test_support::*}; use futures_util::{SinkExt, StreamExt}; use tokio::{ io::AsyncWriteExt, @@ -25,7 +25,7 @@ async fn native_send_accounting_does_not_require_a_server_echo() { let _ = socket.flush().await; }); let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(70, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(70, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; handle.send_text("without echo".to_owned()).unwrap(); assert_send_completed(&mut rx, 70, FrameOpcode::Text, 12).await; @@ -51,7 +51,7 @@ async fn native_local_close_without_peer_close_is_abnormal_once() { // Intentionally drop without flushing tungstenite's queued Close reply. }); let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(71, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(71, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; handle .close(Some(3001), "not acknowledged".to_owned()) @@ -105,7 +105,7 @@ async fn native_fragmented_utf8_and_ping_are_assembled_as_one_message() { }) .await; let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(72, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(72, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; assert_text_message(&mut rx, 72, "🙂").await; assert_text_message(&mut rx, 72, "").await; @@ -154,7 +154,8 @@ async fn native_invalid_frames_and_messages_fail_before_delivery() { }) .await; let (tx, mut rx) = mpsc::channel(8); - let _handle = spawn_connection(73, url, Vec::new(), test_websocket_context(), tx); + let _handle = + spawn_standalone_connection(73, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; match timeout(Duration::from_secs(3), rx.recv()).await.unwrap() { Some(Event::Error { message, .. }) => { @@ -209,7 +210,7 @@ async fn native_message_size_limit_counts_fragments_and_resets_after_delivery() }) .await; let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(76, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(76, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; let event = timeout(Duration::from_secs(10), rx.recv()).await.unwrap(); if overflow { @@ -260,7 +261,7 @@ async fn native_large_incoming_frame_keeps_chunk_offsets_and_utf8() { }) .await; let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(74, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(74, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; assert_text_message(&mut rx, 74, &expected).await; handle.close(Some(1000), String::new()).unwrap(); @@ -307,7 +308,7 @@ async fn native_close_handshake_finishes_while_message_sink_is_blocked() { } } }); - let handle = spawn_connection(75, url, Vec::new(), test_websocket_context(), sink); + let handle = spawn_standalone_connection(75, url, Vec::new(), test_websocket_context(), sink); recv_open_event(&mut rx).await; timeout(Duration::from_secs(3), blocked.notified()) .await diff --git a/moli-websocket/src/runtime.rs b/moli-websocket/src/runtime.rs index 4593c69ab7..7e504dc961 100644 --- a/moli-websocket/src/runtime.rs +++ b/moli-websocket/src/runtime.rs @@ -1,3 +1,4 @@ +use moli_curl::websocket::{CurlWebSocketConnector, CurlWebSocketRuntime}; use std::sync::OnceLock; use crate::commands::command_channel; @@ -9,17 +10,27 @@ use crate::{ }; pub fn spawn_connection( + connector: CurlWebSocketConnector, socket_id: u64, url: String, protocols: Vec, context: ConnectOptions, event_tx: impl Into, ) -> ConnectionHandle { - spawn_native_connection(socket_id, url, protocols, context, event_tx.into(), None) + spawn_native_connection( + Ok(connector), + socket_id, + url, + protocols, + context, + event_tx.into(), + None, + ) } /// Starts a native connection whose browser Open awaits one explicit decision. pub fn spawn_connection_with_handshake_pause( + connector: CurlWebSocketConnector, socket_id: u64, url: String, protocols: Vec, @@ -28,6 +39,7 @@ pub fn spawn_connection_with_handshake_pause( ) -> (ConnectionHandle, HandshakeController) { let (decision, receiver) = tokio::sync::oneshot::channel(); let connection = spawn_native_connection( + Ok(connector), socket_id, url, protocols, @@ -38,7 +50,56 @@ pub fn spawn_connection_with_handshake_pause( (connection, HandshakeController::new(decision)) } +/// Uses the process runtime for callers without a network owner. +pub fn spawn_standalone_connection( + socket_id: u64, + url: String, + protocols: Vec, + context: ConnectOptions, + event_tx: impl Into, +) -> ConnectionHandle { + spawn_native_connection( + standalone_connector(), + socket_id, + url, + protocols, + context, + event_tx.into(), + None, + ) +} + +pub fn spawn_standalone_connection_with_handshake_pause( + socket_id: u64, + url: String, + protocols: Vec, + context: ConnectOptions, + event_tx: impl Into, +) -> (ConnectionHandle, HandshakeController) { + let (decision, receiver) = tokio::sync::oneshot::channel(); + let connection = spawn_native_connection( + standalone_connector(), + socket_id, + url, + protocols, + context, + event_tx.into(), + Some(receiver), + ); + (connection, HandshakeController::new(decision)) +} + +pub(super) fn standalone_connector() -> Result { + static RUNTIME: OnceLock> = OnceLock::new(); + RUNTIME + .get_or_init(|| CurlWebSocketRuntime::new().map_err(|error| error.to_string())) + .as_ref() + .map(CurlWebSocketRuntime::connector) + .map_err(Clone::clone) +} + fn spawn_native_connection( + connector: Result, socket_id: u64, url: String, protocols: Vec, @@ -48,8 +109,16 @@ fn spawn_native_connection( ) -> ConnectionHandle { let (command_tx, command_rx) = command_channel(); let task = websocket_runtime().spawn(async move { + let connector = match connector { + Ok(connector) => connector, + Err(error) => { + drop(command_rx); + let _ = send_error_and_close(&event_tx, socket_id, error).await; + return; + } + }; let _ = run_websocket_connection( - socket_id, url, protocols, context, command_rx, event_tx, decision, + connector, socket_id, url, protocols, context, command_rx, event_tx, decision, ) .await; }); diff --git a/moli-websocket/src/runtime_tests.rs b/moli-websocket/src/runtime_tests.rs index 53b7083173..0d15efb9d7 100644 --- a/moli-websocket/src/runtime_tests.rs +++ b/moli-websocket/src/runtime_tests.rs @@ -19,10 +19,10 @@ async fn next(events: &mut mpsc::Receiver) -> Event { async fn provided_native_connector_shutdown_ends_browser_session() { let (url, server) = spawn_text_echo_websocket_server().await; let runtime = CurlWebSocketRuntime::new().unwrap(); - let mut context = test_websocket_context(); - context.connector = Some(runtime.connector()); + let context = test_websocket_context(); + let connector = runtime.connector(); let (events, mut incoming) = mpsc::channel(8); - let handle = spawn_connection(1000, url, Vec::new(), context, events); + let handle = spawn_connection(connector, 1000, url, Vec::new(), context, events); assert!(matches!(next(&mut incoming).await, Event::Open { .. })); handle.send_text("shared owner".into()).unwrap(); assert!(matches!( @@ -52,11 +52,11 @@ async fn closed_provided_connector_never_falls_back_to_global_owner() { listener.set_nonblocking(true).unwrap(); let url = format!("ws://{}/must-not-connect", listener.local_addr().unwrap()); let runtime = CurlWebSocketRuntime::new().unwrap(); - let mut context = test_websocket_context(); - context.connector = Some(runtime.connector()); + let context = test_websocket_context(); + let connector = runtime.connector(); drop(runtime); let (events, mut incoming) = mpsc::channel(8); - let handle = spawn_connection(1001, url, Vec::new(), context, events); + let handle = spawn_connection(connector, 1001, url, Vec::new(), context, events); assert!( matches!(next(&mut incoming).await, Event::Error { message, .. } if message.contains("curl WebSocket runtime is closed")) ); diff --git a/moli-websocket/src/session/tests.rs b/moli-websocket/src/session/tests.rs index 9a0b7f809d..4dd9096e91 100644 --- a/moli-websocket/src/session/tests.rs +++ b/moli-websocket/src/session/tests.rs @@ -1,5 +1,5 @@ use super::CLOSE_TIMEOUT; -use crate::{Command, Event, FrameOpcode, spawn_connection, test_support::*}; +use crate::{Command, Event, FrameOpcode, spawn_standalone_connection, test_support::*}; use futures_util::{SinkExt, StreamExt}; use tokio::{ net::{TcpListener, TcpSocket}, @@ -39,7 +39,7 @@ async fn native_close_waits_for_queued_data_before_starting_timeout() { socket.flush().await.unwrap(); }); let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(80, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(80, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; handle .send_binary((0..MESSAGE_BYTES).map(|i| (i % 251) as u8).collect()) @@ -78,7 +78,7 @@ async fn native_close_still_times_out_when_peer_never_answers_close() { let _ = finish_rx.await; }); let (tx, mut rx) = mpsc::channel(8); - let handle = spawn_connection(81, url, Vec::new(), test_websocket_context(), tx); + let handle = spawn_standalone_connection(81, url, Vec::new(), test_websocket_context(), tx); recv_open_event(&mut rx).await; handle.close(Some(1000), String::new()).unwrap(); assert_closing(&mut rx, 81).await; @@ -151,7 +151,7 @@ async fn native_fragmented_sends_complete_while_incoming_sink_is_blocked() { } } }); - let handle = spawn_connection(82, url, Vec::new(), test_websocket_context(), sink); + let handle = spawn_standalone_connection(82, url, Vec::new(), test_websocket_context(), sink); recv_open_event(&mut rx).await; timeout(Duration::from_secs(3), blocked.notified()) .await diff --git a/moli-websocket/src/terminal_tests.rs b/moli-websocket/src/terminal_tests.rs index 6046fa79e1..6aafabf520 100644 --- a/moli-websocket/src/terminal_tests.rs +++ b/moli-websocket/src/terminal_tests.rs @@ -106,6 +106,7 @@ async fn failed_handshake_closes_commands_before_blocked_error() { let (commands, receiver) = command_channel(); let (sink, mut events, release) = blocked_terminal_sink(); let task = tokio::spawn(run_websocket_connection( + crate::runtime::standalone_connector().unwrap(), 99, url, Vec::new(), @@ -138,6 +139,7 @@ async fn close_during_handshake_releases_tcp_before_blocked_error() { let (commands, receiver) = command_channel(); let (sink, mut events, release) = blocked_terminal_sink(); let task = tokio::spawn(run_websocket_connection( + crate::runtime::standalone_connector().unwrap(), 100, url, Vec::new(), @@ -197,6 +199,7 @@ async fn rejected_handshake_decision_closes_commands_before_blocked_error() { let (decision, decide) = oneshot::channel(); let (sink, mut events, release) = blocked_terminal_sink(); let task = tokio::spawn(run_websocket_connection( + crate::runtime::standalone_connector().unwrap(), 101, url, Vec::new(), diff --git a/moli-websocket/src/test_support.rs b/moli-websocket/src/test_support.rs index c4ac0b2508..b341d34de1 100644 --- a/moli-websocket/src/test_support.rs +++ b/moli-websocket/src/test_support.rs @@ -16,7 +16,8 @@ use tokio_tungstenite::tungstenite::{ }; use crate::{ - ConnectOptions, Event, FrameOpcode, proxy::websocket_proxy_url_with_env, spawn_connection, + ConnectOptions, Event, FrameOpcode, proxy::websocket_proxy_url_with_env, + spawn_standalone_connection, }; struct InfallibleHandshakeCallback(F); @@ -32,7 +33,6 @@ where pub fn test_websocket_context() -> ConnectOptions { ConnectOptions { - connector: None, origin: "https://example.com".to_owned(), user_agent: "Moli-WebSocket-Test/1.0".to_owned(), extra_headers: Vec::new(), @@ -84,7 +84,8 @@ pub async fn websocket_computed_accept_handshake_failure_message( async fn websocket_handshake_failure_message(url: String, protocols: Vec) -> String { let (event_tx, mut event_rx) = mpsc::channel(32); - let _command_tx = spawn_connection(90, url, protocols, test_websocket_context(), event_tx); + let _command_tx = + spawn_standalone_connection(90, url, protocols, test_websocket_context(), event_tx); recv_handshake_failure_events(&mut event_rx).await } diff --git a/moli-websocket/src/tests.rs b/moli-websocket/src/tests.rs index b16a45032f..cc0ebd8540 100644 --- a/moli-websocket/src/tests.rs +++ b/moli-websocket/src/tests.rs @@ -451,7 +451,8 @@ async fn websocket_last_handle_drop_releases_pending_handshake() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("ws://{}/pending", listener.local_addr().unwrap()); let (event_tx, _event_rx) = mpsc::channel(1); - let handle = spawn_connection(61, url, Vec::new(), test_websocket_context(), event_tx); + let handle = + spawn_standalone_connection(61, url, Vec::new(), test_websocket_context(), event_tx); let retained = handle.clone(); let (mut server, _) = timeout(Duration::from_secs(3), listener.accept()) .await @@ -495,6 +496,7 @@ async fn websocket_rejected_handshake_or_open_event_releases_transport() { assert!(terminal.is_none() || terminal.is_some_and(|result| result.is_err())); }; let client = crate::connection::run_websocket_connection( + crate::runtime::standalone_connector().unwrap(), 64, url, Vec::new(), @@ -535,6 +537,7 @@ async fn websocket_rejected_message_event_releases_transport_and_writer() { assert!(terminal.is_none() || terminal.is_some_and(|result| result.is_err())); }; let client = crate::connection::run_websocket_connection( + crate::runtime::standalone_connector().unwrap(), 65, url, Vec::new(), @@ -614,7 +617,7 @@ async fn websocket_cancel_releases_blocked_sink_and_open_writer() { true } }); - let handle = spawn_connection(62, url, Vec::new(), test_websocket_context(), sink); + let handle = spawn_standalone_connection(62, url, Vec::new(), test_websocket_context(), sink); let retained = handle.clone(); let (stream, _) = timeout(Duration::from_secs(3), listener.accept()) .await @@ -820,7 +823,8 @@ async fn websocket_synthetic_connection_can_receive_frames_and_server_close() { async fn websocket_transport_close_while_connecting_fails_before_open() { let (url, server) = spawn_sleeping_handshake_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let command_tx = spawn_connection(91, url, Vec::new(), test_websocket_context(), event_tx); + let command_tx = + spawn_standalone_connection(91, url, Vec::new(), test_websocket_context(), event_tx); command_tx .close(None, String::new()) @@ -836,7 +840,7 @@ async fn websocket_transport_close_while_connecting_fails_before_open() { async fn websocket_transport_can_pause_after_handshake_before_open() { let (url, server) = spawn_text_echo_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let (command_tx, controller) = spawn_connection_with_handshake_pause( + let (command_tx, controller) = spawn_standalone_connection_with_handshake_pause( 92, url, Vec::new(), @@ -890,7 +894,8 @@ async fn websocket_transport_handshake_applies_context_headers_and_preserves_con ]; context.cookie_header = Some("sid=server".to_owned()); - let command_tx = spawn_connection(1, url, vec!["chat".to_owned()], context, event_tx); + let command_tx = + spawn_standalone_connection(1, url, vec!["chat".to_owned()], context, event_tx); let headers = timeout(Duration::from_secs(3), headers_rx) .await .expect("websocket headers should arrive") @@ -939,7 +944,7 @@ async fn websocket_transport_uses_explicit_http_proxy_connect_without_forwarding context.proxy_bearer_token = Some("proxy-token".to_owned()); let user_agent = context.user_agent.clone(); - let command_tx = spawn_connection(2, url.clone(), Vec::new(), context, event_tx); + let command_tx = spawn_standalone_connection(2, url.clone(), Vec::new(), context, event_tx); let proxy_request = timeout(Duration::from_secs(3), proxy_request_rx) .await .expect("proxy CONNECT should arrive") @@ -985,7 +990,7 @@ async fn websocket_transport_rejects_non_200_http_proxy_connect() { context.http_proxy = Some(proxy_url); context.http_no_proxy = Some(String::new()); - let _command_tx = spawn_connection( + let _command_tx = spawn_standalone_connection( 3, "ws://example.test/socket".to_owned(), Vec::new(), @@ -1016,7 +1021,7 @@ async fn websocket_transport_respects_disabled_tls_verify_for_self_signed_wss() let mut context = test_websocket_context(); context.tls.verify = false; - let command_tx = spawn_connection(3, url, Vec::new(), context, event_tx); + let command_tx = spawn_standalone_connection(3, url, Vec::new(), context, event_tx); let headers = timeout(Duration::from_secs(3), headers_rx) .await .expect("websocket TLS headers should arrive") @@ -1041,7 +1046,7 @@ async fn websocket_transport_wss_allows_server_to_omit_response_subprotocol() { let mut context = test_websocket_context(); context.tls.verify = false; - let _command_tx = spawn_connection( + let _command_tx = spawn_standalone_connection( 4, url, vec!["chat".to_owned(), "superchat".to_owned()], @@ -1175,7 +1180,7 @@ async fn websocket_transport_allows_server_to_omit_response_subprotocol() { let (url, server) = spawn_text_binary_echo_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let command_tx = spawn_connection( + let command_tx = spawn_standalone_connection( 4, url, vec!["chat".to_owned(), "superchat".to_owned()], @@ -1200,7 +1205,8 @@ async fn websocket_transport_allows_server_to_omit_response_subprotocol() { async fn websocket_transport_sends_text_binary_and_reports_buffered_amount_consumption() { let (url, server) = spawn_text_binary_echo_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let command_tx = spawn_connection(4, url, Vec::new(), test_websocket_context(), event_tx); + let command_tx = + spawn_standalone_connection(4, url, Vec::new(), test_websocket_context(), event_tx); let open = recv_open_event(&mut event_rx).await; assert_eq!(open.socket_id, 4); @@ -1231,7 +1237,8 @@ async fn websocket_transport_sends_text_binary_and_reports_buffered_amount_consu async fn websocket_transport_reads_while_sending_many_large_messages() { let (url, server) = spawn_backpressure_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let command_tx = spawn_connection(40, url, Vec::new(), test_websocket_context(), event_tx); + let command_tx = + spawn_standalone_connection(40, url, Vec::new(), test_websocket_context(), event_tx); let open = recv_open_event(&mut event_rx).await; assert_eq!(open.socket_id, 40); @@ -1280,7 +1287,8 @@ async fn websocket_transport_reads_while_sending_many_large_messages() { async fn websocket_transport_reports_server_initiated_close_frame() { let (url, server) = spawn_server_close_websocket_server().await; let (event_tx, mut event_rx) = mpsc::channel(32); - let _command_tx = spawn_connection(5, url, Vec::new(), test_websocket_context(), event_tx); + let _command_tx = + spawn_standalone_connection(5, url, Vec::new(), test_websocket_context(), event_tx); let open = recv_open_event(&mut event_rx).await; assert_eq!(open.socket_id, 5); @@ -1299,7 +1307,8 @@ async fn websocket_transport_handles_close_frame_in_handshake_packet() { ) .await; let (event_tx, mut event_rx) = mpsc::channel(32); - let _command_tx = spawn_connection(6, url, Vec::new(), test_websocket_context(), event_tx); + let _command_tx = + spawn_standalone_connection(6, url, Vec::new(), test_websocket_context(), event_tx); let open = recv_open_event(&mut event_rx).await; assert_eq!(open.socket_id, 6); @@ -1320,7 +1329,7 @@ async fn websocket_dropped_handshake_controller_fails_open_and_releases_transpor assert!(socket.next().await.unwrap().is_err()); }); let (events, mut receiver) = mpsc::channel(4); - let (_connection, controller) = spawn_connection_with_handshake_pause( + let (_connection, controller) = spawn_standalone_connection_with_handshake_pause( 95, url, Vec::new(), @@ -1387,8 +1396,13 @@ async fn websocket_queued_close_precedes_handshake_continuation() { accepted } }); - let (connection, controller) = - spawn_connection_with_handshake_pause(97, url, Vec::new(), test_websocket_context(), sink); + let (connection, controller) = spawn_standalone_connection_with_handshake_pause( + 97, + url, + Vec::new(), + test_websocket_context(), + sink, + ); assert!(matches!( timeout(Duration::from_secs(3), receiver.recv()) .await @@ -1432,7 +1446,7 @@ async fn websocket_native_handshake_generates_keys_and_reports_actual_headers() "{header} belongs to native handshake generation" ); } - let connection = spawn_connection(socket_id, url, Vec::new(), context, events); + let connection = spawn_standalone_connection(socket_id, url, Vec::new(), context, events); let actual = timeout(Duration::from_secs(3), captured) .await .unwrap() diff --git a/moli-websocket/src/tests/tls.rs b/moli-websocket/src/tests/tls.rs index 722005ceca..d60e368c1d 100644 --- a/moli-websocket/src/tests/tls.rs +++ b/moli-websocket/src/tests/tls.rs @@ -12,7 +12,7 @@ fn tls_context(tls: CurlTlsConfig) -> ConnectOptions { async fn exercise_tls_connection(url: String, context: ConnectOptions, opens: bool) { let (events, mut receiver) = mpsc::channel(32); - let connection = spawn_connection(501, url, Vec::new(), context, events); + let connection = spawn_standalone_connection(501, url, Vec::new(), context, events); if !opens { assert!( !recv_handshake_failure_events(&mut receiver) diff --git a/moli-websocket/src/transport.rs b/moli-websocket/src/transport.rs index 7d8b6f042e..2e209cf73e 100644 --- a/moli-websocket/src/transport.rs +++ b/moli-websocket/src/transport.rs @@ -8,11 +8,10 @@ use crate::{ use moli_curl::{ CurlDnsResolution, websocket::{ - CurlWebSocketConnection, CurlWebSocketEvent, CurlWebSocketRequest, CurlWebSocketRuntime, + CurlWebSocketConnection, CurlWebSocketConnector, CurlWebSocketEvent, CurlWebSocketRequest, }, }; use moli_dns_resolver::DnsTarget; -use std::sync::OnceLock; pub(crate) struct HandshakeInfo { pub request_headers: http::HeaderMap, @@ -25,6 +24,7 @@ pub(crate) struct OpenedConnection { } pub(crate) async fn open_websocket_connection( + connector: &CurlWebSocketConnector, request: PreparedWebSocketRequest, context: &ConnectOptions, ) -> Result { @@ -57,18 +57,9 @@ pub(crate) async fn open_websocket_connection( ); } } - let mut connection = match &context.connector { - Some(connector) => connector.connect(native), - None => { - static RUNTIME: OnceLock> = OnceLock::new(); - let runtime = RUNTIME - .get_or_init(|| CurlWebSocketRuntime::new().map_err(|error| error.to_string())) - .as_ref() - .map_err(Clone::clone)?; - runtime.connect(native) - } - } - .map_err(|error| error.to_string())?; + let mut connection = connector + .connect(native) + .map_err(|error| error.to_string())?; match connection.recv().await { Some(CurlWebSocketEvent::Handshake { request: actual, diff --git a/moli-websocket/src/types.rs b/moli-websocket/src/types.rs index 590491781d..584ac80b6b 100644 --- a/moli-websocket/src/types.rs +++ b/moli-websocket/src/types.rs @@ -1,8 +1,5 @@ #[derive(Debug, Clone)] pub struct ConnectOptions { - /// Native runtime for this network scope. None selects the standalone - /// process runtime. A closed provided connector fails without fallback. - pub connector: Option, pub origin: String, pub user_agent: String, pub extra_headers: Vec<(String, String)>,