1use std::{
32 future::Future,
33 sync::{
34 atomic::{AtomicBool, Ordering},
35 Arc, Mutex,
36 },
37 thread,
38 time::{Duration, Instant},
39};
40
41use wasmtime::{Engine, ResourceLimiter, StoreLimits, StoreLimitsBuilder};
42
43pub(crate) const EPOCH_TICK_INTERVAL: Duration = Duration::from_millis(100);
46
47pub const DEFAULT_MEMORY_BYTES: usize = 512 * 1024 * 1024;
50
51pub const DEFAULT_TABLE_ELEMENTS: usize = 100_000;
53
54pub const DEFAULT_INSTANCES: usize = 1_000;
58
59pub(crate) const WALL_DEADLINE_GRACE: Duration = Duration::from_secs(2);
65
66const _: () = assert!(
67 WALL_DEADLINE_GRACE.as_millis() > EPOCH_TICK_INTERVAL.as_millis(),
68 "the wall bound must leave the epoch at least one tick to fire first"
69);
70
71pub const DEFAULT_DEADLINE_SECONDS: u64 = 600;
78
79pub const HOOK_DEFAULT_DEADLINE_SECONDS: u64 = 30;
88
89const _: () = assert!(
90 HOOK_DEFAULT_DEADLINE_SECONDS < DEFAULT_DEADLINE_SECONDS,
91 "a hook call must not be able to stall a session for the capsule-wide budget"
92);
93
94#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
97pub struct ExecutionLimits {
98 pub memory_bytes: usize,
100 pub table_elements: usize,
102 pub instances: usize,
104 pub deadline_seconds: u64,
106}
107
108impl Default for ExecutionLimits {
109 fn default() -> Self {
110 Self {
111 memory_bytes: DEFAULT_MEMORY_BYTES,
112 table_elements: DEFAULT_TABLE_ELEMENTS,
113 instances: DEFAULT_INSTANCES,
114 deadline_seconds: DEFAULT_DEADLINE_SECONDS,
115 }
116 }
117}
118
119impl ExecutionLimits {
120 #[must_use]
123 pub fn resolve(declared: Option<&murmur_artifact::ResourceLimits>) -> Self {
124 let defaults = Self::default();
125 let Some(declared) = declared else {
126 return defaults;
127 };
128 Self {
129 memory_bytes: declared.memory_bytes.unwrap_or(defaults.memory_bytes),
130 table_elements: declared.table_elements.unwrap_or(defaults.table_elements),
131 instances: declared.instances.unwrap_or(defaults.instances),
132 deadline_seconds: declared
133 .deadline_seconds
134 .unwrap_or(defaults.deadline_seconds),
135 }
136 }
137
138 #[must_use]
147 pub fn for_hook_calls(self, declared_deadline_seconds: Option<u64>) -> Self {
148 Self {
149 deadline_seconds: declared_deadline_seconds.unwrap_or(HOOK_DEFAULT_DEADLINE_SECONDS),
150 ..self
151 }
152 }
153
154 #[must_use]
158 pub(crate) fn for_hook_entry(self, entry_deadline_seconds: Option<u64>) -> Self {
159 Self {
160 deadline_seconds: entry_deadline_seconds.unwrap_or(self.deadline_seconds),
161 ..self
162 }
163 }
164
165 pub(crate) fn deadline_ticks(&self) -> u64 {
170 let tick_ms = EPOCH_TICK_INTERVAL.as_millis() as u64;
171 (self.deadline_seconds.saturating_mul(1_000) / tick_ms).max(1)
172 }
173
174 pub(crate) fn wall_bound(&self) -> Duration {
177 Duration::from_secs(self.deadline_seconds).saturating_add(WALL_DEADLINE_GRACE)
178 }
179
180 pub(crate) fn limiter(&self) -> ExecutionLimiter {
182 ExecutionLimiter::new(*self)
183 }
184}
185
186pub(crate) struct ExecutionLimiter {
198 inner: StoreLimits,
199 limits: ExecutionLimits,
200 denial: Option<String>,
202}
203
204impl ExecutionLimiter {
205 fn new(limits: ExecutionLimits) -> Self {
206 let inner = StoreLimitsBuilder::new()
207 .memory_size(limits.memory_bytes)
208 .table_elements(limits.table_elements)
209 .instances(limits.instances)
210 .trap_on_grow_failure(false)
211 .build();
212 Self {
213 inner,
214 limits,
215 denial: None,
216 }
217 }
218
219 pub(crate) fn limits(&self) -> ExecutionLimits {
221 self.limits
222 }
223
224 pub(crate) fn denial(&self) -> Option<&str> {
226 self.denial.as_deref()
227 }
228}
229
230impl ResourceLimiter for ExecutionLimiter {
231 fn memory_growing(
232 &mut self,
233 current: usize,
234 desired: usize,
235 maximum: Option<usize>,
236 ) -> wasmtime::Result<bool> {
237 if self.inner.memory_growing(current, desired, maximum)? {
238 return Ok(true);
239 }
240 let message = format!(
241 "linear memory growth to {desired} bytes exceeds \
242 capabilities.limits.memory_bytes ({})",
243 self.limits.memory_bytes
244 );
245 self.denial = Some(message.clone());
246 Err(wasmtime::Error::msg(message))
247 }
248
249 fn table_growing(
250 &mut self,
251 current: usize,
252 desired: usize,
253 maximum: Option<usize>,
254 ) -> wasmtime::Result<bool> {
255 if self.inner.table_growing(current, desired, maximum)? {
256 return Ok(true);
257 }
258 let message = format!(
259 "table growth to {desired} elements exceeds \
260 capabilities.limits.table_elements ({})",
261 self.limits.table_elements
262 );
263 self.denial = Some(message.clone());
264 Err(wasmtime::Error::msg(message))
265 }
266
267 fn instances(&self) -> usize {
268 self.inner.instances()
269 }
270
271 fn tables(&self) -> usize {
272 self.inner.tables()
273 }
274
275 fn memories(&self) -> usize {
276 self.inner.memories()
277 }
278}
279
280pub(crate) struct EpochTicker {
286 stop: Arc<AtomicBool>,
287}
288
289impl EpochTicker {
290 pub(crate) fn spawn(engine: &Engine) -> Self {
302 let weak = engine.weak();
303 let stop = Arc::new(AtomicBool::new(false));
304 let thread_stop = Arc::clone(&stop);
305
306 thread::spawn(move || {
307 while !thread_stop.load(Ordering::Relaxed) {
308 thread::sleep(EPOCH_TICK_INTERVAL);
309 match weak.upgrade() {
310 Some(engine) => engine.increment_epoch(),
311 None => break,
312 }
313 }
314 });
315
316 Self { stop }
317 }
318}
319
320impl Drop for EpochTicker {
321 fn drop(&mut self) {
325 self.stop.store(true, Ordering::Relaxed);
326 }
327}
328
329#[derive(Debug, Clone, Copy, PartialEq, Eq)]
331pub(crate) struct WallDeadline {
332 pub(crate) elapsed: Duration,
333}
334
335pub(crate) async fn bound_call<F: Future>(
341 bound: Duration,
342 call: F,
343) -> Result<F::Output, WallDeadline> {
344 let started = Instant::now();
345 tokio::time::timeout(bound, call)
346 .await
347 .map_err(|_| WallDeadline {
348 elapsed: started.elapsed(),
349 })
350}
351
352#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
354pub(crate) enum ComponentRole {
355 Capsule,
357 Driver,
360 Tool,
362 Hook,
363 ProcessDriver,
364}
365
366impl ComponentRole {
367 pub(crate) fn wire_name(self) -> &'static str {
369 match self {
370 Self::Capsule => "capsule",
371 Self::Driver => "driver",
372 Self::Tool => "tool",
373 Self::Hook => "hook",
374 Self::ProcessDriver => "process_driver",
375 }
376 }
377}
378
379#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
381pub(crate) struct CallDeadline {
382 pub(crate) component: String,
384 pub(crate) role: ComponentRole,
385 pub(crate) call: String,
388 pub(crate) deadline_seconds: u64,
389 pub(crate) elapsed_ms: u64,
390}
391
392#[derive(Debug, Clone, Default)]
401pub(crate) struct CallDeadlines(Arc<Mutex<Vec<CallDeadline>>>);
402
403impl CallDeadlines {
404 pub(crate) async fn bound<F: Future>(
411 &self,
412 component: &str,
413 role: ComponentRole,
414 call: &str,
415 limits: &ExecutionLimits,
416 future: F,
417 ) -> Result<F::Output, WallDeadline> {
418 let deadline = (limits.deadline_seconds, limits.wall_bound());
419 self.bound_within(component, role, call, deadline, future)
420 .await
421 }
422
423 async fn bound_within<F: Future>(
426 &self,
427 component: &str,
428 role: ComponentRole,
429 call: &str,
430 (deadline_seconds, bound): (u64, Duration),
431 future: F,
432 ) -> Result<F::Output, WallDeadline> {
433 let bounded = bound_call(bound, future).await;
434 if let Err(cut) = &bounded {
435 self.lock().push(CallDeadline {
436 component: component.to_string(),
437 role,
438 call: call.to_string(),
439 deadline_seconds,
440 elapsed_ms: u64::try_from(cut.elapsed.as_millis()).unwrap_or(u64::MAX),
441 });
442 }
443 bounded
444 }
445
446 pub(crate) fn record(&self, cut: CallDeadline) {
448 self.lock().push(cut);
449 }
450
451 pub(crate) fn take(&self) -> Vec<CallDeadline> {
453 std::mem::take(&mut *self.lock())
454 }
455
456 #[cfg(test)]
458 pub(crate) fn push_for_test(&self, cut: CallDeadline) {
459 self.lock().push(cut);
460 }
461
462 fn lock(&self) -> std::sync::MutexGuard<'_, Vec<CallDeadline>> {
463 self.0
464 .lock()
465 .unwrap_or_else(std::sync::PoisonError::into_inner)
466 }
467}
468
469pub(crate) fn wall_deadline_message(subject: &str, seconds: u64) -> String {
472 format!(
473 "{subject} waited on the host past its {seconds}s execution deadline \
474 (capabilities.limits.deadline_seconds) and was stopped"
475 )
476}
477
478pub(crate) fn wall_deadline_hook_message(
481 hook: &str,
482 entry: &str,
483 fn_name: &str,
484 seconds: u64,
485) -> String {
486 format!(
487 "hook '{hook}' {fn_name} waited on the host past its {seconds}s execution deadline \
488 (capabilities.limits.deadline_seconds on the artifacts: entry '{entry}') and was stopped"
489 )
490}
491
492#[derive(Debug, Clone, PartialEq, Eq)]
495pub(crate) enum GuestFailure {
496 DeadlineExceeded { seconds: u64 },
498 ResourceLimit { message: String },
500 Other,
502}
503
504pub(crate) fn classify_guest_failure(
512 error: &wasmtime::Error,
513 limiter: &ExecutionLimiter,
514) -> GuestFailure {
515 if let Some(message) = limiter.denial() {
516 return GuestFailure::ResourceLimit {
517 message: message.to_string(),
518 };
519 }
520 if matches!(
521 error.downcast_ref::<wasmtime::Trap>(),
522 Some(wasmtime::Trap::Interrupt)
523 ) {
524 return GuestFailure::DeadlineExceeded {
525 seconds: limiter.limits().deadline_seconds,
526 };
527 }
528 GuestFailure::Other
529}
530
531impl GuestFailure {
532 pub(crate) fn message(&self, subject: &str, error: &wasmtime::Error) -> String {
539 match self {
540 Self::DeadlineExceeded { seconds } => format!(
541 "{subject} exceeded its {seconds}s execution deadline \
542 (capabilities.limits.deadline_seconds) and was interrupted"
543 ),
544 Self::ResourceLimit { message } => {
545 format!("{subject} exceeded its configured resource limits: {message}")
546 }
547 Self::Other => format!("{subject} trapped: {error}"),
548 }
549 }
550
551 pub(crate) fn hook_message(
555 &self,
556 hook: &str,
557 entry: &str,
558 fn_name: &str,
559 error: &wasmtime::Error,
560 ) -> String {
561 let subject = format!("hook '{hook}' {fn_name}");
562 match self {
563 Self::DeadlineExceeded { seconds } => format!(
564 "{subject} exceeded its {seconds}s execution deadline \
565 (capabilities.limits.deadline_seconds on the artifacts: entry '{entry}') and was \
566 interrupted"
567 ),
568 _ => self.message(&subject, error),
569 }
570 }
571}
572
573#[cfg(test)]
574mod tests {
575 use super::*;
576
577 #[test]
578 fn resolve_uses_defaults_when_manifest_is_silent() {
579 assert_eq!(ExecutionLimits::resolve(None), ExecutionLimits::default());
580 }
581
582 #[test]
583 fn resolve_fills_each_omitted_field_independently() {
584 let declared = murmur_artifact::ResourceLimits {
585 memory_bytes: Some(1024),
586 table_elements: None,
587 instances: None,
588 deadline_seconds: Some(5),
589 };
590 let limits = ExecutionLimits::resolve(Some(&declared));
591
592 assert_eq!(limits.memory_bytes, 1024);
593 assert_eq!(limits.deadline_seconds, 5);
594 assert_eq!(limits.table_elements, DEFAULT_TABLE_ELEMENTS);
595 assert_eq!(limits.instances, DEFAULT_INSTANCES);
596 }
597
598 #[test]
601 fn hook_calls_get_the_lower_deadline_default_when_none_is_declared() {
602 let capsule = ExecutionLimits::resolve(None);
603 let hook = capsule.for_hook_calls(None);
604
605 assert_eq!(capsule.deadline_seconds, DEFAULT_DEADLINE_SECONDS);
606 assert_eq!(hook.deadline_seconds, HOOK_DEFAULT_DEADLINE_SECONDS);
607 assert_eq!(hook.memory_bytes, capsule.memory_bytes);
608 assert_eq!(hook.table_elements, capsule.table_elements);
609 assert_eq!(hook.instances, capsule.instances);
610 }
611
612 #[test]
616 fn an_explicit_deadline_is_honored_for_hook_calls() {
617 let declared = murmur_artifact::ResourceLimits {
618 memory_bytes: None,
619 table_elements: None,
620 instances: None,
621 deadline_seconds: Some(DEFAULT_DEADLINE_SECONDS),
622 };
623 let capsule = ExecutionLimits::resolve(Some(&declared));
624
625 assert_eq!(
626 capsule
627 .for_hook_calls(declared.deadline_seconds)
628 .deadline_seconds,
629 DEFAULT_DEADLINE_SECONDS
630 );
631 assert_eq!(capsule.for_hook_calls(Some(5)).deadline_seconds, 5);
632 }
633
634 #[test]
635 fn deadline_ticks_converts_seconds_and_never_yields_zero() {
636 let limits = ExecutionLimits {
637 deadline_seconds: 1,
638 ..ExecutionLimits::default()
639 };
640 assert_eq!(limits.deadline_ticks(), 10);
642 assert_eq!(ExecutionLimits::default().deadline_ticks(), 6_000);
643 }
644
645 #[test]
649 fn classify_reports_resource_limit_after_a_denial() {
650 let limits = ExecutionLimits {
651 memory_bytes: 1024,
652 ..ExecutionLimits::default()
653 };
654 let mut limiter = limits.limiter();
655 let error = limiter
656 .memory_growing(0, 4096, None)
657 .expect_err("growth past memory_bytes must be denied");
658
659 match classify_guest_failure(&error, &limiter) {
660 GuestFailure::ResourceLimit { message } => {
661 assert!(message.contains("capabilities.limits.memory_bytes"));
662 }
663 other => panic!("expected ResourceLimit, got {other:?}"),
664 }
665 }
666
667 #[test]
668 fn classify_reports_deadline_for_an_interrupt_trap() {
669 let limits = ExecutionLimits {
670 deadline_seconds: 7,
671 ..ExecutionLimits::default()
672 };
673 let limiter = limits.limiter();
674 let error = wasmtime::Error::from(wasmtime::Trap::Interrupt);
675
676 assert_eq!(
677 classify_guest_failure(&error, &limiter),
678 GuestFailure::DeadlineExceeded { seconds: 7 }
679 );
680 }
681
682 #[test]
683 fn a_hook_deadline_names_the_key_on_the_hooks_entry() {
684 let error = wasmtime::Error::from(wasmtime::Trap::Interrupt);
685 let failure = GuestFailure::DeadlineExceeded { seconds: 2 };
686 assert_eq!(
687 failure.hook_message("compact", "local/compact", "on-compaction", &error),
688 "hook 'compact' on-compaction exceeded its 2s execution deadline \
689 (capabilities.limits.deadline_seconds on the artifacts: entry 'local/compact') and \
690 was interrupted"
691 );
692 assert_eq!(
693 failure.message("tool 'echo'", &error),
694 "tool 'echo' exceeded its 2s execution deadline (capabilities.limits.deadline_seconds) \
695 and was interrupted",
696 "every other guest keeps the capsule-wide key"
697 );
698 assert_eq!(
699 GuestFailure::Other.hook_message("h", "local/h", "on-stage", &error),
700 GuestFailure::Other.message("hook 'h' on-stage", &error)
701 );
702 }
703
704 #[test]
707 fn classify_reports_other_for_an_unrelated_trap() {
708 let limiter = ExecutionLimits::default().limiter();
709 let error = wasmtime::Error::from(wasmtime::Trap::UnreachableCodeReached);
710
711 let failure = classify_guest_failure(&error, &limiter);
712 assert_eq!(failure, GuestFailure::Other);
713 assert!(failure
714 .message("tool 'echo'", &error)
715 .starts_with("tool 'echo' trapped: "));
716 }
717
718 #[test]
719 fn the_wall_bound_is_the_deadline_plus_a_positive_grace() {
720 assert!(WALL_DEADLINE_GRACE > Duration::ZERO);
721 let limits = ExecutionLimits {
722 deadline_seconds: 3,
723 ..ExecutionLimits::default()
724 };
725 assert_eq!(
726 limits.wall_bound(),
727 Duration::from_secs(3) + WALL_DEADLINE_GRACE
728 );
729 }
730
731 #[tokio::test]
732 async fn bound_call_returns_a_finished_futures_output() {
733 assert_eq!(
734 bound_call(Duration::from_millis(50), async { 7 }).await,
735 Ok(7)
736 );
737 }
738
739 #[tokio::test]
740 async fn bound_call_stops_a_pending_future_after_the_bound() {
741 let bound = Duration::from_millis(30);
742 let cut = bound_call(bound, std::future::pending::<()>())
743 .await
744 .expect_err("a future that never finishes must be stopped");
745 assert!(cut.elapsed >= bound, "stopped after {:?}", cut.elapsed);
746 }
747
748 #[tokio::test]
751 async fn the_latch_records_one_entry_per_firing() {
752 let deadlines = CallDeadlines::default();
753 let bound = (3, Duration::from_millis(30));
754 let finished = deadlines
755 .bound_within("echo", ComponentRole::Tool, "run", bound, async { 1 })
756 .await;
757 assert_eq!(finished, Ok(1));
758 assert!(deadlines.take().is_empty());
759
760 let cut = deadlines
761 .bound_within(
762 "stall",
763 ComponentRole::Driver,
764 "run",
765 bound,
766 std::future::pending::<()>(),
767 )
768 .await;
769 assert!(cut.is_err());
770 let taken = deadlines.take();
771 assert_eq!(taken.len(), 1);
772 assert_eq!(taken[0].component, "stall");
773 assert_eq!(taken[0].role.wire_name(), "driver");
774 assert_eq!(taken[0].call, "run");
775 assert_eq!(taken[0].deadline_seconds, 3);
776 assert!(taken[0].elapsed_ms >= 30);
777 assert!(deadlines.take().is_empty(), "a take clears the latch");
778 }
779
780 #[test]
781 fn wall_deadline_messages_name_the_key_that_set_the_deadline() {
782 assert_eq!(
783 wall_deadline_message("tool 'stall'", 3),
784 "tool 'stall' waited on the host past its 3s execution deadline \
785 (capabilities.limits.deadline_seconds) and was stopped"
786 );
787 assert_eq!(
788 wall_deadline_hook_message("sleep", "local/sleep", "on-session-start", 2),
789 "hook 'sleep' on-session-start waited on the host past its 2s execution deadline \
790 (capabilities.limits.deadline_seconds on the artifacts: entry 'local/sleep') and was \
791 stopped"
792 );
793 }
794
795 #[test]
796 fn limiter_allows_growth_within_the_cap() {
797 let limits = ExecutionLimits {
798 memory_bytes: 4096,
799 ..ExecutionLimits::default()
800 };
801 let mut limiter = limits.limiter();
802
803 assert!(limiter.memory_growing(0, 4096, None).unwrap());
804 assert!(limiter.denial().is_none());
805 }
806}