feat!: add cooperative execution-time limits across engine, VM, and binding (#539)

- Introduce ExecutionTimer/ExecutionTimerConfig to allow limiting evaluating time.
- To amortize time checking costs, checking interval can be configured via the notion of work units
- A global fallback time limit can be set to universally limit all evaluation in addition to engine level limit setting.
- Implement limnits in interpreter and RVM. In RVM, also handle suspend/resume so that time during pause is not counted.
- Add engine-level APIs to set/clear per-engine timer configuration and apply global fallback defaults.
- Surface execution-time limits through FFI and C# bindings
- Add C# tests and example usage to validate engine overrides, global fallback behavior, and compiled policy enforcement.
- Expand docs for execution-time limit
- Add interpreter YAML cases and VM unit tests for time-limit behavior and deterministic time sources.

Signed-off-by: Anand Krishnamoorthi <anakrish@microsoft.com>

src/rvm/tests/vm.rs

@@ -18,8 +18,17 @@
 
 #[cfg(test)]
 mod tests {
+    use crate::rvm::program::Program;
     use crate::rvm::tests::instruction_parser::{parse_instruction, parse_loop_mode};
     use crate::rvm::tests::test_utils::test_round_trip_serialization;
+    #[cfg(any(test, not(feature = "std")))]
+    use crate::utils::limits::set_time_source;
+    use crate::utils::limits::{
+        acquire_limits_test_lock, fallback_execution_timer_config,
+        set_fallback_execution_timer_config, ExecutionTimerConfig, TimeSource,
+    };
+    use core::num::NonZeroU32;
+    use core::time::Duration;
     #[derive(Debug, Clone, Deserialize, Serialize, Default)]
     struct RuleInfoSpec {
         rule_type: String,
@@ -48,11 +57,122 @@ mod tests {
     use anyhow::Result;
     use serde::{Deserialize, Serialize};
     use std::fs;
+    use std::sync::{Mutex, Once};
     use test_generator::test_resources;
 
     extern crate alloc;
     extern crate std;
 
+    struct FallbackGuard(Option<ExecutionTimerConfig>);
+    impl Drop for FallbackGuard {
+        fn drop(&mut self) {
+            set_fallback_execution_timer_config(self.0);
+        }
+    }
+
+    fn install_fallback_config(config: Option<ExecutionTimerConfig>) -> FallbackGuard {
+        let previous = fallback_execution_timer_config();
+        set_fallback_execution_timer_config(config);
+        FallbackGuard(previous)
+    }
+
+    struct TimeSourceGuard {
+        previous_default: Duration,
+        previous_template: Vec<Duration>,
+    }
+
+    impl Drop for TimeSourceGuard {
+        fn drop(&mut self) {
+            let mut state = TIME_SOURCE_STATE
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            state.default_increment = self.previous_default;
+            state.template_increments = self.previous_template.clone();
+            state.reset_from_template();
+        }
+    }
+
+    fn configure_time_source(
+        increments: Vec<Duration>,
+        default_increment: Duration,
+    ) -> TimeSourceGuard {
+        ensure_time_source_registered();
+
+        let mut state = TIME_SOURCE_STATE
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+
+        let guard = TimeSourceGuard {
+            previous_default: state.default_increment,
+            previous_template: state.template_increments.clone(),
+        };
+
+        state.default_increment = default_increment;
+        state.template_increments = increments;
+        state.reset_from_template();
+
+        guard
+    }
+
+    struct TestTimeSource;
+
+    struct TimeSourceState {
+        current: Duration,
+        started: bool,
+        default_increment: Duration,
+        increments: VecDeque<Duration>,
+        template_increments: Vec<Duration>,
+    }
+
+    impl TimeSourceState {
+        const fn new() -> Self {
+            Self {
+                current: Duration::ZERO,
+                started: false,
+                default_increment: Duration::from_millis(1),
+                increments: VecDeque::new(),
+                template_increments: Vec::new(),
+            }
+        }
+
+        fn reset_from_template(&mut self) {
+            self.current = Duration::ZERO;
+            self.started = false;
+            self.increments = VecDeque::from(self.template_increments.clone());
+        }
+    }
+
+    impl TimeSource for TestTimeSource {
+        fn now(&self) -> Option<Duration> {
+            let mut state = TIME_SOURCE_STATE
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+
+            if !state.started {
+                state.started = true;
+                return Some(state.current);
+            }
+
+            let increment = state
+                .increments
+                .pop_front()
+                .unwrap_or(state.default_increment);
+            state.current = state.current.saturating_add(increment);
+            Some(state.current)
+        }
+    }
+
+    static TEST_TIME_SOURCE: TestTimeSource = TestTimeSource;
+    static TIME_SOURCE_STATE: Mutex<TimeSourceState> = Mutex::new(TimeSourceState::new());
+    static TIME_SOURCE_ONCE: Once = Once::new();
+
+    fn ensure_time_source_registered() {
+        #[cfg(any(test, not(feature = "std")))]
+        TIME_SOURCE_ONCE.call_once(|| {
+            let _ = set_time_source(&TEST_TIME_SOURCE);
+        });
+    }
+
     #[derive(Debug, Clone, Deserialize, Serialize)]
     struct HostAwaitResponseSpec {
         id: crate::Value,
@@ -62,6 +182,13 @@ mod tests {
         values: Vec<crate::Value>,
     }
 
+    fn default_vm_test_execution_timer_config() -> ExecutionTimerConfig {
+        ExecutionTimerConfig {
+            limit: Duration::from_secs(1),
+            check_interval: NonZeroU32::new(100).unwrap_or(NonZeroU32::MIN),
+        }
+    }
+
     #[derive(Debug, Deserialize, Serialize)]
     struct VmTestCase {
         note: String,
@@ -644,6 +771,7 @@ mod tests {
             vm.set_execution_mode(mode);
             vm.set_step_mode(use_step_mode);
             vm.set_strict_builtin_errors(strict);
+            vm.set_execution_timer_config(Some(default_vm_test_execution_timer_config()));
 
             if let Some(data_value) = processed_data.clone() {
                 vm.set_data(data_value)?;
@@ -1001,6 +1129,158 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn vm_execution_time_limit_triggers_error() -> Result<()> {
+        use crate::rvm::instructions::Instruction;
+        use crate::utils::limits::acquire_limits_test_lock;
+        use core::num::NonZeroU32;
+        use core::time::Duration;
+
+        let _lock = acquire_limits_test_lock();
+        let config = ExecutionTimerConfig {
+            limit: Duration::from_nanos(1),
+            check_interval: NonZeroU32::new(1).unwrap(),
+        };
+        let _guard = install_fallback_config(Some(config));
+
+        let mut program = Program::new();
+        program.dispatch_window_size = 2;
+        program.max_rule_window_size = 2;
+        program.entry_points.insert("main".to_string(), 0);
+
+        const INSTRUCTION_COUNT: usize = 60_000;
+        program.instructions = (0..INSTRUCTION_COUNT)
+            .map(|_| Instruction::LoadNull { dest: 0 })
+            .collect();
+        program.instructions.push(Instruction::Return { value: 0 });
+        program.instruction_spans = alloc::vec![None; program.instructions.len()];
+        program.main_entry_point = 0;
+
+        let program = Arc::new(program);
+
+        let mut vm = RegoVM::new();
+        vm.set_max_instructions(usize::MAX);
+        vm.load_program(program);
+
+        let result = vm.execute();
+        assert!(
+            matches!(result, Err(VmError::TimeLimitExceeded { .. })),
+            "expected time limit error but got {result:?}"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn vm_execution_time_limit_override_allows_completion() -> Result<()> {
+        use crate::rvm::instructions::Instruction;
+        use crate::utils::limits::acquire_limits_test_lock;
+        use core::num::NonZeroU32;
+        use core::time::Duration;
+
+        let _lock = acquire_limits_test_lock();
+        let strict_config = ExecutionTimerConfig {
+            limit: Duration::from_nanos(1),
+            check_interval: NonZeroU32::new(1).unwrap(),
+        };
+        let _guard = install_fallback_config(Some(strict_config));
+
+        let mut program = Program::new();
+        program.dispatch_window_size = 2;
+        program.max_rule_window_size = 2;
+        program.entry_points.insert("main".to_string(), 0);
+        program.instructions = alloc::vec![
+            Instruction::LoadNull { dest: 0 },
+            Instruction::Return { value: 0 },
+        ];
+        program.instruction_spans = alloc::vec![None; program.instructions.len()];
+        program.main_entry_point = 0;
+
+        let program = Arc::new(program);
+
+        let mut vm = RegoVM::new();
+        vm.load_program(program);
+
+        let relaxed_config = ExecutionTimerConfig {
+            limit: Duration::from_millis(10),
+            check_interval: NonZeroU32::new(1).unwrap(),
+        };
+        vm.set_execution_timer_config(Some(relaxed_config));
+
+        let result = vm.execute();
+        assert!(
+            result.is_ok(),
+            "expected successful execution, got {result:?}"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn vm_suspend_resume_excludes_suspended_time_from_limit() -> Result<()> {
+        use crate::rvm::instructions::Instruction;
+
+        let _lock = acquire_limits_test_lock();
+        let _guard = install_fallback_config(Some(ExecutionTimerConfig {
+            limit: Duration::from_millis(10),
+            check_interval: NonZeroU32::new(1).unwrap(),
+        }));
+
+        let _time_guard = configure_time_source(
+            alloc::vec![
+                Duration::from_millis(1),
+                Duration::from_millis(1),
+                Duration::from_millis(1),
+                Duration::from_millis(1),
+                Duration::from_millis(100),
+                Duration::from_millis(1),
+            ],
+            Duration::from_millis(1),
+        );
+
+        let mut program = Program::new();
+        program.dispatch_window_size = 3;
+        program.max_rule_window_size = 3;
+        program.entry_points.insert("main".to_string(), 0);
+        program.literals = alloc::vec![Value::from("id"), Value::from(1)];
+        program.instructions = alloc::vec![
+            Instruction::Load {
+                dest: 0,
+                literal_idx: 0
+            },
+            Instruction::Load {
+                dest: 1,
+                literal_idx: 1
+            },
+            Instruction::HostAwait {
+                dest: 2,
+                arg: 1,
+                id: 0
+            },
+            Instruction::Return { value: 2 },
+        ];
+        program.instruction_spans = alloc::vec![None; program.instructions.len()];
+        program.main_entry_point = 0;
+
+        let program = Arc::new(program);
+        let mut vm = RegoVM::new();
+        vm.set_execution_mode(ExecutionMode::Suspendable);
+        vm.load_program(program);
+
+        let _ = vm.execute()?;
+        match vm.execution_state() {
+            ExecutionState::Suspended { reason, .. } => {
+                assert!(matches!(reason, SuspendReason::HostAwait { .. }));
+            }
+            other => panic!("expected suspension, got {other:?}"),
+        }
+
+        let resumed = vm.resume(Some(Value::from(42)))?;
+        assert_eq!(resumed, Value::from(42));
+
+        Ok(())
+    }
+
     #[test_resources("tests/rvm/vm/suites/*.yaml")]
     fn run_vm_test_file(file: &str) {
         run_vm_test_suite(file).unwrap()

src/rvm/vm/errors.rs

@@ -5,6 +5,7 @@ use super::execution_model::SuspendReason;
 use crate::value::Value;
 use alloc::string::String;
 use alloc::vec::Vec;
+use core::time::Duration;
 use thiserror::Error;
 
 /// VM execution errors
@@ -17,7 +18,14 @@ pub enum VmError {
         pc: usize,
     },
 
-    #[error("Execution stopped: exceeded maximum memory limit of {limit} bytes with usage {usage} bytes (pc={pc})")]
+    #[error("Execution exceeded time limit (elapsed={elapsed:?}, limit={limit:?}, pc={pc})")]
+    TimeLimitExceeded {
+        elapsed: Duration,
+        limit: Duration,
+        pc: usize,
+    },
+
+    #[error("Execution exceeded memory limit (usage={usage} bytes, limit={limit} bytes, pc={pc})")]
     MemoryLimitExceeded { usage: u64, limit: u64, pc: usize },
 
     #[error("Literal index {index} out of bounds (pc={pc})")]

src/rvm/vm/execution.rs

@@ -59,6 +59,7 @@ impl RegoVM {
         match self.execution_mode {
             ExecutionMode::RunToCompletion => {
                 self.reset_execution_state();
+                self.reset_execution_timer_state();
 
                 self.validate_vm_state()?;
                 let entry_point_pc_u32 = u32::try_from(entry_point_pc).map_err(|_| {
@@ -73,6 +74,7 @@ impl RegoVM {
             }
             ExecutionMode::Suspendable => {
                 self.reset_execution_state();
+                self.reset_execution_timer_state();
 
                 self.validate_vm_state()?;
                 self.execute_suspendable_entry(entry_point_pc)
@@ -101,6 +103,7 @@ impl RegoVM {
         match self.execution_mode {
             ExecutionMode::RunToCompletion => {
                 self.reset_execution_state();
+                self.reset_execution_timer_state();
 
                 self.validate_vm_state()?;
                 let entry_point_pc_u32 = u32::try_from(entry_point_pc).map_err(|_| {
@@ -115,6 +118,7 @@ impl RegoVM {
             }
             ExecutionMode::Suspendable => {
                 self.reset_execution_state();
+                self.reset_execution_timer_state();
 
                 self.validate_vm_state()?;
                 self.execute_suspendable_entry(entry_point_pc)
@@ -136,6 +140,7 @@ impl RegoVM {
                 });
             }
 
+            self.execution_timer_tick(1)?;
             self.executed_instructions = self.executed_instructions.saturating_add(1);
             let instruction = program.instructions.get(self.pc).cloned().ok_or(
                 VmError::ProgramCounterOutOfBounds {
@@ -168,6 +173,7 @@ impl RegoVM {
 
     fn execute_run_to_completion(&mut self) -> Result<Value> {
         self.reset_execution_state();
+        self.reset_execution_timer_state();
         self.execution_state = ExecutionState::Running;
         match self.jump_to(0_u32) {
             Ok(value) => {
@@ -185,6 +191,7 @@ impl RegoVM {
 
     fn execute_suspendable(&mut self) -> Result<Value> {
         self.reset_execution_state();
+        self.reset_execution_timer_state();
         self.execution_state = ExecutionState::Running;
         match self.run_stackless_from(0) {
             Ok(result) => Ok(result),
@@ -197,6 +204,7 @@ impl RegoVM {
 
     fn execute_suspendable_entry(&mut self, entry_point_pc: usize) -> Result<Value> {
         self.execution_state = ExecutionState::Running;
+        self.reset_execution_timer_state();
         match self.run_stackless_from(entry_point_pc) {
             Ok(result) => Ok(result),
             Err(err) => {
@@ -256,6 +264,7 @@ impl RegoVM {
         }
 
         self.execution_state = ExecutionState::Running;
+        self.restore_execution_timer_after_resume();
 
         let program = self.program.clone();
         self.run_stackless_loop(&program, &mut last_result)?;
@@ -376,6 +385,10 @@ impl RegoVM {
             }
 
             self.pc = frame_pc;
+            if let Err(err) = self.execution_timer_tick(1) {
+                self.execution_state = ExecutionState::Error { error: err.clone() };
+                return Err(err);
+            }
             let instruction = program.instructions.get(self.pc).cloned().ok_or(
                 VmError::ProgramCounterOutOfBounds {
                     pc: self.pc,
@@ -460,6 +473,7 @@ impl RegoVM {
             }
         }
 
+        self.snapshot_execution_timer_on_suspend();
         self.execution_state = ExecutionState::Suspended {
             reason,
             pc: self.pc,

src/rvm/vm/machine.rs

@@ -3,15 +3,21 @@
 
 use crate::rvm::program::Program;
 #[cfg(feature = "allocator-memory-limits")]
-use crate::utils::limits::{self, LimitError};
+use crate::utils::limits;
+use crate::utils::limits::{
+    fallback_execution_timer_config, monotonic_now, ExecutionTimer, ExecutionTimerConfig,
+    LimitError,
+};
 use crate::value::Value;
 use crate::CompiledPolicy;
 use alloc::collections::{btree_map::Entry, BTreeMap, VecDeque};
+#[cfg(feature = "allocator-memory-limits")]
 use alloc::format;
 use alloc::string::String;
 use alloc::sync::Arc;
 use alloc::vec;
 use alloc::vec::Vec;
+use core::time::Duration;
 
 use super::context::{CallRuleContext, ComprehensionContext, LoopContext};
 use super::errors::{Result, VmError};
@@ -105,6 +111,15 @@ pub struct RegoVM {
 
     /// Cache for builtin calls that must stay deterministic across a single evaluation
     pub(super) builtins_cache: BTreeMap<(&'static str, Vec<Value>), Value>,
+
+    /// Optional override for the execution timer configuration
+    pub(super) execution_timer_config: Option<ExecutionTimerConfig>,
+
+    /// Cooperative execution timer used to enforce wall-clock limits
+    pub(super) execution_timer: ExecutionTimer,
+
+    /// Elapsed wall-clock time recorded when the VM entered a suspended state
+    pub(super) execution_timer_elapsed_at_suspend: Option<Duration>,
 }
 
 impl Default for RegoVM {
@@ -116,6 +131,8 @@ impl Default for RegoVM {
 impl RegoVM {
     /// Create a new virtual machine
     pub fn new() -> Self {
+        let fallback_timer = fallback_execution_timer_config();
+
         RegoVM {
             registers: Vec::new(), // Start with no registers - will be resized when program is loaded
             pc: 0,
@@ -143,6 +160,9 @@ impl RegoVM {
             frame_pc_overridden: false,
             strict_builtin_errors: false,
             builtins_cache: BTreeMap::new(),
+            execution_timer_config: None,
+            execution_timer: ExecutionTimer::new(fallback_timer),
+            execution_timer_elapsed_at_suspend: None,
         }
     }
 
@@ -319,6 +339,93 @@ impl RegoVM {
         self.execution_mode
     }
 
+    /// Configure the execution timer to use the supplied configuration, or fall back to the global
+    /// default when `None` is provided.
+    pub fn set_execution_timer_config(&mut self, config: Option<ExecutionTimerConfig>) {
+        self.execution_timer_config = config;
+        self.reset_execution_timer_state();
+    }
+
+    /// Returns the currently configured execution timer, if any.
+    pub const fn execution_timer_config(&self) -> Option<ExecutionTimerConfig> {
+        self.execution_timer_config
+    }
+
+    pub(super) fn reset_execution_timer_state(&mut self) {
+        let config = self.effective_execution_timer_config();
+        self.execution_timer = ExecutionTimer::new(config);
+        self.execution_timer_elapsed_at_suspend = None;
+
+        if config.is_none() {
+            return;
+        }
+
+        if let Some(now) = monotonic_now() {
+            self.execution_timer.start(now);
+        }
+    }
+
+    fn effective_execution_timer_config(&self) -> Option<ExecutionTimerConfig> {
+        self.execution_timer_config
+            .or_else(fallback_execution_timer_config)
+    }
+
+    pub(super) fn execution_timer_tick(&mut self, work_units: u32) -> Result<()> {
+        if self.execution_timer.limit().is_none() {
+            return Ok(());
+        }
+
+        let Some(now) = monotonic_now() else {
+            return Ok(());
+        };
+
+        self.execution_timer
+            .tick(work_units, now)
+            .map_err(|err| match err {
+                LimitError::TimeLimitExceeded { elapsed, limit } => VmError::TimeLimitExceeded {
+                    elapsed,
+                    limit,
+                    pc: self.pc,
+                },
+                LimitError::MemoryLimitExceeded { usage, limit } => VmError::MemoryLimitExceeded {
+                    usage,
+                    limit,
+                    pc: self.pc,
+                },
+            })
+    }
+
+    pub(super) fn snapshot_execution_timer_on_suspend(&mut self) {
+        if self.execution_timer.config().is_none() {
+            self.execution_timer_elapsed_at_suspend = None;
+            return;
+        }
+
+        let Some(now) = monotonic_now() else {
+            self.execution_timer_elapsed_at_suspend = None;
+            return;
+        };
+
+        self.execution_timer_elapsed_at_suspend = self.execution_timer.elapsed(now);
+    }
+
+    pub(super) fn restore_execution_timer_after_resume(&mut self) {
+        if self.execution_timer.config().is_none() {
+            self.execution_timer_elapsed_at_suspend = None;
+            return;
+        }
+
+        let Some(elapsed) = self.execution_timer_elapsed_at_suspend.take() else {
+            return;
+        };
+
+        let Some(now) = monotonic_now() else {
+            return;
+        };
+
+        self.execution_timer.resume_from_elapsed(now, elapsed);
+    }
+
     /// Get the current execution state of the VM
     pub const fn execution_state(&self) -> &ExecutionState {
         &self.execution_state