diff --git a/crates/hiroz-union/src/plugin/wasm/host/ros.rs b/crates/hiroz-union/src/plugin/wasm/host/ros.rs index ccaf17936..18a77d36d 100644 --- a/crates/hiroz-union/src/plugin/wasm/host/ros.rs +++ b/crates/hiroz-union/src/plugin/wasm/host/ros.rs @@ -16,6 +16,35 @@ use super::super::state::{PluginState, ServiceClientData, SubscriptionData}; use super::hu; use hu::plugin::types::PluginError; +/// The longest time that a guest-supplied service timeout may suspend the +/// epoch ticker. +/// +/// `timeout-ms` crosses the WIT boundary as a `u32`, and the guest chooses its +/// value. The host holds a `HostBlockGuard` across the reply wait. Without a +/// clamp, a plugin picks how long the host stops preempting plugins. An +/// unclamped `u32` allows about 49 days. +/// +/// The watchdog exists to preempt a guest that does not yield. The guest must +/// not control it. 60 s is far above any real service call. It is also far +/// below a disabled ticker. +const MAX_GUEST_TIMEOUT: Duration = Duration::from_secs(60); + +/// Caps a guest-supplied timeout. The host writes a warning when it reduces the +/// value, so that the plugin author sees why the timeout changed. +fn clamp_guest_timeout(timeout_ms: u32) -> Duration { + let requested = Duration::from_millis(timeout_ms as u64); + if requested > MAX_GUEST_TIMEOUT { + tracing::warn!( + "plugin asked for a {}ms service timeout; clamping to {}ms", + timeout_ms, + MAX_GUEST_TIMEOUT.as_millis() + ); + MAX_GUEST_TIMEOUT + } else { + requested + } +} + impl PluginState { /// The message type advertised by a live publisher or subscriber on `topic`, /// if any. Used both to build the concrete publish key (`resolve_topic_ke`) @@ -190,11 +219,19 @@ impl hu::plugin::ros::Host for PluginState { // Budget discovery independently; a slow/failed round-trip // shouldn't look like a generic encode failure. const DISCOVERY_TIMEOUT: Duration = Duration::from_millis(2000); - let discovered = tokio::task::block_in_place(|| { - tokio::runtime::Handle::current().block_on( - node.discover_topic_schema_including_subscribers(&topic, DISCOVERY_TIMEOUT), - ) - }) + let discovered = { + // See `subscribe`: wall-clock waits are charged to the + // guest's epoch budget unless the ticker is suspended. + let _epoch = super::super::HostBlockGuard::enter(); + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on( + node.discover_topic_schema_including_subscribers( + &topic, + DISCOVERY_TIMEOUT, + ), + ) + }) + } .map_err(|_| PluginError::NotFound)?; if discovered.schema.type_name != type_name { return Err(PluginError::Invalid(format!( @@ -277,14 +314,18 @@ impl hu::plugin::ros::HostServiceClient for PluginState { // slow/failed discovery round-trip (two get_type_description queries) // can't consume the entire per-call timeout budget. const DISCOVERY_TIMEOUT: Duration = Duration::from_millis(2000); - let (req_schema, resp_schema) = tokio::task::block_in_place(|| { - tokio::runtime::Handle::current().block_on(node.discover_service_schema( - &service_name, - &req_type, - &resp_type, - DISCOVERY_TIMEOUT, - )) - }) + let (req_schema, resp_schema) = { + // See `subscribe`: suspend the epoch ticker across the wait. + let _epoch = super::super::HostBlockGuard::enter(); + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(node.discover_service_schema( + &service_name, + &req_type, + &resp_type, + DISCOVERY_TIMEOUT, + )) + }) + } .map_err(|_| PluginError::NotFound)?; let req_value = parse_yaml_or_json(&request_json).map_err(PluginError::Invalid)?; @@ -300,7 +341,7 @@ impl hu::plugin::ros::HostServiceClient for PluginState { let sn = self.alloc_rep() as i64; let attachment = hiroz::attachment::Attachment::new(sn, gid); - let timeout = Duration::from_millis(timeout_ms as u64); + let timeout = clamp_guest_timeout(timeout_ms); let replies = session .get(&ke) .payload(zenoh::bytes::ZBytes::from(req_cdr)) @@ -309,7 +350,13 @@ impl hu::plugin::ros::HostServiceClient for PluginState { .wait() .map_err(|e| e.to_string())?; - let reply = replies.recv().map_err(|_| PluginError::Timeout)?; + let reply = { + // The caller's own --timeout can exceed the guest's epoch budget, so + // suspend the ticker across the wait (see `subscribe`). + let _epoch = super::super::HostBlockGuard::enter(); + replies.recv() + } + .map_err(|_| PluginError::Timeout)?; let sample = reply.result().map_err(|e| e.to_string())?; let resp_cdr = sample.payload().to_bytes().into_owned(); @@ -347,7 +394,7 @@ impl hu::plugin::ros::HostServiceClient for PluginState { let sn = self.alloc_rep() as i64; let attachment = hiroz::attachment::Attachment::new(sn, gid); - let timeout = Duration::from_millis(timeout_ms as u64); + let timeout = clamp_guest_timeout(timeout_ms); let replies = session .get(&ke) .payload(zenoh::bytes::ZBytes::from(payload)) @@ -356,7 +403,13 @@ impl hu::plugin::ros::HostServiceClient for PluginState { .wait() .map_err(|e| e.to_string())?; - let reply = replies.recv().map_err(|_| PluginError::Timeout)?; + let reply = { + // The caller's own --timeout can exceed the guest's epoch budget, so + // suspend the ticker across the wait (see `subscribe`). + let _epoch = super::super::HostBlockGuard::enter(); + replies.recv() + } + .map_err(|_| PluginError::Timeout)?; let sample = reply.result().map_err(|e| e.to_string())?; Ok(sample.payload().to_bytes().into_owned()) } @@ -532,3 +585,31 @@ mod service_type_name_tests { ); } } + +#[cfg(test)] +mod guest_timeout_tests { + use super::{MAX_GUEST_TIMEOUT, clamp_guest_timeout}; + use std::time::Duration; + + // This clamp stops a guest-supplied `timeout-ms` from deciding how long the + // host suspends the epoch ticker. That suspension covers every plugin in + // the process. A later edit can delete the clamp in one line. + #[test] + fn a_reasonable_timeout_passes_through() { + assert_eq!(clamp_guest_timeout(2_000), Duration::from_millis(2_000)); + } + + #[test] + fn the_maximum_itself_is_not_clamped() { + assert_eq!( + clamp_guest_timeout(MAX_GUEST_TIMEOUT.as_millis() as u32), + MAX_GUEST_TIMEOUT + ); + } + + #[test] + fn an_absurd_timeout_is_clamped() { + // ~49 days, the worst a u32 can ask for. + assert_eq!(clamp_guest_timeout(u32::MAX), MAX_GUEST_TIMEOUT); + } +} diff --git a/crates/hiroz-union/src/plugin/wasm/mod.rs b/crates/hiroz-union/src/plugin/wasm/mod.rs index 4672f1976..50ed46897 100644 --- a/crates/hiroz-union/src/plugin/wasm/mod.rs +++ b/crates/hiroz-union/src/plugin/wasm/mod.rs @@ -15,6 +15,7 @@ pub use host::web_bindgen::hu::plugin::web_types::{HttpRequest, HttpResponse}; use std::collections::HashMap; use std::path::PathBuf; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, OnceLock}; use std::time::Duration; @@ -200,6 +201,74 @@ fn configured_wasm_engine() -> Result { Engine::new(&engine_config).context("creating WASM engine") } +// ─── Epoch budget vs. blocking host calls ──────────────────────────────────── + +/// Number of host calls that block on I/O for a guest at this moment. +/// +/// Every guest dispatch calls `set_epoch_deadline(30)`. The ticker below +/// increments the epoch every 100 ms. A guest therefore gets about 3 s of +/// *wall clock*. +/// +/// The ticker measures wall clock. It does not measure guest execution. A host +/// call that waits on the network spends the guest's budget while the guest +/// does not run. wasmtime then traps the guest as soon as that call returns. +/// The guest never runs the code that the call gave it. +/// +/// This bound is not theoretical. Schema discovery waits for a live publisher +/// and then queries it. On a cold graph that takes seconds. +static HOST_BLOCKING_CALLS: AtomicUsize = AtomicUsize::new(0); + +/// Suspends the epoch ticker while this guard is alive. Hold one around any +/// host call that blocks on I/O. The host then does not charge the wait to the +/// guest's compute budget. +/// +/// The host runs one engine and one ticker. The suspension therefore applies to +/// every guest. It does not apply only to the guest that called the host. +/// +/// Serialised dispatch makes this acceptable today. The CLI runs one plugin in +/// a sequential loop. The TUI iterates the plugins in order. `hu web` holds one +/// mutex over the plugin vector. Two guests cannot run at once, so no second +/// guest exists to preempt. +/// +/// The serialisation bounds the cost. The length of the wait does not bound it. +/// A reader can easily confuse these two reasons, and only the first one holds. +/// Per-plugin locks or a plugin pool would remove the serialisation. The gap +/// then becomes real, and the per-store shape below is the fix. +/// +/// The waits themselves are bounded. Check where each bound comes from. The two +/// discovery calls use a fixed 2 s constant. The service reply waits use +/// `timeout-ms`, and the guest chooses that value. `clamp_guest_timeout` in +/// `host/ros.rs` caps it. An unclamped `u32` would let a plugin choose how long +/// the host stops preempting plugins. +/// +/// The per-store alternative, for whoever needs it: wasmtime supplies +/// `Store::epoch_deadline_callback`, and host functions already hold +/// `&mut PluginState`, which is the store data. A flag there and a callback that +/// returns `UpdateDeadline::Continue` extend the deadline of the blocked guest +/// alone. Every other guest stays preemptible. That design is better, and it +/// costs more work. This counter behaves the same way while dispatch stays +/// serialised. +pub(crate) struct HostBlockGuard(()); + +impl HostBlockGuard { + pub(crate) fn enter() -> Self { + HOST_BLOCKING_CALLS.fetch_add(1, Ordering::SeqCst); + Self(()) + } +} + +impl Drop for HostBlockGuard { + fn drop(&mut self) { + HOST_BLOCKING_CALLS.fetch_sub(1, Ordering::SeqCst); + } +} + +/// Reports whether the epoch ticker may advance now. This function is separate +/// so that a test can check the rule without an engine. +fn epoch_should_tick() -> bool { + HOST_BLOCKING_CALLS.load(Ordering::SeqCst) == 0 +} + /// Process-wide shared WASM engine (and its single epoch-ticker task). Building /// a fresh engine per `load_plugins` call would spawn a new ticker each time — /// the TUI's `reload_plugins` loops, so tickers (each holding an engine clone) @@ -220,7 +289,9 @@ fn shared_wasm_engine() -> Result { handle.spawn(async move { loop { tokio::time::sleep(Duration::from_millis(100)).await; - ticker_engine.increment_epoch(); + if epoch_should_tick() { + ticker_engine.increment_epoch(); + } } }); } else { @@ -232,7 +303,9 @@ fn shared_wasm_engine() -> Result { .spawn(move || { loop { std::thread::sleep(Duration::from_millis(100)); - ticker_engine.increment_epoch(); + if epoch_should_tick() { + ticker_engine.increment_epoch(); + } } }) { @@ -595,3 +668,32 @@ fn plugin_search_dirs() -> Vec { } dirs } + +#[cfg(test)] +mod tests { + use super::{HostBlockGuard, epoch_should_tick}; + + // This counter is the only thing that stops the ticker from charging + // network waits to the guest's compute budget. A later edit can delete one + // `fetch_add` and remove that protection. No other test finds it. + // + // This test is serial by construction. No other unit test in this crate + // takes the guard, because a host call needs a store. + #[test] + fn host_block_guard_suspends_the_epoch_ticker() { + assert!(epoch_should_tick(), "ticker suspended before any guard"); + { + let _outer = HostBlockGuard::enter(); + assert!(!epoch_should_tick(), "guard did not suspend the ticker"); + { + let _inner = HostBlockGuard::enter(); + assert!(!epoch_should_tick(), "nested guard un-suspended it"); + } + assert!( + !epoch_should_tick(), + "dropping the inner guard resumed ticking while the outer is held" + ); + } + assert!(epoch_should_tick(), "ticker never resumed"); + } +}