agentos_runtime/
readiness.rs

1//! Durable, revisioned per-VM readiness with a capacity-one wake lane.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::fmt;
5use std::ops::{BitAnd, BitOr, BitOrAssign, Sub};
6use std::sync::{Arc, Mutex};
7use std::time::{Duration, Instant};
8
9use crate::accounting::{ResourceClass, ResourceLedger};
10use crate::metrics::{ChannelMetricClass, RuntimeMetrics, WakeMetric};
11
12pub type CapabilityId = u64;
13
14#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
15pub struct ReadyFlags(u16);
16
17impl ReadyFlags {
18    pub const READABLE: Self = Self(1 << 0);
19    pub const WRITABLE: Self = Self(1 << 1);
20    pub const ACCEPT: Self = Self(1 << 2);
21    pub const DATAGRAM: Self = Self(1 << 3);
22    pub const END: Self = Self(1 << 4);
23    pub const ERROR: Self = Self(1 << 5);
24    pub const CLOSE: Self = Self(1 << 6);
25
26    pub const fn is_empty(self) -> bool {
27        self.0 == 0
28    }
29
30    pub const fn intersects(self, other: Self) -> bool {
31        self.0 & other.0 != 0
32    }
33
34    pub const fn bits(self) -> u16 {
35        self.0
36    }
37}
38
39impl BitOr for ReadyFlags {
40    type Output = Self;
41
42    fn bitor(self, rhs: Self) -> Self::Output {
43        Self(self.0 | rhs.0)
44    }
45}
46
47impl BitOrAssign for ReadyFlags {
48    fn bitor_assign(&mut self, rhs: Self) {
49        self.0 |= rhs.0;
50    }
51}
52
53impl BitAnd for ReadyFlags {
54    type Output = Self;
55
56    fn bitand(self, rhs: Self) -> Self::Output {
57        Self(self.0 & rhs.0)
58    }
59}
60
61impl Sub for ReadyFlags {
62    type Output = Self;
63
64    fn sub(self, rhs: Self) -> Self::Output {
65        Self(self.0 & !rhs.0)
66    }
67}
68
69#[derive(Clone, Copy, Debug, Eq, PartialEq)]
70pub struct ReadyWake {
71    pub generation: u64,
72    pub epoch: u64,
73}
74
75#[derive(Clone, Copy, Debug, Eq, PartialEq)]
76pub struct ReadyObservation {
77    pub capability_id: CapabilityId,
78    pub capability_generation: u64,
79    pub flags: ReadyFlags,
80    pub revision: u64,
81}
82
83#[derive(Clone, Debug, Eq, PartialEq)]
84pub struct ReadyBatch {
85    pub generation: u64,
86    pub epoch: u64,
87    pub entries: Vec<ReadyObservation>,
88    pub signals_ready: bool,
89    pub timers_ready: bool,
90    pub more: bool,
91}
92
93#[derive(Clone, Copy, Debug, Eq, PartialEq)]
94pub struct ReadyAcknowledgement {
95    pub capability_id: CapabilityId,
96    pub capability_generation: u64,
97    pub observed_revision: u64,
98    pub clear: ReadyFlags,
99}
100
101#[derive(Debug, Clone, Eq, PartialEq)]
102pub enum ReadyError {
103    WrongGeneration {
104        supplied: u64,
105        expected: u64,
106    },
107    StaleWake {
108        supplied: u64,
109        outstanding: Option<u64>,
110    },
111    HandleLimit {
112        limit: usize,
113    },
114    ControlLimit {
115        control: &'static str,
116        limit: usize,
117        config_path: String,
118    },
119    InvalidSignal {
120        signal: i32,
121    },
122    StaleCapabilityGeneration {
123        capability_id: CapabilityId,
124        supplied: u64,
125        expected: u64,
126    },
127    MissingResourceLimit {
128        resource: ResourceClass,
129    },
130    RevisionExhausted {
131        capability_id: CapabilityId,
132    },
133    EpochExhausted,
134    WakeInvariant,
135    WakeDisconnected,
136    Poisoned,
137}
138
139impl fmt::Display for ReadyError {
140    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
141        match self {
142            Self::WrongGeneration { supplied, expected } => write!(
143                formatter,
144                "ERR_AGENTOS_READY_STALE_GENERATION: supplied generation {supplied}, expected {expected}"
145            ),
146            Self::StaleWake {
147                supplied,
148                outstanding,
149            } => write!(
150                formatter,
151                "ERR_AGENTOS_READY_STALE_WAKE: supplied epoch {supplied}, outstanding epoch {outstanding:?}"
152            ),
153            Self::HandleLimit { limit } => write!(
154                formatter,
155                "ERR_AGENTOS_READY_HANDLE_LIMIT: ready handle count exceeded {limit}; raise limits.reactor.maxReadyHandles (VM) or runtime.resources.maxReadyHandles (process)"
156            ),
157            Self::ControlLimit {
158                control,
159                limit,
160                config_path,
161            } => write!(
162                formatter,
163                "ERR_AGENTOS_READY_CONTROL_LIMIT: pending {control} count exceeded {limit}; raise {config_path}"
164            ),
165            Self::InvalidSignal { signal } => write!(
166                formatter,
167                "ERR_AGENTOS_READY_SIGNAL_INVALID: signal {signal} is outside the supported 1..=64 range"
168            ),
169            Self::StaleCapabilityGeneration {
170                capability_id,
171                supplied,
172                expected,
173            } => write!(
174                formatter,
175                "ERR_AGENTOS_READY_STALE_CAPABILITY: capability {capability_id} generation {supplied} does not match live generation {expected}"
176            ),
177            Self::MissingResourceLimit { resource } => write!(
178                formatter,
179                "ERR_AGENTOS_READY_RESOURCE_UNBOUNDED: resource={} has no configured VM limit",
180                resource.name()
181            ),
182            Self::RevisionExhausted { capability_id } => write!(
183                formatter,
184                "ERR_AGENTOS_READY_REVISION_EXHAUSTED: capability {capability_id} exhausted readiness revisions"
185            ),
186            Self::EpochExhausted => formatter.write_str(
187                "ERR_AGENTOS_READY_EPOCH_EXHAUSTED: VM generation exhausted wake epochs",
188            ),
189            Self::WakeInvariant => formatter.write_str(
190                "ERR_AGENTOS_READY_WAKE_INVARIANT: capacity-one wake lane was full while broker state was idle",
191            ),
192            Self::WakeDisconnected => formatter.write_str(
193                "ERR_AGENTOS_READY_WAKE_DISCONNECTED: VM wake consumer disconnected",
194            ),
195            Self::Poisoned => formatter.write_str(
196                "ERR_AGENTOS_READY_STATE_POISONED: VM readiness state lock poisoned",
197            ),
198        }
199    }
200}
201
202impl std::error::Error for ReadyError {}
203
204#[derive(Debug)]
205struct ReadyEntry {
206    capability_generation: u64,
207    flags: ReadyFlags,
208    revision: u64,
209    application_read_interest: bool,
210    ready_since: Option<Instant>,
211}
212
213#[derive(Debug)]
214enum WakeState {
215    Idle,
216    Outstanding { epoch: u64 },
217    Failed(WakeFailure),
218}
219
220#[derive(Clone, Copy, Debug)]
221enum WakeFailure {
222    Invariant,
223    Disconnected,
224}
225
226impl WakeFailure {
227    fn error(self) -> ReadyError {
228        match self {
229            Self::Invariant => ReadyError::WakeInvariant,
230            Self::Disconnected => ReadyError::WakeDisconnected,
231        }
232    }
233}
234
235#[derive(Debug)]
236struct ReadyState {
237    handles: BTreeMap<CapabilityId, ReadyEntry>,
238    last_delivered_capability: Option<CapabilityId>,
239    signals: BTreeSet<i32>,
240    timers: BTreeSet<u64>,
241    wake: WakeState,
242    next_epoch: u64,
243}
244
245#[derive(Clone, Debug)]
246pub struct SessionReadyBroker {
247    generation: u64,
248    max_handles: usize,
249    max_batch_handles: usize,
250    max_pending_timers: usize,
251    timer_limit_config_path: String,
252    state: Arc<Mutex<ReadyState>>,
253    wake_tx: tokio::sync::mpsc::Sender<ReadyWake>,
254    metrics: Option<RuntimeMetrics>,
255}
256
257impl SessionReadyBroker {
258    pub fn new(
259        generation: u64,
260        max_handles: usize,
261    ) -> Result<(Self, tokio::sync::mpsc::Receiver<ReadyWake>), ReadyError> {
262        Self::new_inner(
263            generation,
264            max_handles,
265            max_handles,
266            max_handles,
267            String::from("limits.jsRuntime.maxTimers"),
268            None,
269        )
270    }
271
272    pub fn new_with_metrics(
273        generation: u64,
274        max_handles: usize,
275        metrics: RuntimeMetrics,
276    ) -> Result<(Self, tokio::sync::mpsc::Receiver<ReadyWake>), ReadyError> {
277        Self::new_inner(
278            generation,
279            max_handles,
280            max_handles,
281            max_handles,
282            String::from("limits.jsRuntime.maxTimers"),
283            Some(metrics),
284        )
285    }
286
287    pub fn new_with_resources(
288        generation: u64,
289        resources: Arc<ResourceLedger>,
290        metrics: RuntimeMetrics,
291    ) -> Result<(Self, tokio::sync::mpsc::Receiver<ReadyWake>), ReadyError> {
292        let max_handles = configured_limit(&resources, ResourceClass::ReadyHandles)?;
293        let timer_limit = configured_resource_limit(&resources, ResourceClass::Timers)?;
294        Self::new_inner(
295            generation,
296            max_handles,
297            max_handles,
298            timer_limit.maximum,
299            timer_limit.config_path,
300            Some(metrics),
301        )
302    }
303
304    fn new_inner(
305        generation: u64,
306        max_handles: usize,
307        max_batch_handles: usize,
308        max_pending_timers: usize,
309        timer_limit_config_path: String,
310        metrics: Option<RuntimeMetrics>,
311    ) -> Result<(Self, tokio::sync::mpsc::Receiver<ReadyWake>), ReadyError> {
312        if max_handles == 0 {
313            return Err(ReadyError::HandleLimit { limit: 0 });
314        }
315        let (wake_tx, wake_rx) = tokio::sync::mpsc::channel(1);
316        Ok((
317            Self {
318                generation,
319                max_handles,
320                max_batch_handles,
321                max_pending_timers,
322                timer_limit_config_path,
323                state: Arc::new(Mutex::new(ReadyState {
324                    handles: BTreeMap::new(),
325                    last_delivered_capability: None,
326                    signals: BTreeSet::new(),
327                    timers: BTreeSet::new(),
328                    wake: WakeState::Idle,
329                    next_epoch: 1,
330                })),
331                wake_tx,
332                metrics,
333            },
334            wake_rx,
335        ))
336    }
337
338    pub fn generation(&self) -> u64 {
339        self.generation
340    }
341
342    pub fn max_batch_handles(&self) -> usize {
343        self.max_batch_handles
344    }
345
346    pub fn mark_signal_ready(&self, generation: u64, signal: i32) -> Result<(), ReadyError> {
347        self.validate_generation(generation)?;
348        if !(1..=64).contains(&signal) {
349            return Err(ReadyError::InvalidSignal { signal });
350        }
351        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
352        state.signals.insert(signal);
353        self.schedule_wake_locked(&mut state, false)
354    }
355
356    pub fn mark_timer_ready(&self, generation: u64, timer_id: u64) -> Result<(), ReadyError> {
357        self.validate_generation(generation)?;
358        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
359        if !state.timers.contains(&timer_id) && state.timers.len() >= self.max_pending_timers {
360            return Err(ReadyError::ControlLimit {
361                control: "timers",
362                limit: self.max_pending_timers,
363                config_path: self.timer_limit_config_path.clone(),
364            });
365        }
366        state.timers.insert(timer_id);
367        self.schedule_wake_locked(&mut state, false)
368    }
369
370    pub fn drain_signals(
371        &self,
372        generation: u64,
373        epoch: u64,
374        max: usize,
375    ) -> Result<Vec<i32>, ReadyError> {
376        self.validate_generation(generation)?;
377        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
378        self.validate_epoch(&state, epoch)?;
379        let values = state
380            .signals
381            .iter()
382            .copied()
383            .take(max.max(1))
384            .collect::<Vec<_>>();
385        for value in &values {
386            state.signals.remove(value);
387        }
388        Ok(values)
389    }
390
391    pub fn drain_timers(
392        &self,
393        generation: u64,
394        epoch: u64,
395        max: usize,
396    ) -> Result<Vec<u64>, ReadyError> {
397        self.validate_generation(generation)?;
398        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
399        self.validate_epoch(&state, epoch)?;
400        let values = state
401            .timers
402            .iter()
403            .copied()
404            .take(max.max(1))
405            .collect::<Vec<_>>();
406        for value in &values {
407            state.timers.remove(value);
408        }
409        Ok(values)
410    }
411
412    pub fn mark_ready(
413        &self,
414        generation: u64,
415        capability_id: CapabilityId,
416        capability_generation: u64,
417        mut flags: ReadyFlags,
418    ) -> Result<(), ReadyError> {
419        self.validate_generation(generation)?;
420        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
421        let existing = state.handles.get(&capability_id);
422        if let Some(entry) = existing {
423            validate_capability_generation(
424                capability_id,
425                capability_generation,
426                entry.capability_generation,
427            )?;
428        } else if state.handles.len() >= self.max_handles {
429            return Err(ReadyError::HandleLimit {
430                limit: self.max_handles,
431            });
432        }
433        if existing.is_some_and(|entry| !entry.application_read_interest) {
434            flags = flags - ReadyFlags::READABLE;
435        }
436        if flags.is_empty() {
437            if existing.is_none() {
438                state.handles.insert(
439                    capability_id,
440                    ReadyEntry {
441                        capability_generation,
442                        flags: ReadyFlags::default(),
443                        revision: 0,
444                        application_read_interest: true,
445                        ready_since: None,
446                    },
447                );
448            }
449            return Ok(());
450        }
451        let entry = state.handles.entry(capability_id).or_insert(ReadyEntry {
452            capability_generation,
453            flags: ReadyFlags::default(),
454            revision: 0,
455            application_read_interest: true,
456            ready_since: None,
457        });
458        if entry.flags.is_empty() {
459            entry.ready_since = Some(Instant::now());
460        }
461        entry.revision = entry
462            .revision
463            .checked_add(1)
464            .ok_or(ReadyError::RevisionExhausted { capability_id })?;
465        entry.flags |= flags;
466        if let Some(metrics) = &self.metrics {
467            metrics.record_wake(WakeMetric::Attempted);
468        }
469        let result = self.schedule_wake_locked(&mut state, false);
470        self.observe_ready_state(&state);
471        result
472    }
473
474    pub fn set_application_read_interest(
475        &self,
476        generation: u64,
477        capability_id: CapabilityId,
478        capability_generation: u64,
479        enabled: bool,
480    ) -> Result<(), ReadyError> {
481        self.validate_generation(generation)?;
482        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
483        if !state.handles.contains_key(&capability_id) && state.handles.len() >= self.max_handles {
484            return Err(ReadyError::HandleLimit {
485                limit: self.max_handles,
486            });
487        }
488        let entry = state.handles.entry(capability_id).or_insert(ReadyEntry {
489            capability_generation,
490            flags: ReadyFlags::default(),
491            revision: 0,
492            application_read_interest: enabled,
493            ready_since: None,
494        });
495        validate_capability_generation(
496            capability_id,
497            capability_generation,
498            entry.capability_generation,
499        )?;
500        entry.application_read_interest = enabled;
501        if !enabled && entry.flags.intersects(ReadyFlags::READABLE) {
502            entry.revision = entry
503                .revision
504                .checked_add(1)
505                .ok_or(ReadyError::RevisionExhausted { capability_id })?;
506            entry.flags = entry.flags - ReadyFlags::READABLE;
507            if entry.flags.is_empty() {
508                entry.ready_since = None;
509            }
510        }
511        self.observe_ready_state(&state);
512        Ok(())
513    }
514
515    pub fn remove_capability(
516        &self,
517        generation: u64,
518        capability_id: CapabilityId,
519        capability_generation: u64,
520    ) -> Result<(), ReadyError> {
521        self.validate_generation(generation)?;
522        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
523        if let Some(entry) = state.handles.get(&capability_id) {
524            validate_capability_generation(
525                capability_id,
526                capability_generation,
527                entry.capability_generation,
528            )?;
529        }
530        state.handles.remove(&capability_id);
531        self.observe_ready_state(&state);
532        Ok(())
533    }
534
535    pub fn ready_batch(
536        &self,
537        generation: u64,
538        epoch: u64,
539        max_handles: usize,
540    ) -> Result<ReadyBatch, ReadyError> {
541        self.validate_generation(generation)?;
542        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
543        self.validate_epoch(&state, epoch)?;
544        if let Some(metrics) = &self.metrics {
545            metrics.record_wake(WakeMetric::Delivered);
546        }
547        let limit = max_handles.max(1).min(self.max_batch_handles);
548        let ready_count = state
549            .handles
550            .values()
551            .filter(|entry| !entry.flags.is_empty())
552            .count();
553        let mut entries = Vec::with_capacity(limit.min(ready_count));
554
555        // Continue after the last capability delivered by the previous batch,
556        // then wrap once. A continuously republished low capability therefore
557        // cannot occupy every bounded batch and starve higher capability IDs.
558        if let Some(cursor) = state.last_delivered_capability {
559            for (&capability_id, entry) in state.handles.range((
560                std::ops::Bound::Excluded(cursor),
561                std::ops::Bound::Unbounded,
562            )) {
563                append_ready_observation(&mut entries, limit, capability_id, entry);
564                if entries.len() == limit {
565                    break;
566                }
567            }
568            if entries.len() < limit {
569                for (&capability_id, entry) in state.handles.range(..=cursor) {
570                    append_ready_observation(&mut entries, limit, capability_id, entry);
571                    if entries.len() == limit {
572                        break;
573                    }
574                }
575            }
576        } else {
577            for (&capability_id, entry) in &state.handles {
578                append_ready_observation(&mut entries, limit, capability_id, entry);
579                if entries.len() == limit {
580                    break;
581                }
582            }
583        }
584        if let Some(last) = entries.last() {
585            state.last_delivered_capability = Some(last.capability_id);
586        }
587        self.observe_ready_state(&state);
588        Ok(ReadyBatch {
589            generation,
590            epoch,
591            signals_ready: !state.signals.is_empty(),
592            timers_ready: !state.timers.is_empty(),
593            more: ready_count > entries.len(),
594            entries,
595        })
596    }
597
598    pub fn complete_wake(
599        &self,
600        generation: u64,
601        epoch: u64,
602        acknowledgements: &[ReadyAcknowledgement],
603    ) -> Result<(), ReadyError> {
604        self.validate_generation(generation)?;
605        let mut state = self.state.lock().map_err(|_| ReadyError::Poisoned)?;
606        self.validate_epoch(&state, epoch)?;
607        for acknowledgement in acknowledgements {
608            let Some(entry) = state.handles.get_mut(&acknowledgement.capability_id) else {
609                continue;
610            };
611            if entry.capability_generation != acknowledgement.capability_generation {
612                continue;
613            }
614            if entry.revision == acknowledgement.observed_revision {
615                let mut clear = acknowledgement.clear;
616                if !entry.application_read_interest {
617                    clear |= ReadyFlags::READABLE;
618                }
619                entry.flags = entry.flags - clear;
620                if entry.flags.is_empty() {
621                    entry.ready_since = None;
622                }
623            }
624        }
625        state.wake = WakeState::Idle;
626        let result = self.schedule_wake_locked(&mut state, true);
627        self.observe_ready_state(&state);
628        result
629    }
630
631    pub fn pending_handle_count(&self) -> Result<usize, ReadyError> {
632        Ok(self
633            .state
634            .lock()
635            .map_err(|_| ReadyError::Poisoned)?
636            .handles
637            .values()
638            .filter(|entry| !entry.flags.is_empty())
639            .count())
640    }
641
642    fn validate_generation(&self, generation: u64) -> Result<(), ReadyError> {
643        if generation != self.generation {
644            return Err(ReadyError::WrongGeneration {
645                supplied: generation,
646                expected: self.generation,
647            });
648        }
649        Ok(())
650    }
651
652    fn validate_epoch(&self, state: &ReadyState, epoch: u64) -> Result<(), ReadyError> {
653        let outstanding = match state.wake {
654            WakeState::Outstanding { epoch } => Some(epoch),
655            WakeState::Idle | WakeState::Failed(_) => None,
656        };
657        if outstanding != Some(epoch) {
658            return Err(ReadyError::StaleWake {
659                supplied: epoch,
660                outstanding,
661            });
662        }
663        Ok(())
664    }
665
666    fn schedule_wake_locked(
667        &self,
668        state: &mut ReadyState,
669        rearmed: bool,
670    ) -> Result<(), ReadyError> {
671        if !state.handles.values().any(|entry| !entry.flags.is_empty())
672            && state.signals.is_empty()
673            && state.timers.is_empty()
674        {
675            return Ok(());
676        }
677        match state.wake {
678            WakeState::Idle => {}
679            WakeState::Outstanding { .. } => {
680                if let Some(metrics) = &self.metrics {
681                    metrics.record_wake(WakeMetric::Coalesced);
682                }
683                return Ok(());
684            }
685            WakeState::Failed(failure) => return Err(failure.error()),
686        }
687        let epoch = state.next_epoch;
688        state.next_epoch = state
689            .next_epoch
690            .checked_add(1)
691            .ok_or(ReadyError::EpochExhausted)?;
692        state.wake = WakeState::Outstanding { epoch };
693        match self.wake_tx.try_send(ReadyWake {
694            generation: self.generation,
695            epoch,
696        }) {
697            Ok(()) => {
698                if let Some(metrics) = &self.metrics {
699                    if rearmed {
700                        metrics.record_wake(WakeMetric::Rearmed);
701                    }
702                    metrics.observe_channel(
703                        ChannelMetricClass::ReadyWake,
704                        1,
705                        std::mem::size_of::<ReadyWake>(),
706                    );
707                }
708                Ok(())
709            }
710            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
711                state.wake = WakeState::Failed(WakeFailure::Invariant);
712                Err(ReadyError::WakeInvariant)
713            }
714            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
715                state.wake = WakeState::Failed(WakeFailure::Disconnected);
716                Err(ReadyError::WakeDisconnected)
717            }
718        }
719    }
720
721    fn observe_ready_state(&self, state: &ReadyState) {
722        let Some(metrics) = &self.metrics else {
723            return;
724        };
725        let now = Instant::now();
726        let mut size = 0usize;
727        let mut oldest = Duration::ZERO;
728        for entry in state.handles.values() {
729            if entry.flags.is_empty() {
730                continue;
731            }
732            size += 1;
733            if let Some(ready_since) = entry.ready_since {
734                oldest = oldest.max(now.saturating_duration_since(ready_since));
735            }
736        }
737        metrics.observe_readiness(size, oldest);
738    }
739}
740
741fn append_ready_observation(
742    entries: &mut Vec<ReadyObservation>,
743    limit: usize,
744    capability_id: CapabilityId,
745    entry: &ReadyEntry,
746) {
747    if entries.len() < limit && !entry.flags.is_empty() {
748        entries.push(ReadyObservation {
749            capability_id,
750            capability_generation: entry.capability_generation,
751            flags: entry.flags,
752            revision: entry.revision,
753        });
754    }
755}
756
757fn configured_limit(
758    resources: &ResourceLedger,
759    resource: ResourceClass,
760) -> Result<usize, ReadyError> {
761    resources
762        .usage(resource)
763        .limit
764        .filter(|limit| *limit > 0)
765        .ok_or(ReadyError::MissingResourceLimit { resource })
766}
767
768fn configured_resource_limit(
769    resources: &ResourceLedger,
770    resource: ResourceClass,
771) -> Result<crate::accounting::ResourceLimit, ReadyError> {
772    resources
773        .configured_limit(resource)
774        .filter(|limit| limit.maximum > 0)
775        .ok_or(ReadyError::MissingResourceLimit { resource })
776}
777
778fn validate_capability_generation(
779    capability_id: CapabilityId,
780    supplied: u64,
781    expected: u64,
782) -> Result<(), ReadyError> {
783    if supplied == expected {
784        Ok(())
785    } else {
786        Err(ReadyError::StaleCapabilityGeneration {
787            capability_id,
788            supplied,
789            expected,
790        })
791    }
792}
793
794#[cfg(test)]
795mod tests {
796    use super::*;
797    use crate::accounting::ResourceLimit;
798    use crate::capability::{CapabilityBackend, CapabilityKind, CapabilityRegistry};
799
800    #[tokio::test]
801    async fn repeated_marks_coalesce_to_one_wake() {
802        let (broker, mut wakes) = SessionReadyBroker::new(7, 8).expect("broker");
803        for _ in 0..1_000_000 {
804            broker
805                .mark_ready(7, 41, 1, ReadyFlags::READABLE)
806                .expect("mark ready");
807        }
808        let wake = wakes.recv().await.expect("one wake");
809        assert_eq!(
810            wake,
811            ReadyWake {
812                generation: 7,
813                epoch: 1
814            }
815        );
816        assert!(matches!(
817            wakes.try_recv(),
818            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
819        ));
820        let batch = broker.ready_batch(7, wake.epoch, 8).expect("ready batch");
821        assert_eq!(batch.entries.len(), 1);
822        assert_eq!(batch.entries[0].revision, 1_000_000);
823    }
824
825    #[tokio::test]
826    async fn signals_and_timers_share_one_wake_but_keep_durable_control_state() {
827        let (broker, mut wakes) = SessionReadyBroker::new(8, 8).expect("broker");
828        for _ in 0..10_000 {
829            broker.mark_signal_ready(8, 15).expect("coalesce SIGTERM");
830            broker.mark_timer_ready(8, 91).expect("coalesce timer");
831        }
832        let wake = wakes.recv().await.expect("one control wake");
833        assert!(matches!(
834            wakes.try_recv(),
835            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
836        ));
837        let batch = broker.ready_batch(8, wake.epoch, 8).expect("control batch");
838        assert!(batch.signals_ready);
839        assert!(batch.timers_ready);
840        assert_eq!(broker.drain_signals(8, wake.epoch, 8).unwrap(), vec![15]);
841        assert_eq!(broker.drain_timers(8, wake.epoch, 8).unwrap(), vec![91]);
842        broker
843            .complete_wake(8, wake.epoch, &[])
844            .expect("complete control wake");
845        assert!(matches!(
846            wakes.try_recv(),
847            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
848        ));
849    }
850
851    #[test]
852    fn timer_limit_error_names_the_owning_configuration_path() {
853        let resources = Arc::new(ResourceLedger::root(
854            "vm=timer-limit generation=10",
855            [
856                (
857                    ResourceClass::ReadyHandles,
858                    ResourceLimit::new(2, "limits.reactor.maxReadyHandles"),
859                ),
860                (
861                    ResourceClass::Timers,
862                    ResourceLimit::new(1, "limits.jsRuntime.maxTimers"),
863                ),
864            ],
865        ));
866        let (broker, _wakes) =
867            SessionReadyBroker::new_with_resources(10, resources, RuntimeMetrics::new())
868                .expect("broker");
869        broker.mark_timer_ready(10, 1).expect("first timer");
870
871        let error = broker
872            .mark_timer_ready(10, 2)
873            .expect_err("second unique timer exceeds configured limit");
874
875        assert_eq!(
876            error,
877            ReadyError::ControlLimit {
878                control: "timers",
879                limit: 1,
880                config_path: String::from("limits.jsRuntime.maxTimers"),
881            }
882        );
883        assert!(error
884            .to_string()
885            .ends_with("raise limits.jsRuntime.maxTimers"));
886    }
887
888    #[tokio::test]
889    async fn concurrent_republication_is_not_cleared_by_old_observation() {
890        let (broker, mut wakes) = SessionReadyBroker::new(9, 8).expect("broker");
891        broker
892            .mark_ready(9, 3, 1, ReadyFlags::READABLE)
893            .expect("first mark");
894        let wake = wakes.recv().await.expect("first wake");
895        let batch = broker.ready_batch(9, wake.epoch, 8).expect("batch");
896        broker
897            .mark_ready(9, 3, 1, ReadyFlags::READABLE)
898            .expect("concurrent mark");
899        broker
900            .complete_wake(
901                9,
902                wake.epoch,
903                &[ReadyAcknowledgement {
904                    capability_id: 3,
905                    capability_generation: 1,
906                    observed_revision: batch.entries[0].revision,
907                    clear: ReadyFlags::READABLE,
908                }],
909            )
910            .expect("complete old wake");
911
912        let replacement = wakes.recv().await.expect("replacement wake");
913        let replacement_batch = broker
914            .ready_batch(9, replacement.epoch, 8)
915            .expect("replacement batch");
916        assert_eq!(replacement_batch.entries[0].revision, 2);
917    }
918
919    #[tokio::test]
920    async fn bounded_batches_rotate_past_continuously_hot_low_capability_ids() {
921        let (broker, mut wakes) = SessionReadyBroker::new(10, 8).expect("broker");
922        for capability_id in 1..=8 {
923            broker
924                .mark_ready(10, capability_id, 1, ReadyFlags::READABLE)
925                .expect("mark initial readiness");
926        }
927
928        let first_wake = wakes.recv().await.expect("first wake");
929        let first = broker
930            .ready_batch(10, first_wake.epoch, 2)
931            .expect("first bounded batch");
932        assert_eq!(
933            first
934                .entries
935                .iter()
936                .map(|entry| entry.capability_id)
937                .collect::<Vec<_>>(),
938            vec![1, 2]
939        );
940
941        // Republish the first two handles before acknowledging their old
942        // observations. Their revisions remain pending, but the next batch
943        // must advance to capabilities that have not received a turn yet.
944        for capability_id in 1..=2 {
945            broker
946                .mark_ready(10, capability_id, 1, ReadyFlags::READABLE)
947                .expect("keep low capability hot");
948        }
949        let first_acknowledgements = first
950            .entries
951            .iter()
952            .map(|entry| ReadyAcknowledgement {
953                capability_id: entry.capability_id,
954                capability_generation: entry.capability_generation,
955                observed_revision: entry.revision,
956                clear: ReadyFlags::READABLE,
957            })
958            .collect::<Vec<_>>();
959        broker
960            .complete_wake(10, first_wake.epoch, &first_acknowledgements)
961            .expect("complete first wake");
962
963        let second_wake = wakes.recv().await.expect("second wake");
964        let second = broker
965            .ready_batch(10, second_wake.epoch, 2)
966            .expect("second bounded batch");
967        assert_eq!(
968            second
969                .entries
970                .iter()
971                .map(|entry| entry.capability_id)
972                .collect::<Vec<_>>(),
973            vec![3, 4]
974        );
975        assert!(second.more);
976    }
977
978    #[tokio::test]
979    async fn read_interest_suppresses_readable_publication() {
980        let (broker, mut wakes) = SessionReadyBroker::new(11, 8).expect("broker");
981        broker
982            .mark_ready(11, 5, 1, ReadyFlags::WRITABLE)
983            .expect("create entry");
984        let first = wakes.recv().await.expect("first wake");
985        let batch = broker.ready_batch(11, first.epoch, 8).expect("batch");
986        broker
987            .complete_wake(
988                11,
989                first.epoch,
990                &[ReadyAcknowledgement {
991                    capability_id: 5,
992                    capability_generation: 1,
993                    observed_revision: batch.entries[0].revision,
994                    clear: ReadyFlags::WRITABLE,
995                }],
996            )
997            .expect("complete wake");
998        broker
999            .set_application_read_interest(11, 5, 1, false)
1000            .expect("pause reads");
1001        broker
1002            .mark_ready(11, 5, 1, ReadyFlags::READABLE)
1003            .expect("suppressed readable mark");
1004        assert!(matches!(
1005            wakes.try_recv(),
1006            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
1007        ));
1008    }
1009
1010    #[tokio::test]
1011    async fn read_interest_set_before_first_publication_is_durable() {
1012        let (broker, mut wakes) = SessionReadyBroker::new(12, 8).expect("broker");
1013        broker
1014            .set_application_read_interest(12, 6, 1, false)
1015            .expect("pause before first readiness");
1016        broker
1017            .mark_ready(12, 6, 1, ReadyFlags::READABLE)
1018            .expect("suppressed first readable mark");
1019        assert!(matches!(
1020            wakes.try_recv(),
1021            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
1022        ));
1023
1024        broker
1025            .mark_ready(12, 6, 1, ReadyFlags::END)
1026            .expect("terminal readiness remains observable while reads are paused");
1027        let wake = wakes.recv().await.expect("terminal wake");
1028        let batch = broker.ready_batch(12, wake.epoch, 8).expect("batch");
1029        assert_eq!(batch.entries.len(), 1);
1030        assert_eq!(batch.entries[0].flags, ReadyFlags::END);
1031    }
1032
1033    #[tokio::test]
1034    async fn stale_completion_cannot_mutate_current_epoch() {
1035        let (broker, mut wakes) = SessionReadyBroker::new(13, 8).expect("broker");
1036        broker
1037            .mark_ready(13, 1, 1, ReadyFlags::CLOSE)
1038            .expect("mark close");
1039        let wake = wakes.recv().await.expect("wake");
1040        let error = broker
1041            .complete_wake(13, wake.epoch + 1, &[])
1042            .expect_err("stale epoch must fail");
1043        assert!(matches!(error, ReadyError::StaleWake { .. }));
1044        assert_eq!(broker.pending_handle_count().expect("pending count"), 1);
1045    }
1046
1047    #[test]
1048    fn disconnected_wake_lane_remains_a_hard_error() {
1049        let (broker, wakes) = SessionReadyBroker::new(14, 8).expect("broker");
1050        drop(wakes);
1051
1052        assert!(matches!(
1053            broker.mark_ready(14, 1, 1, ReadyFlags::READABLE),
1054            Err(ReadyError::WakeDisconnected)
1055        ));
1056        assert!(matches!(
1057            broker.mark_ready(14, 1, 1, ReadyFlags::END),
1058            Err(ReadyError::WakeDisconnected)
1059        ));
1060    }
1061
1062    #[tokio::test]
1063    async fn production_metrics_follow_broker_state_without_id_labels() {
1064        let metrics = RuntimeMetrics::new();
1065        let (broker, mut wakes) =
1066            SessionReadyBroker::new_with_metrics(17, 8, metrics.clone()).expect("broker");
1067        broker
1068            .mark_ready(17, 4, 1, ReadyFlags::READABLE)
1069            .expect("first publication");
1070        broker
1071            .mark_ready(17, 4, 1, ReadyFlags::READABLE)
1072            .expect("coalesced publication");
1073        let wake = wakes.recv().await.expect("wake");
1074        let batch = broker.ready_batch(17, wake.epoch, 8).expect("batch");
1075        broker
1076            .complete_wake(
1077                17,
1078                wake.epoch,
1079                &[ReadyAcknowledgement {
1080                    capability_id: 4,
1081                    capability_generation: 1,
1082                    observed_revision: batch.entries[0].revision,
1083                    clear: ReadyFlags::READABLE,
1084                }],
1085            )
1086            .expect("complete");
1087
1088        let snapshot = metrics.snapshot();
1089        assert_eq!(snapshot.wakes[WakeMetric::Attempted.index()], 2);
1090        assert_eq!(snapshot.wakes[WakeMetric::Coalesced.index()], 1);
1091        assert_eq!(snapshot.wakes[WakeMetric::Delivered.index()], 1);
1092        assert_eq!(snapshot.readiness.current_size, 0);
1093        assert_eq!(
1094            snapshot.channels[ChannelMetricClass::ReadyWake.index()].count_high_water,
1095            1
1096        );
1097    }
1098
1099    #[tokio::test]
1100    async fn capability_admission_owns_ready_permits_without_broker_double_charge() {
1101        let resources = Arc::new(ResourceLedger::root(
1102            "vm=readiness-integration generation=23",
1103            [
1104                (
1105                    ResourceClass::Capabilities,
1106                    ResourceLimit::new(4, "limits.reactor.maxCapabilities"),
1107                ),
1108                (
1109                    ResourceClass::ReadyHandles,
1110                    ResourceLimit::new(4, "limits.reactor.maxReadyHandles"),
1111                ),
1112                (
1113                    ResourceClass::Timers,
1114                    ResourceLimit::new(4, "runtime.resources.maxTimers"),
1115                ),
1116            ],
1117        ));
1118        let capabilities = CapabilityRegistry::new(23, Arc::clone(&resources));
1119        let (broker, mut wakes) = SessionReadyBroker::new_with_resources(
1120            23,
1121            Arc::clone(&resources),
1122            RuntimeMetrics::new(),
1123        )
1124        .expect("bounded broker");
1125
1126        let mut leases = Vec::new();
1127        for index in 0..4_u64 {
1128            let lease = capabilities
1129                .reserve(CapabilityKind::Http2Stream)
1130                .expect("admit capability up to configured maximum")
1131                .commit(CapabilityBackend::Native {
1132                    local_id: format!("stream-{index}"),
1133                })
1134                .expect("commit capability");
1135            broker
1136                .mark_ready(23, lease.id(), lease.generation(), ReadyFlags::READABLE)
1137                .expect("readiness publication must consume no second permit");
1138            leases.push(lease);
1139        }
1140        assert_eq!(resources.usage(ResourceClass::ReadyHandles).used, 4);
1141
1142        let wake = wakes.recv().await.expect("one coalesced wake");
1143        let batch = broker.ready_batch(23, wake.epoch, 4).expect("ready batch");
1144        assert_eq!(batch.entries.len(), 4);
1145        let acknowledgements = batch
1146            .entries
1147            .iter()
1148            .map(|entry| ReadyAcknowledgement {
1149                capability_id: entry.capability_id,
1150                capability_generation: entry.capability_generation,
1151                observed_revision: entry.revision,
1152                clear: entry.flags,
1153            })
1154            .collect::<Vec<_>>();
1155        broker
1156            .complete_wake(23, wake.epoch, &acknowledgements)
1157            .expect("complete wake");
1158        drop(broker);
1159        drop(leases);
1160        drop(capabilities);
1161        assert!(resources.is_zero(), "all capability permits must reconcile");
1162    }
1163}
agentos_runtime/readiness.rs

agentos_runtime/
readiness.rs
