Fix fold over canon streams (#307)

air/src/execution_step/air/ap/apply_to_arguments.rs

@@ -15,6 +15,8 @@
  */
 
 use super::*;
+use crate::execution_step::PEEK_ALLOWED_ON_NON_EMPTY;
+
 use air_lambda_parser::LambdaAST;
 use air_parser::ast;
 
@@ -90,10 +92,7 @@ fn apply_scalar_impl(
     let mut result = match scalar {
         ScalarRef::Value(result) => result.clone(),
         ScalarRef::IterableValue(iterator) => {
-            let result = iterator.iterable.peek().expect(
-                "peek always return elements inside fold,\
-            this guaranteed by implementation of next and avoiding empty folds",
-            );
+            let result = iterator.iterable.peek().expect(PEEK_ALLOWED_ON_NON_EMPTY);
             result.into_resolved_result()
         }
     };

air/src/execution_step/air/fold/utils.rs

@@ -16,6 +16,7 @@
 
 use super::*;
 use crate::execution_step::CatchableError;
+use crate::execution_step::PEEK_ALLOWED_ON_NON_EMPTY;
 use crate::JValue;
 use crate::LambdaAST;
 use crate::SecurityTetraplet;
@@ -51,6 +52,10 @@ pub(crate) fn construct_canon_stream_iterable_value<'ctx>(
     exec_ctx: &ExecutionCtx<'ctx>,
 ) -> ExecutionResult<FoldIterableScalar> {
     let canon_stream = exec_ctx.scalars.get_canon_stream(ast_canon_stream.name)?;
+    if canon_stream.is_empty() {
+        return Ok(FoldIterableScalar::Empty);
+    }
+
     // TODO: this one is a relatively long operation and will be refactored in Boxed Value
     let iterable_ingredients = CanonStreamIterableIngredients::init(canon_stream.clone());
     let iterable = Box::new(iterable_ingredients);
@@ -86,7 +91,7 @@ fn create_scalar_iterable<'ctx>(
     match exec_ctx.scalars.get_value(variable_name)? {
         ScalarRef::Value(call_result) => from_value(call_result.clone(), variable_name),
         ScalarRef::IterableValue(fold_state) => {
-            let iterable_value = fold_state.iterable.peek().unwrap();
+            let iterable_value = fold_state.iterable.peek().expect(PEEK_ALLOWED_ON_NON_EMPTY);
             let call_result = iterable_value.into_resolved_result();
             from_value(call_result, variable_name)
         }
@@ -134,7 +139,7 @@ fn create_scalar_lambda_iterable<'ctx>(
             from_jvalue(jvalues, tetraplet, lambda)
         }
         ScalarRef::IterableValue(fold_state) => {
-            let iterable_value = fold_state.iterable.peek().unwrap();
+            let iterable_value = fold_state.iterable.peek().expect(PEEK_ALLOWED_ON_NON_EMPTY);
             let jvalue = iterable_value.apply_lambda(lambda, exec_ctx)?;
             let tetraplet = to_tetraplet(&iterable_value);
 

air/src/execution_step/air/next.rs

@@ -19,6 +19,7 @@ use super::ExecutionCtx;
 use super::ExecutionResult;
 use super::FoldState;
 use super::TraceHandler;
+use crate::execution_step::PEEK_ALLOWED_ON_NON_EMPTY;
 use crate::log_instruction;
 use crate::trace_to_exec_err;
 
@@ -63,7 +64,10 @@ fn maybe_meet_iteration_start<'i>(
 ) -> ExecutionResult<()> {
     if let IterableType::Stream(fold_id) = &fold_state.iterable_type {
         trace_to_exec_err!(
-            trace_ctx.meet_iteration_start(*fold_id, fold_state.iterable.peek().unwrap().pos()),
+            trace_ctx.meet_iteration_start(
+                *fold_id,
+                fold_state.iterable.peek().expect(PEEK_ALLOWED_ON_NON_EMPTY).pos()
+            ),
             next
         )?;
     }

air/src/execution_step/boxed_value/iterable.rs

@@ -15,12 +15,12 @@
  */
 
 mod canon_stream;
-mod json_path_result;
+mod lambda_result;
 mod resolved_call;
 mod vec_resolved_call;
 
 pub(crate) use canon_stream::CanonStreamIterableIngredients;
-pub(crate) use json_path_result::IterableLambdaResult;
+pub(crate) use lambda_result::IterableLambdaResult;
 pub(crate) use resolved_call::IterableResolvedCall;
 pub(crate) use vec_resolved_call::IterableVecResolvedCall;
 

air/src/execution_step/boxed_value/scalar.rs

@@ -17,6 +17,7 @@
 use super::JValuable;
 use crate::execution_step::FoldState;
 use crate::execution_step::RcSecurityTetraplet;
+use crate::execution_step::PEEK_ALLOWED_ON_NON_EMPTY;
 use crate::JValue;
 
 use air_interpreter_data::TracePos;
@@ -42,7 +43,7 @@ impl<'i> ScalarRef<'i> {
         match self {
             ScalarRef::Value(value) => Box::new(value.clone()),
             ScalarRef::IterableValue(fold_state) => {
-                let peeked_value = fold_state.iterable.peek().unwrap();
+                let peeked_value = fold_state.iterable.peek().expect(PEEK_ALLOWED_ON_NON_EMPTY);
                 Box::new(peeked_value)
             }
         }

air/src/execution_step/lambda_applier/utils.rs

@@ -17,6 +17,7 @@
 use super::LambdaError;
 use super::LambdaResult;
 use crate::execution_step::ScalarRef;
+use crate::execution_step::PEEK_ALLOWED_ON_NON_EMPTY;
 use crate::JValue;
 
 pub(super) fn try_jvalue_with_idx(jvalue: &JValue, idx: u32) -> LambdaResult<&JValue> {
@@ -61,8 +62,11 @@ pub(super) fn select_by_scalar<'value, 'i>(
     match scalar_ref {
         Value(lambda_value) => select_by_jvalue(value, &lambda_value.result),
         IterableValue(fold_state) => {
-            // it's safe because iterable always point to valid value
-            let accessor = fold_state.iterable.peek().unwrap().into_resolved_result();
+            let accessor = fold_state
+                .iterable
+                .peek()
+                .expect(PEEK_ALLOWED_ON_NON_EMPTY)
+                .into_resolved_result();
             select_by_jvalue(value, &accessor.result)
         }
     }
@@ -72,8 +76,11 @@ pub(super) fn try_scalar_ref_as_idx(scalar: ScalarRef<'_>) -> LambdaResult<u32>
     match scalar {
         ScalarRef::Value(accessor) => try_jvalue_as_idx(&accessor.result),
         ScalarRef::IterableValue(accessor) => {
-            // it's safe because iterable always point to valid value
-            let accessor = accessor.iterable.peek().unwrap().into_resolved_result();
+            let accessor = accessor
+                .iterable
+                .peek()
+                .expect(PEEK_ALLOWED_ON_NON_EMPTY)
+                .into_resolved_result();
             try_jvalue_as_idx(&accessor.result)
         }
     }

air/src/execution_step/mod.rs

@@ -21,6 +21,9 @@ pub(crate) mod execution_context;
 mod lambda_applier;
 mod resolver;
 
+const PEEK_ALLOWED_ON_NON_EMPTY: &str = "peek always return elements inside fold,\
+            this guaranteed by implementation of next and avoiding empty folds";
+
 pub use errors::CatchableError;
 pub use errors::ExecutionError;
 pub use errors::UncatchableError;

air/tests/test_module/issues/issue_306.rs

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2022 Fluence Labs Limited
+ *
+ * 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
+ *
+ *     http://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.
+ */
+
+use air_interpreter_interface::INTERPRETER_SUCCESS;
+use air_test_utils::prelude::*;
+
+#[test]
+// test for github.com/fluencelabs/aquavm/issues/306
+fn issue_306() {
+    let peer_id_1 = "peer_id_1";
+    let mut peer_vm_1 = create_avm(echo_call_service(), peer_id_1);
+
+    let script = f!(r#"
+        (new $stream
+            (seq
+                (canon "{peer_id_1}" $stream #canon_stream)
+                (fold #canon_stream iterator
+                    (ap iterator $stream))))
+    "#);
+
+    let result = call_vm!(peer_vm_1, <_>::default(), &script, "", "");
+    assert_eq!(result.ret_code, INTERPRETER_SUCCESS)
+}

air/tests/test_module/issues/mod.rs

@@ -29,3 +29,4 @@ mod issue_241;
 mod issue_295;
 mod issue_300;
 mod issue_304;
+mod issue_306;