diff --git a/src/compiler/context.rs b/src/compiler/context.rs index e9f5cf5dd6..eebfcec575 100644 --- a/src/compiler/context.rs +++ b/src/compiler/context.rs @@ -1,11 +1,12 @@ -use super::TimeZone; +use std::ops::ControlFlow; -use super::{Target, state::RuntimeState}; +use super::{ExpressionError, Target, TimeZone, runtime::ExecutionControl, state::RuntimeState}; pub struct Context<'a> { target: &'a mut dyn Target, state: &'a mut RuntimeState, timezone: &'a TimeZone, + execution_control: Option<&'a mut dyn ExecutionControl>, } impl<'a> Context<'a> { @@ -19,6 +20,21 @@ impl<'a> Context<'a> { target, state, timezone, + execution_control: None, + } + } + + pub(crate) fn new_with_control( + target: &'a mut dyn Target, + state: &'a mut RuntimeState, + timezone: &'a TimeZone, + execution_control: &'a mut dyn ExecutionControl, + ) -> Self { + Self { + target, + state, + timezone, + execution_control: Some(execution_control), } } @@ -49,4 +65,25 @@ impl<'a> Context<'a> { pub fn timezone(&self) -> &TimeZone { self.timezone } + + /// Checks whether the embedder has requested that execution stop. + /// + /// VRL calls this between expressions. Functions that perform long-running + /// work can call it at additional safe points. + /// + /// # Errors + /// + /// Returns [`ExpressionError::Interrupted`] when the configured + /// [`ExecutionControl`] requests interruption. With no execution control, + /// this always succeeds. + #[inline] + pub fn checkpoint(&mut self) -> Result<(), ExpressionError> { + match self.execution_control.as_deref_mut() { + Some(control) => match control.checkpoint() { + ControlFlow::Break(()) => Err(ExpressionError::Interrupted), + ControlFlow::Continue(()) => Ok(()), + }, + None => Ok(()), + } + } } diff --git a/src/compiler/expression.rs b/src/compiler/expression.rs index ae2d8c5672..491426be33 100644 --- a/src/compiler/expression.rs +++ b/src/compiler/expression.rs @@ -234,6 +234,8 @@ impl Expression for Expr { Return, Unary, Variable, }; + ctx.checkpoint()?; + match self { Literal(v) => v.resolve(ctx), Container(v) => v.resolve(ctx), diff --git a/src/compiler/expression/assignment.rs b/src/compiler/expression/assignment.rs index 33034cafab..318903d3b3 100644 --- a/src/compiler/expression/assignment.rs +++ b/src/compiler/expression/assignment.rs @@ -4,7 +4,7 @@ use crate::compiler::codes; use crate::compiler::expression::function_call::FunctionCallError::InvalidArgumentKind; use crate::compiler::expression::function_call::InvalidArgumentErrorContext; use crate::compiler::{ - CompileConfig, Context, Expression, Span, TypeDef, + CompileConfig, Context, Expression, ExpressionError, Span, TypeDef, compiler::CompilerError, expression::{Expr, Resolved, assignment::ErrorVariant::InvalidParentPathSegment}, parser::{ @@ -540,6 +540,10 @@ where value } Err(error) => { + if matches!(error, ExpressionError::Interrupted) { + return Err(error); + } + ok.insert(default.clone(), ctx); let value = Value::from(error.to_string()); err.insert(value.clone(), ctx); diff --git a/src/compiler/expression/function_call.rs b/src/compiler/expression/function_call.rs index c20c714bd0..3d92792b7f 100644 --- a/src/compiler/expression/function_call.rs +++ b/src/compiler/expression/function_call.rs @@ -649,7 +649,8 @@ impl FunctionCall { impl Expression for FunctionCall { fn resolve(&self, ctx: &mut Context) -> Resolved { self.expr.resolve(ctx).map_err(|err| match err { - ExpressionError::Abort { .. } + ExpressionError::Interrupted + | ExpressionError::Abort { .. } | ExpressionError::Fallible { .. } | ExpressionError::Missing { .. } => { // propagate the error diff --git a/src/compiler/expression/op.rs b/src/compiler/expression/op.rs index 059fe030f7..7d41bcdb51 100644 --- a/src/compiler/expression/op.rs +++ b/src/compiler/expression/op.rs @@ -3,7 +3,7 @@ use std::fmt; use crate::compiler::codes; use crate::compiler::state::{TypeInfo, TypeState}; use crate::compiler::{ - Context, Expression, TypeDef, + Context, Expression, ExpressionError, TypeDef, expression::{self, Expr, Resolved}, parser::{Node, ast}, value::{ValueError, VrlValueArithmetic}, @@ -128,7 +128,15 @@ impl Expression for Op { use ast::Opcode::{Add, And, Div, Eq, Err, Ge, Gt, Le, Lt, Merge, Mul, Ne, Or, Sub}; match self.opcode { - Err => return self.lhs.resolve(ctx).or_else(|_| self.rhs.resolve(ctx)), + Err => { + return match self.lhs.resolve(ctx) { + std::result::Result::Err(error @ ExpressionError::Interrupted) => { + std::result::Result::Err(error) + } + std::result::Result::Err(_) => self.rhs.resolve(ctx), + result => result, + }; + } Or => { return self .lhs diff --git a/src/compiler/expression_error.rs b/src/compiler/expression_error.rs index 79ac67711f..a50372fa26 100644 --- a/src/compiler/expression_error.rs +++ b/src/compiler/expression_error.rs @@ -1,4 +1,4 @@ -use ExpressionError::{Abort, Error, Fallible, Missing, Return}; +use ExpressionError::{Abort, Error, Fallible, Interrupted, Missing, Return}; use crate::compiler::codes; use crate::diagnostic::{Diagnostic, DiagnosticMessage, Label, Note, Severity, Span}; @@ -8,6 +8,12 @@ pub type Resolved = Result; #[derive(Clone, Debug, Eq, PartialEq)] pub enum ExpressionError { + /// Execution was interrupted by an embedder-provided execution control. + /// + /// Unlike an ordinary expression error, this cannot be caught by VRL's + /// error-coalescing or infallible-assignment expressions. + Interrupted, + Abort { span: Span, message: Option, @@ -59,7 +65,7 @@ impl From for Diagnostic { impl DiagnosticMessage for ExpressionError { fn code(&self) -> usize { match self { - Abort { .. } | Return { .. } | Error { .. } => 0, + Interrupted | Abort { .. } | Return { .. } | Error { .. } => 0, Fallible { .. } => codes::ExprCode::FallibleExpression as usize, Missing { .. } => codes::ExprCode::ExpressionTypeUnavailable as usize, } @@ -67,6 +73,7 @@ impl DiagnosticMessage for ExpressionError { fn message(&self) -> String { match self { + Interrupted => "execution interrupted".to_owned(), Abort { message, .. } => message.clone().unwrap_or_else(|| "aborted".to_owned()), Return { .. } => "return".to_string(), Error { message, .. } => message.clone(), @@ -80,7 +87,7 @@ impl DiagnosticMessage for ExpressionError { Abort { span, .. } => { vec![Label::primary("aborted", span)] } - Return { .. } => Vec::new(), + Interrupted | Return { .. } => Vec::new(), Error { labels, .. } => labels.clone(), Fallible { span } => vec![ Label::primary("expression can result in runtime error", span), @@ -98,7 +105,7 @@ impl DiagnosticMessage for ExpressionError { fn notes(&self) -> Vec { match self { - Return { .. } | Abort { .. } | Missing { .. } => vec![], + Interrupted | Return { .. } | Abort { .. } | Missing { .. } => vec![], Error { notes, .. } => notes.clone(), Fallible { .. } => vec![Note::SeeErrorDocs], } diff --git a/src/compiler/function/closure.rs b/src/compiler/function/closure.rs index 31142e4f81..0676979e70 100644 --- a/src/compiler/function/closure.rs +++ b/src/compiler/function/closure.rs @@ -182,12 +182,10 @@ where err @ Err(_) => err, }; - let value = result?; - cleanup(ctx.state_mut(), key_ident, old_key); cleanup(ctx.state_mut(), value_ident, old_value); - Ok(value) + result } /// Run the closure to completion, given the provided index/value pair, and @@ -211,12 +209,12 @@ where let old_index = insert(ctx.state_mut(), index_ident, index.into()); let old_value = insert(ctx.state_mut(), value_ident, cloned_value); - let value = (self.runner)(ctx)?; + let result = (self.runner)(ctx); cleanup(ctx.state_mut(), index_ident, old_index); cleanup(ctx.state_mut(), value_ident, old_value); - Ok(value) + result } /// Run the closure to completion, given the provided key, and the runtime @@ -233,10 +231,12 @@ where let ident = self.ident(0); let old_key = insert(ctx.state_mut(), ident, cloned_key.into()); - *key = (self.runner)(ctx)?.try_bytes_utf8_lossy()?.into(); + let result = (self.runner)(ctx); cleanup(ctx.state_mut(), ident, old_key); + *key = result?.try_bytes_utf8_lossy()?.into(); + Ok(()) } @@ -254,10 +254,12 @@ where let ident = self.ident(0); let old_value = insert(ctx.state_mut(), ident, cloned_value); - *value = (self.runner)(ctx)?; + let result = (self.runner)(ctx); cleanup(ctx.state_mut(), ident, old_value); + *value = result?; + Ok(()) } diff --git a/src/compiler/runtime.rs b/src/compiler/runtime.rs index 158bb1af1d..f5ed60456e 100644 --- a/src/compiler/runtime.rs +++ b/src/compiler/runtime.rs @@ -1,4 +1,4 @@ -use std::{error::Error, fmt}; +use std::{error::Error, fmt, ops::ControlFlow}; use crate::path::OwnedTargetPath; use crate::value::Value; @@ -10,6 +10,26 @@ use super::{Context, Program, Target, state}; #[allow(clippy::module_name_repetitions)] pub type RuntimeResult = Result; +/// Allows an embedder to cooperatively interrupt VRL execution. +/// +/// VRL invokes [`ExecutionControl::checkpoint`] between expressions. A +/// controller returns [`ControlFlow::Break`] to stop execution or +/// [`ControlFlow::Continue`] to allow it to proceed. The controller owns the +/// interruption policy, such as a deadline, operation budget, or shared +/// cancellation token. +pub trait ExecutionControl { + fn checkpoint(&mut self) -> ControlFlow<()>; +} + +impl ExecutionControl for F +where + F: FnMut() -> ControlFlow<()>, +{ + fn checkpoint(&mut self) -> ControlFlow<()> { + self() + } +} + #[derive(Debug, Default)] pub struct Runtime { state: state::RuntimeState, @@ -18,6 +38,9 @@ pub struct Runtime { /// The error raised if the runtime is terminated. #[derive(Debug, Clone, PartialEq, Eq)] pub enum Terminate { + /// Execution was interrupted by an embedder-provided execution control. + Interrupted, + /// A manual `abort` call. /// /// This is an intentional termination that does not result in an @@ -33,6 +56,7 @@ impl Terminate { #[must_use] pub fn get_expression_error(self) -> ExpressionError { match self { + Terminate::Interrupted => ExpressionError::Interrupted, Terminate::Error(error) | Terminate::Abort(error) => error, } } @@ -41,6 +65,7 @@ impl Terminate { impl fmt::Display for Terminate { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Terminate::Interrupted => f.write_str("execution interrupted"), Terminate::Error(error) | Terminate::Abort(error) => error.fmt(f), } } @@ -98,6 +123,43 @@ impl Runtime { target: &mut dyn Target, program: &Program, timezone: &TimeZone, + ) -> RuntimeResult { + self.resolve_inner(target, program, *timezone, None) + } + + /// Resolves the provided [`Program`] with an embedder-provided execution + /// control. + /// + /// The control is scoped to this invocation and is not stored in the + /// [`Runtime`] or its [`state::RuntimeState`]. Returning + /// [`ControlFlow::Break`] from [`ExecutionControl::checkpoint`] terminates + /// the invocation with [`Terminate::Interrupted`]. + /// + /// This is cooperative interruption: VRL checks between expressions and + /// wherever a function explicitly calls [`Context::checkpoint`]. It does + /// not preempt a single blocking or long-running function call. + /// + /// # Errors + /// + /// Returns [`Terminate::Interrupted`] when the control requests + /// interruption. Other termination conditions are the same as + /// [`Runtime::resolve`]. + pub fn resolve_with_control( + &mut self, + target: &mut dyn Target, + program: &Program, + timezone: &TimeZone, + control: &mut dyn ExecutionControl, + ) -> RuntimeResult { + self.resolve_inner(target, program, *timezone, Some(control)) + } + + fn resolve_inner( + &mut self, + target: &mut dyn Target, + program: &Program, + timezone: TimeZone, + control: Option<&mut dyn ExecutionControl>, ) -> RuntimeResult { // Validate that the path is a value. match target.target_get(&OwnedTargetPath::event_root()) { @@ -114,9 +176,13 @@ impl Runtime { } } - let mut ctx = Context::new(target, &mut self.state, timezone); + let mut ctx = match control { + Some(control) => Context::new_with_control(target, &mut self.state, &timezone, control), + None => Context::new(target, &mut self.state, &timezone), + }; match program.resolve(&mut ctx) { + Err(ExpressionError::Interrupted) => Err(Terminate::Interrupted), Ok(value) | Err(ExpressionError::Return { value, .. }) => Ok(value), Err( err @ (ExpressionError::Abort { .. } @@ -127,3 +193,177 @@ impl Runtime { } } } + +#[cfg(all(test, feature = "stdlib"))] +mod execution_control_tests { + use std::collections::BTreeMap; + use std::ops::ControlFlow; + + use super::{ExecutionControl, Runtime, Terminate, TimeZone}; + use crate::compiler::Program; + use crate::compiler::state::RuntimeState; + use crate::parser::ast::Ident; + use crate::value::Value; + + struct BreakAt { + checkpoint: usize, + break_at: usize, + } + + impl BreakAt { + fn new(break_at: usize) -> Self { + Self { + checkpoint: 0, + break_at, + } + } + } + + impl ExecutionControl for BreakAt { + fn checkpoint(&mut self) -> ControlFlow<()> { + self.checkpoint += 1; + + if self.checkpoint >= self.break_at { + ControlFlow::Break(()) + } else { + ControlFlow::Continue(()) + } + } + } + + fn compile(source: &str) -> Program { + crate::compiler::compile(source, &crate::stdlib::all()) + .expect("program should compile") + .program + } + + fn target() -> Value { + BTreeMap::from([ + ("items".into(), Value::Array(vec![Value::from(1); 100])), + ("value".into(), Value::from("1")), + ]) + .into() + } + + #[test] + fn resolve_without_control_preserves_existing_api() { + let program = compile("1 + 2"); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + + assert_eq!( + runtime.resolve(&mut target, &program, &TimeZone::default()), + Ok(Value::from(3)), + ); + } + + #[test] + fn for_each_loop_returns_interrupted() { + let source = r" + count = 0 + for_each(array!(.items)) -> |_index, _value| { + count = count + 1 + } + count + "; + + let program = compile(source); + + let mut target: Value = + BTreeMap::from([("items".into(), Value::Array(vec![Value::from(1); 5_000]))]).into(); + + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = || ControlFlow::Break(()); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + } + + #[test] + fn interruption_is_not_caught_by_error_coalescing() { + let program = compile("to_int(.value) ?? 2"); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = BreakAt::new(2); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + } + + #[test] + fn interruption_is_not_caught_by_infallible_assignment() { + let program = compile("value, error = to_int(.value)\nvalue"); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = BreakAt::new(2); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + } + + #[test] + fn interruption_is_not_wrapped_by_boolean_or() { + let program = compile("false || true"); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = BreakAt::new(3); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + } + + #[test] + fn control_is_scoped_to_one_resolve_call() { + let program = compile("1"); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = || ControlFlow::Break(()); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + assert_eq!( + runtime.resolve(&mut target, &program, &TimeZone::default()), + Ok(Value::from(1)), + ); + } + + #[test] + fn interruption_restores_closure_variables() { + let source = r#" + item = "outer" + for_each(array!(.items)) -> |_index, item| { + item + } + item + "#; + let program = compile(source); + let mut target = target(); + let mut runtime = Runtime::new(RuntimeState::default()); + let mut control = BreakAt::new(10); + + assert_eq!( + runtime + .resolve_with_control(&mut target, &program, &TimeZone::default(), &mut control,), + Err(Terminate::Interrupted), + ); + assert_eq!( + runtime.state.variable(&Ident::new("item")), + Some(&Value::from("outer")), + ); + } +} diff --git a/src/compiler/value/error.rs b/src/compiler/value/error.rs index e3f56714a1..dd8a8cb2f3 100644 --- a/src/compiler/value/error.rs +++ b/src/compiler/value/error.rs @@ -93,6 +93,10 @@ impl DiagnosticMessage for ValueError { impl From for ExpressionError { fn from(err: ValueError) -> Self { + if let ValueError::Or(ExpressionError::Interrupted) = err { + return Self::Interrupted; + } + Self::Error { message: err.message(), labels: vec![],