From 3e1b1b2be6d82a4e7b661c494b4c864fa0a92bd3 Mon Sep 17 00:00:00 2001 From: Sean Huh Date: Thu, 17 Sep 2026 12:15:45 -0700 Subject: [PATCH] Add async execution engine and AsyncDriver PiperOrigin-RevId: 983346214 --- .../java/dev/cel/runtime/CelRuntimeImpl.java | 8 +- .../java/dev/cel/runtime/planner/BUILD.bazel | 46 + .../cel/runtime/planner/EvalAsyncCall.java | 110 +- .../cel/runtime/planner/EvalConditional.java | 2 +- .../runtime/planner/EvalExhaustiveAnd.java | 14 +- .../planner/EvalExhaustiveConditional.java | 18 +- .../cel/runtime/planner/EvalExhaustiveOr.java | 14 +- .../dev/cel/runtime/planner/EvalHelpers.java | 24 +- .../runtime/planner/EvalLateBoundCall.java | 12 + .../cel/runtime/planner/ExecutionFrame.java | 54 +- .../cel/runtime/planner/PlannedProgram.java | 183 +- .../cel/runtime/planner/ProgramPlanner.java | 42 +- .../java/dev/cel/runtime/planner/BUILD.bazel | 2 + .../planner/ProgramPlannerAsyncTest.java | 1790 +++++++++++++++++ .../runtime/planner/ProgramPlannerTest.java | 17 +- 15 files changed, 2287 insertions(+), 49 deletions(-) create mode 100644 runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerAsyncTest.java diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java index c3bff8dfd..333f72657 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java @@ -191,8 +191,11 @@ public ListenableFuture evalAsync(PartialVars partialVars) { @Override public ListenableFuture evalAsync(Message message) { - throw new UnsupportedOperationException( - "evalAsync is not supported by this Program implementation."); + checkNotNull(message, "message"); + return program.evalAsync( + ProtoMessageActivationFactory.fromProto(message, program.options()), + CelFunctionResolver.EMPTY, + /* partialVars= */ null); } @Override @@ -580,6 +583,7 @@ public CelRuntime build() { container(), options(), lateBoundFunctionNamesBuilder().build(), + runtimeEquality, asyncEvaluationOptions(), asyncExecutor().orElse(null)); setPlanner(planner); diff --git a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel index a8882f539..1e7a78966 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel @@ -61,6 +61,8 @@ java_library( "//runtime:evaluation_exception_builder", "//runtime:function_overload", "//runtime:resolved_overload", + "//runtime:runtime_equality", + "//runtime:runtime_helpers", "@maven//:com_google_code_findbugs_annotations", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", @@ -74,6 +76,9 @@ java_library( tags = [ ], deps = [ + ":async_call_state_tracker", + ":async_completion_coordinator", + ":async_gate", ":error_metadata", ":localized_evaluation_exception", ":planned_interpretable", @@ -82,6 +87,7 @@ java_library( "//common/annotations", "//common/exceptions:runtime_exception", "//common/values", + "//runtime:accumulated_unknowns", "//runtime:activation", "//runtime:async_options", "//runtime:evaluation_exception", @@ -92,6 +98,7 @@ java_library( "//runtime:interpreter_util", "//runtime:partial_vars", "//runtime:program", + "//runtime:runtime_equality", "//runtime:variable_resolver", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", @@ -250,10 +257,19 @@ java_library( name = "eval_async_call", srcs = ["EvalAsyncCall.java"], deps = [ + ":eval_helpers", + ":localized_evaluation_exception", ":planned_interpretable", + "//common:error_codes", "//common/ast", + "//common/exceptions:overload_not_found", + "//common/exceptions:runtime_exception", + "//common/values", + "//runtime:accumulated_unknowns", "//runtime:evaluation_exception", + "//runtime:function_overload", "//runtime:interpretable", + "//runtime:resolved_overload", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", ], @@ -336,6 +352,7 @@ java_library( name = "eval_conditional", srcs = ["EvalConditional.java"], deps = [ + ":eval_helpers", ":planned_interpretable", "//common/ast", "//runtime:accumulated_unknowns", @@ -383,6 +400,7 @@ java_library( ":eval_helpers", ":planned_interpretable", "//common/ast", + "//common/values", "//runtime:accumulated_unknowns", "//runtime:evaluation_exception", "//runtime:interpretable", @@ -439,6 +457,7 @@ java_library( "//runtime:resolved_overload", "//runtime:unknown_attributes", "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", ], ) @@ -446,6 +465,7 @@ java_library( name = "eval_late_bound_call", srcs = ["EvalLateBoundCall.java"], deps = [ + ":eval_async_call", ":eval_helpers", ":planned_interpretable", "//common/ast", @@ -453,6 +473,7 @@ java_library( "//common/values", "//runtime:accumulated_unknowns", "//runtime:evaluation_exception", + "//runtime:function_overload", "//runtime:interpretable", "//runtime:resolved_overload", "@maven//:com_google_guava_guava", @@ -597,6 +618,7 @@ java_library( "PlannedInterpretable.java", ], deps = [ + ":async_call_state_tracker", ":localized_evaluation_exception", "//common:options", "//common/ast", @@ -609,6 +631,7 @@ java_library( "//runtime:partial_vars", "//runtime:resolved_overload", "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", "@maven//:org_jspecify_jspecify", ], ) @@ -666,6 +689,8 @@ cel_android_library( "//runtime:evaluation_exception_builder", "//runtime:function_overload_android", "//runtime:resolved_overload_android", + "//runtime:runtime_equality_android", + "//runtime:runtime_helpers_android", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:org_jspecify_jspecify", "@maven_android//:com_google_guava_guava", @@ -678,6 +703,9 @@ cel_android_library( tags = [ ], deps = [ + ":async_call_state_tracker_android", + ":async_completion_coordinator_android", + ":async_gate_android", ":error_metadata_android", ":localized_evaluation_exception_android", ":planned_interpretable_android", @@ -686,11 +714,13 @@ cel_android_library( "//common/annotations", "//common/exceptions:runtime_exception", "//common/values:values_android", + "//runtime:accumulated_unknowns_android", "//runtime:activation_android", "//runtime:async_options_android", "//runtime:evaluation_exception", "//runtime:evaluation_exception_builder", "//runtime:interpretable_android", + "//runtime:runtime_equality_android", "//runtime:variable_resolver", "//runtime/src/main/java/dev/cel/runtime:evaluation_listener_android", "//runtime/src/main/java/dev/cel/runtime:function_resolver_android", @@ -856,10 +886,19 @@ cel_android_library( name = "eval_async_call_android", srcs = ["EvalAsyncCall.java"], deps = [ + ":eval_helpers_android", + ":localized_evaluation_exception_android", ":planned_interpretable_android", + "//common:error_codes", "//common/ast:ast_android", + "//common/exceptions:overload_not_found", + "//common/exceptions:runtime_exception", + "//common/values:values_android", + "//runtime:accumulated_unknowns_android", "//runtime:evaluation_exception", + "//runtime:function_overload_android", "//runtime:interpretable_android", + "//runtime:resolved_overload_android", "@maven//:com_google_errorprone_error_prone_annotations", "@maven_android//:com_google_guava_guava", ], @@ -942,6 +981,7 @@ cel_android_library( name = "eval_conditional_android", srcs = ["EvalConditional.java"], deps = [ + ":eval_helpers_android", ":planned_interpretable_android", "//common/ast:ast_android", "//runtime:evaluation_exception", @@ -989,6 +1029,7 @@ cel_android_library( ":eval_helpers_android", ":planned_interpretable_android", "//common/ast:ast_android", + "//common/values:values_android", "//runtime:evaluation_exception", "//runtime:interpretable_android", "//runtime/src/main/java/dev/cel/runtime:accumulated_unknowns_android", @@ -1044,6 +1085,7 @@ cel_android_library( "//runtime:interpreter_util_android", "//runtime:resolved_overload_android", "//runtime:unknown_attributes_android", + "@maven//:org_jspecify_jspecify", "@maven_android//:com_google_guava_guava", ], ) @@ -1052,12 +1094,14 @@ cel_android_library( name = "eval_late_bound_call_android", srcs = ["EvalLateBoundCall.java"], deps = [ + ":eval_async_call_android", ":eval_helpers_android", ":planned_interpretable_android", "//common/ast:ast_android", "//common/exceptions:overload_not_found", "//common/values:values_android", "//runtime:evaluation_exception", + "//runtime:function_overload_android", "//runtime:interpretable_android", "//runtime:resolved_overload_android", "//runtime/src/main/java/dev/cel/runtime:accumulated_unknowns_android", @@ -1198,6 +1242,7 @@ cel_android_library( "PlannedInterpretable.java", ], deps = [ + ":async_call_state_tracker_android", ":localized_evaluation_exception_android", "//common:options", "//common/ast:ast_android", @@ -1211,5 +1256,6 @@ cel_android_library( "//runtime/src/main/java/dev/cel/runtime:partial_vars_android", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:org_jspecify_jspecify", + "@maven_android//:com_google_guava_guava", ], ) diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java index 3b88564aa..cc7ae428d 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java @@ -15,10 +15,20 @@ package dev.cel.runtime.planner; import static com.google.common.base.Preconditions.checkNotNull; +import static dev.cel.runtime.planner.EvalHelpers.evalStrictly; +import com.google.common.collect.ImmutableList; import com.google.errorprone.annotations.Immutable; +import dev.cel.common.CelErrorCode; import dev.cel.common.ast.CelExpr; +import dev.cel.common.exceptions.CelOverloadNotFoundException; +import dev.cel.common.exceptions.CelRuntimeException; +import dev.cel.common.values.CelValueConverter; +import dev.cel.runtime.AccumulatedUnknowns; +import dev.cel.runtime.CelAsyncFunctionOverload; import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelFunctionOverload; +import dev.cel.runtime.CelResolvedOverload; import dev.cel.runtime.GlobalResolver; /** Evaluates an asynchronous function call within a planned program. */ @@ -26,22 +36,106 @@ final class EvalAsyncCall extends PlannedInterpretable { private final String functionName; + private final CelResolvedOverload resolvedOverload; + private final CelAsyncFunctionOverload overload; - static EvalAsyncCall create(CelExpr expr, String functionName) { - return new EvalAsyncCall(expr, functionName); + @SuppressWarnings("Immutable") // Array not mutated + private final PlannedInterpretable[] args; + + private final CelValueConverter celValueConverter; + + static EvalAsyncCall create( + CelExpr expr, + String functionName, + CelResolvedOverload resolvedOverload, + CelAsyncFunctionOverload overload, + PlannedInterpretable[] args, + CelValueConverter celValueConverter) { + return new EvalAsyncCall( + expr, functionName, resolvedOverload, overload, args, celValueConverter); } @Override Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEvaluationException { - throw new CelEvaluationException( - String.format( - "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only" - + " supported via evalAsync.", - functionName)); + if (!frame.isAsync()) { + throw new CelEvaluationException( + String.format( + "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only" + + " supported via evalAsync.", + functionName)); + } + + Object[] evaluatedArgs = new Object[args.length]; + AccumulatedUnknowns accumulatedUnknowns = null; + + for (int i = 0; i < args.length; i++) { + Object argVal = evalStrictly(args[i], resolver, frame); + accumulatedUnknowns = AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, argVal); + evaluatedArgs[i] = argVal; + } + + if (accumulatedUnknowns != null) { + return accumulatedUnknowns; + } + + if (!CelFunctionOverload.canHandle( + evaluatedArgs, resolvedOverload.getParameterTypes(), resolvedOverload.isStrict())) { + throw new LocalizedEvaluationException( + new CelOverloadNotFoundException( + functionName, ImmutableList.of(resolvedOverload.getOverloadId())), + expr().id()); + } + + return dispatchAsync( + expr().id(), + functionName, + resolvedOverload.getOverloadId(), + evaluatedArgs, + overload, + celValueConverter, + frame); + } + + static Object dispatchAsync( + long exprId, + String functionName, + String overloadId, + Object[] evaluatedArgs, + CelAsyncFunctionOverload overload, + CelValueConverter celValueConverter, + ExecutionFrame frame) + throws CelEvaluationException { + if (!frame.isAsync()) { + throw new CelEvaluationException( + String.format( + "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only" + + " supported via evalAsync.", + functionName)); + } + try { + return frame + .asyncTracker() + .recordOrGet( + exprId, functionName, overloadId, evaluatedArgs, overload, celValueConverter); + } catch (CelRuntimeException e) { + throw new LocalizedEvaluationException(e, exprId); + } catch (RuntimeException e) { + throw new LocalizedEvaluationException(e, CelErrorCode.INTERNAL_ERROR, exprId); + } } - private EvalAsyncCall(CelExpr expr, String functionName) { + private EvalAsyncCall( + CelExpr expr, + String functionName, + CelResolvedOverload resolvedOverload, + CelAsyncFunctionOverload overload, + PlannedInterpretable[] args, + CelValueConverter celValueConverter) { super(expr); this.functionName = checkNotNull(functionName); + this.resolvedOverload = checkNotNull(resolvedOverload); + this.overload = checkNotNull(overload); + this.args = checkNotNull(args); + this.celValueConverter = checkNotNull(celValueConverter); } } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalConditional.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalConditional.java index c2d730cdf..0263c7666 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalConditional.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalConditional.java @@ -30,7 +30,7 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEva PlannedInterpretable condition = args[0]; PlannedInterpretable truthy = args[1]; PlannedInterpretable falsy = args[2]; - Object condResult = condition.eval(resolver, frame); + Object condResult = EvalHelpers.evalStrictly(condition, resolver, frame); if (condResult instanceof AccumulatedUnknowns) { return condResult; } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveAnd.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveAnd.java index ac3d07200..88b8964ef 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveAnd.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveAnd.java @@ -15,6 +15,7 @@ package dev.cel.runtime.planner; import static dev.cel.runtime.planner.EvalHelpers.evalNonstrictly; +import static dev.cel.runtime.planner.EvalHelpers.maybeMergeAsyncUnknowns; import com.google.errorprone.annotations.Immutable; import dev.cel.common.ast.CelExpr; @@ -37,11 +38,13 @@ final class EvalExhaustiveAnd extends PlannedInterpretable { @Override Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { AccumulatedUnknowns accumulatedUnknowns = null; + AccumulatedUnknowns asyncUnknowns = null; ErrorValue errorValue = null; boolean hasFalse = false; for (PlannedInterpretable arg : args) { Object argVal = evalNonstrictly(arg, resolver, frame); + asyncUnknowns = maybeMergeAsyncUnknowns(asyncUnknowns, argVal); if (argVal instanceof Boolean) { if (!((boolean) argVal)) { hasFalse = true; @@ -55,10 +58,7 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { } if (argVal instanceof AccumulatedUnknowns) { - accumulatedUnknowns = - accumulatedUnknowns == null - ? (AccumulatedUnknowns) argVal - : accumulatedUnknowns.merge((AccumulatedUnknowns) argVal); + accumulatedUnknowns = AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, argVal); } else if (argVal instanceof ErrorValue) { if (errorValue == null) { errorValue = (ErrorValue) argVal; @@ -66,6 +66,12 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { } } + if (asyncUnknowns != null) { + return accumulatedUnknowns != null + ? AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, asyncUnknowns) + : asyncUnknowns; + } + if (hasFalse) { return false; } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveConditional.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveConditional.java index 01e242c0f..7a6a20bc8 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveConditional.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveConditional.java @@ -15,11 +15,12 @@ package dev.cel.runtime.planner; import static dev.cel.runtime.planner.EvalHelpers.evalNonstrictly; +import static dev.cel.runtime.planner.EvalHelpers.maybeMergeAsyncUnknowns; import com.google.errorprone.annotations.Immutable; import dev.cel.common.ast.CelExpr; +import dev.cel.common.values.ErrorValue; import dev.cel.runtime.AccumulatedUnknowns; -import dev.cel.runtime.CelEvaluationException; import dev.cel.runtime.GlobalResolver; /** @@ -36,16 +37,25 @@ final class EvalExhaustiveConditional extends PlannedInterpretable { private final PlannedInterpretable[] args; @Override - Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEvaluationException { + Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { PlannedInterpretable condition = args[0]; PlannedInterpretable truthy = args[1]; PlannedInterpretable falsy = args[2]; - Object condResult = condition.eval(resolver, frame); + Object condResult = evalNonstrictly(condition, resolver, frame); Object truthyVal = evalNonstrictly(truthy, resolver, frame); Object falsyVal = evalNonstrictly(falsy, resolver, frame); - if (condResult instanceof AccumulatedUnknowns) { + AccumulatedUnknowns asyncUnknowns = maybeMergeAsyncUnknowns(null, condResult); + asyncUnknowns = maybeMergeAsyncUnknowns(asyncUnknowns, truthyVal); + asyncUnknowns = maybeMergeAsyncUnknowns(asyncUnknowns, falsyVal); + if (asyncUnknowns != null) { + return condResult instanceof AccumulatedUnknowns + ? AccumulatedUnknowns.maybeMerge((AccumulatedUnknowns) condResult, asyncUnknowns) + : asyncUnknowns; + } + + if (condResult instanceof AccumulatedUnknowns || condResult instanceof ErrorValue) { return condResult; } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveOr.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveOr.java index 07164f8c7..5fa06633b 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveOr.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalExhaustiveOr.java @@ -15,6 +15,7 @@ package dev.cel.runtime.planner; import static dev.cel.runtime.planner.EvalHelpers.evalNonstrictly; +import static dev.cel.runtime.planner.EvalHelpers.maybeMergeAsyncUnknowns; import com.google.errorprone.annotations.Immutable; import dev.cel.common.ast.CelExpr; @@ -37,11 +38,13 @@ final class EvalExhaustiveOr extends PlannedInterpretable { @Override Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { AccumulatedUnknowns accumulatedUnknowns = null; + AccumulatedUnknowns asyncUnknowns = null; ErrorValue errorValue = null; boolean hasTrue = false; for (PlannedInterpretable arg : args) { Object argVal = evalNonstrictly(arg, resolver, frame); + asyncUnknowns = maybeMergeAsyncUnknowns(asyncUnknowns, argVal); if (argVal instanceof Boolean) { if ((boolean) argVal) { hasTrue = true; @@ -55,10 +58,7 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { } if (argVal instanceof AccumulatedUnknowns) { - accumulatedUnknowns = - accumulatedUnknowns == null - ? (AccumulatedUnknowns) argVal - : accumulatedUnknowns.merge((AccumulatedUnknowns) argVal); + accumulatedUnknowns = AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, argVal); } else if (argVal instanceof ErrorValue) { if (errorValue == null) { errorValue = (ErrorValue) argVal; @@ -66,6 +66,12 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) { } } + if (asyncUnknowns != null) { + return accumulatedUnknowns != null + ? AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, asyncUnknowns) + : asyncUnknowns; + } + if (hasTrue) { return true; } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalHelpers.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalHelpers.java index 1b8d61234..5361d64a4 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalHelpers.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalHelpers.java @@ -25,9 +25,21 @@ import dev.cel.runtime.CelUnknownSet; import dev.cel.runtime.GlobalResolver; import dev.cel.runtime.InterpreterUtil; +import org.jspecify.annotations.Nullable; final class EvalHelpers { + static @Nullable AccumulatedUnknowns maybeMergeAsyncUnknowns( + @Nullable AccumulatedUnknowns accumulator, Object value) { + if (value instanceof AccumulatedUnknowns) { + AccumulatedUnknowns unknowns = (AccumulatedUnknowns) value; + if (!unknowns.callIds().isEmpty()) { + return AccumulatedUnknowns.maybeMerge(accumulator, unknowns); + } + } + return accumulator; + } + static Object evalNonstrictly( PlannedInterpretable interpretable, GlobalResolver resolver, ExecutionFrame frame) { try { @@ -44,7 +56,17 @@ static Object evalNonstrictly( static Object evalStrictly( PlannedInterpretable interpretable, GlobalResolver resolver, ExecutionFrame frame) { try { - return interpretable.eval(resolver, frame); + Object val = interpretable.eval(resolver, frame); + if (val instanceof ErrorValue) { + ErrorValue errorValue = (ErrorValue) val; + Exception cause = errorValue.value(); + if (cause instanceof LocalizedEvaluationException) { + throw (LocalizedEvaluationException) cause; + } + throw new LocalizedEvaluationException( + cause, CelErrorCode.INTERNAL_ERROR, errorValue.exprId()); + } + return val; } catch (LocalizedEvaluationException e) { // Already localized - propagate as-is to preserve inner expression ID throw e; diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalLateBoundCall.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalLateBoundCall.java index 719b4af21..0ab57e095 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/EvalLateBoundCall.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalLateBoundCall.java @@ -21,6 +21,7 @@ import dev.cel.common.exceptions.CelOverloadNotFoundException; import dev.cel.common.values.CelValueConverter; import dev.cel.runtime.AccumulatedUnknowns; +import dev.cel.runtime.CelAsyncFunctionOverload; import dev.cel.runtime.CelEvaluationException; import dev.cel.runtime.CelResolvedOverload; import dev.cel.runtime.GlobalResolver; @@ -56,6 +57,17 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEva .findOverload(functionName, overloadIds, argVals) .orElseThrow(() -> new CelOverloadNotFoundException(functionName, overloadIds)); + if (resolvedOverload.getDefinition() instanceof CelAsyncFunctionOverload) { + return EvalAsyncCall.dispatchAsync( + expr().id(), + functionName, + resolvedOverload.getOverloadId(), + argVals, + (CelAsyncFunctionOverload) resolvedOverload.getDefinition(), + celValueConverter, + frame); + } + return EvalHelpers.dispatch(functionName, resolvedOverload, celValueConverter, argVals); } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/ExecutionFrame.java b/runtime/src/main/java/dev/cel/runtime/planner/ExecutionFrame.java index b67f5520c..a3c422916 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/ExecutionFrame.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/ExecutionFrame.java @@ -14,6 +14,9 @@ package dev.cel.runtime.planner; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; + import dev.cel.common.CelOptions; import dev.cel.common.exceptions.CelIterationLimitExceededException; import dev.cel.runtime.CelEvaluationException; @@ -30,10 +33,39 @@ final class ExecutionFrame { private final int comprehensionIterationLimit; private final CelFunctionResolver functionResolver; - private final PartialVars partialVars; + private final @Nullable PartialVars partialVars; private final @Nullable CelEvaluationListener listener; + private final @Nullable AsyncCallStateTracker asyncTracker; private int iterationCount; - private BlockMemoizer blockMemoizer; + private @Nullable BlockMemoizer blockMemoizer; + + static ExecutionFrame create( + CelFunctionResolver functionResolver, + CelOptions celOptions, + @Nullable PartialVars partialVars, + @Nullable CelEvaluationListener listener) { + return new ExecutionFrame( + functionResolver, + celOptions.comprehensionMaxIterations(), + partialVars, + listener, + /* asyncTracker= */ null); + } + + static ExecutionFrame createForAsync( + CelFunctionResolver functionResolver, + CelOptions celOptions, + @Nullable PartialVars partialVars, + @Nullable CelEvaluationListener listener, + AsyncCallStateTracker asyncTracker) { + checkNotNull(asyncTracker, "asyncTracker"); + return new ExecutionFrame( + functionResolver, + celOptions.comprehensionMaxIterations(), + partialVars, + listener, + asyncTracker); + } Optional findOverload( String functionName, Collection overloadIds, Object[] args) @@ -64,13 +96,13 @@ BlockMemoizer getBlockMemoizer() { return blockMemoizer; } - static ExecutionFrame create( - CelFunctionResolver functionResolver, - CelOptions celOptions, - @Nullable PartialVars partialVars, - @Nullable CelEvaluationListener listener) { - return new ExecutionFrame( - functionResolver, celOptions.comprehensionMaxIterations(), partialVars, listener); + boolean isAsync() { + return asyncTracker != null; + } + + AsyncCallStateTracker asyncTracker() { + checkState(asyncTracker != null, "Not in async execution mode"); + return asyncTracker; } Optional partialVars() { @@ -85,10 +117,12 @@ private ExecutionFrame( CelFunctionResolver functionResolver, int limit, @Nullable PartialVars partialVars, - @Nullable CelEvaluationListener listener) { + @Nullable CelEvaluationListener listener, + @Nullable AsyncCallStateTracker asyncTracker) { this.comprehensionIterationLimit = limit; this.functionResolver = functionResolver; this.partialVars = partialVars; this.listener = listener; + this.asyncTracker = asyncTracker; } } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java index 2f007923e..d46baad70 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java @@ -14,14 +14,19 @@ package dev.cel.runtime.planner; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.util.concurrent.MoreExecutors.directExecutor; + import com.google.auto.value.AutoValue; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.SettableFuture; import com.google.errorprone.annotations.Immutable; import dev.cel.common.CelOptions; import dev.cel.common.annotations.Internal; import dev.cel.common.exceptions.CelRuntimeException; import dev.cel.common.values.ErrorValue; +import dev.cel.runtime.AccumulatedUnknowns; import dev.cel.runtime.Activation; import dev.cel.runtime.CelAsyncEvaluationOptions; import dev.cel.runtime.CelEvaluationException; @@ -33,8 +38,11 @@ import dev.cel.runtime.InterpreterUtil; import dev.cel.runtime.PartialVars; import dev.cel.runtime.Program; +import dev.cel.runtime.RuntimeEquality; +import dev.cel.runtime.planner.AsyncCompletionCoordinator.WaitResult; import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; import org.jspecify.annotations.Nullable; /** @@ -53,6 +61,8 @@ public abstract class PlannedProgram implements Program { public abstract CelOptions options(); + abstract RuntimeEquality runtimeEquality(); + // CelAsyncEvaluationOptions is an immutable value object. @SuppressWarnings("Immutable") @AutoValue.CopyAnnotations @@ -67,10 +77,16 @@ static PlannedProgram create( PlannedInterpretable interpretable, ErrorMetadata metadata, CelOptions options, + RuntimeEquality runtimeEquality, CelAsyncEvaluationOptions asyncOptions, @Nullable ListeningExecutorService asyncExecutor) { return new AutoValue_PlannedProgram( - interpretable, metadata, options, asyncOptions, Optional.ofNullable(asyncExecutor)); + interpretable, + metadata, + options, + runtimeEquality, + asyncOptions, + Optional.ofNullable(asyncExecutor)); } @Override @@ -131,34 +147,73 @@ public Object eval(PartialVars partialVars) throws CelEvaluationException { @Override public ListenableFuture evalAsync() { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + return evalAsync(GlobalResolver.EMPTY, CelFunctionResolver.EMPTY, /* partialVars= */ null); } @Override public ListenableFuture evalAsync(Map mapValue) { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + checkNotNull(mapValue, "mapValue"); + return evalAsync( + Activation.copyOf(mapValue), CelFunctionResolver.EMPTY, /* partialVars= */ null); } @Override public ListenableFuture evalAsync( Map mapValue, CelFunctionResolver lateBoundFunctionResolver) { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + checkNotNull(mapValue, "mapValue"); + checkNotNull(lateBoundFunctionResolver, "lateBoundFunctionResolver"); + return evalAsync( + Activation.copyOf(mapValue), lateBoundFunctionResolver, /* partialVars= */ null); } @Override public ListenableFuture evalAsync(CelVariableResolver resolver) { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + checkNotNull(resolver, "resolver"); + return evalAsync( + (name) -> resolver.find(name).orElse(null), + CelFunctionResolver.EMPTY, + /* partialVars= */ null); } @Override public ListenableFuture evalAsync( CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver) { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + checkNotNull(resolver, "resolver"); + checkNotNull(lateBoundFunctionResolver, "lateBoundFunctionResolver"); + return evalAsync( + (name) -> resolver.find(name).orElse(null), + lateBoundFunctionResolver, + /* partialVars= */ null); } @Override public ListenableFuture evalAsync(PartialVars partialVars) { - throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + checkNotNull(partialVars, "partialVars"); + return evalAsync( + (name) -> partialVars.resolver().find(name).orElse(null), + CelFunctionResolver.EMPTY, + partialVars); + } + + public ListenableFuture evalAsync( + GlobalResolver resolver, + CelFunctionResolver lateBoundResolver, + @Nullable PartialVars partialVars) { + checkNotNull(resolver, "resolver"); + checkNotNull(lateBoundResolver, "lateBoundResolver"); + ListeningExecutorService effectiveExecutor = + asyncExecutor() + .orElseThrow( + () -> + new IllegalStateException( + "No async executor was configured for evalAsync. You must provide a" + + " ListeningExecutorService when configuring the CelRuntime (via" + + " setAsyncExecutor).")); + + AsyncDriver driver = + new AsyncDriver(resolver, lateBoundResolver, partialVars, effectiveExecutor); + driver.step(); + return driver.resultFuture; } public Object evalOrThrow( @@ -201,10 +256,14 @@ private CelEvaluationException newCelEvaluationException(long exprId, Throwable LocalizedEvaluationException localized = (LocalizedEvaluationException) e; exprId = localized.exprId(); Throwable cause = localized.getCause(); + if (cause instanceof CelEvaluationException) { + return (CelEvaluationException) cause; + } if (cause instanceof CelRuntimeException) { builder = CelEvaluationExceptionBuilder.newBuilder((CelRuntimeException) cause); } else { - builder = CelEvaluationExceptionBuilder.newBuilder(cause.getMessage()).setCause(cause); + Throwable innerCause = cause.getCause() != null ? cause.getCause() : cause; + builder = CelEvaluationExceptionBuilder.newBuilder(cause.getMessage()).setCause(innerCause); } } else if (e instanceof CelRuntimeException) { builder = CelEvaluationExceptionBuilder.newBuilder((CelRuntimeException) e); @@ -220,5 +279,113 @@ private CelEvaluationException newCelEvaluationException(long exprId, Throwable return builder.setMetadata(metadata(), exprId).build(); } + private final class AsyncDriver { + private final GlobalResolver resolver; + private final CelFunctionResolver lateBoundResolver; + private final @Nullable PartialVars partialVars; + private final ListeningExecutorService executor; + private final SettableFuture resultFuture = SettableFuture.create(); + private final AsyncCallStateTracker tracker = AsyncCallStateTracker.create(runtimeEquality()); + private final AsyncGate gate = AsyncGate.create(asyncOptions().maxConcurrency()); + private final AsyncCompletionCoordinator coordinator; + private final AtomicInteger iterationCount = new AtomicInteger(); + + private void step() { + try { + while (true) { + int maxIterations = asyncOptions().maxIterations(); + if (maxIterations >= 0 && iterationCount.incrementAndGet() > maxIterations) { + fail( + new CelEvaluationException( + "Exceeded maximum async evaluation iterations: " + maxIterations)); + return; + } + + ExecutionFrame frame = + ExecutionFrame.createForAsync( + lateBoundResolver, options(), partialVars, /* listener= */ null, tracker); + Object evalResult = interpretable().eval(resolver, frame); + + if (evalResult instanceof ErrorValue) { + ErrorValue errorValue = (ErrorValue) evalResult; + fail(newCelEvaluationException(errorValue.exprId(), errorValue.value())); + return; + } + + if (evalResult instanceof AccumulatedUnknowns) { + AccumulatedUnknowns unknowns = (AccumulatedUnknowns) evalResult; + if (unknowns.callIds().isEmpty()) { + complete(InterpreterUtil.maybeAdaptToCelUnknownSet(evalResult)); + return; + } + + tracker.dispatchPendingCalls( + unknowns.callIds(), + executor, + gate, + coordinator, + asyncOptions().observer().orElse(null)); + + WaitResult waitResult = coordinator.waitForCompletions(this::step); + switch (waitResult) { + case REEVALUATE_NOW: + continue; + case NO_OUTSTANDING_WORK: + fail( + new CelEvaluationException( + "Asynchronous evaluation stalled: unresolved async calls remain but no" + + " tasks are in-flight.")); + return; + case REGISTERED: + case CANCELLED: + return; + } + } + + complete(InterpreterUtil.maybeAdaptToCelUnknownSet(evalResult)); + return; + } + } catch (Throwable t) { + fail(newCelEvaluationException(interpretable().expr().id(), t)); + } + } + + private void complete(Object value) { + tracker.cancelInFlight(); + resultFuture.set(value); + } + + private void fail(Throwable t) { + tracker.cancelInFlight(); + resultFuture.setException(t); + } + + private AsyncDriver( + GlobalResolver resolver, + CelFunctionResolver lateBoundResolver, + @Nullable PartialVars partialVars, + ListeningExecutorService executor) { + this.resolver = resolver; + this.lateBoundResolver = lateBoundResolver; + this.partialVars = partialVars; + this.executor = executor; + this.coordinator = + AsyncCompletionCoordinator.create( + asyncOptions(), + gate, + executor, + t -> fail(newCelEvaluationException(interpretable().expr().id(), t))); + this.resultFuture.addListener( + () -> { + if (resultFuture.isCancelled()) { + gate.cancel(); + coordinator.cancel(); + tracker.cancelInFlight(); + } + }, + directExecutor()); + } + } + PlannedProgram() {} } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java b/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java index 47b7cf552..dcb5e96e0 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java @@ -54,6 +54,8 @@ import dev.cel.runtime.CelEvaluationExceptionBuilder; import dev.cel.runtime.CelResolvedOverload; import dev.cel.runtime.DefaultDispatcher; +import dev.cel.runtime.RuntimeEquality; +import dev.cel.runtime.RuntimeHelpers; import java.util.Arrays; import java.util.HashMap; import java.util.NoSuchElementException; @@ -84,6 +86,8 @@ public final class ProgramPlanner { @SuppressWarnings("Immutable") private final @Nullable ListeningExecutorService asyncExecutor; + private final RuntimeEquality runtimeEquality; + /** * Plans a {@link PlannedProgram} from the provided parsed-only or type-checked {@link * CelAbstractSyntaxTree}. @@ -106,7 +110,7 @@ public PlannedProgram plan(CelAbstractSyntaxTree ast) throws CelEvaluationExcept } return PlannedProgram.create( - plannedInterpretable, errorMetadata, options, asyncOptions, asyncExecutor); + plannedInterpretable, errorMetadata, options, runtimeEquality, asyncOptions, asyncExecutor); } private PlannedInterpretable plan(CelExpr celExpr, PlannerContext ctx) { @@ -334,7 +338,13 @@ private PlannedInterpretable planCall(CelExpr expr, PlannerContext ctx) { } if (resolvedOverload.getDefinition() instanceof CelAsyncFunctionOverload) { - return EvalAsyncCall.create(expr, functionName); + return EvalAsyncCall.create( + expr, + functionName, + resolvedOverload, + (CelAsyncFunctionOverload) resolvedOverload.getDefinition(), + evaluatedArgs, + celValueConverter); } switch (argCount) { @@ -732,6 +742,31 @@ public static ProgramPlanner newPlanner( ImmutableSet lateBoundFunctionNames, CelAsyncEvaluationOptions asyncOptions, @Nullable ListeningExecutorService asyncExecutor) { + return newPlanner( + typeProvider, + valueProvider, + dispatcher, + celValueConverter, + container, + options, + lateBoundFunctionNames, + RuntimeEquality.create(RuntimeHelpers.create(), options), + asyncOptions, + asyncExecutor); + } + + @SuppressWarnings("TooManyParameters") + public static ProgramPlanner newPlanner( + CelTypeProvider typeProvider, + CelValueProvider valueProvider, + DefaultDispatcher dispatcher, + CelValueConverter celValueConverter, + CelContainer container, + CelOptions options, + ImmutableSet lateBoundFunctionNames, + RuntimeEquality runtimeEquality, + CelAsyncEvaluationOptions asyncOptions, + @Nullable ListeningExecutorService asyncExecutor) { return new ProgramPlanner( typeProvider, valueProvider, @@ -740,6 +775,7 @@ public static ProgramPlanner newPlanner( container, options, lateBoundFunctionNames, + runtimeEquality, asyncOptions, asyncExecutor); } @@ -752,6 +788,7 @@ private ProgramPlanner( CelContainer container, CelOptions options, ImmutableSet lateBoundFunctionNames, + RuntimeEquality runtimeEquality, CelAsyncEvaluationOptions asyncOptions, @Nullable ListeningExecutorService asyncExecutor) { this.typeProvider = typeProvider; @@ -761,6 +798,7 @@ private ProgramPlanner( this.container = container; this.options = options; this.lateBoundFunctionNames = lateBoundFunctionNames; + this.runtimeEquality = checkNotNull(runtimeEquality); this.asyncOptions = checkNotNull(asyncOptions); this.asyncExecutor = asyncExecutor; this.attributeFactory = diff --git a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel index 53240ff87..59e72ce53 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel @@ -24,6 +24,7 @@ java_library( "//common:options", "//common/ast", "//common/exceptions:divide_by_zero", + "//common/exceptions:overload_not_found", "//common/exceptions:runtime_exception", "//common/internal:cel_descriptor_pools", "//common/internal:default_message_factory", @@ -62,6 +63,7 @@ java_library( "@maven//:com_google_code_findbugs_annotations", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", + "@maven//:com_google_protobuf_protobuf_java", "@maven//:com_google_testparameterinjector_test_parameter_injector", "@maven//:junit_junit", "@maven//:org_jspecify_jspecify", diff --git a/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerAsyncTest.java b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerAsyncTest.java new file mode 100644 index 000000000..aff42b754 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerAsyncTest.java @@ -0,0 +1,1790 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev.cel.runtime.planner; + +import static com.google.common.base.Preconditions.checkState; +import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.Futures.immediateFailedFuture; +import static com.google.common.util.concurrent.Futures.immediateFuture; +import static com.google.common.util.concurrent.MoreExecutors.directExecutor; +import static com.google.common.util.concurrent.MoreExecutors.listeningDecorator; +import static com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; +import static dev.cel.common.CelFunctionDecl.newFunctionDeclaration; +import static dev.cel.common.CelOverloadDecl.newGlobalOverload; +import static dev.cel.common.CelOverloadDecl.newMemberOverload; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.ForwardingListeningExecutorService; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.SettableFuture; +import com.google.errorprone.annotations.Immutable; +import javax.annotation.concurrent.ThreadSafe; +import com.google.protobuf.Any; +import com.google.testing.junit.testparameterinjector.TestParameter; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import dev.cel.common.CelAbstractSyntaxTree; +import dev.cel.common.CelErrorCode; +import dev.cel.common.CelOptions; +import dev.cel.common.exceptions.CelDivideByZeroException; +import dev.cel.common.exceptions.CelOverloadNotFoundException; +import dev.cel.common.types.ListType; +import dev.cel.common.types.MapType; +import dev.cel.common.types.OpaqueType; +import dev.cel.common.types.SimpleType; +import dev.cel.compiler.CelCompiler; +import dev.cel.compiler.CelCompilerFactory; +import dev.cel.expr.conformance.proto3.TestAllTypes; +import dev.cel.parser.CelStandardMacro; +import dev.cel.runtime.AccumulatedUnknowns; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncDrainAction; +import dev.cel.runtime.CelAsyncDrainStrategy; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import dev.cel.runtime.CelAsyncObserver; +import dev.cel.runtime.CelAttribute; +import dev.cel.runtime.CelAttributePattern; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.CelFunctionBinding; +import dev.cel.runtime.CelLateFunctionBindings; +import dev.cel.runtime.CelRuntime; +import dev.cel.runtime.CelRuntime.Program; +import dev.cel.runtime.CelRuntimeBuilder; +import dev.cel.runtime.CelRuntimeFactory; +import dev.cel.runtime.CelUnknownSet; +import dev.cel.runtime.CelVariableResolver; +import dev.cel.runtime.PartialVars; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.jspecify.annotations.Nullable; +import org.junit.After; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +public final class ProgramPlannerAsyncTest { + + private static final CelCompiler CEL_COMPILER = + CelCompilerFactory.standardCelCompilerBuilder() + .setStandardMacros(CelStandardMacro.STANDARD_MACROS) + .setOptions(CelOptions.current().build()) + .addVar("x", SimpleType.INT) + .addVar("y", SimpleType.INT) + .addVar("dx", SimpleType.DOUBLE) + .addVar("list_var", ListType.create(SimpleType.INT)) + .addVar("map_var", MapType.create(SimpleType.STRING, SimpleType.INT)) + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncSquare", + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT), + newGlobalOverload("asyncSquare_double", SimpleType.DOUBLE, SimpleType.DOUBLE)), + newFunctionDeclaration( + "asyncAdd", + newGlobalOverload( + "asyncAdd_int_int", SimpleType.INT, SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "asyncSum3", + newGlobalOverload( + "asyncSum3_int", + SimpleType.INT, + SimpleType.INT, + SimpleType.INT, + SimpleType.INT)), + newFunctionDeclaration( + "asyncIsEven", + newGlobalOverload("asyncIsEven_int", SimpleType.BOOL, SimpleType.INT)), + newFunctionDeclaration( + "asyncFail", newGlobalOverload("asyncFail_int", SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "asyncNullReturn", + newGlobalOverload("asyncNullReturn_int", SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "asyncSyncThrow", + newGlobalOverload("asyncSyncThrow_int", SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "asyncListSize", + newGlobalOverload( + "asyncListSize_list", SimpleType.INT, ListType.create(SimpleType.INT))), + newFunctionDeclaration( + "asyncMapSize", + newGlobalOverload( + "asyncMapSize_map", + SimpleType.INT, + MapType.create(SimpleType.STRING, SimpleType.INT))), + newFunctionDeclaration( + "lateAdd", + newGlobalOverload( + "lateAdd_int_int", SimpleType.INT, SimpleType.INT, SimpleType.INT))) + .build(); + + private static final CelFunctionBinding ASYNC_SQUARE_INT = + CelFunctionBinding.fromAsync( + "asyncSquare_int", Long.class, (Long arg) -> immediateFuture(arg * arg)); + + private static final CelFunctionBinding ASYNC_ADD_INT = + CelFunctionBinding.fromAsync( + "asyncAdd_int_int", Long.class, Long.class, (a, b) -> immediateFuture(a + b)); + + private final ListeningExecutorService executor = + listeningDecorator(Executors.newFixedThreadPool(4)); + + @After + public void tearDown() { + executor.shutdownNow(); + } + + private enum BasicAsyncCase { + UNARY("asyncSquare(4) + 1", 17L), + BINARY("asyncAdd(10, 20) * 2", 60L), + NESTED("asyncSquare(asyncSquare(3))", 81L); + + private final String expression; + private final long expectedResult; + + BasicAsyncCase(String expression, long expectedResult) { + this.expression = expression; + this.expectedResult = expectedResult; + } + } + + @Test + public void evalAsync_basicFunctions_evaluatesSuccessfully( + @TestParameter BasicAsyncCase testCase, @TestParameter boolean parsedOnly) throws Exception { + CelAbstractSyntaxTree ast = + parsedOnly + ? CEL_COMPILER.parse(testCase.expression).getAst() + : CEL_COMPILER.compile(testCase.expression).getAst(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + parsedOnly ? "asyncSquare" : "asyncSquare_int", + Long.class, + (Long arg) -> immediateFuture(arg * arg)), + CelFunctionBinding.fromAsync( + parsedOnly ? "asyncAdd" : "asyncAdd_int_int", + Long.class, + Long.class, + (Long a, Long b) -> immediateFuture(a + b))) + .build(); + Program program = runtime.createProgram(ast); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(testCase.expectedResult); + } + + @Test + public void evalAsync_varargsFunction_evaluatesSuccessfully() throws Exception { + Program program = + createProgram( + "asyncSum3(1, 2, 3)", + CelFunctionBinding.fromAsync( + "asyncSum3_int", + ImmutableList.of(Long.class, Long.class, Long.class), + (Object[] args) -> + immediateFuture((Long) args[0] + (Long) args[1] + (Long) args[2]))); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(6L); + } + + @Test + public void evalAsync_syncProgram_evaluatesSuccessfully() throws Exception { + Program program = createProgram("1 + 2 * 3"); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(7L); + } + + private enum ActivationOverloadCase { + MAP, + MAP_WITH_LATE_BOUND_RESOLVER, + VARIABLE_RESOLVER, + VARIABLE_RESOLVER_WITH_LATE_BOUND_RESOLVER, + PARTIAL_VARS + } + + @Test + public void evalAsync_activationOverloads_evaluatesCorrectly( + @TestParameter ActivationOverloadCase overloadCase) throws Exception { + boolean useLateBound = + overloadCase == ActivationOverloadCase.MAP_WITH_LATE_BOUND_RESOLVER + || overloadCase == ActivationOverloadCase.VARIABLE_RESOLVER_WITH_LATE_BOUND_RESOLVER; + CelAbstractSyntaxTree ast = + CEL_COMPILER + .compile(useLateBound ? "asyncSquare(x) + lateAdd(y, 0)" : "asyncSquare(x) + y") + .getAst(); + CelRuntimeBuilder runtimeBuilder = + plannerRuntimeBuilder().addFunctionBindings(ASYNC_SQUARE_INT); + if (useLateBound) { + runtimeBuilder.addLateBoundFunctions("lateAdd"); + } + Program program = runtimeBuilder.build().createProgram(ast); + ImmutableMap mapActivation = ImmutableMap.of("x", 5L, "y", 10L); + CelVariableResolver varResolver = (name) -> Optional.ofNullable(mapActivation.get(name)); + CelLateFunctionBindings lateBoundResolver = + CelLateFunctionBindings.from( + CelFunctionBinding.from("lateAdd_int_int", Long.class, Long.class, Long::sum)); + + ListenableFuture future; + switch (overloadCase) { + case MAP: + future = program.evalAsync(mapActivation); + break; + case MAP_WITH_LATE_BOUND_RESOLVER: + future = program.evalAsync(mapActivation, lateBoundResolver); + break; + case VARIABLE_RESOLVER: + future = program.evalAsync(varResolver); + break; + case VARIABLE_RESOLVER_WITH_LATE_BOUND_RESOLVER: + future = program.evalAsync(varResolver, lateBoundResolver); + break; + case PARTIAL_VARS: + future = + program.evalAsync( + PartialVars.of(varResolver, CelAttributePattern.fromQualifiedIdentifier("unused"))); + break; + default: + throw new AssertionError(overloadCase); + } + + assertThat(future.get(5, SECONDS)).isEqualTo(35L); + } + + @Test + public void evalAsync_protoMessage_evaluatesCorrectly() throws Exception { + CelCompiler protoCompiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addVar("single_int64", SimpleType.INT) + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncSquare", + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT))) + .build(); + CelAbstractSyntaxTree protoAst = + protoCompiler.compile("asyncSquare(single_int64) - 1").getAst(); + Program protoProgram = + plannerRuntimeBuilder() + .addMessageTypes(TestAllTypes.getDescriptor()) + .addFunctionBindings(ASYNC_SQUARE_INT) + .build() + .createProgram(protoAst); + TestAllTypes message = TestAllTypes.newBuilder().setSingleInt64(6L).build(); + + Object protoResult = protoProgram.evalAsync(message).get(5, SECONDS); + + assertThat(protoResult).isEqualTo(35L); + } + + @Test + public void evalAsync_nullProtoMessage_throwsNullPointerException() throws Exception { + Program program = createProgram("1"); + + assertThrows(NullPointerException.class, () -> program.evalAsync((TestAllTypes) null)); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_conditionalBranching_onlyEvaluatesTakenBranch() throws Exception { + AtomicInteger untakenBranchCalls = new AtomicInteger(); + Program program = + createProgram( + "asyncIsEven(4) ? asyncSquare(3) : asyncFail(1)", + CelFunctionBinding.fromAsync( + "asyncIsEven_int", Long.class, (Long arg) -> immediateFuture(arg % 2 == 0)), + ASYNC_SQUARE_INT, + CelFunctionBinding.fromAsync( + "asyncFail_int", + Long.class, + (Long arg) -> { + untakenBranchCalls.incrementAndGet(); + return immediateFailedFuture(new AssertionError("Should not be called")); + })); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(9L); + assertThat(untakenBranchCalls.get()).isEqualTo(0); + } + + private enum ShortCircuitOperator { + OR("asyncIsEven(2) || (asyncSquare(10) == 100)", true), + AND("asyncIsEven(3) && (asyncSquare(10) == 100)", false); + + private final String expression; + private final boolean expectedResult; + + ShortCircuitOperator(String expression, boolean expectedResult) { + this.expression = expression; + this.expectedResult = expectedResult; + } + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_shortCircuiting_cancelsUnneededSibling( + @TestParameter ShortCircuitOperator op) throws Exception { + SettableFuture firstBranchFuture = SettableFuture.create(); + SettableFuture slowSibling = SettableFuture.create(); + CountDownLatch siblingStarted = new CountDownLatch(1); + CountDownLatch siblingCancelledLatch = new CountDownLatch(1); + slowSibling.addListener( + () -> { + if (slowSibling.isCancelled()) { + siblingCancelledLatch.countDown(); + } + }, + directExecutor()); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainNone()) + .build(); + Program program = + createProgram( + op.expression, + options, + CelFunctionBinding.fromAsync( + "asyncIsEven_int", Long.class, (Long arg) -> firstBranchFuture), + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + siblingStarted.countDown(); + return slowSibling; + })); + + ListenableFuture evalFuture = program.evalAsync(); + boolean started = siblingStarted.await(5, SECONDS); + firstBranchFuture.set(op.expectedResult); + + assertThat(started).isTrue(); + assertThat(evalFuture.get(5, SECONDS)).isEqualTo(op.expectedResult); + assertThat(siblingCancelledLatch.await(5, SECONDS)).isTrue(); + } + + @Immutable + @SuppressWarnings("Immutable") + private enum ExhaustiveEvalCase { + OR_LEFT_TRUE("asyncIsEven(2) || (asyncSquare(3) == 9)", true, 1), + AND_LEFT_FALSE("asyncIsEven(3) && (asyncSquare(3) == 9)", false, 1), + CONDITIONAL_ASYNC_PRED_TRUE("asyncIsEven(2) ? asyncSquare(3) : asyncSquare(4)", 9L, 2), + CONDITIONAL_SYNC_PRED_FALSE("false ? asyncSquare(3) : asyncSquare(4)", 16L, 2); + + private final String expression; + private final Object expectedResult; + private final int expectedSquareCalls; + + ExhaustiveEvalCase(String expression, Object expectedResult, int expectedSquareCalls) { + this.expression = expression; + this.expectedResult = expectedResult; + this.expectedSquareCalls = expectedSquareCalls; + } + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_exhaustiveEval_waitsForAllBranchesBeforeCompleting( + @TestParameter ExhaustiveEvalCase testCase) throws Exception { + AtomicInteger squareCalls = new AtomicInteger(); + CelAbstractSyntaxTree ast = CEL_COMPILER.compile(testCase.expression).getAst(); + CelRuntime runtime = + plannerRuntimeBuilder() + .setOptions( + CelOptions.current() + .enableHeterogeneousNumericComparisons(true) + .enableShortCircuiting(false) + .build()) + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncIsEven_int", Long.class, (Long arg) -> immediateFuture(arg % 2 == 0)), + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> + executor.submit( + () -> { + Thread.sleep(10); + squareCalls.incrementAndGet(); + return (Object) (arg * arg); + }))) + .build(); + Program program = runtime.createProgram(ast); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(testCase.expectedResult); + assertThat(squareCalls.get()).isEqualTo(testCase.expectedSquareCalls); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_shortCircuitedUnknownWithAsyncCall_doesNotLeakAsyncCall() throws Exception { + AtomicInteger callCounter = new AtomicInteger(); + Program program = + createProgram( + "((x + asyncSquare(3) == 0) && false) ? 0 : x", + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + callCounter.incrementAndGet(); + return immediateFuture(arg * arg); + })); + + Object result = + program + .evalAsync(PartialVars.of(CelAttributePattern.fromQualifiedIdentifier("x"))) + .get(5, SECONDS); + + assertThat(result).isInstanceOf(CelUnknownSet.class); + assertThat(callCounter.get()).isEqualTo(0); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_exhaustiveConditionalWithFailingCondition_evaluatesBranchesThenFails( + @TestParameter({ + "(1 / 0 == 0) ? asyncSquare(3) : asyncSquare(4)", + "(true && (1 / 0 == 0)) ? asyncSquare(3) : asyncSquare(4)" + }) + String expression) + throws Exception { + AtomicInteger squareCalls = new AtomicInteger(); + CelAbstractSyntaxTree ast = CEL_COMPILER.compile(expression).getAst(); + Program program = + plannerRuntimeBuilder() + .setOptions( + CelOptions.current() + .enableHeterogeneousNumericComparisons(true) + .enableShortCircuiting(false) + .build()) + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + squareCalls.incrementAndGet(); + return immediateFuture(arg * arg); + })) + .build() + .createProgram(ast); + + ListenableFuture future = program.evalAsync(); + + ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + CelEvaluationException evalEx = (CelEvaluationException) e.getCause(); + assertThat(evalEx.getErrorCode()).isEqualTo(CelErrorCode.DIVIDE_BY_ZERO); + assertThat(squareCalls.get()).isEqualTo(2); + } + + @Test + public void evalAsync_strictOperatorWrappingNonStrictErrorValue_throwsDivideByZero( + @TestParameter({"[true && (1 / 0 == 0)]", "!(true && (1 / 0 == 0))"}) String expression) + throws Exception { + Program program = createProgram(expression); + + ListenableFuture future = program.evalAsync(); + + ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + CelEvaluationException evalEx = (CelEvaluationException) e.getCause(); + assertThat(evalEx.getErrorCode()).isEqualTo(CelErrorCode.DIVIDE_BY_ZERO); + } + + @Test + public void evalAsync_lateBoundAsyncFunction_evaluatesCorrectly() throws Exception { + CelAbstractSyntaxTree ast = CEL_COMPILER.compile("asyncSquare(21)").getAst(); + Program program = + plannerRuntimeBuilder().addLateBoundFunctions("asyncSquare").build().createProgram(ast); + CelLateFunctionBindings lateBindings = + CelLateFunctionBindings.from( + CelFunctionBinding.fromAsync( + "asyncSquare_int", Long.class, (Long arg) -> immediateFuture(arg * arg))); + + Object result = program.evalAsync(ImmutableMap.of(), lateBindings).get(5, SECONDS); + + assertThat(result).isEqualTo(441L); + } + + @Test + public void eval_withAsyncFunction_throwsCelEvaluationException( + @TestParameter({"asyncSquare(4)", "1 + asyncSquare(5)"}) String expression) throws Exception { + Program program = createProgram(expression, ASYNC_SQUARE_INT); + + CelEvaluationException ex = assertThrows(CelEvaluationException.class, program::eval); + + assertThat(ex) + .hasMessageThat() + .contains("Async function 'asyncSquare' evaluated in synchronous mode."); + assertThat(ex).hasCauseThat().isNull(); + } + + @Test + public void trace_withAsyncFunction_throwsCelEvaluationException() throws Exception { + Program program = createProgram("asyncSquare(4)", ASYNC_SQUARE_INT); + + CelEvaluationException ex = + assertThrows(CelEvaluationException.class, () -> program.trace((expr, res) -> {})); + + assertThat(ex).hasMessageThat().contains("Async function 'asyncSquare'"); + } + + @Test + public void evalAsync_noExecutorProvided_throwsIllegalStateException() throws Exception { + CelAbstractSyntaxTree ast = CEL_COMPILER.compile("1 + 1").getAst(); + CelRuntime runtime = CelRuntimeFactory.plannerRuntimeBuilder().build(); + Program program = runtime.createProgram(ast); + + IllegalStateException e = assertThrows(IllegalStateException.class, program::evalAsync); + + assertThat(e).hasMessageThat().contains("No async executor was configured"); + } + + @Test + public void evalAsync_withSufficientIterations_succeeds() throws Exception { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setMaxIterations(2).build(); + Program program = createProgram("asyncSquare(5)", options, ASYNC_SQUARE_INT); + + ListenableFuture future = program.evalAsync(); + + assertThat(future.get(5, SECONDS)).isEqualTo(25L); + } + + @Test + public void evalAsync_exceedsMaxIterations_throwsCelEvaluationException() throws Exception { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setMaxIterations(1).build(); + Program program = createProgram("asyncSquare(5)", options, ASYNC_SQUARE_INT); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e) + .hasCauseThat() + .hasMessageThat() + .contains("Exceeded maximum async evaluation iterations: 1"); + } + + @Test + public void evalAsync_withObserverOnSuccess_recordsLifecycleEvents() throws Exception { + RecordingObserver observer = new RecordingObserver(); + Program program = + createProgram( + "asyncSquare(3)", + CelAsyncEvaluationOptions.builder().setObserver(observer).build(), + ASYNC_SQUARE_INT); + + ListenableFuture future = program.evalAsync(); + + assertThat(future.get(5, SECONDS)).isEqualTo(9L); + assertThat(observer.startedFunctionName()).hasValue("asyncSquare"); + assertThat(observer.startedArgs()).hasValue(ImmutableList.of(3L)); + assertThat(observer.finishedFunctionName()).hasValue("asyncSquare"); + assertThat(observer.finishedResult()).hasValue(9L); + assertThat(observer.finishedError()).isEmpty(); + } + + private enum FailureScenario { + ASYNC_FAILURE("asyncFail(1)", 1L, "simulated error"), + SYNC_THROW("asyncSyncThrow(2)", 2L, "sync throw"); + + private final String expression; + private final long expectedArg; + private final String expectedErrorSubstring; + + FailureScenario(String expression, long expectedArg, String expectedErrorSubstring) { + this.expression = expression; + this.expectedArg = expectedArg; + this.expectedErrorSubstring = expectedErrorSubstring; + } + } + + @Test + public void evalAsync_withObserverOnFailure_recordsLifecycleEvents( + @TestParameter FailureScenario scenario) throws Exception { + RecordingObserver observer = new RecordingObserver(); + Program program = + createProgram( + scenario.expression, + CelAsyncEvaluationOptions.builder().setObserver(observer).build(), + CelFunctionBinding.fromAsync( + "asyncFail_int", + Long.class, + (Long arg) -> + immediateFailedFuture(new IllegalArgumentException("simulated error"))), + CelFunctionBinding.fromAsync( + "asyncSyncThrow_int", + Long.class, + (Long arg) -> { + throw new IllegalStateException("sync throw"); + })); + + ListenableFuture future = program.evalAsync(); + + assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + assertThat(observer.startedArgs()).hasValue(ImmutableList.of(scenario.expectedArg)); + assertThat(observer.finishedResult()).isEmpty(); + assertThat(observer.finishedError().map(Throwable::getMessage)) + .hasValue(scenario.expectedErrorSubstring); + } + + @Test + public void evalAsync_observerThrowsInStartCallback_failsEvaluationAndReleasesPermits() + throws Exception { + CelAsyncObserver observer = + new CelAsyncObserver() { + private final AtomicBoolean first = new AtomicBoolean(true); + + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) { + if (first.compareAndSet(true, false)) { + throw new RuntimeException("observer start failure"); + } + } + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) {} + }; + Program program = + createProgram( + "asyncSquare(2) + asyncSquare(3)", + CelAsyncEvaluationOptions.builder().setMaxConcurrency(1).setObserver(observer).build(), + ASYNC_SQUARE_INT); + + ListenableFuture future = program.evalAsync(); + + ExecutionException startEx = + assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + assertThat(startEx).hasCauseThat().hasMessageThat().contains("observer start failure"); + } + + @Test + public void evalAsync_observerThrowsInFinishCallback_completesEvaluationAndReleasesPermits() + throws Exception { + CelAsyncObserver observer = + new CelAsyncObserver() { + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) {} + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + throw new AssertionError("observer finish error"); + } + }; + Program program = + createProgram( + "asyncSquare(2) + asyncSquare(3)", + CelAsyncEvaluationOptions.builder().setMaxConcurrency(1).setObserver(observer).build(), + ASYNC_SQUARE_INT); + + ListenableFuture future = program.evalAsync(); + + assertThat(future.get(5, SECONDS)).isEqualTo(13L); + } + + @Test + public void evalAsync_applyAsyncReturnsNull_failsWithDescriptiveException() throws Exception { + Program program = + createProgram( + "asyncNullReturn(1)", + CelFunctionBinding.fromAsync("asyncNullReturn_int", Long.class, (Long arg) -> null)); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("returned a null ListenableFuture"); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_argumentEvaluatesToErrorValue_propagatesWithoutInvokingAsyncFunction() + throws Exception { + AtomicInteger asyncCalls = new AtomicInteger(); + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncAnd", + newGlobalOverload( + "asyncAnd_bool_bool", SimpleType.BOOL, SimpleType.BOOL, SimpleType.BOOL))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncAnd_bool_bool", + Boolean.class, + Boolean.class, + (Boolean a, Boolean b) -> { + asyncCalls.incrementAndGet(); + return immediateFuture(a && b); + })) + .build(); + Program program = + runtime.createProgram( + compiler + .compile("asyncAnd(true && (1 / 0 == 0), asyncAnd(true, true) && ([true][1]))") + .getAst()); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(((CelEvaluationException) e.getCause()).getErrorCode()) + .isEqualTo(CelErrorCode.DIVIDE_BY_ZERO); + assertThat(e).hasCauseThat().hasMessageThat().contains("/ by zero"); + assertThat(asyncCalls.get()).isEqualTo(0); + } + + @Test + public void evalAsync_parsedOnlyTypeOrArityMismatch_throwsOverloadNotFoundException( + @TestParameter({"asyncSquare('bad')", "asyncSquare(4, 'extra')"}) String expression) + throws Exception { + CelAbstractSyntaxTree ast = CEL_COMPILER.parse(expression).getAst(); + Program program = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncSquare", Long.class, (Long arg) -> immediateFuture(arg * arg))) + .build() + .createProgram(ast); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasCauseThat().isInstanceOf(CelOverloadNotFoundException.class); + } + + private enum SiblingFailureMode { + ASYNC_FAILED_FUTURE, + SYNC_RUNTIME_EXCEPTION, + PASS2_DIVISION_BY_ZERO, + DRAIN_STRATEGY_EXCEPTION + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_failureModes_cancelsInFlightSiblingsAndPropagatesCelEvaluationException( + @TestParameter SiblingFailureMode failureMode) throws Exception { + SettableFuture inFlightSibling = SettableFuture.create(); + CountDownLatch siblingStarted = new CountDownLatch(1); + CountDownLatch siblingCancelledLatch = new CountDownLatch(1); + inFlightSibling.addListener( + () -> { + if (inFlightSibling.isCancelled()) { + siblingCancelledLatch.countDown(); + } + }, + directExecutor()); + CelAsyncDrainStrategy drainStrategy = + failureMode == SiblingFailureMode.DRAIN_STRATEGY_EXCEPTION + ? (completedBatch, activeCount) -> { + throw new IllegalStateException("custom drain strategy failure"); + } + : CelAsyncDrainStrategy.drainNone(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(drainStrategy).build(); + String expr = + failureMode == SiblingFailureMode.PASS2_DIVISION_BY_ZERO + ? "asyncSquare(10) + (1 / asyncSquare(0))" + : "asyncSquare(10) + asyncFail(5)"; + Program program = + createProgram( + expr, + options, + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + if (arg == 10L) { + siblingStarted.countDown(); + return inFlightSibling; + } + try { + siblingStarted.await(5, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return immediateFuture(0L); + }), + CelFunctionBinding.fromAsync( + "asyncFail_int", + Long.class, + (Long arg) -> { + try { + siblingStarted.await(5, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + checkState( + failureMode != SiblingFailureMode.SYNC_RUNTIME_EXCEPTION, + "synchronous crash"); + if (failureMode == SiblingFailureMode.DRAIN_STRATEGY_EXCEPTION) { + return immediateFuture(1L); + } + return immediateFailedFuture( + new IllegalArgumentException("simulated async error")); + })); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(siblingCancelledLatch.await(5, SECONDS)).isTrue(); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_futureCancellation_cancelsInFlightGateAndDebounceTimer() throws Exception { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + SettableFuture task1Future = SettableFuture.create(); + SettableFuture task2Future = SettableFuture.create(); + SettableFuture task3Future = SettableFuture.create(); + CountDownLatch task1Started = new CountDownLatch(1); + CountDownLatch task2Started = new CountDownLatch(1); + CountDownLatch task2Cancelled = new CountDownLatch(1); + AtomicInteger tasksSubmitted = new AtomicInteger(); + AtomicInteger task4Executed = new AtomicInteger(); + ListeningExecutorService trackingExecutor = + new ForwardingListeningExecutorService() { + @Override + protected ListeningExecutorService delegate() { + return executor; + } + + @Override + public void execute(Runnable command) { + tasksSubmitted.incrementAndGet(); + super.execute(command); + } + }; + task2Future.addListener( + () -> { + if (task2Future.isCancelled()) { + task2Cancelled.countDown(); + } + }, + directExecutor()); + CelAsyncDrainStrategy longDebounceStrategy = + (completedBatch, activeCount) -> CelAsyncDrainAction.waitDuration(Duration.ofMinutes(10)); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(longDebounceStrategy) + .setScheduledExecutorService(scheduler) + .setMaxConcurrency(2) + .build(); + CelAbstractSyntaxTree ast = + CEL_COMPILER + .compile("asyncSquare(1) + asyncSquare(2) + asyncSquare(3) + asyncSquare(4)") + .getAst(); + Program program = + plannerRuntimeBuilder() + .setAsyncExecutor(trackingExecutor) + .setAsyncEvaluationOptions(options) + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + if (arg == 1L) { + task1Started.countDown(); + return task1Future; + } + if (arg == 2L) { + task2Started.countDown(); + return task2Future; + } + if (arg == 3L) { + return task3Future; + } + task4Executed.incrementAndGet(); + return immediateFuture(arg * arg); + })) + .build() + .createProgram(ast); + + ListenableFuture evalFuture = program.evalAsync(); + boolean started = task1Started.await(5, SECONDS) && task2Started.await(5, SECONDS); + task1Future.set(1L); + ScheduledFuture scheduledTask = null; + long deadline = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + while (System.nanoTime() < deadline) { + scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + if (scheduledTask != null) { + break; + } + Thread.sleep(5); + } + evalFuture.cancel(/* mayInterruptIfRunning= */ true); + while (scheduledTask != null + && !scheduledTask.isCancelled() + && System.nanoTime() < deadline) { + Thread.sleep(5); + } + task3Future.set(9L); + + assertThat(started).isTrue(); + assertThat(scheduledTask).isNotNull(); + assertThat(evalFuture.isCancelled()).isTrue(); + assertThat(scheduledTask.isCancelled()).isTrue(); + assertThat(task2Cancelled.await(5, SECONDS)).isTrue(); + assertThat(task4Executed.get()).isEqualTo(0); + assertThat(tasksSubmitted.get()).isEqualTo(2); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_evalExceptionInPass2_doesNotExecuteRemainingCallsUnderMaxConcurrency() + throws Exception { + SettableFuture call1Future = SettableFuture.create(); + CountDownLatch call1Started = new CountDownLatch(1); + AtomicInteger remainingCallsExecuted = new AtomicInteger(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setMaxConcurrency(1) + .setDrainStrategy(CelAsyncDrainStrategy.drainNone()) + .build(); + Program program = + createProgram( + "(1 / asyncSquare(1)) + asyncSquare(2) + asyncSquare(3)", + options, + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + if (arg == 1L) { + call1Started.countDown(); + return call1Future; + } + remainingCallsExecuted.incrementAndGet(); + return immediateFuture(arg * arg); + })); + + ListenableFuture future = program.evalAsync(); + boolean started = call1Started.await(5, SECONDS); + call1Future.set(0L); + + ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + + assertThat(started).isTrue(); + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("/ by zero"); + assertThat(remainingCallsExecuted.get()).isEqualTo(0); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_memoization_deduplicatesIntAcrossPasses() throws Exception { + AtomicInteger callCounter = new AtomicInteger(); + Program program = + createProgram( + "asyncSquare(3) + asyncSquare(asyncSquare(4))", + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + callCounter.incrementAndGet(); + return immediateFuture(arg * arg); + })); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(265L); + assertThat(callCounter.get()).isEqualTo(3); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_memoization_normalizesDoubleNanAcrossPasses() throws Exception { + AtomicInteger callCounter = new AtomicInteger(); + Program program = + createProgram( + "asyncSquare(dx) + asyncSquare(asyncSquare(2.0))", + CelFunctionBinding.fromAsync( + "asyncSquare_double", + Double.class, + (Double arg) -> { + if (Double.isNaN(arg)) { + callCounter.incrementAndGet(); + } + return immediateFuture(arg * arg); + })); + + Object result = program.evalAsync(ImmutableMap.of("dx", Double.NaN)).get(5, SECONDS); + + assertThat((Double) result).isNaN(); + assertThat(callCounter.get()).isEqualTo(1); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_memoization_normalizesNegativeZeroAcrossPasses() throws Exception { + AtomicInteger callCounter = new AtomicInteger(); + Program program = + createProgram( + "asyncSquare(dx) + asyncSquare(asyncSquare(2.0))", + CelFunctionBinding.fromAsync( + "asyncSquare_double", + Double.class, + (Double arg) -> { + if (arg == 0.0d) { + callCounter.incrementAndGet(); + } + return immediateFuture(arg * arg); + })); + + Object result = program.evalAsync(ImmutableMap.of("dx", -0.0d)).get(5, SECONDS); + + assertThat(result).isEqualTo(16.0d); + assertThat(callCounter.get()).isEqualTo(1); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_protoDifferencerEquality_deduplicatesEquivalentProtoArgs() + throws Exception { + AtomicInteger protoCalls = new AtomicInteger(); + AtomicInteger msgLookups = new AtomicInteger(); + CelCompiler protoCompiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addMessageTypes(TestAllTypes.getDescriptor()) + .addVar("msg", OpaqueType.create("cel.expr.conformance.proto3.TestAllTypes")) + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncProtoVal", + newGlobalOverload( + "asyncProtoVal_msg", + SimpleType.INT, + OpaqueType.create("cel.expr.conformance.proto3.TestAllTypes"))), + newFunctionDeclaration( + "asyncSquare", + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT))) + .build(); + CelAbstractSyntaxTree ast = + protoCompiler.compile("asyncProtoVal(msg) + asyncSquare(asyncSquare(2))").getAst(); + Program program = + plannerRuntimeBuilder() + .setOptions( + CelOptions.current() + .enableHeterogeneousNumericComparisons(true) + .enableProtoDifferencerEquality(true) + .build()) + .addMessageTypes(TestAllTypes.getDescriptor()) + .addFunctionBindings( + ASYNC_SQUARE_INT, + CelFunctionBinding.fromAsync( + "asyncProtoVal_msg", + TestAllTypes.class, + (TestAllTypes msg) -> { + protoCalls.incrementAndGet(); + return immediateFuture(26L); + })) + .build() + .createProgram(ast); + TestAllTypes inner1 = TestAllTypes.newBuilder().setSingleInt32(1).setSingleInt64(2L).build(); + TestAllTypes inner2 = TestAllTypes.newBuilder().setSingleInt64(2L).setSingleInt32(1).build(); + TestAllTypes msg1 = + TestAllTypes.newBuilder() + .setSingleAny( + Any.newBuilder() + .setTypeUrl("type.googleapis.com/cel.expr.conformance.proto3.TestAllTypes") + .setValue(inner1.toByteString())) + .build(); + TestAllTypes msg2 = + TestAllTypes.newBuilder() + .setSingleAny( + Any.newBuilder() + .setTypeUrl("type.googleapis.com/cel.expr.conformance.proto3.TestAllTypes") + .setValue(inner2.toByteString().concat(inner1.toByteString()))) + .build(); + CelVariableResolver resolver = + (String name) -> + name.equals("msg") + ? Optional.of(msgLookups.getAndIncrement() == 0 ? msg1 : msg2) + : Optional.empty(); + + Object result = program.evalAsync(resolver).get(5, SECONDS); + + assertThat(result).isEqualTo(42L); + assertThat(protoCalls.get()).isEqualTo(1); + } + + @Test + @SuppressWarnings({"Immutable", "rawtypes"}) // Test only + public void evalAsync_containerCacheKeys_memoizesAcrossPasses() throws Exception { + AtomicInteger listCalls = new AtomicInteger(); + AtomicInteger mapCalls = new AtomicInteger(); + Program program = + createProgram( + "asyncListSize([1, 2]) + asyncMapSize({'a': 1}) + asyncSquare(asyncSquare(3))", + ASYNC_SQUARE_INT, + CelFunctionBinding.fromAsync( + "asyncListSize_list", + List.class, + (List arg) -> { + listCalls.incrementAndGet(); + return immediateFuture((long) arg.size()); + }), + CelFunctionBinding.fromAsync( + "asyncMapSize_map", + Map.class, + (Map arg) -> { + mapCalls.incrementAndGet(); + return immediateFuture((long) arg.size()); + })); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(84L); + assertThat(listCalls.get()).isEqualTo(1); + assertThat(mapCalls.get()).isEqualTo(1); + } + + @Test + public void evalAsync_memberFunction_evaluatesSuccessfully() throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addVar("x", SimpleType.INT) + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncMemberSquare", + newMemberOverload("int_asyncMemberSquare", SimpleType.INT, SimpleType.INT))) + .build(); + Program memberProgram = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "int_asyncMemberSquare", Long.class, (Long arg) -> immediateFuture(arg * arg))) + .build() + .createProgram(compiler.compile("x.asyncMemberSquare()").getAst()); + + Object result = memberProgram.evalAsync(ImmutableMap.of("x", 7L)).get(5, SECONDS); + + assertThat(result).isEqualTo(49L); + } + + @Test + public void evalAsync_stalledEvaluation_throwsCelEvaluationException() throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "stalledCall", newGlobalOverload("stalledCall_overload", SimpleType.INT))) + .build(); + Program stalledProgram = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.from( + "stalledCall_overload", + ImmutableList.of(), + (args) -> AccumulatedUnknowns.createForAsyncCall(10L, 9999L))) + .build() + .createProgram(compiler.compile("stalledCall()").getAst()); + + ExecutionException stalledEx = + assertThrows(ExecutionException.class, () -> stalledProgram.evalAsync().get(5, SECONDS)); + + assertThat(stalledEx) + .hasCauseThat() + .hasMessageThat() + .contains("Asynchronous evaluation stalled"); + } + + @Test + public void evalAsync_nonStrictErrorAtRoot_throwsCelEvaluationException() throws Exception { + Program nonStrictErrorProgram = createProgram("true && (1 / 0 == 0)"); + + ExecutionException errEx = + assertThrows( + ExecutionException.class, () -> nonStrictErrorProgram.evalAsync().get(5, SECONDS)); + + assertThat(errEx).hasCauseThat().hasMessageThat().contains("/ by zero"); + } + + @Immutable + @SuppressWarnings("Immutable") + private enum DrainStrategyCase { + DRAIN_ALL(CelAsyncDrainStrategy.drainAll()), + DRAIN_NONE(CelAsyncDrainStrategy.drainNone()), + DRAIN_READY_ZERO(CelAsyncDrainStrategy.drainReady(Duration.ZERO)), + DRAIN_READY_50MS(CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50))); + + private final CelAsyncDrainStrategy strategy; + + DrainStrategyCase(CelAsyncDrainStrategy strategy) { + this.strategy = strategy; + } + } + + @Test + public void evalAsync_drainStrategies_evaluatesCorrectly( + @TestParameter DrainStrategyCase testCase) throws Exception { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(testCase.strategy).build(); + Program program = createProgram("asyncSquare(3) + asyncSquare(4)", options, ASYNC_SQUARE_INT); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(25L); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_maxConcurrency_limitsSimultaneousInFlightCalls() throws Exception { + AtomicInteger currentInFlight = new AtomicInteger(); + AtomicInteger maxObservedInFlight = new AtomicInteger(); + CountDownLatch twoInFlightLatch = new CountDownLatch(2); + SettableFuture barrier = SettableFuture.create(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setMaxConcurrency(2).build(); + Program program = + createProgram( + "asyncSquare(1) + asyncSquare(2) + asyncSquare(3) + asyncSquare(4)", + options, + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + int active = currentInFlight.incrementAndGet(); + maxObservedInFlight.accumulateAndGet(active, Math::max); + twoInFlightLatch.countDown(); + return executor.submit( + () -> { + try { + barrier.get(5, SECONDS); + return arg * arg; + } finally { + currentInFlight.decrementAndGet(); + } + }); + })); + + ListenableFuture future = program.evalAsync(); + boolean twoStarted = twoInFlightLatch.await(5, SECONDS); + barrier.set(null); + + assertThat(twoStarted).isTrue(); + assertThat(future.get(5, SECONDS)).isEqualTo(30L); + assertThat(maxObservedInFlight.get()).isAtMost(2); + } + + @Test + public void evalAsync_partialVars_returnsCelUnknownSet( + @TestParameter({"x", "asyncSquare(3) + x"}) String expression) throws Exception { + Program program = createProgram(expression, ASYNC_SQUARE_INT); + + Object result = + program.evalAsync(PartialVars.of(CelAttributePattern.create("x"))).get(5, SECONDS); + + assertThat(result).isInstanceOf(CelUnknownSet.class); + assertThat(((CelUnknownSet) result).attributes()).containsExactly(CelAttribute.create("x")); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_partialVarsReachedInPass2_cancelsInFlightSiblingAndReturnsUnknownSet() + throws Exception { + SettableFuture condFuture = SettableFuture.create(); + SettableFuture slowSibling = SettableFuture.create(); + CountDownLatch siblingStarted = new CountDownLatch(1); + CountDownLatch siblingCancelledLatch = new CountDownLatch(1); + slowSibling.addListener( + () -> { + if (slowSibling.isCancelled()) { + siblingCancelledLatch.countDown(); + } + }, + directExecutor()); + Program shortCircuitUnknownProgram = + createProgram( + "(asyncIsEven(2) || (asyncSquare(10) == 100)) ? x : 0", + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainNone()) + .build(), + CelFunctionBinding.fromAsync("asyncIsEven_int", Long.class, (Long arg) -> condFuture), + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + siblingStarted.countDown(); + return slowSibling; + })); + + ListenableFuture scFuture = + shortCircuitUnknownProgram.evalAsync(PartialVars.of(CelAttributePattern.create("x"))); + boolean started = siblingStarted.await(5, SECONDS); + condFuture.set(true); + + assertThat(started).isTrue(); + assertThat(scFuture.get(5, SECONDS)).isInstanceOf(CelUnknownSet.class); + assertThat(siblingCancelledLatch.await(5, SECONDS)).isTrue(); + } + + @Test + public void evalAsync_deepSequentialChainOnDirectExecutor_completesWithoutStackOverflow() + throws Exception { + StringBuilder expr = new StringBuilder("0"); + for (int i = 0; i < 40; i++) { + expr = new StringBuilder("asyncAdd(").append(expr).append(", 1)"); + } + CelAbstractSyntaxTree ast = CEL_COMPILER.compile(expr.toString()).getAst(); + CelRuntime runtime = + CelRuntimeFactory.plannerRuntimeBuilder() + .setAsyncExecutor(newDirectExecutorService()) + .addFunctionBindings(ASYNC_ADD_INT) + .build(); + Program program = runtime.createProgram(ast); + + Object result = program.evalAsync().get(5, SECONDS); + + assertThat(result).isEqualTo(40L); + } + + @Test + public void evalAsync_concurrentProgramInvocations_isolatesStatePerRun() throws Exception { + Program program = createProgram("asyncSquare(x) + 1", ASYNC_SQUARE_INT); + List> futures = new ArrayList<>(); + + for (long i = 1; i <= 10; i++) { + futures.add(program.evalAsync(ImmutableMap.of("x", i))); + } + + long deadlineNanos = System.nanoTime() + Duration.ofSeconds(5).toNanos(); + for (int i = 0; i < futures.size(); i++) { + long expected = (long) (i + 1) * (i + 1) + 1; + long remainingNanos = Math.max(1L, deadlineNanos - System.nanoTime()); + assertThat(futures.get(i).get(remainingNanos, NANOSECONDS)).isEqualTo(expected); + } + } + + @Test + public void evalAsync_argEvaluatesToLocalizedErrorValue_propagatesLocalizedError() + throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncCheck", + newGlobalOverload("asyncCheck_bool", SimpleType.BOOL, SimpleType.BOOL))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncCheck_bool", Boolean.class, (Boolean arg) -> immediateFuture(arg))) + .build(); + Program program = + runtime.createProgram(compiler.compile("asyncCheck(true && (1 / 0 == 1))").getAst()); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasCauseThat().isInstanceOf(CelDivideByZeroException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("/ by zero"); + } + + @Test + public void evalAsync_argEvaluatesToNonLocalizedErrorValue_wrapsAndPropagatesError() + throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncCheck", + newGlobalOverload("asyncCheck_bool", SimpleType.BOOL, SimpleType.BOOL)), + newFunctionDeclaration( + "syncCrash", newGlobalOverload("syncCrash_overload", SimpleType.BOOL))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncCheck_bool", Boolean.class, (Boolean arg) -> immediateFuture(arg)), + CelFunctionBinding.from( + "syncCrash_overload", + ImmutableList.of(), + (args) -> { + throw new IllegalStateException("synchronous function crash"); + })) + .build(); + Program program = + runtime.createProgram(compiler.compile("asyncCheck(true && syncCrash())").getAst()); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e) + .hasCauseThat() + .hasCauseThat() + .hasMessageThat() + .contains("synchronous function crash"); + } + + @Test + public void evalAsync_argEvaluatesToCelRuntimeExceptionErrorValue_wrapsWithErrorCode() + throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncCheck", + newGlobalOverload("asyncCheck_bool", SimpleType.BOOL, SimpleType.BOOL))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncCheck_bool", Boolean.class, (Boolean arg) -> immediateFuture(arg))) + .build(); + Program program = + runtime.createProgram(compiler.compile("asyncCheck(true && [true][1])").getAst()); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(((CelEvaluationException) e.getCause()).getErrorCode()) + .isEqualTo(CelErrorCode.INDEX_OUT_OF_BOUNDS); + } + + @Test + public void evalAsync_asyncFunctionFailsWithCelRuntimeException_preservesErrorCode() + throws Exception { + Program program = + createProgram( + "asyncSquare(5)", + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> + immediateFailedFuture( + new CelDivideByZeroException(new ArithmeticException("/ by zero"))))); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(((CelEvaluationException) e.getCause()).getErrorCode()) + .isEqualTo(CelErrorCode.DIVIDE_BY_ZERO); + } + + @Test + public void evalAsync_argEvaluatesToGenericErrorValue_wrapsAsInternalError() throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addVar("x", SimpleType.DYN) + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncCheck", + newGlobalOverload("asyncCheck_bool", SimpleType.BOOL, SimpleType.BOOL))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncCheck_bool", Boolean.class, (Boolean arg) -> immediateFuture(arg))) + .build(); + Program program = runtime.createProgram(compiler.compile("asyncCheck(x && true)").getAst()); + + ExecutionException e = + assertThrows( + ExecutionException.class, + () -> program.evalAsync(ImmutableMap.of("x", 1L)).get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(((CelEvaluationException) e.getCause()).getErrorCode()) + .isEqualTo(CelErrorCode.INTERNAL_ERROR); + assertThat(e).hasCauseThat().hasCauseThat().isInstanceOf(IllegalArgumentException.class); + assertThat(e) + .hasCauseThat() + .hasCauseThat() + .hasMessageThat() + .contains("Expected boolean value, found: 1"); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_syncFunctionThrowsWithCause_unwrapsOriginalCause() throws Exception { + IllegalArgumentException rootCause = new IllegalArgumentException("nested root cause"); + Program program = + createProgram( + "asyncSyncThrow(1)", + CelFunctionBinding.from( + "asyncSyncThrow_int", + Long.class, + (Long arg) -> { + throw new RuntimeException("outer wrapper", rootCause); + })); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasCauseThat().hasCauseThat().isSameInstanceAs(rootCause); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_maxIterationsExceeded_cancelsInFlightCalls() throws Exception { + SettableFuture inFlightSibling = SettableFuture.create(); + SettableFuture call1Future = SettableFuture.create(); + CountDownLatch siblingStarted = new CountDownLatch(1); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setMaxIterations(1).build(); + Program program = + createProgram( + "asyncSquare(asyncSquare(2)) + asyncSquare(10)", + options, + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + if (arg == 10L) { + siblingStarted.countDown(); + return inFlightSibling; + } + return call1Future; + })); + + ListenableFuture future = program.evalAsync(); + boolean started = siblingStarted.await(5, SECONDS); + call1Future.set(4L); + ExecutionException e = assertThrows(ExecutionException.class, () -> future.get(5, SECONDS)); + + assertThat(started).isTrue(); + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e) + .hasCauseThat() + .hasMessageThat() + .contains("Exceeded maximum async evaluation iterations: 1"); + assertThat(inFlightSibling.isCancelled()).isTrue(); + } + + @Test + public void evalAsync_unboundedMaxIterations_succeeds() throws Exception { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setMaxIterations(-1).build(); + Program program = createProgram("asyncSquare(5)", options, ASYNC_SQUARE_INT); + + ListenableFuture future = program.evalAsync(); + + assertThat(future.get(5, SECONDS)).isEqualTo(25L); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void + evalAsync_strictAsyncFunctionWithImmediateErrorAndAsyncSibling_doesNotDispatchSibling() + throws Exception { + AtomicInteger siblingDispatched = new AtomicInteger(); + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncAdd", + newGlobalOverload( + "asyncAdd_int_int", SimpleType.INT, SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "asyncSquare", + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT))) + .build(); + CelRuntime runtime = + plannerRuntimeBuilder() + .addFunctionBindings( + CelFunctionBinding.fromAsync( + "asyncAdd_int_int", + Long.class, + Long.class, + (Long a, Long b) -> immediateFuture(a + b)), + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + siblingDispatched.incrementAndGet(); + return immediateFuture(arg * arg); + })) + .build(); + Program program = + runtime.createProgram( + compiler.compile("asyncAdd(true ? (1 / 0) : 0, asyncSquare(10))").getAst()); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("/ by zero"); + assertThat(siblingDispatched.get()).isEqualTo(0); + } + + @Test + public void evalAsync_errorValueAtRoot_throwsCelEvaluationException() throws Exception { + Program program = + createProgram( + "(1 / 0 == 0) || (asyncSquare(10) == 0)", + CelFunctionBinding.fromAsync( + "asyncSquare_int", Long.class, (Long arg) -> immediateFuture(100L))); + + ExecutionException e = + assertThrows(ExecutionException.class, () -> program.evalAsync().get(5, SECONDS)); + + assertThat(e).hasCauseThat().isInstanceOf(CelEvaluationException.class); + assertThat(e).hasCauseThat().hasMessageThat().contains("/ by zero"); + } + + @Test + @SuppressWarnings("Immutable") // Test only + public void evalAsync_asyncCallWithUnknownArgument_returnsUnknownSetWithoutDispatchingCall() + throws Exception { + AtomicInteger callCounter = new AtomicInteger(); + Program program = + createProgram( + "asyncSquare(x)", + CelFunctionBinding.fromAsync( + "asyncSquare_int", + Long.class, + (Long arg) -> { + callCounter.incrementAndGet(); + return immediateFuture(arg * arg); + })); + + Object result = + program + .evalAsync(PartialVars.of(CelAttributePattern.fromQualifiedIdentifier("x"))) + .get(5, SECONDS); + + assertThat(result).isInstanceOf(CelUnknownSet.class); + assertThat(callCounter.get()).isEqualTo(0); + } + + private CelRuntimeBuilder plannerRuntimeBuilder() { + return CelRuntimeFactory.plannerRuntimeBuilder().setAsyncExecutor(executor); + } + + private Program createProgram(String expression, CelFunctionBinding... bindings) + throws Exception { + CelAbstractSyntaxTree ast = CEL_COMPILER.compile(expression).getAst(); + return plannerRuntimeBuilder().addFunctionBindings(bindings).build().createProgram(ast); + } + + private Program createProgram( + String expression, CelAsyncEvaluationOptions asyncOptions, CelFunctionBinding... bindings) + throws Exception { + CelAbstractSyntaxTree ast = CEL_COMPILER.compile(expression).getAst(); + return plannerRuntimeBuilder() + .setAsyncEvaluationOptions(asyncOptions) + .addFunctionBindings(bindings) + .build() + .createProgram(ast); + } + + @ThreadSafe + private static final class RecordingObserver implements CelAsyncObserver { + private final CopyOnWriteArrayList startedCalls = new CopyOnWriteArrayList<>(); + private final CopyOnWriteArrayList finishedCalls = new CopyOnWriteArrayList<>(); + + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) { + startedCalls.add(new RecordedCall(call, args, /* result= */ null, /* error= */ null)); + } + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + finishedCalls.add(new RecordedCall(call, ImmutableList.of(), result, error)); + } + + Optional startedFunctionName() { + return startedCalls.isEmpty() + ? Optional.empty() + : Optional.of(startedCalls.get(0).call.functionName()); + } + + Optional> startedArgs() { + return startedCalls.isEmpty() ? Optional.empty() : Optional.of(startedCalls.get(0).args); + } + + Optional finishedFunctionName() { + return finishedCalls.isEmpty() + ? Optional.empty() + : Optional.of(finishedCalls.get(0).call.functionName()); + } + + Optional finishedResult() { + return finishedCalls.isEmpty() + ? Optional.empty() + : Optional.ofNullable(finishedCalls.get(0).result); + } + + Optional finishedError() { + return finishedCalls.isEmpty() + ? Optional.empty() + : Optional.ofNullable(finishedCalls.get(0).error); + } + + private RecordingObserver() {} + } + + @Immutable + @SuppressWarnings("Immutable") + private static final class RecordedCall { + private final CelAsyncCall call; + private final ImmutableList args; + private final @Nullable Object result; + private final @Nullable Throwable error; + + private RecordedCall( + CelAsyncCall call, + ImmutableList args, + @Nullable Object result, + @Nullable Throwable error) { + this.call = call; + this.args = args; + this.result = result; + this.error = error; + } + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java index ebf8e1cdb..ac6d14d66 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java +++ b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java @@ -16,6 +16,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.Futures.immediateFuture; import static com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; import static dev.cel.common.CelFunctionDecl.newFunctionDeclaration; import static dev.cel.common.CelOverloadDecl.newGlobalOverload; @@ -28,7 +29,6 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.primitives.UnsignedLong; -import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.testing.junit.testparameterinjector.TestParameter; import com.google.testing.junit.testparameterinjector.TestParameterInjector; @@ -1264,21 +1264,27 @@ public void newPlanner_nullAsyncOptions_throwsNullPointerException() { } @Test - public void plan_asyncFunction_evalSynchronously_throwsCelEvaluationException() throws Exception { + public void plan_asyncFunction_evalSynchronously_throwsCelEvaluationException( + @TestParameter({"asyncSquare(5)", "add(1, asyncSquare(5))"}) String expression) + throws Exception { CelCompiler compiler = CelCompilerFactory.standardCelCompilerBuilder() .addFunctionDeclarations( newFunctionDeclaration( "asyncSquare", - newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT))) + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT)), + newFunctionDeclaration( + "add", + newGlobalOverload("add_int", SimpleType.INT, SimpleType.INT, SimpleType.INT))) .build(); - CelAbstractSyntaxTree ast = compiler.compile("asyncSquare(5)").getAst(); + CelAbstractSyntaxTree ast = compiler.compile(expression).getAst(); DefaultDispatcher.Builder dispatcher = DefaultDispatcher.newBuilder(); addBindingsToDispatcher( dispatcher, ImmutableList.of( CelFunctionBinding.fromAsync( - "asyncSquare_int", Long.class, (Long arg) -> Futures.immediateFuture(arg * arg)))); + "asyncSquare_int", Long.class, (Long arg) -> immediateFuture(arg * arg)), + CelFunctionBinding.from("add_int", Long.class, Long.class, (Long a, Long b) -> a + b))); ProgramPlanner planner = ProgramPlanner.newPlanner( TYPE_PROVIDER, @@ -1297,6 +1303,7 @@ public void plan_asyncFunction_evalSynchronously_throwsCelEvaluationException() assertThat(e) .hasMessageThat() .contains("Async function 'asyncSquare' evaluated in synchronous mode."); + assertThat(e).hasCauseThat().isNull(); } @Test