From 19ceff01c0055e6ffa6d298aa0cf60bec7d1348f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 18 Aug 2026 02:37:44 +0000 Subject: [PATCH] Fix hung WASM wedging the host and reap OS children on chat stop. Drop the kernel write lock across WASM invoke and audit JSONL I/O so one stuck skill cannot freeze mediate/charge/court. Dispatch timeout now increments the wasmtime epoch to trap the guest; stdio uses the same 30s bound. Chat stop (llm_release_by_process) and end_process SIGKILL isolation_spawn_os children without waiting under the lock. Retarget orch_window_force to the production 6/8 window so kernel-ci can pass. Co-authored-by: wu1w --- crates/tevarn-kernel-host/src/main.rs | 87 ++++++++---- crates/tevarn-kernel/src/audit.rs | 173 ++++++++++++++++------- crates/tevarn-kernel/src/isolation.rs | 78 +++++++++- crates/tevarn-kernel/src/kernel.rs | 149 ++++++++++++++++++- crates/tevarn-kernel/src/loop_guard.rs | 16 ++- crates/tevarn-kernel/src/wasm_runtime.rs | 151 +++++++++++++++++++- 6 files changed, 559 insertions(+), 95 deletions(-) diff --git a/crates/tevarn-kernel-host/src/main.rs b/crates/tevarn-kernel-host/src/main.rs index 83a137f4..d8b5411b 100644 --- a/crates/tevarn-kernel-host/src/main.rs +++ b/crates/tevarn-kernel-host/src/main.rs @@ -2785,36 +2785,7 @@ async fn handle_connection(runtime: Arc, stream: TcpStream) { // clients (UI panel 500s / ping timeout) when workers blocked on locks. // Also: if the blocking pool saturates while every worker awaits // spawn_blocking, accept() never runs → host looks "up" but dead. - let rt = runtime.clone(); - let resp = match tokio::time::timeout( - DISPATCH_TIMEOUT, - tokio::task::spawn_blocking(move || dispatch(&rt, &line)), - ) - .await - { - Ok(Ok(v)) => v, - Ok(Err(e)) => { - warn!("dispatch join failed: {e}"); - err_resp( - serde_json::Value::Null, - -32603, - format!("dispatch join: {e}"), - None, - ) - } - Err(_) => { - warn!("dispatch timed out after {DISPATCH_TIMEOUT:?}"); - err_resp( - serde_json::Value::Null, - -32603, - format!( - "dispatch timeout after {}s (kernel busy / lock)", - DISPATCH_TIMEOUT.as_secs() - ), - None, - ) - } - }; + let resp = dispatch_with_timeout(runtime.clone(), line).await; let mut out = serde_json::to_string(&resp).unwrap_or_else(|_| { r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"serialize"}}"# .into() @@ -2842,6 +2813,55 @@ async fn run_tcp(runtime: Arc, addr: SocketAddr) -> anyhow::Result<()> } } +async fn isolation_reap_loop(runtime: Arc) { + // Poll exited children even when the Python dispatcher is idle. + // max_age stays 600s so a long-running intended child is not killed early. + let mut interval = tokio::time::interval(Duration::from_secs(5)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + let rt = runtime.clone(); + let _ = tokio::task::spawn_blocking(move || { + rt.kernel().isolation_reap(Some(600.0)); + }) + .await; + } +} + +async fn dispatch_with_timeout(runtime: Arc, line: String) -> Value { + let rt = runtime.clone(); + match tokio::time::timeout( + DISPATCH_TIMEOUT, + tokio::task::spawn_blocking(move || dispatch(&rt, &line)), + ) + .await + { + Ok(Ok(v)) => v, + Ok(Err(e)) => { + warn!("dispatch join failed: {e}"); + err_resp( + serde_json::Value::Null, + -32603, + format!("dispatch join: {e}"), + None, + ) + } + Err(_) => { + warn!("dispatch timed out after {DISPATCH_TIMEOUT:?}"); + runtime.kernel().wasm_interrupt(); + err_resp( + serde_json::Value::Null, + -32603, + format!( + "dispatch timeout after {}s (kernel busy / lock)", + DISPATCH_TIMEOUT.as_secs() + ), + None, + ) + } + } +} + async fn run_stdio(runtime: Arc) -> anyhow::Result<()> { use tokio::io::stdin; info!("tevarn-kernel-host stdio mode"); @@ -2852,7 +2872,7 @@ async fn run_stdio(runtime: Arc) -> anyhow::Result<()> { if line.is_empty() { continue; } - let resp = dispatch(&runtime, &line); + let resp = dispatch_with_timeout(runtime.clone(), line).await; let mut out = serde_json::to_string(&resp).unwrap_or_default(); out.push('\n'); stdout.write_all(out.as_bytes()).await?; @@ -2893,6 +2913,11 @@ async fn async_main() -> anyhow::Result<()> { DISPATCH_TIMEOUT.as_secs() ); + let reap_rt = runtime.clone(); + tokio::spawn(async move { + isolation_reap_loop(reap_rt).await; + }); + if args.stdio { run_stdio(runtime).await } else { diff --git a/crates/tevarn-kernel/src/audit.rs b/crates/tevarn-kernel/src/audit.rs index 046e3684..2a10f0c0 100644 --- a/crates/tevarn-kernel/src/audit.rs +++ b/crates/tevarn-kernel/src/audit.rs @@ -3,9 +3,10 @@ use std::fs::{self, File, OpenOptions}; use std::io::{BufRead, BufReader, Write}; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Mutex; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{self, SyncSender}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -126,23 +127,38 @@ impl KernelEvent { } } -/// Append-only JSONL audit store with size-based rotation + optional WORM. -/// -/// WORM (`TEVARN_AUDIT_WORM=1`): rotated segments are never deleted; new -/// segments get monotonic names `.worm.`. External anchor file holds -/// signed tip hash for offline verification. -pub struct AuditEventStore { +enum AuditMsg { + Event(Value), + Flush(SyncSender<()>), +} + +/// Disk + rotation state. Shared with a dedicated writer thread so the +/// kernel write lock is never held across audit I/O. +struct AuditDisk { path: PathBuf, lock: Mutex<()>, max_bytes: u64, keep_segments: u32, - worm: bool, + worm: AtomicBool, anchor_path: PathBuf, // audit-fix: anchor 降频状态（事件计数 / 上次 anchor 时刻 epoch millis） anchor_events: AtomicU64, anchor_last_ms: AtomicU64, } +/// Append-only JSONL audit store with size-based rotation + optional WORM. +/// +/// WORM (`TEVARN_AUDIT_WORM=1`): rotated segments are never deleted; new +/// segments get monotonic names `.worm.`. External anchor file holds +/// signed tip hash for offline verification. +/// +/// `append` is non-blocking (bounded queue). A stuck disk cannot freeze +/// mediate / charge / court. Fail-open: a full queue drops the line. +pub struct AuditEventStore { + disk: Arc, + tx: SyncSender, +} + impl AuditEventStore { pub fn new(path: impl Into) -> Self { let path = path.into(); @@ -162,16 +178,33 @@ impl AuditEventStore { }) .unwrap_or(false); let anchor_path = path.with_extension("anchor.json"); - Self { + let disk = Arc::new(AuditDisk { path, lock: Mutex::new(()), max_bytes, keep_segments, - worm, + worm: AtomicBool::new(worm), anchor_path, anchor_events: AtomicU64::new(0), anchor_last_ms: AtomicU64::new(0), - } + }); + let (tx, rx) = mpsc::sync_channel::(512); + let writer = disk.clone(); + let _ = std::thread::Builder::new() + .name("tevarn-audit".into()) + .spawn(move || { + while let Ok(msg) = rx.recv() { + match msg { + AuditMsg::Event(v) => { + writer.write_event(&v); + } + AuditMsg::Flush(ack) => { + let _ = ack.send(()); + } + } + } + }); + Self { disk, tx } } pub fn default_path() -> PathBuf { @@ -179,17 +212,86 @@ impl AuditEventStore { } pub fn path(&self) -> &Path { - &self.path + &self.disk.path } pub fn worm(&self) -> bool { - self.worm + self.disk.worm.load(Ordering::Relaxed) } pub fn set_worm(&mut self, on: bool) { - self.worm = on; + self.disk.worm.store(on, Ordering::Relaxed); + } + + /// Block until queued events are written (verify / tests). Never used + /// on the mediate hot path. + pub fn flush(&self) { + let (ack_tx, ack_rx) = mpsc::sync_channel(1); + if self.tx.send(AuditMsg::Flush(ack_tx)).is_ok() { + let _ = ack_rx.recv_timeout(Duration::from_secs(2)); + } + } + + /// Enqueue an event. Never blocks the kernel lock on disk I/O. + /// Fail-open: a full or dead queue drops the line (or writes inline). + pub fn append(&self, event: &Value) -> bool { + match self.tx.try_send(AuditMsg::Event(event.clone())) { + Ok(()) => true, + Err(mpsc::TrySendError::Full(_)) => { + tracing::warn!("audit queue full; dropping event (fail-open)"); + false + } + Err(mpsc::TrySendError::Disconnected(_)) => self.disk.write_event(event), + } + } + + /// External anchor: tip hash + monotonic seq file for offline integrity checks. + pub fn write_anchor(&self, tip_hash: &str, prev_hash: Option<&str>) -> bool { + self.disk.write_anchor(tip_hash, prev_hash) } + pub fn read_anchor(&self) -> Option { + self.flush(); + self.disk.read_anchor() + } + + /// Verify anchor tip matches active file tail hash. + pub fn verify_anchor(&self) -> Value { + self.flush(); + let tail = self.disk.load_tail_hash(); + let anchor = self.disk.read_anchor(); + let tip = anchor + .as_ref() + .and_then(|a| a.get("tip_hash")) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + let ok = match (&tail, &tip) { + (Some(t), Some(a)) => t == a, + (None, None) => true, + _ => false, + }; + serde_json::json!({ + "ok": ok, + "worm": self.worm(), + "tail_hash": tail, + "anchor_tip": tip, + "anchor_path": self.disk.anchor_path.display().to_string(), + "audit_path": self.disk.path.display().to_string(), + }) + } + + pub fn load_tail_hash(&self) -> Option { + self.flush(); + self.disk.load_tail_hash() + } + + pub fn verify_file_chain(&self) -> (bool, i64) { + self.flush(); + self.disk.verify_file_chain() + } +} + +impl AuditDisk { fn rotate_if_needed(&self) { let meta = match fs::metadata(&self.path) { Ok(m) => m, @@ -199,7 +301,7 @@ impl AuditEventStore { return; } let base = self.path.to_string_lossy().to_string(); - if self.worm { + if self.worm.load(Ordering::Relaxed) { // WORM: never delete; seal active file under unique name let sealed = format!( "{base}.worm.{}", @@ -227,7 +329,7 @@ impl AuditEventStore { let _ = fs::rename(&self.path, &rotated); } - pub fn append(&self, event: &Value) -> bool { + fn write_event(&self, event: &Value) -> bool { let _g = self.lock.lock().unwrap_or_else(|e| e.into_inner()); if let Some(parent) = self.path.parent() { let _ = fs::create_dir_all(parent); @@ -260,13 +362,12 @@ impl AuditEventStore { true } - /// External anchor: tip hash + monotonic seq file for offline integrity checks. - pub fn write_anchor(&self, tip_hash: &str, prev_hash: Option<&str>) -> bool { + fn write_anchor(&self, tip_hash: &str, prev_hash: Option<&str>) -> bool { let body = serde_json::json!({ "tip_hash": tip_hash, "prev_hash": prev_hash.unwrap_or(""), "path": self.path.display().to_string(), - "worm": self.worm, + "worm": self.worm.load(Ordering::Relaxed), "anchored_at": SystemTime::now() .duration_since(UNIX_EPOCH) .map(|d| d.as_secs_f64()) @@ -296,36 +397,12 @@ impl AuditEventStore { .is_ok() } - pub fn read_anchor(&self) -> Option { + fn read_anchor(&self) -> Option { let s = fs::read_to_string(&self.anchor_path).ok()?; serde_json::from_str(&s).ok() } - /// Verify anchor tip matches active file tail hash. - pub fn verify_anchor(&self) -> Value { - let tail = self.load_tail_hash(); - let anchor = self.read_anchor(); - let tip = anchor - .as_ref() - .and_then(|a| a.get("tip_hash")) - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); - let ok = match (&tail, &tip) { - (Some(t), Some(a)) => t == a, - (None, None) => true, - _ => false, - }; - serde_json::json!({ - "ok": ok, - "worm": self.worm, - "tail_hash": tail, - "anchor_tip": tip, - "anchor_path": self.anchor_path.display().to_string(), - "audit_path": self.path.display().to_string(), - }) - } - - pub fn load_tail_hash(&self) -> Option { + fn load_tail_hash(&self) -> Option { let f = File::open(&self.path).ok()?; let reader = BufReader::new(f); let mut last: Option = None; @@ -343,7 +420,7 @@ impl AuditEventStore { last } - pub fn verify_file_chain(&self) -> (bool, i64) { + fn verify_file_chain(&self) -> (bool, i64) { let Ok(f) = File::open(&self.path) else { return (false, 0); }; diff --git a/crates/tevarn-kernel/src/isolation.rs b/crates/tevarn-kernel/src/isolation.rs index adfecdfb..cee3c5a1 100644 --- a/crates/tevarn-kernel/src/isolation.rs +++ b/crates/tevarn-kernel/src/isolation.rs @@ -434,14 +434,21 @@ impl IsolationSupervisor { } /// Kill OS child (if owned) then mark ledger killed. + /// + /// Does not `wait()` on the calling thread: a wedged child must not + /// hold the kernel write lock (chat stop / end_process). pub fn kill(&mut self, handle_id: &str) -> Option { if let Some(mut child) = self.children.remove(handle_id) { - // audit-fix: unix 下先整组 SIGKILL（pgid==child pid），再 kill+wait - // 回收组长本体，防孙进程泄漏 + // unix: 先整组 SIGKILL（pgid==child pid），再 kill；wait 放到 + // 分离线程，防 D 状态/僵尸回收卡住 host。 #[cfg(unix)] Self::kill_process_group(child.id()); let _ = child.kill(); - let _ = child.wait(); + let _ = std::thread::Builder::new() + .name("iso-wait".into()) + .spawn(move || { + let _ = child.wait(); + }); } else if let Some(h) = self.handles.get(handle_id) { if h.status == "running" { if let Some(pid) = h.os_pid { @@ -681,4 +688,69 @@ mod tests { assert_eq!(completed.exit_code, Some(0)); } + fn pid_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + #[cfg(windows)] + { + Command::new("tasklist") + .args(["/FI", &format!("PID eq {pid}"), "/NH"]) + .stdin(Stdio::null()) + .output() + .map(|o| String::from_utf8_lossy(&o.stdout).contains(&pid.to_string())) + .unwrap_or(false) + } + #[cfg(not(windows))] + { + Command::new("kill") + .args(["-0", &pid.to_string()]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .map(|st| st.success()) + .unwrap_or(false) + } + } + + #[test] + fn drop_process_kills_os_child() { + let mut s = IsolationSupervisor::new(); + s.set_process_profile("p1", IsolationProfile::Off); + #[cfg(windows)] + let cmd = "ping -n 30 127.0.0.1"; + #[cfg(not(windows))] + let cmd = "sleep 30"; + let h = s.spawn_os("p1", cmd, "os").expect("spawn"); + let pid = h.os_pid.expect("os pid"); + assert!(pid_alive(pid), "child should be running"); + s.drop_process("p1"); + let mut dead = false; + for _ in 0..40 { + std::thread::sleep(std::time::Duration::from_millis(50)); + if !pid_alive(pid) { + dead = true; + break; + } + } + assert!(dead, "drop_process must SIGKILL the OS child"); + assert_eq!(s.status()["os_children"], 0); + } + + #[test] + fn kill_returns_without_waiting_on_child() { + let mut s = IsolationSupervisor::new(); + s.set_process_profile("p1", IsolationProfile::Off); + #[cfg(windows)] + let cmd = "ping -n 30 127.0.0.1"; + #[cfg(not(windows))] + let cmd = "sleep 30"; + let h = s.spawn_os("p1", cmd, "os").expect("spawn"); + let t0 = std::time::Instant::now(); + let killed = s.kill(&h.id).expect("kill"); + assert!(t0.elapsed() < std::time::Duration::from_secs(2)); + assert_eq!(killed.status, "killed"); + } + } diff --git a/crates/tevarn-kernel/src/kernel.rs b/crates/tevarn-kernel/src/kernel.rs index 1d3e2595..2cd9abe3 100644 --- a/crates/tevarn-kernel/src/kernel.rs +++ b/crates/tevarn-kernel/src/kernel.rs @@ -1699,6 +1699,9 @@ impl AgentKernel { pub fn llm_release_by_process(&self, process_id: &str) -> usize { let mut g = self.inner.write(); let n = g.llm.release_by_process(process_id); + // Chat stop calls this RPC (not end_process). Reap isolation OS + // children here so Stop does not wait for the 600s dispatcher tick. + g.isolation.drop_process(process_id); if n > 0 { Self::emit_locked( &mut g, @@ -4050,11 +4053,26 @@ impl AgentKernel { entry: &str, params: Value, ) -> KernelResult { - let mut g = self.inner.write(); - match g.wasm.invoke(module_id, entry, ¶ms) { - Ok(r) => Ok(json!(r)), - Err(e) => Err(KernelError::Invalid(e)), + // Snapshot under a short lock, then run the guest with the kernel + // lock released so a hung skill cannot freeze mediate/charge/court. + let job = { + let g = self.inner.read(); + g.wasm + .prepare_invoke(module_id) + .map_err(KernelError::Invalid)? + }; + let (result, effects) = job.run(entry, ¶ms); + { + let mut g = self.inner.write(); + g.wasm.apply_invoke_effects(effects); } + Ok(json!(result)) + } + + /// Trap in-flight WASM guests (host dispatch timeout). Does not take + /// the write lock and does not introduce a user-facing gate. + pub fn wasm_interrupt(&self) { + self.inner.read().wasm.increment_epoch(); } pub fn wasm_unload(&self, module_id: &str) -> KernelResult { @@ -5003,4 +5021,127 @@ mod tests { let cleared = kernel.decide_tool("file_write", Some(&args), Some(&p.id), None, None); assert_eq!(cleared.verdict, "ask"); } + + #[test] + fn wasm_invoke_does_not_hold_kernel_lock() { + use std::sync::Arc; + let kernel = Arc::new(k()); + let wat = br#"(module + (func (export "main") + (loop $l (br $l))) + )"#; + // Modest fuel so the guest cannot outlive the test if epoch races. + let m = kernel + .wasm_load( + "spin", + std::str::from_utf8(wat).unwrap(), + Some(2_000_000), + Some(2), + ) + .unwrap(); + let mid = m["id"].as_str().unwrap().to_string(); + kernel.wasm_activate(&mid).unwrap(); + let k_run = kernel.clone(); + let mid_run = mid.clone(); + let invoke = std::thread::spawn(move || k_run.wasm_invoke(&mid_run, "main", json!({}))); + // While the guest spins, mediate / charge must still proceed. + let p = kernel + .create_process( + "main", + None, + None, + Some(vec!["file_read".into()]), + Some(100), + None, + ) + .unwrap(); + let t0 = std::time::Instant::now(); + assert!(kernel.mediate(&p.id, "tool_call", "file_read", None).is_ok()); + assert!(t0.elapsed() < std::time::Duration::from_secs(2)); + kernel.wasm_interrupt(); + std::thread::sleep(std::time::Duration::from_millis(20)); + kernel.wasm_interrupt(); + let _ = invoke.join(); + } + + fn isolation_pid_alive(pid: u32) -> bool { + if pid == 0 { + return false; + } + std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .map(|st| st.success()) + .unwrap_or(false) + } + + #[cfg(unix)] + #[test] + fn end_process_reaps_os_children() { + let kernel = k(); + let p = kernel + .create_process( + "main", + None, + None, + Some(vec!["terminal".into()]), + None, + None, + ) + .unwrap(); + kernel.isolation_set_profile(&p.id, "off"); + let h = kernel + .isolation_spawn_os(&p.id, "sleep 30", Some("os")) + .unwrap(); + let pid = h["os_pid"].as_u64().unwrap() as u32; + assert!(isolation_pid_alive(pid)); + kernel + .end_process(&p.id, "killed", Some("chat_stop")) + .unwrap(); + let mut dead = false; + for _ in 0..40 { + std::thread::sleep(std::time::Duration::from_millis(50)); + if !isolation_pid_alive(pid) { + dead = true; + break; + } + } + assert!(dead, "end_process must SIGKILL isolation_spawn_os children"); + } + + #[cfg(unix)] + #[test] + fn llm_release_by_process_reaps_os_children() { + // Chat stop calls llm_release_by_process, not end_process. + let kernel = k(); + let p = kernel + .create_process( + "main", + None, + None, + Some(vec!["terminal".into()]), + None, + None, + ) + .unwrap(); + kernel.isolation_set_profile(&p.id, "off"); + let h = kernel + .isolation_spawn_os(&p.id, "sleep 30", Some("os")) + .unwrap(); + let pid = h["os_pid"].as_u64().unwrap() as u32; + assert!(isolation_pid_alive(pid)); + let _ = kernel.llm_release_by_process(&p.id); + let mut dead = false; + for _ in 0..40 { + std::thread::sleep(std::time::Duration::from_millis(50)); + if !isolation_pid_alive(pid) { + dead = true; + break; + } + } + assert!(dead, "chat-stop llm_release_by_process must SIGKILL OS children"); + } } diff --git a/crates/tevarn-kernel/src/loop_guard.rs b/crates/tevarn-kernel/src/loop_guard.rs index 7153b686..a9673991 100644 --- a/crates/tevarn-kernel/src/loop_guard.rs +++ b/crates/tevarn-kernel/src/loop_guard.rs @@ -819,16 +819,20 @@ mod tests { "p1", LoopGuardConfig::for_role(false, RoleKind::Steward, None), ); - for _ in 0..3 { + // Production window: ≥6 orch-heavy of last 8 (relaxed from 3/5 so + // multi-round hire/dispatch does not 熔断 a long chat). + for i in 0..5 { let d = g.begin_round( "p1", &["crew_steward".into(), "crew_steward".into()], ); - if matches!(d, GuardDecision::ForceFinal { .. }) { - return; - } + assert!( + matches!(d, GuardDecision::Allow { .. }), + "round {} must stay allow under 6/8 window, got {:?}", + i + 1, + d.to_dict() + ); } - // 3 orch rounds should trip let d = g.begin_round("p1", &["crew_steward".into()]); assert!( matches!( @@ -836,7 +840,7 @@ mod tests { GuardDecision::ForceFinal { code: ref c, .. - } if c == "orch_window_thrash" || c == "max_tool_rounds" + } if c == "orch_window_thrash" ), "got {:?}", d.to_dict() diff --git a/crates/tevarn-kernel/src/wasm_runtime.rs b/crates/tevarn-kernel/src/wasm_runtime.rs index 17325db2..286cbe90 100644 --- a/crates/tevarn-kernel/src/wasm_runtime.rs +++ b/crates/tevarn-kernel/src/wasm_runtime.rs @@ -151,13 +151,21 @@ fn make_engine() -> Engine { let mut config = Config::new(); // Fuel: instruction metering hard stop let _ = config.consume_fuel(true); - // Epoch is optional; fuel is primary budget + // Epoch: host dispatch timeout increments this to trap a hung guest + // without holding the kernel write lock for the whole invoke. + let _ = config.epoch_interruption(true); let _ = config.wasm_bulk_memory(true); let _ = config.wasm_multi_value(true); // Cranelift is default compiler backend Engine::new(&config).unwrap_or_else(|e| { - tracing::error!("wasmtime Engine::new failed ({e}); using Engine::default()"); - Engine::default() + tracing::error!("wasmtime Engine::new failed ({e}); retrying configured engine"); + let mut retry = Config::new(); + let _ = retry.consume_fuel(true); + let _ = retry.epoch_interruption(true); + Engine::new(&retry).unwrap_or_else(|e2| { + tracing::error!("wasmtime retry failed ({e2}); using Engine::default()"); + Engine::default() + }) }) } @@ -200,6 +208,71 @@ pub struct WasmInvokeResult { pub engine: String, } +/// Snapshot of one invoke so the kernel lock can be dropped while the guest runs. +pub struct PreparedInvoke { + engine: Engine, + compiled: Option, + meta: WasmModule, + blob: Vec, + memory: Vec, +} + +/// Mutations to apply after an unlocked invoke returns. +#[derive(Debug, Clone)] +pub struct InvokeEffects { + pub module_id: String, + pub status: Option, + pub engine_name: Option, + pub current_pages: Option, + pub memory_bytes_used: Option, + pub memory: Option>, +} + +impl PreparedInvoke { + /// Run wasmtime / hostcall on this snapshot. Does not touch the live runtime. + pub fn run(self, entry: &str, params: &Value) -> (WasmInvokeResult, InvokeEffects) { + let id = self.meta.id.clone(); + let status_before = self.meta.status.clone(); + let mut mini = WasmRuntime { + engine: self.engine, + modules: HashMap::from([(id.clone(), self.meta.clone())]), + blobs: HashMap::from([(id.clone(), self.blob)]), + compiled: self.compiled.into_iter().map(|m| (id.clone(), m)).collect(), + memory: HashMap::from([(id.clone(), self.memory)]), + default_fuel: self.meta.fuel_limit, + default_mem_pages: self.meta.memory_pages_limit, + default_max_ops: self.meta.max_ops, + }; + let result = match mini.invoke(&id, entry, params) { + Ok(r) => r, + Err(e) => WasmInvokeResult { + module_id: id.clone(), + ok: false, + fuel_used: 0, + ops_executed: 0, + max_stack: 0, + hostcalls: vec![], + output: Value::Null, + error: Some(e), + engine: "none".into(), + }, + }; + let after = mini.modules.get(&id).cloned(); + let effects = InvokeEffects { + module_id: id.clone(), + status: after + .as_ref() + .map(|m| m.status.clone()) + .filter(|s| s != &status_before), + engine_name: after.as_ref().map(|m| m.engine.clone()), + current_pages: after.as_ref().map(|m| m.current_pages), + memory_bytes_used: after.as_ref().map(|m| m.memory_bytes_used), + memory: mini.memory.remove(&id), + }; + (result, effects) + } +} + // ── host state for wasmtime ──────────────────────────────── struct HostState { @@ -512,6 +585,52 @@ impl WasmRuntime { Ok(m.clone()) } + /// Bump the engine epoch so any in-flight guest with `set_epoch_deadline` + /// traps. Cheap and lock-free relative to WASM execution. + pub fn increment_epoch(&self) { + self.engine.increment_epoch(); + } + + /// Clone invoke inputs so the caller can drop the kernel write lock. + pub fn prepare_invoke(&self, module_id: &str) -> Result { + let meta = self + .modules + .get(module_id) + .cloned() + .ok_or_else(|| format!("unknown module {module_id}"))?; + if meta.status != "active" && meta.status != "loaded" { + return Err(format!("module status {}", meta.status)); + } + Ok(PreparedInvoke { + engine: self.engine.clone(), + compiled: self.compiled.get(module_id).cloned(), + meta, + blob: self.blobs.get(module_id).cloned().unwrap_or_default(), + memory: self.memory.get(module_id).cloned().unwrap_or_default(), + }) + } + + pub fn apply_invoke_effects(&mut self, effects: InvokeEffects) { + let id = &effects.module_id; + if let Some(mm) = self.modules.get_mut(id) { + if let Some(s) = effects.status { + mm.status = s; + } + if let Some(e) = effects.engine_name { + mm.engine = e; + } + if let Some(p) = effects.current_pages { + mm.current_pages = p; + } + if let Some(b) = effects.memory_bytes_used { + mm.memory_bytes_used = b; + } + } + if let Some(mem) = effects.memory { + self.memory.insert(id.clone(), mem); + } + } + pub fn invoke( &mut self, module_id: &str, @@ -601,6 +720,8 @@ impl WasmRuntime { store .set_fuel(m.fuel_limit) .map_err(|e| format!("set_fuel: {e}"))?; + // Trap when the host increments the shared engine epoch (dispatch timeout). + store.set_epoch_deadline(1); let mut linker = Linker::new(&self.engine); define_env_imports(&mut linker)?; @@ -1145,6 +1266,7 @@ impl WasmRuntime { "wasmtime", "cranelift", "fuel", + "epoch_interruption", "store_limits_memory", "wat", "env.log", @@ -1336,4 +1458,27 @@ mod tests { .unwrap(); assert!(!r.ok); } + + #[test] + fn epoch_increment_traps_infinite_loop() { + let mut rt = WasmRuntime::default(); + let wat = br#"(module + (func (export "main") + (loop $l (br $l))) + )"#; + let m = rt.load("spin", wat, Some(u64::MAX / 4), Some(2)).unwrap(); + assert!(m.wasmtime_ready); + rt.activate(&m.id).unwrap(); + let job = rt.prepare_invoke(&m.id).unwrap(); + let engine = rt.engine.clone(); + let handle = std::thread::spawn(move || job.run("main", &json!({}))); + std::thread::sleep(std::time::Duration::from_millis(30)); + engine.increment_epoch(); + let (r, _) = handle.join().expect("invoke thread"); + assert!(!r.ok, "epoch interrupt must trap the guest: {:?}", r.error); + assert!( + r.error.is_some(), + "trapped invoke should surface an error" + ); + } }