Add optional execution-cancellation breakpoint

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(()),
+        }
+    }
 }

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),

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);

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

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

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<Value, ExpressionError>;
 
 #[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<String>,
@@ -59,14 +65,15 @@ impl From<ExpressionError> 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,
         }
     }
 
     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<Note> {
         match self {
-            Return { .. } | Abort { .. } | Missing { .. } => vec![],
+            Interrupted | Return { .. } | Abort { .. } | Missing { .. } => vec![],
             Error { notes, .. } => notes.clone(),
             Fallible { .. } => vec![Note::SeeErrorDocs],
         }

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(())
     }