diff --git a/crates/bridge/src/queue_tracker.rs b/crates/bridge/src/queue_tracker.rs index 7ef9130f9e..d9dd8ed6eb 100644 --- a/crates/bridge/src/queue_tracker.rs +++ b/crates/bridge/src/queue_tracker.rs @@ -82,6 +82,7 @@ pub enum TrackedLimit { PendingWasmSignals, PendingSidecarResponses, OutboundSidecarRequests, + GuestExecutions, VmProcesses, VmOpenFds, VmPipes, @@ -119,6 +120,7 @@ impl TrackedLimit { TrackedLimit::PendingWasmSignals => "pending_wasm_signals", TrackedLimit::PendingSidecarResponses => "pending_sidecar_responses", TrackedLimit::OutboundSidecarRequests => "outbound_sidecar_requests", + TrackedLimit::GuestExecutions => "guest_executions", TrackedLimit::VmProcesses => "vm_processes", TrackedLimit::VmOpenFds => "vm_open_fds", TrackedLimit::VmPipes => "vm_pipes", @@ -154,7 +156,8 @@ impl TrackedLimit { | TrackedLimit::PendingWasmSignals | TrackedLimit::PendingSidecarResponses | TrackedLimit::OutboundSidecarRequests => LimitCategory::Queue, - TrackedLimit::VmProcesses + TrackedLimit::GuestExecutions + | TrackedLimit::VmProcesses | TrackedLimit::VmOpenFds | TrackedLimit::VmPipes | TrackedLimit::VmPtys diff --git a/crates/v8-runtime/src/session.rs b/crates/v8-runtime/src/session.rs index c5139ba709..0651dcc00a 100644 --- a/crates/v8-runtime/src/session.rs +++ b/crates/v8-runtime/src/session.rs @@ -10,7 +10,7 @@ use std::time::{Duration, Instant}; #[cfg(not(test))] use agentos_bridge::queue_tracker::warn_limit_exhausted; -use agentos_bridge::queue_tracker::{register_queue, QueueGauge, TrackedLimit}; +use agentos_bridge::queue_tracker::{register_limit, register_queue, QueueGauge, TrackedLimit}; use agentos_bridge::{bridge_contract, BridgeCallConvention}; use agentos_runtime::accounting::{Reservation, ResourceClass, ResourceLedger}; use agentos_runtime::metrics::{ExecutorMetricClass, RuntimeMetrics}; @@ -1043,8 +1043,24 @@ impl SessionShutdown { } } -/// Concurrency slot tracker shared across session threads -type SlotControl = Arc<(Mutex, Condvar)>; +/// Concurrency slot tracker shared across session threads. +struct SlotControlState { + active: Mutex, + cvar: Condvar, + gauge: Arc, +} + +impl SlotControlState { + fn new(maximum: usize) -> Self { + Self { + active: Mutex::new(0), + cvar: Condvar::new(), + gauge: register_limit(TrackedLimit::GuestExecutions, maximum), + } + } +} + +type SlotControl = Arc; /// An admitted V8 executor slot. It is acquired before spawning or assigning /// an OS thread and remains owned by that generation until the thread exits. @@ -1061,8 +1077,8 @@ impl SessionSlotPermit { maximum: usize, metrics: RuntimeMetrics, ) -> Result { - let (lock, _) = &**control; - let mut active = lock + let mut active = control + .active .lock() .map_err(|_| String::from("ERR_AGENTOS_VM_EXECUTOR_POISONED: slot lock poisoned"))?; if *active >= maximum { @@ -1071,6 +1087,7 @@ impl SessionSlotPermit { )); } *active += 1; + control.gauge.observe_depth(*active); metrics.observe_executor(ExecutorMetricClass::Vm, *active, 0); Ok(Self { control: Arc::clone(control), @@ -1081,13 +1098,13 @@ impl SessionSlotPermit { impl Drop for SessionSlotPermit { fn drop(&mut self) { - let (lock, cvar) = &*self.control; - match lock.lock() { + match self.control.active.lock() { Ok(mut active) if *active > 0 => { *active -= 1; + self.control.gauge.observe_depth(*active); self.metrics .observe_executor(ExecutorMetricClass::Vm, *active, 0); - cvar.notify_all(); + self.control.cvar.notify_all(); } Ok(_) => eprintln!( "ERR_AGENTOS_VM_EXECUTOR_ACCOUNTING_UNDERFLOW: executor permit released at zero" @@ -1261,7 +1278,7 @@ impl SessionManager { sessions: HashMap::new(), quarantined: Vec::new(), max_concurrency, - slot_control: Arc::new((Mutex::new(0), Condvar::new())), + slot_control: Arc::new(SlotControlState::new(max_concurrency)), event_tx: event_tx.into(), call_id_router, shared_call_id: Arc::new(AtomicU64::new(1)), @@ -2010,8 +2027,7 @@ impl SessionManager { /// Number of sessions that have acquired a concurrency slot. #[allow(dead_code)] pub fn active_slot_count(&self) -> usize { - let (lock, _) = &*self.slot_control; - *lock.lock().unwrap() + *self.slot_control.active.lock().unwrap() } pub fn session_output_generation(&self, session_id: &str) -> Option { @@ -3956,7 +3972,7 @@ mod tests { #[test] fn vm_executor_permits_report_active_and_high_water_metrics() { - let control: SlotControl = Arc::new((Mutex::new(0), Condvar::new())); + let control = Arc::new(SlotControlState::new(2)); let metrics = RuntimeMetrics::new(); let first = SessionSlotPermit::try_acquire(&control, 2, metrics.clone()) @@ -3981,6 +3997,80 @@ mod tests { assert_eq!(released.high_water, 2); } + #[test] + fn guest_execution_slots_are_tracked_and_warn_near_capacity() { + const SUBPROCESS_ENV: &str = "AGENTOS_GUEST_EXECUTION_SLOT_GAUGE_SUBPROCESS"; + if std::env::var_os(SUBPROCESS_ENV).is_none() { + let test_name = + "session::tests::guest_execution_slots_are_tracked_and_warn_near_capacity"; + let output = + std::process::Command::new(std::env::current_exe().expect("current test binary")) + .arg(test_name) + .arg("--exact") + .arg("--nocapture") + .env(SUBPROCESS_ENV, "1") + .output() + .unwrap_or_else(|error| panic!("spawn isolated test {test_name}: {error}")); + assert!( + output.status.success(), + "isolated test {test_name} failed with status {:?}\nstdout:\n{}\nstderr:\n{}", + output.status.code(), + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + ); + return; + } + + let warnings = Arc::new(Mutex::new(Vec::new())); + let warning_sink = Arc::clone(&warnings); + agentos_bridge::queue_tracker::set_limit_warning_handler(Box::new(move |warning| { + if warning.name == TrackedLimit::GuestExecutions { + warning_sink + .lock() + .expect("warning sink mutex") + .push(warning.clone()); + } + })); + + let control = Arc::new(SlotControlState::new(5)); + let metrics = RuntimeMetrics::new(); + let mut permits = Vec::new(); + for active in 1..=5 { + permits.push( + SessionSlotPermit::try_acquire(&control, 5, metrics.clone()) + .expect("acquire guest execution slot"), + ); + assert_eq!(control.gauge.depth(), active); + } + + assert_eq!(control.gauge.depth(), 5); + assert_eq!(control.gauge.high_water(), 5); + assert_eq!(control.gauge.capacity(), 5); + assert!(agentos_bridge::queue_tracker::queue_snapshot() + .iter() + .any(|stat| { + stat.name == TrackedLimit::GuestExecutions + && stat.depth == 5 + && stat.high_water == 5 + && stat.capacity == 5 + })); + let warnings = warnings.lock().expect("warning sink mutex"); + assert_eq!(warnings.len(), 1); + assert_eq!(warnings[0].observed, 4); + assert_eq!(warnings[0].capacity, 5); + drop(warnings); + + let rejection = SessionSlotPermit::try_acquire(&control, 5, metrics) + .err() + .expect("hard guest execution limit must reject"); + assert!(rejection.contains("ERR_AGENTOS_VM_EXECUTOR_LIMIT")); + assert_eq!(control.gauge.depth(), 5); + + drop(permits); + assert_eq!(control.gauge.depth(), 0); + assert_eq!(control.gauge.high_water(), 5); + } + #[test] fn configured_executor_and_command_bounds_drive_session_manager() { const SUBPROCESS_ENV: &str = "AGENTOS_V8_CONFIGURED_SESSION_MANAGER_SUBPROCESS"; diff --git a/docs/content/docs/resource-limits.mdx b/docs/content/docs/resource-limits.mdx index 98f53eb568..93503369ac 100644 --- a/docs/content/docs/resource-limits.mdx +++ b/docs/content/docs/resource-limits.mdx @@ -98,6 +98,9 @@ bounded queues are tracked in a central limit registry that: - **Warns before the limit is hit.** As usage crosses ~80% of a cap, the runtime emits a structured warning (once per crossing, re-armed only after it recovers), so a slow consumer or a runaway guest is visible *before* it fails. +- **Tracks process-wide guest execution admission.** The `guest_executions` + resource gauge reports the active count, high-water mark, and configured + capacity for the bounded V8 executor slots. - **Never drops data silently.** Internal queues either apply backpressure or fail with a typed error naming the exhausted limit and the setting used to raise it. A rejected event is not popped and forgotten.