refactor(curl): organize runtime and protocol modules

moli-curl/README.md

@@ -1,4 +1,27 @@
-# Native WebSocket owner diagnostics
+# Native HTTP and WebSocket runtime
+
+## Module boundaries
+
+| Module | Responsibility |
+| --- | --- |
+| `runtime.rs`, `runtime/owner.rs` | Runtime ownership and the single native thread: submit handling, shutdown, `perform`, completion dispatch and `poll` |
+| `http.rs`, `http/registry.rs`, `http/scheduling.rs` | HTTP request capability, priority/origin admission, DNS/deadlines and transfer completion |
+| `websocket/connector.rs`, `websocket/connection.rs` | WebSocket submission capability and the caller/owner I/O endpoints, including independent send completion |
+| `websocket/registry.rs`, `websocket/session.rs` | Resident native connections, handshake and frame I/O |
+| `websocket/readiness.rs` | I/O admission, extra socket interests and applying the shared poll's readiness results |
+| `websocket/standalone.rs` | Convenience owner using the same runtime for standalone callers |
+| `dns_adapter.rs`, `tls.rs` | Shared DNS and TLS configuration |
+
+The owner dispatches to concrete HTTP and WebSocket registries. HTTP completion
+removes a transfer; WebSocket handshake completion keeps its handle resident.
+Both registries preserve the easy handle's error details when interpreting
+`CURLMSG_DONE`. Only `runtime/owner.rs` calls `perform`, `messages` and `poll`.
+Read that file first when investigating scheduling or thread shutdown.
+
+Public exports remain at `moli_curl::*` for HTTP/runtime and
+`moli_curl::websocket::*` for WebSocket. Neither protocol's request capability
+owns the native thread. Shared owner counters live in `runtime/diagnostics.rs`;
+the diagnostics switch and log fields below remain compatible.
 
 `moli-curl` owns native connections and frame I/O. Browser message assembly,
 UTF-8 checks, command admission and the close handshake live in `moli-websocket`.

moli-curl/src/http.rs

@@ -0,0 +1,154 @@
+//! HTTP submission types and owner-local request scheduling.
+//! The shared runtime drives libcurl; this module owns HTTP jobs until completion.
+
+pub(crate) mod registry;
+mod scheduling;
+
+use crate::{
+    CurlDnsResolution, CurlTransferId,
+    runtime::{CurlRuntimeCommand, identity::next_transfer_id},
+};
+use anyhow::{Result, anyhow};
+use crossbeam_channel::Sender;
+use curl::{
+    easy::{Easy2, Handler},
+    multi::MultiWaker,
+};
+use std::{
+    fmt,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+    time::Instant,
+};
+
+/// Origin key used by the curl scheduler for per-origin active transfer caps.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CurlOriginKey {
+    pub scheme: String,
+    pub host: String,
+    pub port: Option<u16>,
+}
+
+/// A configured curl transfer plus scheduler metadata.
+pub struct CurlMultiJob<H: Handler, C> {
+    pub easy: Easy2<H>,
+    pub context: C,
+    pub origin: Option<CurlOriginKey>,
+    /// Absolute deadline for the whole scheduler-owned transfer attempt.
+    ///
+    /// libcurl cannot account for time spent in Moli's priority queue or in
+    /// the shared DNS residence because both happen before the easy handle is
+    /// added to the multi handle. The owner enforces this deadline in those
+    /// residences and gives libcurl only the remaining duration.
+    pub deadline: Option<Instant>,
+    /// DNS ownership chosen by the caller before this transfer enters curl.
+    ///
+    /// A curl-managed policy preserves libcurl's resolver behavior. A shared
+    /// origin policy parks the transfer outside the curl multi handle set until
+    /// the bounded system resolver publishes an answer.
+    pub dns_resolution: CurlDnsResolution,
+    /// Higher values start before lower values when jobs are queued.
+    pub priority: u8,
+    pub label: String,
+}
+
+impl<H: Handler, C> fmt::Debug for CurlMultiJob<H, C> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("CurlMultiJob")
+            .field("origin", &self.origin)
+            .field("deadline", &self.deadline)
+            .field("dns_resolution", &self.dns_resolution)
+            .field("priority", &self.priority)
+            .field("label", &self.label)
+            .finish_non_exhaustive()
+    }
+}
+
+/// Completion emitted by `CurlMultiRuntime`.
+pub struct CurlMultiCompletion<H: Handler, C> {
+    pub transfer_id: CurlTransferId,
+    pub easy: Option<Easy2<H>>,
+    pub context: C,
+    pub result: Result<()>,
+}
+
+impl<H: Handler, C> fmt::Debug for CurlMultiCompletion<H, C> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("CurlMultiCompletion")
+            .field("transfer_id", &self.transfer_id)
+            .field("has_easy", &self.easy.is_some())
+            .field("result", &self.result.as_ref().map(|_| ()))
+            .finish_non_exhaustive()
+    }
+}
+
+/// Error returned when a job cannot be submitted and is returned to the caller.
+pub struct CurlSubmitError<H: Handler, C> {
+    pub job: CurlMultiJob<H, C>,
+    pub error: anyhow::Error,
+}
+
+impl<H: Handler, C> fmt::Debug for CurlSubmitError<H, C> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("CurlSubmitError")
+            .field("job", &self.job)
+            .field("error", &self.error)
+            .finish()
+    }
+}
+
+/// Submits HTTP work to one owner. Clones carry no shutdown or join authority.
+#[derive(Debug)]
+pub struct CurlHttpSender<H: Handler + Send + 'static, C: Send + 'static> {
+    pub(crate) command_tx: Sender<CurlRuntimeCommand<H, C>>,
+    pub(crate) owner_waker: MultiWaker,
+    pub(crate) shutdown_requested: Arc<AtomicBool>,
+}
+
+impl<H: Handler + Send + 'static, C: Send + 'static> Clone for CurlHttpSender<H, C> {
+    fn clone(&self) -> Self {
+        Self {
+            command_tx: self.command_tx.clone(),
+            owner_waker: self.owner_waker.clone(),
+            shutdown_requested: self.shutdown_requested.clone(),
+        }
+    }
+}
+
+impl<H: Handler + Send + 'static, C: Send + 'static> CurlHttpSender<H, C> {
+    pub fn submit(
+        &self,
+        job: CurlMultiJob<H, C>,
+    ) -> std::result::Result<CurlTransferId, CurlSubmitError<H, C>> {
+        if self.shutdown_requested.load(Ordering::SeqCst) {
+            return Err(CurlSubmitError {
+                job,
+                error: anyhow!("curl multi runtime is shutting down"),
+            });
+        }
+        let transfer_id = match next_transfer_id() {
+            Ok(transfer_id) => transfer_id,
+            Err(error) => return Err(CurlSubmitError { job, error }),
+        };
+        match self
+            .command_tx
+            .send(CurlRuntimeCommand::Request { transfer_id, job })
+        {
+            Ok(()) => {
+                let _ = self.owner_waker.wakeup();
+                Ok(transfer_id)
+            }
+            Err(error) => {
+                let CurlRuntimeCommand::Request { job, .. } = error.into_inner() else {
+                    unreachable!("submit only sends request commands");
+                };
+                Err(CurlSubmitError {
+                    job,
+                    error: anyhow!("curl multi runtime is shutting down"),
+                })
+            }
+        }
+    }
+}

moli-curl/src/http/registry.rs

@@ -0,0 +1,568 @@
+//! HTTP jobs move through the priority queue, DNS and active easy handles.
+//! The runtime alone calls perform/messages/poll and dispatches completions here.
+
+use super::{
+    CurlMultiCompletion, CurlMultiJob,
+    scheduling::{
+        CurlActiveTransfer, CurlPendingJob, active_origin_count, enqueue_existing_pending_job,
+        enqueue_pending_job, job_is_eligible, pending_origin_count, take_expired_pending_jobs,
+        take_transfers_in_notification_order,
+    },
+};
+use crate::{
+    CurlMultiRuntimeConfig, CurlTransferId,
+    dns_adapter::{CurlDnsOwnerCompletion, CurlDnsOwnerResidence, CurlDnsReady},
+};
+use anyhow::{Context, Result, anyhow};
+use crossbeam_channel::{Receiver, Sender};
+use curl::{
+    easy::Handler,
+    multi::{Message, Multi},
+};
+use std::{
+    collections::{HashMap, VecDeque},
+    num::NonZeroUsize,
+    sync::OnceLock,
+    time::{Duration, Instant},
+};
+
+pub(crate) struct HttpRegistry<H: Handler, C> {
+    config: CurlMultiRuntimeConfig,
+    completion_tx: Sender<CurlMultiCompletion<H, C>>,
+    closed: bool,
+    pending: VecDeque<CurlPendingJob<H, C>>,
+    dns: CurlDnsOwnerResidence<CurlTransferId, CurlPendingJob<H, C>>,
+    active: HashMap<CurlTransferId, CurlActiveTransfer<H, C>>,
+}
+
+impl<H: Handler, C> HttpRegistry<H, C> {
+    pub(crate) fn new(
+        config: CurlMultiRuntimeConfig,
+        completion_tx: Sender<CurlMultiCompletion<H, C>>,
+    ) -> Self {
+        Self {
+            config,
+            completion_tx,
+            closed: false,
+            pending: VecDeque::new(),
+            dns: CurlDnsOwnerResidence::default(),
+            active: HashMap::new(),
+        }
+    }
+
+    pub(crate) fn has_active(&self) -> bool {
+        !self.active.is_empty()
+    }
+
+    /// DNS-only work can wait on its completion channel without polling curl.
+    pub(crate) fn has_curl_work(&self) -> bool {
+        self.has_active() || !self.pending.is_empty()
+    }
+
+    pub(crate) fn next_deadline(&self) -> Option<Instant> {
+        self.pending
+            .iter()
+            .filter_map(|pending| pending.job.deadline)
+            .chain(self.dns.next_deadline(|pending| pending.job.deadline))
+            .min()
+    }
+
+    pub(crate) fn dns_completions(&self) -> &Receiver<CurlDnsOwnerCompletion<CurlTransferId>> {
+        self.dns.completion_receiver()
+    }
+
+    pub(crate) fn advance(&mut self, multi: &mut Multi) {
+        self.drain_dns_completions();
+        self.expire_waiting_jobs();
+        self.start_eligible_jobs(multi);
+    }
+
+    pub(crate) fn contains(&self, id: CurlTransferId) -> bool {
+        self.active.contains_key(&id)
+    }
+
+    pub(crate) fn completion_result(
+        &self,
+        id: CurlTransferId,
+        message: &Message<'_>,
+    ) -> Option<Result<(), curl::Error>> {
+        message.result_for2(&self.active.get(&id)?.handle)
+    }
+
+    pub(crate) fn complete(
+        &mut self,
+        multi: &mut Multi,
+        completed: Vec<(CurlTransferId, Result<(), curl::Error>)>,
+    ) {
+        for (transfer_id, active, result) in
+            take_transfers_in_notification_order(&mut self.active, completed)
+        {
+            self.finish_active_transfer(multi, transfer_id, active, result.map_err(Into::into));
+        }
+    }
+
+    pub(crate) fn admit(&mut self, transfer_id: CurlTransferId, job: CurlMultiJob<H, C>) {
+        if self.closed {
+            self.send_completion(CurlMultiCompletion {
+                transfer_id,
+                easy: Some(job.easy),
+                context: job.context,
+                result: Err(anyhow!("curl multi runtime is shutting down")),
+            });
+            return;
+        }
+        if curl_runtime_trace_enabled() {
+            let origin = job.origin.as_ref();
+            tracing::info!(
+                target: "moli_cdp_nav_timing",
+                transfer_id = %transfer_id,
+                label = %job.label,
+                origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
+                origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
+                origin_port = ?origin.and_then(|origin| origin.port),
+                origin = ?job.origin,
+                priority = job.priority,
+                pending_before = self.pending.len(),
+                pending_same_origin_before = origin
+                    .map(|origin| pending_origin_count(&self.pending, origin))
+                    .unwrap_or(0),
+                stage = "curl_runtime_job_queued",
+            );
+        }
+        enqueue_pending_job(&mut self.pending, transfer_id, job);
+    }
+
+    pub(crate) fn shutdown(&mut self, multi: &mut Multi) {
+        if self.closed {
+            return;
+        }
+        self.closed = true;
+        while let Some(pending) = self.pending.pop_front() {
+            let CurlPendingJob {
+                transfer_id, job, ..
+            } = pending;
+            self.send_completion(CurlMultiCompletion {
+                transfer_id,
+                easy: Some(job.easy),
+                context: job.context,
+                result: Err(anyhow!("curl multi runtime is shutting down")),
+            });
+        }
+        for pending in self.dns.drain() {
+            let CurlPendingJob {
+                transfer_id, job, ..
+            } = pending;
+            let _ = self.completion_tx.send(CurlMultiCompletion {
+                transfer_id,
+                easy: Some(job.easy),
+                context: job.context,
+                result: Err(anyhow!(
+                    "curl multi runtime DNS request cancelled during shutdown"
+                )),
+            });
+        }
+        for (transfer_id, active) in self.active.drain() {
+            let easy = multi.remove2(active.handle).ok();
+            let _ = self.completion_tx.send(CurlMultiCompletion {
+                transfer_id,
+                easy,
+                context: active.context,
+                result: Err(anyhow!(
+                    "curl multi runtime request cancelled during shutdown"
+                )),
+            });
+        }
+    }
+
+    fn start_eligible_jobs(&mut self, multi: &mut Multi) {
+        loop {
+            if self.closed || self.active.len() >= self.config.max_active.get() {
+                return;
+            }
+            let Some(index) = self.pending.iter().position(|pending| {
+                job_is_eligible(
+                    pending.job.origin.as_ref(),
+                    &self.active,
+                    self.config.max_host_active,
+                )
+            }) else {
+                return;
+            };
+            let pending = self
+                .pending
+                .remove(index)
+                .expect("pending curl job index should exist");
+            let dns_target = pending.job.dns_resolution.target().cloned();
+            match dns_target {
+                Some(target) => self
+                    .dns
+                    .start(pending.transfer_id, pending, target, multi.waker()),
+                None => self.start_job(multi, pending),
+            }
+        }
+    }
+
+    fn drain_dns_completions(&mut self) {
+        while let Some(ready) = self.dns.try_claim_next() {
+            self.handle_dns_completion(ready);
+        }
+    }
+
+    pub(crate) fn claim_dns_completion(
+        &mut self,
+        completion: CurlDnsOwnerCompletion<CurlTransferId>,
+    ) {
+        let Some(ready) = self.dns.claim(completion) else {
+            return;
+        };
+        self.handle_dns_completion(ready);
+    }
+
+    fn handle_dns_completion(&mut self, ready: CurlDnsReady<CurlPendingJob<H, C>>) {
+        let mut pending = ready.pending;
+        if self.closed {
+            let CurlPendingJob {
+                transfer_id, job, ..
+            } = pending;
+            self.send_completion(CurlMultiCompletion {
+                transfer_id,
+                easy: Some(job.easy),
+                context: job.context,
+                result: Err(anyhow!("curl multi runtime is shutting down")),
+            });
+            return;
+        }
+        if pending.deadline_reached(Instant::now()) {
+            self.complete_timed_out_job(pending, "while waiting for DNS");
+            return;
+        }
+        match ready.result {
+            Ok(addresses) => {
+                if let Err(error) = pending
+                    .job
+                    .dns_resolution
+                    .install(&mut pending.job.easy, addresses.as_ref())
+                {
+                    let CurlPendingJob {
+                        transfer_id, job, ..
+                    } = pending;
+                    self.send_completion(CurlMultiCompletion {
+                        transfer_id,
+                        easy: Some(job.easy),
+                        context: job.context,
+                        result: Err(error),
+                    });
+                    return;
+                }
+                enqueue_existing_pending_job(&mut self.pending, pending);
+            }
+            Err(error) => {
+                let CurlPendingJob {
+                    transfer_id, job, ..
+                } = pending;
+                self.send_completion(CurlMultiCompletion {
+                    transfer_id,
+                    easy: Some(job.easy),
+                    context: job.context,
+                    result: Err(anyhow!(error.to_string())),
+                });
+            }
+        }
+    }
+
+    fn start_job(&mut self, multi: &mut Multi, pending: CurlPendingJob<H, C>) {
+        if pending.deadline_reached(Instant::now()) {
+            self.complete_timed_out_job(pending, "while waiting to start");
+            return;
+        }
+        let transfer_id = pending.transfer_id;
+        let queued_for = pending.enqueued_at.elapsed();
+        let mut job = pending.job;
+        let label = job.label.clone();
+        if let Some(deadline) = job.deadline {
+            let Some(remaining) = curl_timeout_for_deadline(deadline, Instant::now()) else {
+                self.send_completion(CurlMultiCompletion {
+                    transfer_id,
+                    easy: Some(job.easy),
+                    context: job.context,
+                    result: Err(curl_runtime_timeout_error("while waiting to start")),
+                });
+                return;
+            };
+            if let Err(error) = job.easy.timeout(remaining) {
+                self.send_completion(CurlMultiCompletion {
+                    transfer_id,
+                    easy: Some(job.easy),
+                    context: job.context,
+                    result: Err(error).context("failed to apply remaining curl request deadline"),
+                });
+                return;
+            }
+        }
+        match multi
+            .add2(job.easy)
+            .with_context(|| anyhow!("failed to add curl easy handle for {label}"))
+        {
+            Ok(mut handle) => {
+                handle
+                    .set_token(transfer_id.token())
+                    .expect("active curl handle must accept its transfer token");
+                if curl_runtime_trace_enabled() {
+                    let origin = job.origin.as_ref();
+                    tracing::info!(
+                        target: "moli_cdp_nav_timing",
+                        transfer_id = %transfer_id,
+                        label = %label,
+                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
+                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
+                        origin_port = ?origin.and_then(|origin| origin.port),
+                        origin = ?job.origin,
+                        priority = job.priority,
+                        queued_ms = queued_for.as_millis(),
+                        active_before = self.active.len(),
+                        active_same_origin_before = origin
+                            .map(|origin| active_origin_count(&self.active, origin))
+                            .unwrap_or(0),
+                        pending_after = self.pending.len(),
+                        pending_same_origin_after = origin
+                            .map(|origin| pending_origin_count(&self.pending, origin))
+                            .unwrap_or(0),
+                        max_active = self.config.max_active.get(),
+                        max_host_active = ?self.config.max_host_active.map(NonZeroUsize::get),
+                        max_host_connections = ?self.config.max_host_connections.map(NonZeroUsize::get),
+                        max_total_connections = ?self.config.max_total_connections.map(NonZeroUsize::get),
+                        max_concurrent_streams = ?self.config.max_concurrent_streams.map(NonZeroUsize::get),
+                        multiplex = self.config.multiplex,
+                        stage = "curl_runtime_job_start",
+                    );
+                }
+                let previous = self.active.insert(
+                    transfer_id,
+                    CurlActiveTransfer {
+                        handle,
+                        context: job.context,
+                        origin: job.origin,
+                        priority: job.priority,
+                        label,
+                        started_at: Instant::now(),
+                        queued_for,
+                    },
+                );
+                assert!(previous.is_none(), "curl transfer identity is unique");
+            }
+            Err(error) => self.send_completion(CurlMultiCompletion {
+                transfer_id,
+                easy: None,
+                context: job.context,
+                result: Err(error),
+            }),
+        }
+    }
+
+    fn expire_waiting_jobs(&mut self) {
+        let now = Instant::now();
+        for pending in take_expired_pending_jobs(&mut self.pending, now) {
+            self.complete_timed_out_job(pending, "while waiting in the scheduler");
+        }
+        for pending in self.dns.take_expired(now, |pending| pending.job.deadline) {
+            self.complete_timed_out_job(pending, "while waiting for DNS");
+        }
+    }
+
+    fn complete_timed_out_job(&self, pending: CurlPendingJob<H, C>, stage: &'static str) {
+        let CurlPendingJob {
+            transfer_id, job, ..
+        } = pending;
+        self.send_completion(CurlMultiCompletion {
+            transfer_id,
+            easy: Some(job.easy),
+            context: job.context,
+            result: Err(curl_runtime_timeout_error(stage)),
+        });
+    }
+
+    fn finish_active_transfer(
+        &self,
+        multi: &mut Multi,
+        transfer_id: CurlTransferId,
+        active: CurlActiveTransfer<H, C>,
+        result: Result<()>,
+    ) {
+        let easy = match multi.remove2(active.handle) {
+            Ok(easy) => Some(easy),
+            Err(error) => {
+                self.send_completion(CurlMultiCompletion {
+                    transfer_id,
+                    easy: None,
+                    context: active.context,
+                    result: Err(anyhow!(
+                        "failed to remove curl easy handle for {}: {error}",
+                        active.label
+                    )),
+                });
+                return;
+            }
+        };
+        if curl_runtime_trace_enabled() {
+            let origin = active.origin.as_ref();
+            match &result {
+                Ok(()) => {
+                    tracing::info!(
+                        target: "moli_cdp_nav_timing",
+                        transfer_id = %transfer_id,
+                        label = %active.label,
+                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
+                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
+                        origin_port = ?origin.and_then(|origin| origin.port),
+                        origin = ?active.origin,
+                        priority = active.priority,
+                        ok = true,
+                        active_ms = active.started_at.elapsed().as_millis(),
+                        queued_ms = active.queued_for.as_millis(),
+                        active_remaining = self.active.len(),
+                        active_same_origin_remaining = origin
+                            .map(|origin| active_origin_count(&self.active, origin))
+                            .unwrap_or(0),
+                        pending_after = self.pending.len(),
+                        pending_same_origin_after = origin
+                            .map(|origin| pending_origin_count(&self.pending, origin))
+                            .unwrap_or(0),
+                        stage = "curl_runtime_job_done",
+                    );
+                }
+                Err(error) => {
+                    tracing::info!(
+                        target: "moli_cdp_nav_timing",
+                        transfer_id = %transfer_id,
+                        label = %active.label,
+                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
+                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
+                        origin_port = ?origin.and_then(|origin| origin.port),
+                        origin = ?active.origin,
+                        priority = active.priority,
+                        ok = false,
+                        error = %error,
+                        active_ms = active.started_at.elapsed().as_millis(),
+                        queued_ms = active.queued_for.as_millis(),
+                        active_remaining = self.active.len(),
+                        active_same_origin_remaining = origin
+                            .map(|origin| active_origin_count(&self.active, origin))
+                            .unwrap_or(0),
+                        pending_after = self.pending.len(),
+                        pending_same_origin_after = origin
+                            .map(|origin| pending_origin_count(&self.pending, origin))
+                            .unwrap_or(0),
+                        stage = "curl_runtime_job_done",
+                    );
+                }
+            }
+        }
+        let result = result.with_context(|| {
+            anyhow!(
+                "curl request failed for {} after active={}ms queued={}ms",
+                active.label,
+                active.started_at.elapsed().as_millis(),
+                active.queued_for.as_millis()
+            )
+        });
+        self.send_completion(CurlMultiCompletion {
+            transfer_id,
+            easy,
+            context: active.context,
+            result,
+        });
+    }
+
+    fn send_completion(&self, completion: CurlMultiCompletion<H, C>) {
+        let _ = self.completion_tx.send(completion);
+    }
+}
+
+fn curl_runtime_timeout_error(stage: &str) -> anyhow::Error {
+    anyhow!("curl multi runtime request timed out {stage}")
+}
+
+fn curl_timeout_for_deadline(deadline: Instant, now: Instant) -> Option<Duration> {
+    let remaining = deadline.saturating_duration_since(now);
+    // curl-rust converts CURLOPT_TIMEOUT_MS with `Duration::as_millis()`. A
+    // positive sub-millisecond value would therefore become zero, which
+    // libcurl interprets as disabling the timeout entirely.
+    (remaining >= Duration::from_millis(1)).then_some(remaining)
+}
+
+fn curl_runtime_trace_enabled() -> bool {
+    static ENABLED: OnceLock<bool> = OnceLock::new();
+    *ENABLED.get_or_init(|| {
+        env_flag_enabled("MOLI_CDP_NAV_TIMING") || env_flag_enabled("MOLI_CURL_RUNTIME_TRACE")
+    })
+}
+
+fn env_flag_enabled(name: &str) -> bool {
+    std::env::var(name).is_ok_and(|value| {
+        let value = value.trim();
+        !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false")
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{CurlDnsResolution, CurlOriginKey};
+    use curl::easy::Easy2;
+    #[derive(Debug)]
+    struct TestHandler;
+
+    impl Handler for TestHandler {}
+
+    fn test_job(
+        label: &str,
+        priority: u8,
+        origin: Option<CurlOriginKey>,
+    ) -> CurlMultiJob<TestHandler, String> {
+        CurlMultiJob {
+            easy: Easy2::new(TestHandler),
+            context: label.to_owned(),
+            origin,
+            deadline: None,
+            dns_resolution: CurlDnsResolution::curl_managed(),
+            priority,
+            label: label.to_owned(),
+        }
+    }
+
+    fn test_transfer_id(sequence: usize) -> CurlTransferId {
+        CurlTransferId::from_token(sequence).expect("test transfer ID is non-zero")
+    }
+
+    #[test]
+    fn active_transfer_wait_is_capped_by_the_earliest_queued_deadline() {
+        let now = Instant::now();
+        let later = now + Duration::from_secs(2);
+        let earlier = now + Duration::from_secs(1);
+        let mut later_job = test_job("later", 1, None);
+        later_job.deadline = Some(later);
+        let mut earlier_job = test_job("earlier", 1, None);
+        earlier_job.deadline = Some(earlier);
+        let mut pending = VecDeque::new();
+        enqueue_pending_job(&mut pending, test_transfer_id(1), later_job);
+        enqueue_pending_job(&mut pending, test_transfer_id(2), earlier_job);
+        let (completion_tx, _) = crossbeam_channel::unbounded();
+        let mut registry = HttpRegistry::new(CurlMultiRuntimeConfig::default(), completion_tx);
+        registry.pending = pending;
+
+        assert_eq!(registry.next_deadline(), Some(earlier));
+    }
+
+    #[test]
+    fn sub_millisecond_deadline_never_disables_the_libcurl_timeout() {
+        let now = Instant::now();
+        assert_eq!(
+            curl_timeout_for_deadline(now + Duration::from_micros(999), now),
+            None
+        );
+        assert_eq!(
+            curl_timeout_for_deadline(now + Duration::from_millis(1), now),
+            Some(Duration::from_millis(1))
+        );
+    }
+}

moli-curl/src/http/scheduling.rs

@@ -1,3 +1,5 @@
+//! Priority ordering, per-origin admission and ordered HTTP completions.
+
 use std::{
     collections::{HashMap, VecDeque},
     hash::Hash,
@@ -7,38 +9,8 @@ use std::{
 
 use curl::{easy::Handler, multi::Easy2Handle};
 
-use crate::dns_adapter::CurlDnsOwnerResidence;
-
 use super::{CurlMultiJob, CurlOriginKey, CurlTransferId};
 
-pub(super) struct CurlOwnerState<H: Handler, C> {
-    pub(super) closed: bool,
-    pub(super) pending: VecDeque<CurlPendingJob<H, C>>,
-    pub(super) dns: CurlDnsOwnerResidence<CurlTransferId, CurlPendingJob<H, C>>,
-    pub(super) active: HashMap<CurlTransferId, CurlActiveTransfer<H, C>>,
-}
-
-impl<H: Handler, C> Default for CurlOwnerState<H, C> {
-    fn default() -> Self {
-        Self {
-            closed: false,
-            pending: VecDeque::new(),
-            dns: CurlDnsOwnerResidence::default(),
-            active: HashMap::new(),
-        }
-    }
-}
-
-impl<H: Handler, C> CurlOwnerState<H, C> {
-    pub(super) fn next_waiting_deadline(&self) -> Option<Instant> {
-        self.pending
-            .iter()
-            .filter_map(|pending| pending.job.deadline)
-            .chain(self.dns.next_deadline(|pending| pending.job.deadline))
-            .min()
-    }
-}
-
 pub(super) struct CurlActiveTransfer<H: Handler, C> {
     pub(super) handle: Easy2Handle<H>,
     pub(super) context: C,
@@ -109,11 +81,11 @@ pub(super) fn enqueue_existing_pending_job<H: Handler, C>(
 
 pub(super) fn job_is_eligible<H: Handler, C>(
     origin: Option<&CurlOriginKey>,
-    state: &CurlOwnerState<H, C>,
+    active: &HashMap<CurlTransferId, CurlActiveTransfer<H, C>>,
     max_active_per_host: Option<NonZeroUsize>,
 ) -> bool {
     match (origin, max_active_per_host) {
-        (Some(origin), Some(limit)) => active_origin_count(&state.active, origin) < limit.get(),
+        (Some(origin), Some(limit)) => active_origin_count(active, origin) < limit.get(),
         _ => true,
     }
 }
@@ -272,28 +244,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn active_transfer_wait_is_capped_by_the_earliest_queued_deadline() {
-        let now = Instant::now();
-        let later = now + Duration::from_secs(2);
-        let earlier = now + Duration::from_secs(1);
-        let mut later_job = test_job("later", 1, None);
-        later_job.deadline = Some(later);
-        let mut earlier_job = test_job("earlier", 1, None);
-        earlier_job.deadline = Some(earlier);
-        let mut pending = VecDeque::new();
-        enqueue_pending_job(&mut pending, test_transfer_id(1), later_job);
-        enqueue_pending_job(&mut pending, test_transfer_id(2), earlier_job);
-        let state = CurlOwnerState {
-            closed: false,
-            pending,
-            dns: CurlDnsOwnerResidence::default(),
-            active: HashMap::new(),
-        };
-
-        assert_eq!(state.next_waiting_deadline(), Some(earlier));
-    }
-
     #[test]
     fn completed_jobs_preserve_libcurl_notification_order() {
         let mut active = HashMap::from([
@@ -354,15 +304,10 @@ mod tests {
             started_at: Instant::now(),
             queued_for: Duration::ZERO,
         };
-        let state = CurlOwnerState {
-            closed: false,
-            pending: VecDeque::new(),
-            dns: CurlDnsOwnerResidence::default(),
-            active: HashMap::from([(test_transfer_id(1), active)]),
-        };
+        let active = HashMap::from([(test_transfer_id(1), active)]);
         let cap = NonZeroUsize::new(1);
 
-        assert!(!job_is_eligible(Some(&capped_origin), &state, cap));
-        assert!(job_is_eligible(Some(&other_origin), &state, cap));
+        assert!(!job_is_eligible(Some(&capped_origin), &active, cap));
+        assert!(job_is_eligible(Some(&other_origin), &active, cap));
     }
 }

moli-curl/src/lib.rs

@@ -1,13 +1,12 @@
 //! Shared libcurl multi scheduler for Moli network requests.
 
 mod dns_adapter;
+mod http;
 mod runtime;
 mod tls;
 pub mod websocket;
 
 pub use dns_adapter::CurlDnsResolution;
-pub use runtime::{
-    CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig,
-    CurlOriginKey, CurlSubmitError, CurlTransferId,
-};
+pub use http::{CurlHttpSender, CurlMultiCompletion, CurlMultiJob, CurlOriginKey, CurlSubmitError};
+pub use runtime::{CurlMultiRuntime, CurlMultiRuntimeConfig, CurlTransferId};
 pub use tls::CurlTlsConfig;

moli-curl/src/runtime.rs

@@ -1,109 +1,36 @@
+//! Owns the common native thread; HTTP and WebSocket handles only submit work.
+
 mod config;
+pub(crate) mod diagnostics;
 pub(crate) mod identity;
 mod owner;
-mod residence;
+#[cfg(test)]
+mod tests;
 
+use crate::{CurlHttpSender, CurlMultiCompletion, CurlMultiJob, websocket::CurlWebSocketConnector};
+use anyhow::{Context, Result};
+use crossbeam_channel::Receiver;
+use curl::easy::Handler;
+use parking_lot::Mutex;
 use std::{
-    fmt,
     sync::{
         Arc,
         atomic::{AtomicBool, Ordering},
     },
     thread,
-    time::Instant,
 };
 
-use anyhow::{Context, Result, anyhow};
-use crossbeam_channel::{Receiver, Sender};
-use curl::{
-    easy::{Easy2, Handler},
-    multi::MultiWaker,
-};
-use parking_lot::Mutex;
-
-use crate::dns_adapter::CurlDnsResolution;
-use crate::websocket::CurlWebSocketConnector;
-
 pub use config::CurlMultiRuntimeConfig;
 pub use identity::CurlTransferId;
+use owner::CurlRuntimeOwner;
 
-use identity::next_transfer_id;
-use owner::{CurlRuntimeCommand, CurlRuntimeOwner};
-
-/// Origin key used by the curl scheduler for per-origin active transfer caps.
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct CurlOriginKey {
-    pub scheme: String,
-    pub host: String,
-    pub port: Option<u16>,
-}
-
-/// A configured curl transfer plus scheduler metadata.
-pub struct CurlMultiJob<H: Handler, C> {
-    pub easy: Easy2<H>,
-    pub context: C,
-    pub origin: Option<CurlOriginKey>,
-    /// Absolute deadline for the whole scheduler-owned transfer attempt.
-    ///
-    /// libcurl cannot account for time spent in Moli's priority queue or in
-    /// the shared DNS residence because both happen before the easy handle is
-    /// added to the multi handle. The owner enforces this deadline in those
-    /// residences and gives libcurl only the remaining duration.
-    pub deadline: Option<Instant>,
-    /// DNS ownership chosen by the caller before this transfer enters curl.
-    ///
-    /// A curl-managed policy preserves libcurl's resolver behavior. A shared
-    /// origin policy parks the transfer outside the curl multi handle set until
-    /// the bounded system resolver publishes an answer.
-    pub dns_resolution: CurlDnsResolution,
-    /// Higher values start before lower values when jobs are queued.
-    pub priority: u8,
-    pub label: String,
-}
-
-impl<H: Handler, C> fmt::Debug for CurlMultiJob<H, C> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("CurlMultiJob")
-            .field("origin", &self.origin)
-            .field("deadline", &self.deadline)
-            .field("dns_resolution", &self.dns_resolution)
-            .field("priority", &self.priority)
-            .field("label", &self.label)
-            .finish_non_exhaustive()
-    }
-}
-
-/// Completion emitted by `CurlMultiRuntime`.
-pub struct CurlMultiCompletion<H: Handler, C> {
-    pub transfer_id: CurlTransferId,
-    pub easy: Option<Easy2<H>>,
-    pub context: C,
-    pub result: Result<()>,
-}
-
-impl<H: Handler, C> fmt::Debug for CurlMultiCompletion<H, C> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("CurlMultiCompletion")
-            .field("transfer_id", &self.transfer_id)
-            .field("has_easy", &self.easy.is_some())
-            .field("result", &self.result.as_ref().map(|_| ()))
-            .finish_non_exhaustive()
-    }
-}
-
-/// Error returned when a job cannot be submitted and is returned to the caller.
-pub struct CurlSubmitError<H: Handler, C> {
-    pub job: CurlMultiJob<H, C>,
-    pub error: anyhow::Error,
-}
-
-impl<H: Handler, C> fmt::Debug for CurlSubmitError<H, C> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("CurlSubmitError")
-            .field("job", &self.job)
-            .field("error", &self.error)
-            .finish()
-    }
+#[derive(Debug)]
+pub(crate) enum CurlRuntimeCommand<H: Handler, C> {
+    Request {
+        transfer_id: CurlTransferId,
+        job: CurlMultiJob<H, C>,
+    },
+    Shutdown,
 }
 
 /// Owns one native thread. Request handles cannot extend its lifetime or join it.
@@ -116,24 +43,6 @@ pub struct CurlMultiRuntime<H: Handler + Send + 'static, C: Send + 'static> {
     owner_handle: Mutex<Option<thread::JoinHandle<()>>>,
 }
 
-/// Submits HTTP work to one owner. Clones carry no shutdown or join authority.
-#[derive(Debug)]
-pub struct CurlHttpSender<H: Handler + Send + 'static, C: Send + 'static> {
-    command_tx: Sender<CurlRuntimeCommand<H, C>>,
-    owner_waker: MultiWaker,
-    shutdown_requested: Arc<AtomicBool>,
-}
-
-impl<H: Handler + Send + 'static, C: Send + 'static> Clone for CurlHttpSender<H, C> {
-    fn clone(&self) -> Self {
-        Self {
-            command_tx: self.command_tx.clone(),
-            owner_waker: self.owner_waker.clone(),
-            shutdown_requested: self.shutdown_requested.clone(),
-        }
-    }
-}
-
 impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
     pub fn new(
         config: CurlMultiRuntimeConfig,
@@ -146,20 +55,24 @@ impl<H: Handler + Send + 'static, C: Send + 'static> CurlMultiRuntime<H, C> {
         let shutdown_requested = Arc::new(AtomicBool::new(false));
         #[cfg(test)]
         let owner_started = Arc::new(AtomicBool::new(false));
-        let owner = CurlRuntimeOwner::new(
-            config,
-            command_rx,
-            completion_tx,
-            waker_tx,
-            Arc::clone(&shutdown_requested),
-            websocket_rx,
-            #[cfg(test)]
-            Arc::clone(&owner_started),
-        );
-        let thread_name = owner.thread_name().to_owned();
+        let thread_name = config.thread_name.clone();
+        let owner_shutdown = shutdown_requested.clone();
+        #[cfg(test)]
+        let started = owner_started.clone();
         let owner_handle = thread::Builder::new()
             .name(thread_name)
-            .spawn(move || owner.run())
+            .spawn(move || {
+                CurlRuntimeOwner::run(
+                    config,
+                    command_rx,
+                    completion_tx,
+                    waker_tx,
+                    owner_shutdown,
+                    websocket_rx,
+                    #[cfg(test)]
+                    started,
+                )
+            })
             .context("failed to spawn curl multi runtime owner thread")?;
         let owner_waker = waker_rx
             .recv()
@@ -213,144 +126,3 @@ impl<H: Handler + Send + 'static, C: Send + 'static> Drop for CurlMultiRuntime<H
         self.shutdown();
     }
 }
-
-impl<H: Handler + Send + 'static, C: Send + 'static> CurlHttpSender<H, C> {
-    pub fn submit(
-        &self,
-        job: CurlMultiJob<H, C>,
-    ) -> std::result::Result<CurlTransferId, CurlSubmitError<H, C>> {
-        if self.shutdown_requested.load(Ordering::SeqCst) {
-            return Err(CurlSubmitError {
-                job,
-                error: anyhow!("curl multi runtime is shutting down"),
-            });
-        }
-        let transfer_id = match next_transfer_id() {
-            Ok(transfer_id) => transfer_id,
-            Err(error) => return Err(CurlSubmitError { job, error }),
-        };
-        match self
-            .command_tx
-            .send(CurlRuntimeCommand::Request { transfer_id, job })
-        {
-            Ok(()) => {
-                let _ = self.owner_waker.wakeup();
-                Ok(transfer_id)
-            }
-            Err(error) => {
-                let CurlRuntimeCommand::Request { job, .. } = error.into_inner() else {
-                    unreachable!("submit only sends request commands");
-                };
-                Err(CurlSubmitError {
-                    job,
-                    error: anyhow!("curl multi runtime is shutting down"),
-                })
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::{
-        io::{Read, Write},
-        net::TcpListener,
-        time::Duration,
-    };
-
-    use super::*;
-
-    #[derive(Debug)]
-    struct TestHandler;
-
-    impl Handler for TestHandler {}
-
-    #[test]
-    fn submitted_identity_reaches_the_matching_runtime_completion() {
-        let listener = TcpListener::bind(("127.0.0.1", 0))
-            .expect("test HTTP listener should bind to a local port");
-        let address = listener
-            .local_addr()
-            .expect("test HTTP listener should have an address");
-        let server = thread::spawn(move || {
-            let (mut stream, _) = listener
-                .accept()
-                .expect("curl should connect to the test HTTP listener");
-            stream
-                .set_read_timeout(Some(Duration::from_secs(5)))
-                .expect("test HTTP connection should accept a read timeout");
-            let mut request = [0; 4096];
-            let _ = stream
-                .read(&mut request)
-                .expect("test HTTP request should be readable");
-            stream
-                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
-                .expect("test HTTP response should be writable");
-        });
-
-        let (runtime, completion_rx) = CurlMultiRuntime::new(CurlMultiRuntimeConfig {
-            poll_interval: Duration::from_millis(5),
-            ..CurlMultiRuntimeConfig::default()
-        })
-        .expect("test curl runtime should start");
-        let mut easy = Easy2::new(TestHandler);
-        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(),
-                origin: None,
-                deadline: None,
-                dns_resolution: CurlDnsResolution::curl_managed(),
-                priority: 1,
-                label: "identity-test".to_owned(),
-            })
-            .expect("test curl transfer should be accepted");
-
-        let completion = completion_rx
-            .recv_timeout(Duration::from_secs(5))
-            .expect("test curl transfer should reach terminal completion");
-        assert_eq!(completion.transfer_id, transfer_id);
-        assert_eq!(completion.context, "matching-context");
-        assert!(completion.easy.is_some());
-        completion
-            .result
-            .expect("test curl transfer should complete successfully");
-
-        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::<TestHandler, Vec<u8>>::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)
-        ));
-    }
-}

moli-curl/src/runtime/diagnostics.rs

@@ -3,7 +3,7 @@
 use std::time::{Duration, Instant};
 
 #[derive(Default, Debug)]
-pub(super) struct Counters {
+pub(crate) struct Counters {
     pub turns: u64,
     pub progressed_turns: u64,
     pub read_bytes: u64,
@@ -47,10 +47,10 @@ struct Window {
     counters: Counters,
 }
 
-pub(super) struct Diagnostics(Option<Window>);
+pub(crate) struct Diagnostics(Option<Window>);
 
 impl Diagnostics {
-    pub(super) fn from_env() -> Self {
+    pub(crate) fn from_env() -> Self {
         Self::new(
             std::env::var("MOLI_CURL_WEBSOCKET_DIAGNOSTICS").is_ok_and(|value| {
                 let value = value.trim();
@@ -66,15 +66,15 @@ impl Diagnostics {
         }))
     }
 
-    pub(super) fn counters(&mut self) -> Option<&mut Counters> {
+    pub(crate) fn counters(&mut self) -> Option<&mut Counters> {
         self.0.as_mut().map(|window| &mut window.counters)
     }
 
-    pub(super) fn poll_start(&self) -> Option<Instant> {
+    pub(crate) fn poll_start(&self) -> Option<Instant> {
         self.0.as_ref().map(|_| Instant::now())
     }
 
-    pub(super) fn polled(&mut self, start: Option<Instant>, timeout: Duration, progressed: bool) {
+    pub(crate) fn polled(&mut self, start: Option<Instant>, timeout: Duration, progressed: bool) {
         if let Some(start) = start {
             self.0
                 .as_mut()
@@ -84,7 +84,7 @@ impl Diagnostics {
         }
     }
 
-    pub(super) fn report(&mut self, sessions: usize, finished: bool) {
+    pub(crate) fn report(&mut self, sessions: usize, finished: bool) {
         let Some(window) = &mut self.0 else {
             return;
         };

moli-curl/src/runtime/owner.rs

@@ -1,62 +1,55 @@
+//! One native owner drives both protocols. Registries own their easy handles;
+//! only this loop performs curl work, drains CURLMSG_DONE and waits for readiness.
+
 use std::{
-    num::NonZeroUsize,
     sync::{
-        Arc, OnceLock,
+        Arc,
         atomic::{AtomicBool, Ordering},
     },
     time::{Duration, Instant},
 };
 
-use anyhow::{Context, Result, anyhow};
-use crossbeam_channel::{self, Receiver, Sender};
+use crossbeam_channel::{Receiver, Sender};
 use curl::{
     easy::Handler,
     multi::{Multi, MultiWaker},
 };
 use tracing::debug;
 
-use crate::dns_adapter::{CurlDnsOwnerCompletion, CurlDnsReady};
-use crate::websocket::{Submission, registry::WebSocketRegistry};
-
 use super::{
-    CurlMultiCompletion, CurlMultiJob, CurlMultiRuntimeConfig, CurlTransferId,
+    CurlMultiRuntimeConfig, CurlRuntimeCommand,
     config::{make_runtime_multi, runtime_wait_timeout},
-    residence::{
-        CurlActiveTransfer, CurlOwnerState, CurlPendingJob, active_origin_count,
-        enqueue_existing_pending_job, enqueue_pending_job, job_is_eligible, pending_origin_count,
-        take_expired_pending_jobs, take_transfers_in_notification_order,
-    },
+    diagnostics::Diagnostics,
+};
+use crate::{
+    CurlMultiCompletion, CurlTransferId,
+    dns_adapter::CurlDnsOwnerCompletion,
+    http::registry::HttpRegistry,
+    websocket::{Submission, registry::WebSocketRegistry},
 };
 
-#[derive(Debug)]
-pub(super) enum CurlRuntimeCommand<H: Handler, C> {
-    Request {
-        transfer_id: CurlTransferId,
-        job: CurlMultiJob<H, C>,
-    },
-    Shutdown,
-}
-
 enum CurlOwnerEvent<H: Handler, C> {
-    Command(std::result::Result<CurlRuntimeCommand<H, C>, crossbeam_channel::RecvError>),
-    Dns(std::result::Result<CurlDnsOwnerCompletion<CurlTransferId>, crossbeam_channel::RecvError>),
-    WebSocket(std::result::Result<Submission, crossbeam_channel::RecvError>),
+    Command(Result<CurlRuntimeCommand<H, C>, crossbeam_channel::RecvError>),
+    Dns(Result<CurlDnsOwnerCompletion<CurlTransferId>, crossbeam_channel::RecvError>),
+    WebSocket(Result<Submission, crossbeam_channel::RecvError>),
     Deadline,
 }
 
-pub(super) struct CurlRuntimeOwner<H: Handler + Send + 'static, C: Send + 'static> {
-    config: CurlMultiRuntimeConfig,
+pub(super) struct CurlRuntimeOwner<H: Handler, C> {
     command_rx: Receiver<CurlRuntimeCommand<H, C>>,
-    completion_tx: Sender<CurlMultiCompletion<H, C>>,
-    waker_tx: Sender<MultiWaker>,
     shutdown_requested: Arc<AtomicBool>,
-    websocket_rx: Receiver<Submission>,
-    #[cfg(test)]
-    owner_started: Arc<AtomicBool>,
+    closed: bool,
+    poll_interval: Duration,
+    diagnostics: Diagnostics,
+    http: HttpRegistry<H, C>,
+    websockets: WebSocketRegistry,
+    // Drop the registries' easy handles before their Multi, including on unwind.
+    multi: Multi,
 }
 
-impl<H: Handler + Send + 'static, C: Send + 'static> CurlRuntimeOwner<H, C> {
-    pub(super) fn new(
+impl<H: Handler, C> CurlRuntimeOwner<H, C> {
+    /// Construct all native state on its owner thread, including the WS cache.
+    pub(super) fn run(
         config: CurlMultiRuntimeConfig,
         command_rx: Receiver<CurlRuntimeCommand<H, C>>,
         completion_tx: Sender<CurlMultiCompletion<H, C>>,
@@ -64,640 +57,176 @@ impl<H: Handler + Send + 'static, C: Send + 'static> CurlRuntimeOwner<H, C> {
         shutdown_requested: Arc<AtomicBool>,
         websocket_rx: Receiver<Submission>,
         #[cfg(test)] owner_started: Arc<AtomicBool>,
-    ) -> Self {
-        Self {
-            config,
-            command_rx,
-            completion_tx,
-            waker_tx,
-            shutdown_requested,
-            websocket_rx,
-            #[cfg(test)]
-            owner_started,
-        }
-    }
-
-    pub(super) fn thread_name(&self) -> &str {
-        &self.config.thread_name
-    }
-
-    pub(super) fn run(self) {
+    ) {
         #[cfg(test)]
-        self.owner_started.store(true, Ordering::SeqCst);
-        let mut multi = make_runtime_multi(&self.config);
-        let _ = self.waker_tx.send(multi.waker());
-        let mut state = CurlOwnerState::default();
-        let mut websockets = WebSocketRegistry::new(self.websocket_rx.clone());
+        owner_started.store(true, Ordering::SeqCst);
+        let multi = make_runtime_multi(&config);
+        let _ = waker_tx.send(multi.waker());
+        let mut owner = Self {
+            command_rx,
+            shutdown_requested,
+            closed: false,
+            poll_interval: config.poll_interval,
+            diagnostics: Diagnostics::from_env(),
+            http: HttpRegistry::new(config, completion_tx),
+            websockets: WebSocketRegistry::new(websocket_rx),
+            multi,
+        };
+        owner.drive();
+    }
 
+    fn drive(&mut self) {
         loop {
-            self.drain_commands(&mut state, &mut multi);
-            if state.closed {
-                websockets.shutdown(&mut multi);
+            self.drain_commands();
+            self.http.advance(&mut self.multi);
+            self.process_completed_transfers();
+            let progressed = self
+                .websockets
+                .advance(&mut self.multi, &mut self.diagnostics);
+            if let Some(counters) = self.diagnostics.counters() {
+                counters.turns += 1;
+                counters.progressed_turns += u64::from(progressed);
             }
-            self.drain_dns_completions(&mut state);
-            self.expire_waiting_jobs(&mut state);
-            self.start_eligible_jobs(&mut state, &mut multi);
-            self.process_completed_transfers(&mut state, &mut multi, &mut websockets);
-            let progressed = websockets.advance(&mut multi);
 
-            if state.closed
-                && state.pending.is_empty()
-                && state.dns.is_empty()
-                && state.active.is_empty()
-                && self.command_rx.is_empty()
-            {
+            // close() retires both registries. Drain racing HTTP submissions so
+            // every accepted job still receives its terminal completion.
+            if self.closed && self.command_rx.is_empty() {
+                self.diagnostics.report(0, true);
                 return;
             }
 
             let runnable = progressed || !self.command_rx.is_empty();
-            if !runnable
-                && state.active.is_empty()
-                && state.pending.is_empty()
-                && websockets.is_empty()
-            {
-                self.wait_for_next_owner_event(&mut state, &mut multi, &mut websockets);
+            if !runnable && !self.http.has_curl_work() && self.websockets.is_empty() {
+                self.wait_for_next_owner_event();
             } else {
-                self.wait_for_curl_progress(&mut multi, &state, &mut websockets, runnable);
+                self.wait_for_curl_progress(runnable);
             }
         }
     }
 
-    fn drain_commands(&self, state: &mut CurlOwnerState<H, C>, multi: &mut Multi) {
+    fn drain_commands(&mut self) {
         // Continuous HTTP submissions must not starve native I/O or WebSockets.
         for _ in 0..256 {
             match self.command_rx.try_recv() {
-                Ok(command) => self.handle_command(state, multi, command),
+                Ok(command) => self.handle_command(command),
                 Err(crossbeam_channel::TryRecvError::Empty) => break,
                 Err(crossbeam_channel::TryRecvError::Disconnected) => {
-                    self.close(state, multi);
+                    self.close();
                     break;
                 }
             }
         }
     }
 
-    fn wait_for_next_owner_event(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        multi: &mut Multi,
-        websockets: &mut WebSocketRegistry,
-    ) {
-        if state.closed {
+    fn handle_command(&mut self, command: CurlRuntimeCommand<H, C>) {
+        match command {
+            CurlRuntimeCommand::Request { transfer_id, job } => self.http.admit(transfer_id, job),
+            CurlRuntimeCommand::Shutdown => self.close(),
+        }
+    }
+
+    fn close(&mut self) {
+        if self.closed {
             return;
         }
-        let event = if let Some(deadline) = state.dns.next_deadline(|pending| pending.job.deadline)
-        {
+        self.closed = true;
+        self.shutdown_requested.store(true, Ordering::SeqCst);
+        self.http.shutdown(&mut self.multi);
+        self.websockets.shutdown(&mut self.multi);
+    }
+
+    fn wait_for_next_owner_event(&mut self) {
+        if self.closed {
+            return;
+        }
+        let event = if let Some(deadline) = self.http.next_deadline() {
             let deadline_rx =
                 crossbeam_channel::after(deadline.saturating_duration_since(Instant::now()));
             crossbeam_channel::select! {
                 recv(self.command_rx) -> command => CurlOwnerEvent::Command(command),
-                recv(websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission),
-                recv(state.dns.completion_receiver()) -> completion => CurlOwnerEvent::Dns(completion),
+                recv(self.websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission),
+                recv(self.http.dns_completions()) -> completion => CurlOwnerEvent::Dns(completion),
                 recv(deadline_rx) -> _ => CurlOwnerEvent::Deadline,
             }
         } else {
             crossbeam_channel::select! {
                 recv(self.command_rx) -> command => CurlOwnerEvent::Command(command),
-                recv(websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission),
-                recv(state.dns.completion_receiver()) -> completion => CurlOwnerEvent::Dns(completion),
+                recv(self.websockets.submissions()) -> submission => CurlOwnerEvent::WebSocket(submission),
+                recv(self.http.dns_completions()) -> completion => CurlOwnerEvent::Dns(completion),
             }
         };
         match event {
-            CurlOwnerEvent::WebSocket(Ok(submission)) => websockets.admit(submission, multi),
-            CurlOwnerEvent::WebSocket(Err(_)) => self.close(state, multi),
-            CurlOwnerEvent::Command(command) => match command {
-                Ok(command) => self.handle_command(state, multi, command),
-                Err(_) => self.close(state, multi),
-            },
-            CurlOwnerEvent::Dns(completion) => {
-                if let Ok(completion) = completion {
-                    self.claim_dns_completion(state, completion);
-                }
+            CurlOwnerEvent::Command(Ok(command)) => self.handle_command(command),
+            CurlOwnerEvent::WebSocket(Ok(submission)) => {
+                self.websockets.admit(submission, &mut self.multi)
             }
-            CurlOwnerEvent::Deadline => {}
+            CurlOwnerEvent::Command(Err(_)) | CurlOwnerEvent::WebSocket(Err(_)) => self.close(),
+            CurlOwnerEvent::Dns(Ok(completion)) => self.http.claim_dns_completion(completion),
+            CurlOwnerEvent::Dns(Err(_)) | CurlOwnerEvent::Deadline => {}
         }
     }
 
-    fn handle_command(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        multi: &mut Multi,
-        command: CurlRuntimeCommand<H, C>,
-    ) {
-        match command {
-            CurlRuntimeCommand::Request { transfer_id, job } if state.closed => {
-                self.send_completion(CurlMultiCompletion {
-                    transfer_id,
-                    easy: Some(job.easy),
-                    context: job.context,
-                    result: Err(anyhow!("curl multi runtime is shutting down")),
-                });
-            }
-            CurlRuntimeCommand::Request { transfer_id, job } => {
-                self.admit_job(state, transfer_id, job)
-            }
-            CurlRuntimeCommand::Shutdown => self.close(state, multi),
-        }
-    }
-
-    fn admit_job(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        transfer_id: CurlTransferId,
-        job: CurlMultiJob<H, C>,
-    ) {
-        if curl_runtime_trace_enabled() {
-            let origin = job.origin.as_ref();
-            tracing::info!(
-                target: "moli_cdp_nav_timing",
-                transfer_id = %transfer_id,
-                label = %job.label,
-                origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
-                origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
-                origin_port = ?origin.and_then(|origin| origin.port),
-                origin = ?job.origin,
-                priority = job.priority,
-                pending_before = state.pending.len(),
-                pending_same_origin_before = origin
-                    .map(|origin| pending_origin_count(&state.pending, origin))
-                    .unwrap_or(0),
-                stage = "curl_runtime_job_queued",
-            );
-        }
-        enqueue_pending_job(&mut state.pending, transfer_id, job);
-    }
-
-    fn close(&self, state: &mut CurlOwnerState<H, C>, multi: &mut Multi) {
-        if state.closed {
-            return;
-        }
-        state.closed = true;
-        self.shutdown_requested.store(true, Ordering::SeqCst);
-        while let Some(pending) = state.pending.pop_front() {
-            let CurlPendingJob {
-                transfer_id, job, ..
-            } = pending;
-            self.send_completion(CurlMultiCompletion {
-                transfer_id,
-                easy: Some(job.easy),
-                context: job.context,
-                result: Err(anyhow!("curl multi runtime is shutting down")),
-            });
-        }
-        for pending in state.dns.drain() {
-            let CurlPendingJob {
-                transfer_id, job, ..
-            } = pending;
-            self.send_completion(CurlMultiCompletion {
-                transfer_id,
-                easy: Some(job.easy),
-                context: job.context,
-                result: Err(anyhow!(
-                    "curl multi runtime DNS request cancelled during shutdown"
-                )),
-            });
-        }
-        for (transfer_id, active) in state.active.drain() {
-            let easy = multi.remove2(active.handle).ok();
-            self.send_completion(CurlMultiCompletion {
-                transfer_id,
-                easy,
-                context: active.context,
-                result: Err(anyhow!(
-                    "curl multi runtime request cancelled during shutdown"
-                )),
-            });
-        }
-    }
-
-    fn start_eligible_jobs(&self, state: &mut CurlOwnerState<H, C>, multi: &mut Multi) {
-        loop {
-            if state.closed || state.active.len() >= self.config.max_active.get() {
-                return;
-            }
-            let Some(index) = state.pending.iter().position(|pending| {
-                job_is_eligible(
-                    pending.job.origin.as_ref(),
-                    state,
-                    self.config.max_host_active,
-                )
-            }) else {
-                return;
-            };
-            let pending = state
-                .pending
-                .remove(index)
-                .expect("pending curl job index should exist");
-            let dns_target = pending.job.dns_resolution.target().cloned();
-            match dns_target {
-                Some(target) => {
-                    state
-                        .dns
-                        .start(pending.transfer_id, pending, target, multi.waker())
-                }
-                None => self.start_job(state, multi, pending),
-            }
-        }
-    }
-
-    fn drain_dns_completions(&self, state: &mut CurlOwnerState<H, C>) {
-        while let Some(ready) = state.dns.try_claim_next() {
-            self.handle_dns_completion(state, ready);
-        }
-    }
-
-    fn claim_dns_completion(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        completion: CurlDnsOwnerCompletion<CurlTransferId>,
-    ) {
-        let Some(ready) = state.dns.claim(completion) else {
-            return;
-        };
-        self.handle_dns_completion(state, ready);
-    }
-
-    fn handle_dns_completion(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        ready: CurlDnsReady<CurlPendingJob<H, C>>,
-    ) {
-        let mut pending = ready.pending;
-        if state.closed {
-            let CurlPendingJob {
-                transfer_id, job, ..
-            } = pending;
-            self.send_completion(CurlMultiCompletion {
-                transfer_id,
-                easy: Some(job.easy),
-                context: job.context,
-                result: Err(anyhow!("curl multi runtime is shutting down")),
-            });
-            return;
-        }
-        if pending.deadline_reached(Instant::now()) {
-            self.complete_timed_out_job(pending, "while waiting for DNS");
-            return;
-        }
-        match ready.result {
-            Ok(addresses) => {
-                if let Err(error) = pending
-                    .job
-                    .dns_resolution
-                    .install(&mut pending.job.easy, addresses.as_ref())
-                {
-                    let CurlPendingJob {
-                        transfer_id, job, ..
-                    } = pending;
-                    self.send_completion(CurlMultiCompletion {
-                        transfer_id,
-                        easy: Some(job.easy),
-                        context: job.context,
-                        result: Err(error),
-                    });
-                    return;
-                }
-                enqueue_existing_pending_job(&mut state.pending, pending);
-            }
-            Err(error) => {
-                let CurlPendingJob {
-                    transfer_id, job, ..
-                } = pending;
-                self.send_completion(CurlMultiCompletion {
-                    transfer_id,
-                    easy: Some(job.easy),
-                    context: job.context,
-                    result: Err(anyhow!(error.to_string())),
-                });
-            }
-        }
-    }
-
-    fn start_job(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        multi: &mut Multi,
-        pending: CurlPendingJob<H, C>,
-    ) {
-        if pending.deadline_reached(Instant::now()) {
-            self.complete_timed_out_job(pending, "while waiting to start");
-            return;
-        }
-        let transfer_id = pending.transfer_id;
-        let queued_for = pending.enqueued_at.elapsed();
-        let mut job = pending.job;
-        let label = job.label.clone();
-        if let Some(deadline) = job.deadline {
-            let Some(remaining) = curl_timeout_for_deadline(deadline, Instant::now()) else {
-                self.send_completion(CurlMultiCompletion {
-                    transfer_id,
-                    easy: Some(job.easy),
-                    context: job.context,
-                    result: Err(curl_runtime_timeout_error("while waiting to start")),
-                });
-                return;
-            };
-            if let Err(error) = job.easy.timeout(remaining) {
-                self.send_completion(CurlMultiCompletion {
-                    transfer_id,
-                    easy: Some(job.easy),
-                    context: job.context,
-                    result: Err(error).context("failed to apply remaining curl request deadline"),
-                });
-                return;
-            }
-        }
-        match multi
-            .add2(job.easy)
-            .with_context(|| anyhow!("failed to add curl easy handle for {label}"))
-        {
-            Ok(mut handle) => {
-                handle
-                    .set_token(transfer_id.token())
-                    .expect("active curl handle must accept its transfer token");
-                if curl_runtime_trace_enabled() {
-                    let origin = job.origin.as_ref();
-                    tracing::info!(
-                        target: "moli_cdp_nav_timing",
-                        transfer_id = %transfer_id,
-                        label = %label,
-                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
-                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
-                        origin_port = ?origin.and_then(|origin| origin.port),
-                        origin = ?job.origin,
-                        priority = job.priority,
-                        queued_ms = queued_for.as_millis(),
-                        active_before = state.active.len(),
-                        active_same_origin_before = origin
-                            .map(|origin| active_origin_count(&state.active, origin))
-                            .unwrap_or(0),
-                        pending_after = state.pending.len(),
-                        pending_same_origin_after = origin
-                            .map(|origin| pending_origin_count(&state.pending, origin))
-                            .unwrap_or(0),
-                        max_active = self.config.max_active.get(),
-                        max_host_active = ?self.config.max_host_active.map(NonZeroUsize::get),
-                        max_host_connections = ?self.config.max_host_connections.map(NonZeroUsize::get),
-                        max_total_connections = ?self.config.max_total_connections.map(NonZeroUsize::get),
-                        max_concurrent_streams = ?self.config.max_concurrent_streams.map(NonZeroUsize::get),
-                        multiplex = self.config.multiplex,
-                        stage = "curl_runtime_job_start",
-                    );
-                }
-                let previous = state.active.insert(
-                    transfer_id,
-                    CurlActiveTransfer {
-                        handle,
-                        context: job.context,
-                        origin: job.origin,
-                        priority: job.priority,
-                        label,
-                        started_at: Instant::now(),
-                        queued_for,
-                    },
-                );
-                assert!(previous.is_none(), "curl transfer identity is unique");
-            }
-            Err(error) => self.send_completion(CurlMultiCompletion {
-                transfer_id,
-                easy: None,
-                context: job.context,
-                result: Err(error),
-            }),
-        }
-    }
-
-    fn process_completed_transfers(
-        &self,
-        state: &mut CurlOwnerState<H, C>,
-        multi: &mut Multi,
-        websockets: &mut WebSocketRegistry,
-    ) {
-        if let Err(error) = multi.perform() {
+    fn process_completed_transfers(&mut self) {
+        if let Err(error) = self.multi.perform() {
             debug!("curl multi runtime perform failed: {error}");
-            websockets.fail_sessions(multi, &error.to_string());
+            self.websockets
+                .fail_sessions(&mut self.multi, &error.to_string());
         }
         // Drain CURLMSG_DONE once. A WebSocket DONE starts its open residence;
         // HTTP DONE removes a finished transfer, preserving notification order.
         let mut completed = Vec::new();
-        multi.messages(|message| {
+        self.multi.messages(|message| {
             let Some(id) = message.token().ok().and_then(CurlTransferId::from_token) else {
                 return;
             };
-            // Keep the easy handle's error buffer as well as the CURLcode.
-            let result = match state.active.get(&id) {
-                Some(transfer) => message.result_for2(&transfer.handle),
-                None => websockets.handshake_result(id, &message),
-            };
+            // Both registries use result_for2 to retain the easy error buffer.
+            let result = self
+                .http
+                .completion_result(id, &message)
+                .or_else(|| self.websockets.handshake_result(id, &message));
             if let Some(result) = result {
                 completed.push((id, result));
             }
         });
         for (id, result) in &completed {
-            if !state.active.contains_key(id) {
-                websockets.complete_handshake(*id, result.clone(), multi);
+            if !self.http.contains(*id) {
+                self.websockets
+                    .complete_handshake(*id, result.clone(), &mut self.multi);
             }
         }
-        for (transfer_id, active, result) in
-            take_transfers_in_notification_order(&mut state.active, completed)
-        {
-            self.finish_active_transfer(
-                state,
-                multi,
-                transfer_id,
-                active,
-                result.map_err(Into::into),
-            );
-        }
+        self.http.complete(&mut self.multi, completed);
     }
 
-    fn expire_waiting_jobs(&self, state: &mut CurlOwnerState<H, C>) {
-        let now = Instant::now();
-        for pending in take_expired_pending_jobs(&mut state.pending, now) {
-            self.complete_timed_out_job(pending, "while waiting in the scheduler");
-        }
-        for pending in state.dns.take_expired(now, |pending| pending.job.deadline) {
-            self.complete_timed_out_job(pending, "while waiting for DNS");
-        }
-    }
-
-    fn complete_timed_out_job(&self, pending: CurlPendingJob<H, C>, stage: &'static str) {
-        let CurlPendingJob {
-            transfer_id, job, ..
-        } = pending;
-        self.send_completion(CurlMultiCompletion {
-            transfer_id,
-            easy: Some(job.easy),
-            context: job.context,
-            result: Err(curl_runtime_timeout_error(stage)),
-        });
-    }
-
-    fn finish_active_transfer(
-        &self,
-        state: &CurlOwnerState<H, C>,
-        multi: &mut Multi,
-        transfer_id: CurlTransferId,
-        active: CurlActiveTransfer<H, C>,
-        result: Result<()>,
-    ) {
-        let easy = match multi.remove2(active.handle) {
-            Ok(easy) => Some(easy),
-            Err(error) => {
-                self.send_completion(CurlMultiCompletion {
-                    transfer_id,
-                    easy: None,
-                    context: active.context,
-                    result: Err(anyhow!(
-                        "failed to remove curl easy handle for {}: {error}",
-                        active.label
-                    )),
-                });
-                return;
-            }
-        };
-        if curl_runtime_trace_enabled() {
-            let origin = active.origin.as_ref();
-            match &result {
-                Ok(()) => {
-                    tracing::info!(
-                        target: "moli_cdp_nav_timing",
-                        transfer_id = %transfer_id,
-                        label = %active.label,
-                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
-                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
-                        origin_port = ?origin.and_then(|origin| origin.port),
-                        origin = ?active.origin,
-                        priority = active.priority,
-                        ok = true,
-                        active_ms = active.started_at.elapsed().as_millis(),
-                        queued_ms = active.queued_for.as_millis(),
-                        active_remaining = state.active.len(),
-                        active_same_origin_remaining = origin
-                            .map(|origin| active_origin_count(&state.active, origin))
-                            .unwrap_or(0),
-                        pending_after = state.pending.len(),
-                        pending_same_origin_after = origin
-                            .map(|origin| pending_origin_count(&state.pending, origin))
-                            .unwrap_or(0),
-                        stage = "curl_runtime_job_done",
-                    );
-                }
-                Err(error) => {
-                    tracing::info!(
-                        target: "moli_cdp_nav_timing",
-                        transfer_id = %transfer_id,
-                        label = %active.label,
-                        origin_scheme = origin.map(|origin| origin.scheme.as_str()).unwrap_or(""),
-                        origin_host = origin.map(|origin| origin.host.as_str()).unwrap_or(""),
-                        origin_port = ?origin.and_then(|origin| origin.port),
-                        origin = ?active.origin,
-                        priority = active.priority,
-                        ok = false,
-                        error = %error,
-                        active_ms = active.started_at.elapsed().as_millis(),
-                        queued_ms = active.queued_for.as_millis(),
-                        active_remaining = state.active.len(),
-                        active_same_origin_remaining = origin
-                            .map(|origin| active_origin_count(&state.active, origin))
-                            .unwrap_or(0),
-                        pending_after = state.pending.len(),
-                        pending_same_origin_after = origin
-                            .map(|origin| pending_origin_count(&state.pending, origin))
-                            .unwrap_or(0),
-                        stage = "curl_runtime_job_done",
-                    );
-                }
-            }
-        }
-        let result = result.with_context(|| {
-            anyhow!(
-                "curl request failed for {} after active={}ms queued={}ms",
-                active.label,
-                active.started_at.elapsed().as_millis(),
-                active.queued_for.as_millis()
-            )
-        });
-        self.send_completion(CurlMultiCompletion {
-            transfer_id,
-            easy,
-            context: active.context,
-            result,
-        });
-    }
-
-    fn wait_for_curl_progress(
-        &self,
-        multi: &mut Multi,
-        state: &CurlOwnerState<H, C>,
-        websockets: &mut WebSocketRegistry,
-        progressed: bool,
-    ) {
+    fn wait_for_curl_progress(&mut self, runnable: bool) {
         // HTTP cancellation uses its configured progress interval. Idle WS
         // sessions use socket/waker readiness and need not inherit that cadence.
-        let interval = if state.active.is_empty() {
-            Duration::from_secs(1)
+        let interval = if self.http.has_active() {
+            self.poll_interval
         } else {
-            self.config.poll_interval
+            Duration::from_secs(1)
         };
-        let mut wait_timeout = runtime_wait_timeout(multi, interval).unwrap_or(interval);
-        if let Some(deadline) = state.next_waiting_deadline() {
-            wait_timeout = wait_timeout.min(deadline.saturating_duration_since(Instant::now()));
+        let mut timeout = runtime_wait_timeout(&self.multi, interval).unwrap_or(interval);
+        for deadline in [self.http.next_deadline(), self.websockets.next_deadline()]
+            .into_iter()
+            .flatten()
+        {
+            timeout = timeout.min(deadline.saturating_duration_since(Instant::now()));
         }
-        if let Some(deadline) = websockets.next_deadline() {
-            wait_timeout = wait_timeout.min(deadline.saturating_duration_since(Instant::now()));
+        if runnable {
+            timeout = Duration::ZERO;
         }
-        if progressed {
-            wait_timeout = Duration::ZERO;
+        // libcurl adds HTTP/handshake sockets and its waker to these open WS fds.
+        let fds = self.websockets.poll_fds();
+        let started = self.diagnostics.poll_start();
+        let result = self.multi.poll(fds, timeout);
+        self.diagnostics.polled(started, timeout, runnable);
+        match result {
+            Ok(_) => self.websockets.apply_readiness(),
+            Err(error) => self
+                .websockets
+                .fail_sessions(&mut self.multi, &error.to_string()),
         }
-        websockets.wait(multi, wait_timeout, progressed);
-    }
-
-    fn send_completion(&self, completion: CurlMultiCompletion<H, C>) {
-        let _ = self.completion_tx.send(completion);
-    }
-}
-
-fn curl_runtime_timeout_error(stage: &str) -> anyhow::Error {
-    anyhow!("curl multi runtime request timed out {stage}")
-}
-
-fn curl_timeout_for_deadline(deadline: Instant, now: Instant) -> Option<Duration> {
-    let remaining = deadline.saturating_duration_since(now);
-    // curl-rust converts CURLOPT_TIMEOUT_MS with `Duration::as_millis()`. A
-    // positive sub-millisecond value would therefore become zero, which
-    // libcurl interprets as disabling the timeout entirely.
-    (remaining >= Duration::from_millis(1)).then_some(remaining)
-}
-
-fn curl_runtime_trace_enabled() -> bool {
-    static ENABLED: OnceLock<bool> = OnceLock::new();
-    *ENABLED.get_or_init(|| {
-        env_flag_enabled("MOLI_CDP_NAV_TIMING") || env_flag_enabled("MOLI_CURL_RUNTIME_TRACE")
-    })
-}
-
-fn env_flag_enabled(name: &str) -> bool {
-    std::env::var(name).is_ok_and(|value| {
-        let value = value.trim();
-        !value.is_empty() && value != "0" && !value.eq_ignore_ascii_case("false")
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn sub_millisecond_deadline_never_disables_the_libcurl_timeout() {
-        let now = Instant::now();
-        assert_eq!(
-            curl_timeout_for_deadline(now + Duration::from_micros(999), now),
-            None
-        );
-        assert_eq!(
-            curl_timeout_for_deadline(now + Duration::from_millis(1), now),
-            Some(Duration::from_millis(1))
-        );
+        self.diagnostics
+            .report(self.websockets.session_count(), false);
     }
 }

moli-curl/src/runtime/tests.rs

@@ -0,0 +1,103 @@
+use std::{
+    io::{Read, Write},
+    net::TcpListener,
+    time::Duration,
+};
+
+use super::*;
+use crate::{CurlDnsResolution, CurlMultiJob};
+use curl::easy::Easy2;
+
+#[derive(Debug)]
+struct TestHandler;
+
+impl Handler for TestHandler {}
+
+#[test]
+fn submitted_identity_reaches_the_matching_runtime_completion() {
+    let listener = TcpListener::bind(("127.0.0.1", 0))
+        .expect("test HTTP listener should bind to a local port");
+    let address = listener
+        .local_addr()
+        .expect("test HTTP listener should have an address");
+    let server = thread::spawn(move || {
+        let (mut stream, _) = listener
+            .accept()
+            .expect("curl should connect to the test HTTP listener");
+        stream
+            .set_read_timeout(Some(Duration::from_secs(5)))
+            .expect("test HTTP connection should accept a read timeout");
+        let mut request = [0; 4096];
+        let _ = stream
+            .read(&mut request)
+            .expect("test HTTP request should be readable");
+        stream
+            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
+            .expect("test HTTP response should be writable");
+    });
+
+    let (runtime, completion_rx) = CurlMultiRuntime::new(CurlMultiRuntimeConfig {
+        poll_interval: Duration::from_millis(5),
+        ..CurlMultiRuntimeConfig::default()
+    })
+    .expect("test curl runtime should start");
+    let mut easy = Easy2::new(TestHandler);
+    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(),
+            origin: None,
+            deadline: None,
+            dns_resolution: CurlDnsResolution::curl_managed(),
+            priority: 1,
+            label: "identity-test".to_owned(),
+        })
+        .expect("test curl transfer should be accepted");
+
+    let completion = completion_rx
+        .recv_timeout(Duration::from_secs(5))
+        .expect("test curl transfer should reach terminal completion");
+    assert_eq!(completion.transfer_id, transfer_id);
+    assert_eq!(completion.context, "matching-context");
+    assert!(completion.easy.is_some());
+    completion
+        .result
+        .expect("test curl transfer should complete successfully");
+
+    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::<TestHandler, Vec<u8>>::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)
+    ));
+}

moli-curl/src/websocket.rs

@@ -1,42 +1,36 @@
-//! Persistent native WebSocket sessions, owned by a separate libcurl multi thread.
+//! Native WebSocket frames on the shared libcurl runtime.
 //!
 //! This layer transports frames. Browser handshake policy, message assembly and
 //! close-handshake semantics belong to the caller. Dropping the receiver cancels
 //! its session, including DNS and handshake work, independently of queue capacity.
 //!
-//! Internally, the shared runtime drives DNS and scheduling; session owns the native
-//! handle and its Opening/Open/ReceivedClose lifecycle; scheduling holds I/O
-//! admission state and maps polled sockets back to their sessions. Returning
+//! Connector submits work; connection holds the caller/owner I/O endpoints;
+//! registry owns DNS and attached sessions; session drives handshake and frame
+//! I/O. Readiness maps the runtime's poll results back to sessions. Returning
 //! AGAIN parks that I/O until its socket is signalled. Application wakeups only
 //! resume paused work, such as a new frame or restored receive capacity.
 
+mod connection;
 mod connection_pool;
-mod diagnostics;
+mod connector;
+mod readiness;
 pub(crate) mod registry;
 mod request;
-mod scheduling;
 mod session;
+mod standalone;
 #[cfg(test)]
 mod tests;
 
-use std::{
-    sync::{
-        Arc,
-        atomic::{AtomicBool, Ordering},
-    },
-    time::Duration,
-};
+use crate::{CurlDnsResolution, CurlTlsConfig};
+use anyhow::{Result, bail};
+use std::time::Duration;
 
-use anyhow::{Context, Result, bail};
-use curl::multi::MultiWaker;
-use parking_lot::Mutex;
-use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot};
-
-use crate::{
-    CurlDnsResolution, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlTlsConfig, CurlTransferId,
-    runtime::identity::next_transfer_id,
-};
+use connection::SessionIo;
+pub use connection::{CurlWebSocketConnection, CurlWebSocketSender};
+pub use connector::CurlWebSocketConnector;
+pub(crate) use connector::Submission;
 pub use curl::easy::{WsFlags, WsFrame};
+pub use standalone::CurlWebSocketRuntime;
 
 /// Fragment large messages above this layer to bound native write residence.
 pub const MAX_SEND_FRAME_BYTES: usize = 64 * 1024;
@@ -120,287 +114,3 @@ impl CurlWebSocketSend {
         Ok(())
     }
 }
-
-#[derive(Debug)]
-struct Control {
-    cancelled: AtomicBool,
-    closed: AtomicBool,
-    reading: AtomicBool,
-    send: Mutex<Option<PendingSend>>,
-    terminal: Mutex<Option<std::result::Result<(), String>>>,
-    waker: MultiWaker,
-    #[cfg(test)]
-    read_blocked: tokio::sync::Notify,
-    #[cfg(test)]
-    write_blocked: tokio::sync::Notify,
-    #[cfg(test)]
-    read_attempts: std::sync::atomic::AtomicUsize,
-    #[cfg(test)]
-    receive_allocations: std::sync::atomic::AtomicUsize,
-    #[cfg(test)]
-    read_waiting: tokio::sync::Notify,
-    #[cfg(test)]
-    owner_thread: Mutex<Option<std::thread::ThreadId>>,
-}
-
-impl Control {
-    fn wake(&self) {
-        let _ = self.waker.wakeup();
-    }
-    fn cancel(&self) {
-        self.cancelled.store(true, Ordering::Release);
-        self.wake();
-    }
-}
-
-#[derive(Debug)]
-struct PendingSend {
-    frame: CurlWebSocketSend,
-    offset: usize,
-    completed: oneshot::Sender<usize>,
-}
-
-/// Cloneable write/control capability; it does not keep a dropped receiver alive.
-#[derive(Clone, Debug)]
-pub struct CurlWebSocketSender {
-    control: Arc<Control>,
-}
-
-impl CurlWebSocketSender {
-    /// Writes one frame and returns its payload size once libcurl consumes it.
-    ///
-    /// Only one frame may be pending per connection; overlapping sends return an
-    /// error. The caller owns data/control ordering and any message queue.
-    /// Completion is independent of reads and received event delivery.
-    ///
-    /// Dropping an unpolled future submits nothing. Once submitted, the frame
-    /// remains pending until written or the connection is cancelled/closed, even
-    /// if this future is dropped. Transport failure or cancellation returns an error.
-    pub async fn send_frame(&self, frame: CurlWebSocketSend) -> Result<usize> {
-        frame.validate()?;
-        let completion = {
-            let mut send = self.control.send.lock();
-            if self.control.closed.load(Ordering::Acquire)
-                || self.control.cancelled.load(Ordering::Acquire)
-            {
-                bail!("WebSocket transport is closed");
-            }
-            if send.is_some() {
-                bail!("WebSocket frame is already pending");
-            }
-            let (completed, completion) = oneshot::channel();
-            *send = Some(PendingSend {
-                frame,
-                offset: 0,
-                completed,
-            });
-            completion
-        };
-        self.control.wake();
-        completion
-            .await
-            .context("WebSocket transport closed before frame completion")
-    }
-
-    /// Handshake delivery starts paused, so application data cannot outrun Open.
-    /// Pausing does not retract queued chunks or a read already in progress.
-    /// Send completions progress independently of reads and received events.
-    pub fn set_reading(&self, enabled: bool) {
-        self.control.reading.store(enabled, Ordering::Release);
-        self.control.wake();
-    }
-
-    pub fn cancel(&self) {
-        self.control.cancel();
-    }
-}
-
-#[derive(Debug)]
-pub struct CurlWebSocketConnection {
-    id: CurlTransferId,
-    sender: CurlWebSocketSender,
-    events: mpsc::Receiver<CurlWebSocketEvent>,
-    terminal_delivered: bool,
-}
-
-impl CurlWebSocketConnection {
-    pub fn id(&self) -> CurlTransferId {
-        self.id
-    }
-    pub fn sender(&self) -> CurlWebSocketSender {
-        self.sender.clone()
-    }
-
-    pub async fn recv(&mut self) -> Option<CurlWebSocketEvent> {
-        if let Some(event) = self.events.recv().await {
-            self.sender.control.wake();
-            return Some(event);
-        }
-        if self.terminal_delivered {
-            return None;
-        }
-        self.terminal_delivered = true;
-        let result = self
-            .sender
-            .control
-            .terminal
-            .lock()
-            .take()
-            .unwrap_or_else(|| Err("WebSocket runtime stopped".to_owned()));
-        Some(CurlWebSocketEvent::Closed { result })
-    }
-}
-
-impl Drop for CurlWebSocketConnection {
-    fn drop(&mut self) {
-        self.sender.cancel();
-    }
-}
-
-struct SessionIo {
-    events: mpsc::Sender<CurlWebSocketEvent>,
-    control: Arc<Control>,
-    _slot: OwnedSemaphorePermit,
-}
-
-impl SessionIo {
-    fn cancelled(&self) -> bool {
-        self.control.cancelled.load(Ordering::Acquire)
-    }
-    fn finish(&self, result: std::result::Result<(), String>) {
-        *self.control.terminal.lock() = Some(result);
-    }
-}
-
-impl Drop for SessionIo {
-    fn drop(&mut self) {
-        self.control.closed.store(true, Ordering::Release);
-        // Settle the pending write before closing the receive event channel.
-        self.control.send.lock().take();
-    }
-}
-
-pub(crate) struct Submission {
-    id: CurlTransferId,
-    request: CurlWebSocketRequest,
-    io: SessionIo,
-}
-
-/// Admission capability for one native owner. Clones do not keep that owner
-/// alive; shutting down its runtime closes connections and rejects new ones.
-#[derive(Clone, Debug)]
-pub struct CurlWebSocketConnector {
-    inner: Arc<ConnectorInner>,
-}
-
-#[derive(Debug)]
-struct ConnectorInner {
-    submissions: crossbeam_channel::Sender<Submission>,
-    slots: Arc<Semaphore>,
-    waker: MultiWaker,
-    shutdown: Arc<AtomicBool>,
-}
-
-impl CurlWebSocketConnector {
-    pub(crate) fn channel() -> (
-        crossbeam_channel::Sender<Submission>,
-        crossbeam_channel::Receiver<Submission>,
-    ) {
-        crossbeam_channel::bounded(SESSION_CAPACITY)
-    }
-
-    pub(crate) fn new(
-        submissions: crossbeam_channel::Sender<Submission>,
-        waker: MultiWaker,
-        shutdown: Arc<AtomicBool>,
-    ) -> Self {
-        Self {
-            inner: Arc::new(ConnectorInner {
-                submissions,
-                slots: Arc::new(Semaphore::new(SESSION_CAPACITY)),
-                waker,
-                shutdown,
-            }),
-        }
-    }
-
-    pub fn connect(&self, request: CurlWebSocketRequest) -> Result<CurlWebSocketConnection> {
-        if self.inner.shutdown.load(Ordering::Acquire) {
-            bail!("curl WebSocket runtime is closed");
-        }
-        let slot = self
-            .inner
-            .slots
-            .clone()
-            .try_acquire_owned()
-            .context("too many curl WebSocket sessions")?;
-        let id = next_transfer_id()?;
-        let control = Arc::new(Control {
-            cancelled: AtomicBool::new(false),
-            closed: AtomicBool::new(false),
-            reading: AtomicBool::new(false),
-            send: Mutex::new(None),
-            terminal: Mutex::new(None),
-            waker: self.inner.waker.clone(),
-            #[cfg(test)]
-            read_blocked: tokio::sync::Notify::new(),
-            #[cfg(test)]
-            write_blocked: tokio::sync::Notify::new(),
-            #[cfg(test)]
-            read_attempts: std::sync::atomic::AtomicUsize::new(0),
-            #[cfg(test)]
-            receive_allocations: std::sync::atomic::AtomicUsize::new(0),
-            #[cfg(test)]
-            read_waiting: tokio::sync::Notify::new(),
-            #[cfg(test)]
-            owner_thread: Mutex::new(None),
-        });
-        let (event_tx, events) = mpsc::channel(MAX_PENDING_EVENTS);
-        let io = SessionIo {
-            events: event_tx,
-            control: control.clone(),
-            _slot: slot,
-        };
-        self.inner
-            .submissions
-            .try_send(Submission { id, request, io })
-            .map_err(|_| anyhow::anyhow!("curl WebSocket runtime cannot accept a session"))?;
-        control.wake();
-        Ok(CurlWebSocketConnection {
-            id,
-            sender: CurlWebSocketSender { control },
-            events,
-            terminal_delivered: false,
-        })
-    }
-}
-
-/// Standalone owner for callers without an existing HTTP runtime. Uses the
-/// same Multi driver as HTTP, with no HTTP submissions.
-#[derive(Debug)]
-pub struct CurlWebSocketRuntime {
-    runtime: CurlMultiRuntime<StandaloneHandler, ()>,
-}
-
-#[derive(Debug)]
-struct StandaloneHandler;
-impl curl::easy::Handler for StandaloneHandler {}
-
-impl CurlWebSocketRuntime {
-    pub fn new() -> Result<Self> {
-        let (runtime, _) = CurlMultiRuntime::new(CurlMultiRuntimeConfig {
-            thread_name: "moli-curl-websocket".to_owned(),
-            poll_interval: Duration::from_secs(1),
-            ..CurlMultiRuntimeConfig::default()
-        })?;
-        Ok(Self { runtime })
-    }
-
-    pub fn connector(&self) -> CurlWebSocketConnector {
-        self.runtime.websocket_connector()
-    }
-
-    pub fn connect(&self, request: CurlWebSocketRequest) -> Result<CurlWebSocketConnection> {
-        self.connector().connect(request)
-    }
-}

moli-curl/src/websocket/connection.rs

@@ -0,0 +1,214 @@
+//! The caller's receive/write handles and the owner's I/O endpoint share one
+//! lifetime. Pending send completion stays independent of received event capacity.
+
+use super::{CurlWebSocketEvent, CurlWebSocketSend, MAX_PENDING_EVENTS};
+use crate::CurlTransferId;
+use anyhow::{Context, Result, bail};
+use curl::multi::MultiWaker;
+use parking_lot::Mutex;
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+use tokio::sync::{OwnedSemaphorePermit, mpsc, oneshot};
+
+#[derive(Debug)]
+pub(super) struct Control {
+    cancelled: AtomicBool,
+    closed: AtomicBool,
+    pub(super) reading: AtomicBool,
+    pub(super) send: Mutex<Option<PendingSend>>,
+    terminal: Mutex<Option<std::result::Result<(), String>>>,
+    waker: MultiWaker,
+    #[cfg(test)]
+    pub(super) read_blocked: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) write_blocked: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) read_attempts: std::sync::atomic::AtomicUsize,
+    #[cfg(test)]
+    pub(super) receive_allocations: std::sync::atomic::AtomicUsize,
+    #[cfg(test)]
+    pub(super) read_waiting: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) owner_thread: Mutex<Option<std::thread::ThreadId>>,
+}
+
+impl Control {
+    fn wake(&self) {
+        let _ = self.waker.wakeup();
+    }
+    fn cancel(&self) {
+        self.cancelled.store(true, Ordering::Release);
+        self.wake();
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct PendingSend {
+    pub(super) frame: CurlWebSocketSend,
+    pub(super) offset: usize,
+    pub(super) completed: oneshot::Sender<usize>,
+}
+
+/// Cloneable write/control capability; it does not keep a dropped receiver alive.
+#[derive(Clone, Debug)]
+pub struct CurlWebSocketSender {
+    pub(super) control: Arc<Control>,
+}
+
+impl CurlWebSocketSender {
+    /// Writes one frame and returns its payload size once libcurl consumes it.
+    ///
+    /// Only one frame may be pending per connection; overlapping sends return an
+    /// error. The caller owns data/control ordering and any message queue.
+    /// Completion is independent of reads and received event delivery.
+    ///
+    /// Dropping an unpolled future submits nothing. Once submitted, the frame
+    /// remains pending until written or the connection is cancelled/closed, even
+    /// if this future is dropped. Transport failure or cancellation returns an error.
+    pub async fn send_frame(&self, frame: CurlWebSocketSend) -> Result<usize> {
+        frame.validate()?;
+        let completion = {
+            let mut send = self.control.send.lock();
+            if self.control.closed.load(Ordering::Acquire)
+                || self.control.cancelled.load(Ordering::Acquire)
+            {
+                bail!("WebSocket transport is closed");
+            }
+            if send.is_some() {
+                bail!("WebSocket frame is already pending");
+            }
+            let (completed, completion) = oneshot::channel();
+            *send = Some(PendingSend {
+                frame,
+                offset: 0,
+                completed,
+            });
+            completion
+        };
+        self.control.wake();
+        completion
+            .await
+            .context("WebSocket transport closed before frame completion")
+    }
+
+    /// Handshake delivery starts paused, so application data cannot outrun Open.
+    /// Pausing does not retract queued chunks or a read already in progress.
+    /// Send completions progress independently of reads and received events.
+    pub fn set_reading(&self, enabled: bool) {
+        self.control.reading.store(enabled, Ordering::Release);
+        self.control.wake();
+    }
+
+    pub fn cancel(&self) {
+        self.control.cancel();
+    }
+}
+
+#[derive(Debug)]
+pub struct CurlWebSocketConnection {
+    id: CurlTransferId,
+    sender: CurlWebSocketSender,
+    pub(super) events: mpsc::Receiver<CurlWebSocketEvent>,
+    terminal_delivered: bool,
+}
+
+impl CurlWebSocketConnection {
+    pub(super) fn channel(
+        id: CurlTransferId,
+        waker: MultiWaker,
+        slot: OwnedSemaphorePermit,
+    ) -> (Self, SessionIo) {
+        let control = Arc::new(Control {
+            cancelled: AtomicBool::new(false),
+            closed: AtomicBool::new(false),
+            reading: AtomicBool::new(false),
+            send: Mutex::new(None),
+            terminal: Mutex::new(None),
+            waker,
+            #[cfg(test)]
+            read_blocked: tokio::sync::Notify::new(),
+            #[cfg(test)]
+            write_blocked: tokio::sync::Notify::new(),
+            #[cfg(test)]
+            read_attempts: std::sync::atomic::AtomicUsize::new(0),
+            #[cfg(test)]
+            receive_allocations: std::sync::atomic::AtomicUsize::new(0),
+            #[cfg(test)]
+            read_waiting: tokio::sync::Notify::new(),
+            #[cfg(test)]
+            owner_thread: Mutex::new(None),
+        });
+        let (event_tx, events) = mpsc::channel(MAX_PENDING_EVENTS);
+        let io = SessionIo {
+            events: event_tx,
+            control: control.clone(),
+            _slot: slot,
+        };
+        (
+            Self {
+                id,
+                sender: CurlWebSocketSender { control },
+                events,
+                terminal_delivered: false,
+            },
+            io,
+        )
+    }
+
+    pub fn id(&self) -> CurlTransferId {
+        self.id
+    }
+    pub fn sender(&self) -> CurlWebSocketSender {
+        self.sender.clone()
+    }
+
+    pub async fn recv(&mut self) -> Option<CurlWebSocketEvent> {
+        if let Some(event) = self.events.recv().await {
+            self.sender.control.wake();
+            return Some(event);
+        }
+        if self.terminal_delivered {
+            return None;
+        }
+        self.terminal_delivered = true;
+        let result = self
+            .sender
+            .control
+            .terminal
+            .lock()
+            .take()
+            .unwrap_or_else(|| Err("WebSocket runtime stopped".to_owned()));
+        Some(CurlWebSocketEvent::Closed { result })
+    }
+}
+
+impl Drop for CurlWebSocketConnection {
+    fn drop(&mut self) {
+        self.sender.cancel();
+    }
+}
+
+pub(super) struct SessionIo {
+    pub(super) events: mpsc::Sender<CurlWebSocketEvent>,
+    pub(super) control: Arc<Control>,
+    _slot: OwnedSemaphorePermit,
+}
+
+impl SessionIo {
+    pub(super) fn cancelled(&self) -> bool {
+        self.control.cancelled.load(Ordering::Acquire)
+    }
+    pub(super) fn finish(&self, result: std::result::Result<(), String>) {
+        *self.control.terminal.lock() = Some(result);
+    }
+}
+
+impl Drop for SessionIo {
+    fn drop(&mut self) {
+        self.control.closed.store(true, Ordering::Release);
+        // Settle the pending write before closing the receive event channel.
+        self.control.send.lock().take();
+    }
+}

moli-curl/src/websocket/connector.rs

@@ -0,0 +1,78 @@
+//! Submission capability for a specific owner, without thread ownership.
+
+use super::{
+    CurlWebSocketConnection, CurlWebSocketRequest, SESSION_CAPACITY, connection::SessionIo,
+};
+use crate::{CurlTransferId, runtime::identity::next_transfer_id};
+use anyhow::{Context, Result, bail};
+use curl::multi::MultiWaker;
+use std::sync::{
+    Arc,
+    atomic::{AtomicBool, Ordering},
+};
+use tokio::sync::Semaphore;
+
+pub(crate) struct Submission {
+    pub(super) id: CurlTransferId,
+    pub(super) request: CurlWebSocketRequest,
+    pub(super) io: SessionIo,
+}
+
+/// Admission capability for one native owner. Clones do not keep that owner
+/// alive; shutting down its runtime closes connections and rejects new ones.
+#[derive(Clone, Debug)]
+pub struct CurlWebSocketConnector {
+    inner: Arc<ConnectorInner>,
+}
+
+#[derive(Debug)]
+struct ConnectorInner {
+    submissions: crossbeam_channel::Sender<Submission>,
+    slots: Arc<Semaphore>,
+    waker: MultiWaker,
+    shutdown: Arc<AtomicBool>,
+}
+
+impl CurlWebSocketConnector {
+    pub(crate) fn channel() -> (
+        crossbeam_channel::Sender<Submission>,
+        crossbeam_channel::Receiver<Submission>,
+    ) {
+        crossbeam_channel::bounded(SESSION_CAPACITY)
+    }
+
+    pub(crate) fn new(
+        submissions: crossbeam_channel::Sender<Submission>,
+        waker: MultiWaker,
+        shutdown: Arc<AtomicBool>,
+    ) -> Self {
+        Self {
+            inner: Arc::new(ConnectorInner {
+                submissions,
+                slots: Arc::new(Semaphore::new(SESSION_CAPACITY)),
+                waker,
+                shutdown,
+            }),
+        }
+    }
+
+    pub fn connect(&self, request: CurlWebSocketRequest) -> Result<CurlWebSocketConnection> {
+        if self.inner.shutdown.load(Ordering::Acquire) {
+            bail!("curl WebSocket runtime is closed");
+        }
+        let slot = self
+            .inner
+            .slots
+            .clone()
+            .try_acquire_owned()
+            .context("too many curl WebSocket sessions")?;
+        let id = next_transfer_id()?;
+        let (connection, io) = CurlWebSocketConnection::channel(id, self.inner.waker.clone(), slot);
+        self.inner
+            .submissions
+            .try_send(Submission { id, request, io })
+            .map_err(|_| anyhow::anyhow!("curl WebSocket runtime cannot accept a session"))?;
+        let _ = self.inner.waker.wakeup();
+        Ok(connection)
+    }
+}

moli-curl/src/websocket/readiness.rs

@@ -2,11 +2,11 @@
 //! work/capacity; WaitingForSocket I/O has already returned AGAIN. A successful
 //! operation stays Runnable so libcurl's buffered data can be drained as well.
 
-use std::{collections::HashMap, time::Duration};
+use std::collections::HashMap;
 
-use curl::multi::{Multi, WaitFd};
+use curl::multi::WaitFd;
 
-use super::{diagnostics::Diagnostics, session::Session};
+use super::session::Session;
 use crate::CurlTransferId;
 
 #[derive(Default, PartialEq, Eq)]
@@ -47,22 +47,15 @@ impl IoState {
 }
 
 /// Keeps each extra fd paired with its session across curl_multi_poll.
-/// libcurl also polls opening handshakes and the cross-thread waker here.
+/// The runtime polls these alongside HTTP/handshake sockets and curl's waker.
 #[derive(Default)]
-pub(super) struct SocketPoll {
+pub(super) struct SocketReadiness {
     ids: Vec<CurlTransferId>,
     fds: Vec<WaitFd>,
 }
 
-impl SocketPoll {
-    pub(super) fn wait(
-        &mut self,
-        multi: &Multi,
-        sessions: &mut HashMap<CurlTransferId, Session>,
-        timeout: Duration,
-        progressed: bool,
-        diagnostics: &mut Diagnostics,
-    ) -> Result<(), curl::MultiError> {
+impl SocketReadiness {
+    pub(super) fn prepare(&mut self, sessions: &HashMap<CurlTransferId, Session>) -> &mut [WaitFd] {
         self.ids.clear();
         self.fds.clear();
         for (id, session) in sessions.iter() {
@@ -71,10 +64,10 @@ impl SocketPoll {
                 self.fds.push(fd);
             }
         }
-        let started = diagnostics.poll_start();
-        let result = multi.poll(&mut self.fds, timeout);
-        diagnostics.polled(started, timeout, progressed);
-        result?;
+        &mut self.fds
+    }
+
+    pub(super) fn dispatch(&self, sessions: &mut HashMap<CurlTransferId, Session>) {
         for (id, fd) in self.ids.iter().zip(&self.fds) {
             if fd.received_read() || fd.received_write() {
                 // A socket event permits a retry, not guaranteed progress.
@@ -86,6 +79,5 @@ impl SocketPoll {
                     .socket_ready();
             }
         }
-        Ok(())
     }
 }

moli-curl/src/websocket/registry.rs

@@ -1,23 +1,24 @@
 //! Persistent WebSocket residences on the common HTTP/WebSocket Multi.
 //! The runtime owns perform, completion dispatch and the single wait loop.
 
-use std::{
-    collections::HashMap,
-    rc::Rc,
-    time::{Duration, Instant},
-};
+use std::{collections::HashMap, rc::Rc, time::Instant};
 
-use curl::{easy::Easy2, multi::Multi};
+use curl::{
+    easy::Easy2,
+    multi::{Multi, WaitFd},
+};
 
 use super::{
     SessionIo, Submission,
     connection_pool::ConnectionPool,
-    diagnostics::Diagnostics,
+    readiness::SocketReadiness,
     request::{self, Handshake},
-    scheduling::SocketPoll,
     session::{Session, Step},
 };
-use crate::{CurlDnsResolution, CurlTransferId, dns_adapter::CurlDnsOwnerResidence};
+use crate::{
+    CurlDnsResolution, CurlTransferId, dns_adapter::CurlDnsOwnerResidence,
+    runtime::diagnostics::Diagnostics,
+};
 
 struct Pending {
     id: CurlTransferId,
@@ -31,9 +32,8 @@ pub(crate) struct WebSocketRegistry {
     submissions: crossbeam_channel::Receiver<Submission>,
     sessions: HashMap<CurlTransferId, Session>,
     dns: CurlDnsOwnerResidence<CurlTransferId, Pending>,
-    poll: SocketPoll,
+    readiness: SocketReadiness,
     receive: Vec<u8>,
-    diagnostics: Diagnostics,
     pool: Option<Rc<ConnectionPool>>,
     closed: bool,
 }
@@ -44,9 +44,8 @@ impl WebSocketRegistry {
             submissions,
             sessions: HashMap::new(),
             dns: CurlDnsOwnerResidence::default(),
-            poll: SocketPoll::default(),
+            readiness: SocketReadiness::default(),
             receive: Vec::new(),
-            diagnostics: Diagnostics::from_env(),
             pool: None,
             closed: false,
         }
@@ -76,7 +75,6 @@ impl WebSocketRegistry {
                 .io
                 .finish(Err("curl WebSocket runtime shut down".to_owned()));
         }
-        self.diagnostics.report(0, true);
     }
 
     pub(crate) fn admit(&mut self, submission: Submission, multi: &mut Multi) {
@@ -188,7 +186,7 @@ impl WebSocketRegistry {
         self.sessions.get(&id)?.handshake_result(message)
     }
 
-    pub(crate) fn advance(&mut self, multi: &mut Multi) -> bool {
+    pub(crate) fn advance(&mut self, multi: &mut Multi, diagnostics: &mut Diagnostics) -> bool {
         for _ in 0..super::SESSION_CAPACITY {
             let Ok(submission) = self.submissions.try_recv() else {
                 break;
@@ -199,7 +197,7 @@ impl WebSocketRegistry {
         let mut retired = Vec::new();
         let mut progressed = false;
         for (id, session) in &mut self.sessions {
-            match session.advance(&mut self.receive, &mut self.diagnostics) {
+            match session.advance(&mut self.receive, diagnostics) {
                 Ok(Step::Progress) => progressed = true,
                 Ok(Step::Idle) => {}
                 terminal => retired.push((*id, terminal)),
@@ -208,10 +206,6 @@ impl WebSocketRegistry {
         for (id, step) in retired {
             self.apply_step(id, step, multi);
         }
-        if let Some(counters) = self.diagnostics.counters() {
-            counters.turns += 1;
-            counters.progressed_turns += u64::from(progressed);
-        }
         progressed
     }
 
@@ -234,17 +228,16 @@ impl WebSocketRegistry {
             .min()
     }
 
-    pub(crate) fn wait(&mut self, multi: &mut Multi, timeout: Duration, progressed: bool) {
-        if let Err(error) = self.poll.wait(
-            multi,
-            &mut self.sessions,
-            timeout,
-            progressed,
-            &mut self.diagnostics,
-        ) {
-            self.fail_sessions(multi, &error.to_string());
-        }
-        self.diagnostics.report(self.sessions.len(), false);
+    pub(crate) fn session_count(&self) -> usize {
+        self.sessions.len()
+    }
+
+    pub(crate) fn poll_fds(&mut self) -> &mut [WaitFd] {
+        self.readiness.prepare(&self.sessions)
+    }
+
+    pub(crate) fn apply_readiness(&mut self) {
+        self.readiness.dispatch(&mut self.sessions);
     }
 
     pub(crate) fn fail_sessions(&mut self, multi: &mut Multi, error: &str) {

moli-curl/src/websocket/session.rs

@@ -8,9 +8,9 @@ use curl::{
     multi::{Easy2Handle, Multi, WaitFd},
 };
 
-use super::diagnostics::Diagnostics;
-use super::{CurlWebSocketEvent, SessionIo, WsFlags, request::Handshake, scheduling::IoState};
+use super::{CurlWebSocketEvent, SessionIo, WsFlags, readiness::IoState, request::Handshake};
 use crate::CurlTransferId;
+use crate::runtime::diagnostics::Diagnostics;
 
 const CHUNK_BYTES: usize = 16 * 1024;
 const IO_BUDGET: usize = 8;

moli-curl/src/websocket/standalone.rs

@@ -0,0 +1,36 @@
+//! Convenience owner for callers without a shared network runtime.
+
+use super::{CurlWebSocketConnection, CurlWebSocketConnector, CurlWebSocketRequest};
+use crate::{CurlMultiRuntime, CurlMultiRuntimeConfig};
+use anyhow::Result;
+use std::time::Duration;
+
+/// Standalone owner for callers without an existing HTTP runtime. Uses the
+/// same Multi driver as HTTP, with no HTTP submissions.
+#[derive(Debug)]
+pub struct CurlWebSocketRuntime {
+    runtime: CurlMultiRuntime<StandaloneHandler, ()>,
+}
+
+#[derive(Debug)]
+struct StandaloneHandler;
+impl curl::easy::Handler for StandaloneHandler {}
+
+impl CurlWebSocketRuntime {
+    pub fn new() -> Result<Self> {
+        let (runtime, _) = CurlMultiRuntime::new(CurlMultiRuntimeConfig {
+            thread_name: "moli-curl-websocket".to_owned(),
+            poll_interval: Duration::from_secs(1),
+            ..CurlMultiRuntimeConfig::default()
+        })?;
+        Ok(Self { runtime })
+    }
+
+    pub fn connector(&self) -> CurlWebSocketConnector {
+        self.runtime.websocket_connector()
+    }
+
+    pub fn connect(&self, request: CurlWebSocketRequest) -> Result<CurlWebSocketConnection> {
+        self.connector().connect(request)
+    }
+}

moli-curl/src/websocket/tests.rs

@@ -1,10 +1,11 @@
 use std::{
     io::{Read, Write},
     net::{TcpListener, TcpStream},
+    sync::{Arc, atomic::Ordering},
     thread,
 };
 
-use tokio::time::timeout;
+use tokio::{sync::oneshot, time::timeout};
 use tokio_tungstenite::tungstenite::{self, Message, handshake::derive_accept_key};
 
 use super::*;
@@ -324,7 +325,7 @@ async fn native_full_delivery_queue_does_not_block_other_sessions_or_cancellatio
     let mut blocked = runtime.connect(CurlWebSocketRequest::new(url)).unwrap();
     opened(&mut blocked).await;
     blocked.sender().set_reading(true);
-    timeout(DEADLINE, blocked.sender.control.read_blocked.notified())
+    timeout(DEADLINE, blocked.sender().control.read_blocked.notified())
         .await
         .unwrap();
     assert_eq!(blocked.events.len(), MAX_PENDING_EVENTS);
@@ -359,9 +360,12 @@ async fn native_queue_capacity_restoration_wakes_owner() {
     let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap();
     opened(&mut connection).await;
     connection.sender().set_reading(true);
-    timeout(DEADLINE, connection.sender.control.read_blocked.notified())
-        .await
-        .unwrap();
+    timeout(
+        DEADLINE,
+        connection.sender().control.read_blocked.notified(),
+    )
+    .await
+    .unwrap();
     for expected in 0..32 {
         assert!(
             matches!(event(&mut connection).await, CurlWebSocketEvent::Chunk { data, .. } if data == [expected])

moli-curl/src/websocket/tests/readiness.rs

@@ -84,7 +84,7 @@ async fn native_socket_readiness_is_serviced_during_continuous_traffic() {
         .unwrap();
     opened(&mut quiet).await;
     quiet.sender().set_reading(true);
-    timeout(DEADLINE, quiet.sender.control.read_waiting.notified())
+    timeout(DEADLINE, quiet.sender().control.read_waiting.notified())
         .await
         .unwrap();
     let mut active = runtime.connect(CurlWebSocketRequest::new(url)).unwrap();
@@ -181,9 +181,12 @@ async fn native_peer_eof_wakes_waiting_reader() {
     let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap();
     opened(&mut connection).await;
     connection.sender().set_reading(true);
-    timeout(DEADLINE, connection.sender.control.read_waiting.notified())
-        .await
-        .unwrap();
+    timeout(
+        DEADLINE,
+        connection.sender().control.read_waiting.notified(),
+    )
+    .await
+    .unwrap();
     finish_tx.send(()).unwrap();
     assert!(matches!(
         event(&mut connection).await,

moli-curl/src/websocket/tests/shared.rs

@@ -3,7 +3,7 @@ use std::num::NonZeroUsize;
 use curl::easy::{Easy2, Handler, WriteError};
 
 use super::*;
-use crate::{CurlMultiJob, CurlOriginKey};
+use crate::{CurlMultiJob, CurlMultiRuntime, CurlMultiRuntimeConfig, CurlOriginKey};
 
 #[derive(Default, Debug)]
 struct HttpCapture {
@@ -149,7 +149,7 @@ async fn shared_owner_keeps_http_and_websocket_pools_independent() {
         assert_eq!(easy.get_ref().body, b"ok");
         for mut connection in connections {
             assert_eq!(
-                *connection.sender.control.owner_thread.lock(),
+                *connection.sender().control.owner_thread.lock(),
                 easy.get_ref().owner
             );
             connection.sender().set_reading(true);

moli-curl/src/websocket/tests/storage.rs

@@ -18,9 +18,12 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_
         let mut connection = runtime.connect(CurlWebSocketRequest::new(url)).unwrap();
         opened(&mut connection).await;
         connection.sender().set_reading(true);
-        timeout(DEADLINE, connection.sender.control.read_waiting.notified())
-            .await
-            .unwrap();
+        timeout(
+            DEADLINE,
+            connection.sender().control.read_waiting.notified(),
+        )
+        .await
+        .unwrap();
         connections.push(connection);
         peers.push(peer);
         writers.push(write);
@@ -30,7 +33,7 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_
             .iter()
             .map(|connection| {
                 connection
-                    .sender
+                    .sender()
                     .control
                     .receive_allocations
                     .load(Ordering::Acquire)
@@ -53,7 +56,7 @@ async fn native_receive_storage_is_shared_across_idle_probes_and_transferred_to_
         retained.push(data);
         timeout(
             DEADLINE,
-            connections[index].sender.control.read_waiting.notified(),
+            connections[index].sender().control.read_waiting.notified(),
         )
         .await
         .unwrap();