Function TestConcurrentEvalCancelDuringDebounce

cel/program_async_test.go:783–814

(t *testing.T)

}

func TestConcurrentEvalCancelDuringDebounce(t *testing.T) {
	// Tests context cancellation while awaiting a debounce timeout in the completion drain loop.
	prg := mustProgram(t, `delayed_rpc("first", 10) + delayed_rpc("second", 1000)`,
		cel.Function("delayed_rpc",
			cel.Overload("delayed_rpc_string_int", []*cel.Type{cel.StringType, cel.IntType}, cel.StringType,
				cel.AsyncBinding(func(ctx context.Context, args ...ref.Val) ref.Val {
					msg := string(args[0].(types.String))
					delayMs := time.Duration(int64(args[1].(types.Int))) * time.Millisecond
					time.Sleep(delayMs)
					return types.String(msg)
				}),
			),
		),
		cel.ConcurrentDrainStrategy(async.DrainReady(10*time.Second)),
	)

	ctx, cancel := context.WithCancel(context.Background())
	resCh := prg.ConcurrentEval(ctx, cel.NoVars())

	// Wait for the first call (10ms) to complete and enter the 10-second debounce wait.
	time.Sleep(30 * time.Millisecond)
	cancel()

	select {
	case res := <-resCh:
		if res.Err == nil || !errors.Is(res.Err, context.Canceled) {
			t.Fatalf("ConcurrentEval() error = %v, want context.Canceled", res.Err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("ConcurrentEval() timed out waiting for cancellation during debounce")
	}
}

func TestConcurrentEvalRecover(t *testing.T) {
	env, err := cel.NewEnv(