agentos_bridge/
queue_tracker.rs

1//! Centralized bounded-queue usage tracker.
2//!
3//! secure-exec streams guest output through a *chain* of bounded queues: the
4//! V8 -> host event channel, the sidecar stdout/stdin frame queues, and so on.
5//! Each queue applies backpressure when full (it parks the producer until the
6//! consumer drains) rather than crashing, but backpressure is invisible: a slow
7//! host consumer silently stalls a session with nothing in the logs.
8//!
9//! This module gives that whole chain a single, inspectable home:
10//!
11//! * Every bounded queue registers a [`QueueGauge`] (with a stable name and its
12//!   capacity) in a process-global [`QueueRegistry`].
13//! * Producers report depth as they enqueue (either by an exact count for
14//!   manually-tracked queues via [`TrackedSyncSender`], or by sampling the live
15//!   depth of a Tokio channel via [`QueueGauge::observe_depth`]).
16//! * When a queue crosses [`WARN_FILL_PERCENT`] of capacity the gauge emits a
17//!   single `warn!`, so "the consumer is falling behind" shows up *before* the
18//!   queue saturates and backpressure stalls the session. It re-arms once the
19//!   queue drains back below [`REARM_FILL_PERCENT`].
20//! * [`queue_snapshot`] returns the live depth / high-water / capacity of every
21//!   registered queue for debugging or a status endpoint.
22
23use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
24use std::sync::mpsc::{
25    Receiver, RecvError, RecvTimeoutError, SendError, SyncSender, TryRecvError, TrySendError,
26};
27use std::sync::{Arc, Mutex, OnceLock, Weak};
28use std::time::Duration;
29
30/// Fill fraction (percent of capacity) at or above which a queue is considered
31/// "near full" and emits a warning. Edge-triggered so a steadily-full queue logs
32/// once, not on every enqueue.
33pub const WARN_FILL_PERCENT: usize = 80;
34
35/// Fill fraction a near-full queue must drain back below before it will warn
36/// again. The gap to [`WARN_FILL_PERCENT`] provides hysteresis so a queue
37/// hovering at the threshold does not flap.
38pub const REARM_FILL_PERCENT: usize = 50;
39
40/// What class of bounded resource a gauge tracks. Lets a snapshot / a host hook
41/// group and reason about limits beyond just queues.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum LimitCategory {
44    /// A bounded channel/buffer with enqueue/dequeue flow (the default).
45    Queue,
46    /// A saturating resource counter (fds, processes, sockets, bytes in use).
47    Resource,
48    /// A memory/heap envelope.
49    Memory,
50    /// A CPU or wall-clock execution budget.
51    Cpu,
52}
53
54impl LimitCategory {
55    /// Stable lowercase tag for logs and snapshots.
56    pub fn as_str(self) -> &'static str {
57        match self {
58            LimitCategory::Queue => "queue",
59            LimitCategory::Resource => "resource",
60            LimitCategory::Memory => "memory",
61            LimitCategory::Cpu => "cpu",
62        }
63    }
64}
65
66/// Stable catalog of tracked limits that may emit near-capacity or exhaustion
67/// warnings. Keep `website/src/content/docs/docs/features/resource-limits.mdx`
68/// in sync when adding, removing, or renaming variants so host-visible warning
69/// names and the documented constants do not drift.
70#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
71pub enum TrackedLimit {
72    JavascriptEventChannel,
73    V8SessionFrames,
74    SidecarStdinFrames,
75    SidecarStdoutFrames,
76    CompletedSidecarResponses,
77    PendingProcessEvents,
78    PendingProcessEventBytes,
79    PendingExecutionEvents,
80    PendingExecutionEventBytes,
81    PendingKernelStdinBytes,
82    PendingWasmSignals,
83    PendingSidecarResponses,
84    OutboundSidecarRequests,
85    VmProcesses,
86    VmOpenFds,
87    VmPipes,
88    VmPtys,
89    VmSockets,
90    VmConnections,
91    VmSocketBufferedBytes,
92    VmSocketDatagramQueueLen,
93    VmFilesystemBytes,
94    VmInodes,
95    VmRecursiveFsDepth,
96    VmRecursiveFsEntries,
97    V8HeapBytes,
98    V8CpuTimeMs,
99    V8WallClockMs,
100    WasmFuelMs,
101    WasmMemoryBytes,
102}
103
104impl TrackedLimit {
105    /// Stable lowercase tag emitted in logs, snapshots, and host
106    /// `limit_warning` events.
107    pub fn as_str(self) -> &'static str {
108        match self {
109            TrackedLimit::JavascriptEventChannel => "javascript_event_channel",
110            TrackedLimit::V8SessionFrames => "v8_session_frames",
111            TrackedLimit::SidecarStdinFrames => "sidecar_stdin_frames",
112            TrackedLimit::SidecarStdoutFrames => "sidecar_stdout_frames",
113            TrackedLimit::CompletedSidecarResponses => "completed_sidecar_responses",
114            TrackedLimit::PendingProcessEvents => "pending_process_events",
115            TrackedLimit::PendingProcessEventBytes => "pending_process_event_bytes",
116            TrackedLimit::PendingExecutionEvents => "pending_execution_events",
117            TrackedLimit::PendingExecutionEventBytes => "pending_execution_event_bytes",
118            TrackedLimit::PendingKernelStdinBytes => "pending_kernel_stdin_bytes",
119            TrackedLimit::PendingWasmSignals => "pending_wasm_signals",
120            TrackedLimit::PendingSidecarResponses => "pending_sidecar_responses",
121            TrackedLimit::OutboundSidecarRequests => "outbound_sidecar_requests",
122            TrackedLimit::VmProcesses => "vm_processes",
123            TrackedLimit::VmOpenFds => "vm_open_fds",
124            TrackedLimit::VmPipes => "vm_pipes",
125            TrackedLimit::VmPtys => "vm_ptys",
126            TrackedLimit::VmSockets => "vm_sockets",
127            TrackedLimit::VmConnections => "vm_connections",
128            TrackedLimit::VmSocketBufferedBytes => "vm_socket_buffered_bytes",
129            TrackedLimit::VmSocketDatagramQueueLen => "vm_socket_datagram_queue_len",
130            TrackedLimit::VmFilesystemBytes => "vm_filesystem_bytes",
131            TrackedLimit::VmInodes => "vm_inodes",
132            TrackedLimit::VmRecursiveFsDepth => "vm_recursive_fs_depth",
133            TrackedLimit::VmRecursiveFsEntries => "vm_recursive_fs_entries",
134            TrackedLimit::V8HeapBytes => "v8_heap_bytes",
135            TrackedLimit::V8CpuTimeMs => "v8_cpu_time_ms",
136            TrackedLimit::V8WallClockMs => "v8_wall_clock_ms",
137            TrackedLimit::WasmFuelMs => "wasm_fuel_ms",
138            TrackedLimit::WasmMemoryBytes => "wasm_memory_bytes",
139        }
140    }
141
142    pub fn category(self) -> LimitCategory {
143        match self {
144            TrackedLimit::JavascriptEventChannel
145            | TrackedLimit::V8SessionFrames
146            | TrackedLimit::SidecarStdinFrames
147            | TrackedLimit::SidecarStdoutFrames
148            | TrackedLimit::CompletedSidecarResponses
149            | TrackedLimit::PendingProcessEvents
150            | TrackedLimit::PendingProcessEventBytes
151            | TrackedLimit::PendingExecutionEvents
152            | TrackedLimit::PendingExecutionEventBytes
153            | TrackedLimit::PendingKernelStdinBytes
154            | TrackedLimit::PendingWasmSignals
155            | TrackedLimit::PendingSidecarResponses
156            | TrackedLimit::OutboundSidecarRequests => LimitCategory::Queue,
157            TrackedLimit::VmProcesses
158            | TrackedLimit::VmOpenFds
159            | TrackedLimit::VmPipes
160            | TrackedLimit::VmPtys
161            | TrackedLimit::VmSockets
162            | TrackedLimit::VmConnections
163            | TrackedLimit::VmSocketBufferedBytes
164            | TrackedLimit::VmSocketDatagramQueueLen
165            | TrackedLimit::VmFilesystemBytes
166            | TrackedLimit::VmInodes
167            | TrackedLimit::VmRecursiveFsDepth
168            | TrackedLimit::VmRecursiveFsEntries => LimitCategory::Resource,
169            TrackedLimit::V8HeapBytes | TrackedLimit::WasmMemoryBytes => LimitCategory::Memory,
170            TrackedLimit::V8CpuTimeMs | TrackedLimit::V8WallClockMs | TrackedLimit::WasmFuelMs => {
171                LimitCategory::Cpu
172            }
173        }
174    }
175}
176
177/// A near-capacity event for one limit, delivered to the global warning sink at
178/// the same edge as the `tracing::warn!`. This is the structured payload a host
179/// hook (e.g. agentOS `onLimitWarning`) is built from.
180#[derive(Debug, Clone)]
181pub struct LimitWarning {
182    pub name: TrackedLimit,
183    pub category: LimitCategory,
184    pub observed: usize,
185    pub capacity: usize,
186    pub fill_percent: usize,
187}
188
189type LimitWarningHandler = Arc<dyn Fn(&LimitWarning) + Send + Sync>;
190
191fn warning_handler_slot() -> &'static Mutex<Option<LimitWarningHandler>> {
192    static HANDLER: OnceLock<Mutex<Option<LimitWarningHandler>>> = OnceLock::new();
193    HANDLER.get_or_init(|| Mutex::new(None))
194}
195
196/// Install a process-global sink that is invoked on the same edge-triggered,
197/// hysteresis-gated boundary as the `tracing::warn!` whenever a tracked limit
198/// crosses [`WARN_FILL_PERCENT`]. The sidecar uses this to forward limit warnings
199/// to the host as structured events (the `onLimitWarning` hook). The handler must
200/// be cheap and non-blocking; it runs on the producer's thread.
201pub fn set_limit_warning_handler(handler: Box<dyn Fn(&LimitWarning) + Send + Sync>) {
202    if let Ok(mut slot) = warning_handler_slot().lock() {
203        *slot = Some(Arc::from(handler));
204    }
205}
206
207fn dispatch_warning(warning: &LimitWarning) {
208    // Clone the handler Arc out and DROP the registry mutex before invoking it,
209    // so the handler never runs while we hold the global lock. The sink can be
210    // reached while a kernel lock is held (e.g. fd_tables -> warning_mutex ->
211    // handler); keeping the invocation outside the mutex avoids a lock-order
212    // hazard if the handler ever does non-trivial work.
213    let handler = match warning_handler_slot().lock() {
214        Ok(slot) => slot.as_ref().cloned(),
215        Err(_) => None,
216    };
217    if let Some(handler) = handler {
218        handler(warning);
219    }
220}
221
222/// Emit a structured/logged warning for a limit that has already been exhausted.
223/// Use this for runtime caps such as CPU or heap exhaustion where there is no
224/// continuously sampled queue depth to observe before the terminal edge.
225pub fn warn_limit_exhausted(name: TrackedLimit, observed: usize, capacity: usize) {
226    let fill_percent = observed
227        .saturating_mul(100)
228        .checked_div(capacity)
229        .unwrap_or(0);
230    let category = name.category();
231    tracing::warn!(
232        limit = name.as_str(),
233        category = category.as_str(),
234        observed,
235        capacity,
236        fill_percent,
237        "bounded limit exhausted"
238    );
239    dispatch_warning(&LimitWarning {
240        name,
241        category,
242        observed,
243        capacity,
244        fill_percent,
245    });
246}
247
248/// Live usage gauge for a single bounded queue.
249///
250/// Cloneable handles share one gauge through an [`Arc`]; the registry keeps a
251/// [`Weak`] so a gauge is auto-pruned from snapshots once its queue is dropped.
252#[derive(Debug)]
253pub struct QueueGauge {
254    name: TrackedLimit,
255    category: LimitCategory,
256    capacity: usize,
257    depth: AtomicUsize,
258    high_water: AtomicUsize,
259    warned: AtomicBool,
260}
261
262impl QueueGauge {
263    fn new(name: TrackedLimit, capacity: usize, category: LimitCategory) -> Self {
264        Self {
265            name,
266            category,
267            capacity,
268            depth: AtomicUsize::new(0),
269            high_water: AtomicUsize::new(0),
270            warned: AtomicBool::new(false),
271        }
272    }
273
274    /// Stable limit name (used in logs and snapshots).
275    pub fn name(&self) -> TrackedLimit {
276        self.name
277    }
278
279    /// The class of bounded resource this gauge tracks.
280    pub fn category(&self) -> LimitCategory {
281        self.category
282    }
283
284    /// Configured queue capacity (slots). `0` means unbounded / untracked.
285    pub fn capacity(&self) -> usize {
286        self.capacity
287    }
288
289    /// Current observed depth.
290    pub fn depth(&self) -> usize {
291        self.depth.load(Ordering::Acquire)
292    }
293
294    /// Highest depth observed over the gauge's lifetime.
295    pub fn high_water(&self) -> usize {
296        self.high_water.load(Ordering::Acquire)
297    }
298
299    /// Fill fraction (0–100) at the given depth. Saturates rather than dividing
300    /// by zero for untracked (capacity 0) queues.
301    fn fill_percent(&self, depth: usize) -> usize {
302        depth
303            .saturating_mul(100)
304            .checked_div(self.capacity)
305            .unwrap_or(0)
306    }
307
308    /// Record a new depth: refresh the high-water mark and emit an edge-triggered
309    /// near-capacity warning (or recovery debug line).
310    fn evaluate(&self, depth: usize) {
311        self.high_water.fetch_max(depth, Ordering::AcqRel);
312        if self.capacity == 0 {
313            return;
314        }
315        let percent = self.fill_percent(depth);
316        if percent >= WARN_FILL_PERCENT {
317            if !self.warned.swap(true, Ordering::AcqRel) {
318                tracing::warn!(
319                    limit = self.name.as_str(),
320                    category = self.category.as_str(),
321                    observed = depth,
322                    capacity = self.capacity,
323                    fill_percent = percent,
324                    "bounded limit near capacity"
325                );
326                // Same edge as the log: notify the structured warning sink so the
327                // host can surface it (e.g. an onLimitWarning hook). Edge-triggered
328                // + hysteresis keep this from firing more than once per crossing.
329                dispatch_warning(&LimitWarning {
330                    name: self.name,
331                    category: self.category,
332                    observed: depth,
333                    capacity: self.capacity,
334                    fill_percent: percent,
335                });
336            }
337        } else if percent <= REARM_FILL_PERCENT && self.warned.swap(false, Ordering::AcqRel) {
338            tracing::debug!(
339                limit = self.name.as_str(),
340                category = self.category.as_str(),
341                depth,
342                capacity = self.capacity,
343                fill_percent = percent,
344                "bounded limit drained back below threshold"
345            );
346        }
347    }
348
349    /// Report the queue's exact current depth (for queues whose backing channel
350    /// exposes its live length, e.g. a Tokio mpsc via `max_capacity - capacity`).
351    pub fn observe_depth(&self, depth: usize) {
352        self.depth.store(depth, Ordering::Release);
353        self.evaluate(depth);
354    }
355
356    /// Account for one item entering the queue (for manually-tracked queues).
357    pub fn record_enqueue(&self) {
358        let depth = self.depth.fetch_add(1, Ordering::AcqRel) + 1;
359        self.evaluate(depth);
360    }
361
362    /// Account for one item leaving the queue. Saturates at zero so a stray
363    /// dequeue can never underflow the depth counter. Re-evaluates so a gauge
364    /// that latched "warned" while full re-arms once the queue drains back below
365    /// the re-arm threshold, even if the producer has since gone idle.
366    pub fn record_dequeue(&self) {
367        let mut current = self.depth.load(Ordering::Acquire);
368        loop {
369            if current == 0 {
370                return;
371            }
372            match self.depth.compare_exchange_weak(
373                current,
374                current - 1,
375                Ordering::AcqRel,
376                Ordering::Acquire,
377            ) {
378                Ok(_) => {
379                    self.evaluate(current - 1);
380                    break;
381                }
382                Err(actual) => current = actual,
383            }
384        }
385    }
386}
387
388/// Immutable view of a tracked limit's usage, returned by [`queue_snapshot`].
389#[derive(Debug, Clone, PartialEq, Eq)]
390pub struct QueueSnapshot {
391    pub name: TrackedLimit,
392    pub category: LimitCategory,
393    pub depth: usize,
394    pub high_water: usize,
395    pub capacity: usize,
396    pub fill_percent: usize,
397}
398
399/// Process-global registry of every live [`QueueGauge`].
400#[derive(Default)]
401pub struct QueueRegistry {
402    gauges: Mutex<Vec<Weak<QueueGauge>>>,
403}
404
405impl QueueRegistry {
406    /// The shared registry. All `secure-exec` bounded queues register here so
407    /// their usage can be inspected from one place.
408    pub fn global() -> &'static QueueRegistry {
409        static REGISTRY: OnceLock<QueueRegistry> = OnceLock::new();
410        REGISTRY.get_or_init(QueueRegistry::default)
411    }
412
413    /// Register a new bounded limit and return its gauge. Dropping the returned
414    /// `Arc` (and all clones) removes the limit from future snapshots.
415    pub fn register(&self, name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
416        let category = name.category();
417        let gauge = Arc::new(QueueGauge::new(name, capacity, category));
418        let mut gauges = self.gauges.lock().expect("queue registry mutex poisoned");
419        gauges.retain(|weak| weak.strong_count() > 0);
420        gauges.push(Arc::downgrade(&gauge));
421        gauge
422    }
423
424    /// Snapshot the live usage of every registered queue, pruning dead entries.
425    pub fn snapshot(&self) -> Vec<QueueSnapshot> {
426        let mut gauges = self.gauges.lock().expect("queue registry mutex poisoned");
427        gauges.retain(|weak| weak.strong_count() > 0);
428        gauges
429            .iter()
430            .filter_map(Weak::upgrade)
431            .map(|gauge| {
432                let depth = gauge.depth();
433                QueueSnapshot {
434                    name: gauge.name(),
435                    category: gauge.category(),
436                    depth,
437                    high_water: gauge.high_water(),
438                    capacity: gauge.capacity(),
439                    fill_percent: gauge.fill_percent(depth),
440                }
441            })
442            .collect()
443    }
444}
445
446/// Register a bounded queue (the [`LimitCategory::Queue`] case) with the global
447/// registry. Convenience over [`QueueRegistry::global`] + [`QueueRegistry::register`].
448pub fn register_queue(name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
449    debug_assert_eq!(name.category(), LimitCategory::Queue);
450    QueueRegistry::global().register(name, capacity)
451}
452
453/// Register a non-queue bounded limit (a saturating resource or memory envelope)
454/// with the global registry, so it shares the same approach-warning + snapshot
455/// machinery as queues. Observe usage with [`QueueGauge::observe_depth`].
456pub fn register_limit(name: TrackedLimit, capacity: usize) -> Arc<QueueGauge> {
457    QueueRegistry::global().register(name, capacity)
458}
459
460/// Snapshot every registered queue from the global registry.
461pub fn queue_snapshot() -> Vec<QueueSnapshot> {
462    QueueRegistry::global().snapshot()
463}
464
465/// Emit a `debug!` line for every registered queue. Useful for an on-demand dump
466/// of the queue chain when diagnosing a stall.
467pub fn log_queue_snapshot() {
468    for stat in queue_snapshot() {
469        tracing::debug!(
470            limit = stat.name.as_str(),
471            category = stat.category.as_str(),
472            depth = stat.depth,
473            high_water = stat.high_water,
474            capacity = stat.capacity,
475            fill_percent = stat.fill_percent,
476            "limit usage"
477        );
478    }
479}
480
481/// A `std::sync::mpsc::SyncSender` that feeds a [`QueueGauge`] as items flow
482/// through it, so a queue whose backing channel cannot report its own length
483/// still participates in the centralized tracker.
484///
485/// `send` keeps the underlying blocking-backpressure semantics; it just records
486/// the enqueue first so near-capacity warnings fire as the queue fills.
487#[derive(Debug)]
488pub struct TrackedSyncSender<T> {
489    inner: SyncSender<T>,
490    gauge: Arc<QueueGauge>,
491}
492
493impl<T> Clone for TrackedSyncSender<T> {
494    fn clone(&self) -> Self {
495        Self {
496            inner: self.inner.clone(),
497            gauge: Arc::clone(&self.gauge),
498        }
499    }
500}
501
502impl<T> TrackedSyncSender<T> {
503    /// Blocking send: record the enqueue, then hand off to the bounded channel
504    /// (which parks the caller until a slot is free: clean backpressure).
505    pub fn send(&self, value: T) -> Result<(), SendError<T>> {
506        self.gauge.record_enqueue();
507        self.inner.send(value)
508    }
509
510    /// Non-blocking send: record the enqueue only on success. Lets a caller with
511    /// its own deadline poll instead of parking indefinitely on a full queue.
512    pub fn try_send(&self, value: T) -> Result<(), TrySendError<T>> {
513        match self.inner.try_send(value) {
514            Ok(()) => {
515                self.gauge.record_enqueue();
516                Ok(())
517            }
518            Err(error) => Err(error),
519        }
520    }
521
522    /// The gauge backing this sender.
523    pub fn gauge(&self) -> &Arc<QueueGauge> {
524        &self.gauge
525    }
526}
527
528/// Receiver half of a [`tracked_sync_channel`]; records a dequeue for every
529/// item it yields so the gauge depth tracks the real backlog.
530#[derive(Debug)]
531pub struct TrackedReceiver<T> {
532    inner: Receiver<T>,
533    gauge: Arc<QueueGauge>,
534}
535
536impl<T> TrackedReceiver<T> {
537    /// Blocking receive that decrements the gauge for the item it returns.
538    pub fn recv(&self) -> Result<T, RecvError> {
539        let value = self.inner.recv()?;
540        self.gauge.record_dequeue();
541        Ok(value)
542    }
543
544    pub fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError> {
545        let value = self.inner.recv_timeout(timeout)?;
546        self.gauge.record_dequeue();
547        Ok(value)
548    }
549
550    pub fn try_recv(&self) -> Result<T, TryRecvError> {
551        let value = self.inner.try_recv()?;
552        self.gauge.record_dequeue();
553        Ok(value)
554    }
555}
556
557/// Create a bounded `std::sync::mpsc` sync-channel whose depth is tracked by a
558/// registered [`QueueGauge`]. Drop-in for `std::sync::mpsc::sync_channel` plus
559/// centralized usage tracking + near-capacity warnings.
560pub fn tracked_sync_channel<T>(
561    name: TrackedLimit,
562    capacity: usize,
563) -> (TrackedSyncSender<T>, TrackedReceiver<T>) {
564    let (tx, rx) = std::sync::mpsc::sync_channel(capacity);
565    let gauge = register_queue(name, capacity);
566    (
567        TrackedSyncSender {
568            inner: tx,
569            gauge: Arc::clone(&gauge),
570        },
571        TrackedReceiver { inner: rx, gauge },
572    )
573}
574
575#[cfg(test)]
576mod tests {
577    use super::*;
578
579    fn warning_handler_test_lock() -> &'static Mutex<()> {
580        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
581        LOCK.get_or_init(|| Mutex::new(()))
582    }
583
584    #[test]
585    fn gauge_tracks_depth_and_high_water() {
586        let gauge = QueueGauge::new(
587            TrackedLimit::JavascriptEventChannel,
588            10,
589            LimitCategory::Queue,
590        );
591        assert_eq!(gauge.depth(), 0);
592        gauge.record_enqueue();
593        gauge.record_enqueue();
594        assert_eq!(gauge.depth(), 2);
595        assert_eq!(gauge.high_water(), 2);
596        gauge.record_dequeue();
597        assert_eq!(gauge.depth(), 1);
598        // High-water never regresses.
599        assert_eq!(gauge.high_water(), 2);
600        // Dequeue never underflows below zero.
601        gauge.record_dequeue();
602        gauge.record_dequeue();
603        assert_eq!(gauge.depth(), 0);
604    }
605
606    #[test]
607    fn gauge_warn_flag_is_edge_triggered_with_hysteresis() {
608        let gauge = QueueGauge::new(TrackedLimit::V8SessionFrames, 10, LimitCategory::Queue);
609        // Below 80%: not warned.
610        gauge.observe_depth(7);
611        assert!(!gauge.warned.load(Ordering::Acquire));
612        // Cross 80%: warned.
613        gauge.observe_depth(8);
614        assert!(gauge.warned.load(Ordering::Acquire));
615        // Still near full: stays armed (single warning, not re-fired).
616        gauge.observe_depth(9);
617        assert!(gauge.warned.load(Ordering::Acquire));
618        // Drain to <=50%: re-arms.
619        gauge.observe_depth(5);
620        assert!(!gauge.warned.load(Ordering::Acquire));
621    }
622
623    #[test]
624    fn gauge_rearms_on_dequeue_drain() {
625        // record_enqueue/record_dequeue gauges (TrackedSyncSender/Receiver) must
626        // also re-arm as they drain, not only stay latched after the producer idles.
627        let gauge = QueueGauge::new(TrackedLimit::SidecarStdoutFrames, 10, LimitCategory::Queue);
628        for _ in 0..9 {
629            gauge.record_enqueue(); // climbs to 90% -> warned
630        }
631        assert_eq!(gauge.depth(), 9);
632        assert!(gauge.warned.load(Ordering::Acquire));
633        for _ in 0..6 {
634            gauge.record_dequeue(); // drains to 30% (<=50%) -> re-arm on dequeue
635        }
636        assert_eq!(gauge.depth(), 3);
637        assert!(!gauge.warned.load(Ordering::Acquire));
638    }
639
640    #[test]
641    fn tracked_channel_reports_usage_through_registry() {
642        let (tx, rx) = tracked_sync_channel::<u32>(TrackedLimit::SidecarStdoutFrames, 4);
643        tx.send(1).unwrap();
644        tx.send(2).unwrap();
645
646        let snapshot = queue_snapshot();
647        let entry = snapshot
648            .iter()
649            .find(|stat| stat.name == TrackedLimit::SidecarStdoutFrames)
650            .expect("registered queue should appear in snapshot");
651        assert_eq!(entry.depth, 2);
652        assert_eq!(entry.capacity, 4);
653        assert_eq!(entry.high_water, 2);
654        assert_eq!(entry.fill_percent, 50);
655        assert_eq!(entry.category, LimitCategory::Queue);
656
657        assert_eq!(rx.recv().unwrap(), 1);
658        assert_eq!(tx.gauge().depth(), 1);
659
660        // Dropping the channel removes it from later snapshots.
661        drop(tx);
662        drop(rx);
663        assert!(queue_snapshot()
664            .iter()
665            .all(|stat| stat.name != TrackedLimit::SidecarStdoutFrames));
666    }
667
668    #[test]
669    fn warning_sink_fires_once_per_crossing() {
670        let _handler_guard = warning_handler_test_lock()
671            .lock()
672            .expect("warning-handler test lock");
673        let captured: Arc<Mutex<Vec<LimitWarning>>> = Arc::new(Mutex::new(Vec::new()));
674        let sink = Arc::clone(&captured);
675        // The handler is global; filter by our unique name so a gauge from a
676        // concurrently-running test can never pollute this assertion.
677        set_limit_warning_handler(Box::new(move |warning| {
678            if warning.name == TrackedLimit::VmPipes {
679                sink.lock().expect("sink mutex").push(warning.clone());
680            }
681        }));
682
683        let gauge = register_limit(TrackedLimit::VmPipes, 10);
684        gauge.observe_depth(7); // below 80%: no warning
685        assert!(captured.lock().unwrap().is_empty());
686        gauge.observe_depth(9); // crosses 80%: fires once
687        gauge.observe_depth(10); // still near full: must NOT re-fire (edge-triggered)
688
689        let warnings = captured.lock().unwrap();
690        assert_eq!(
691            warnings.len(),
692            1,
693            "warning sink must fire once per crossing"
694        );
695        assert_eq!(warnings[0].category, LimitCategory::Resource);
696        assert_eq!(warnings[0].capacity, 10);
697        assert!(warnings[0].fill_percent >= WARN_FILL_PERCENT);
698    }
699
700    #[test]
701    fn exhausted_warning_sink_fires_immediately() {
702        let _handler_guard = warning_handler_test_lock()
703            .lock()
704            .expect("warning-handler test lock");
705        let captured: Arc<Mutex<Vec<LimitWarning>>> = Arc::new(Mutex::new(Vec::new()));
706        let sink = Arc::clone(&captured);
707        set_limit_warning_handler(Box::new(move |warning| {
708            if warning.name == TrackedLimit::V8CpuTimeMs {
709                sink.lock().expect("sink mutex").push(warning.clone());
710            }
711        }));
712
713        warn_limit_exhausted(TrackedLimit::V8CpuTimeMs, 30_000, 30_000);
714
715        let warnings = captured.lock().unwrap();
716        assert_eq!(warnings.len(), 1);
717        assert_eq!(warnings[0].category, LimitCategory::Cpu);
718        assert_eq!(warnings[0].fill_percent, 100);
719    }
720}
agentos_bridge/queue_tracker.rs

agentos_bridge/
queue_tracker.rs
