fix(hu): stop charging host I/O waits to the guest's epoch budget

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);
+    }
+}

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> {
     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<Engine> {
             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<Engine> {
                 .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<PathBuf> {
     }
     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");
+    }
+}