fix(execution-engine): treat non-defined stream as empty in canon (#347)

This PR forces `canon` instruction to treat streams as empty, it's crucial for deterministic behaviour.

Closes #346.

air/src/execution_step/air/canon.rs

@@ -19,9 +19,9 @@ use super::ExecutionResult;
 use super::TraceHandler;
 use crate::execution_step::boxed_value::CanonStream;
 use crate::execution_step::Generation;
+use crate::execution_step::Stream;
 use crate::log_instruction;
 use crate::trace_to_exec_err;
-use crate::CatchableError;
 use crate::ExecutionError;
 use crate::UncatchableError;
 
@@ -30,7 +30,7 @@ use air_interpreter_data::TracePos;
 use air_parser::ast;
 use air_trace_handler::MergerCanonResult;
 
-use std::rc::Rc;
+use std::borrow::Cow;
 
 impl<'i> super::ExecutableInstruction<'i> for ast::Canon<'i> {
     #[tracing::instrument(level = "debug", skip(exec_ctx, trace_ctx))]
@@ -90,14 +90,7 @@ fn create_canon_stream_from_pos(
     ast_canon: &ast::Canon<'_>,
     exec_ctx: &ExecutionCtx<'_>,
 ) -> ExecutionResult<CanonStream> {
-    let stream = exec_ctx
-        .streams
-        .get(ast_canon.stream.name, ast_canon.stream.position)
-        .ok_or_else(|| {
-            ExecutionError::Catchable(Rc::new(CatchableError::StreamsForCanonNotFound(
-                ast_canon.stream.name.to_string(),
-            )))
-        })?;
+    let stream = get_stream_or_default(ast_canon, exec_ctx);
 
     let values = stream_elements_pos
         .iter()
@@ -145,16 +138,9 @@ fn create_canon_stream_from_name(
     peer_id: String,
     exec_ctx: &ExecutionCtx<'_>,
 ) -> ExecutionResult<StreamWithPositions> {
-    let stream = exec_ctx
-        .streams
-        .get(ast_canon.stream.name, ast_canon.stream.position)
-        .ok_or_else(|| {
-            ExecutionError::Catchable(Rc::new(CatchableError::StreamsForCanonNotFound(
-                ast_canon.stream.name.to_string(),
-            )))
-        })?;
+    let stream = get_stream_or_default(ast_canon, exec_ctx);
 
-    let canon_stream = CanonStream::from_stream(stream, peer_id);
+    let canon_stream = CanonStream::from_stream(stream.as_ref(), peer_id);
     let stream_elements_pos = stream
         .iter(Generation::Last)
         // it's always safe to iter over all generations
@@ -169,3 +155,14 @@ fn create_canon_stream_from_name(
 
     Ok(result)
 }
+
+/// This function gets a stream from context or return a default empty stream,
+/// it's crucial for deterministic behaviour, for more info see
+/// github.com/fluencelabs/aquavm/issues/346
+fn get_stream_or_default<'ctx, 'value>(
+    ast_canon: &ast::Canon<'_>,
+    exec_ctx: &'ctx ExecutionCtx<'value>,
+) -> Cow<'ctx, Stream> {
+    let maybe_stream = exec_ctx.streams.get(ast_canon.stream.name, ast_canon.stream.position);
+    maybe_stream.map(Cow::Borrowed).unwrap_or_default()
+}

air/src/execution_step/errors/catchable_errors.rs

@@ -88,10 +88,6 @@ pub enum CatchableError {
     #[error("variable with name '{0}' was cleared by new and then wasn't set")]
     VariableWasNotInitializedAfterNew(String),
 
-    /// Canon instruction can't canonicalize a stream since it's been found.
-    #[error("stream with name {0} wasn't defined, so canon instruction can't canonicalize it")]
-    StreamsForCanonNotFound(String),
-
     /// This error type is occurred when the length functor applied to a value of non-array type.
     #[error("the length functor could applied only to an array-like value, but it's applied to '{0}'")]
     LengthFunctorAppliedToNotArray(JValue),

air/tests/test_module/features/join_behaviour/join_behaviour.rs

@@ -248,7 +248,7 @@ fn fold_with_join_behaviour() {
 }
 
 #[test]
-fn canon_with_join_behaviour() {
+fn canon_with_empty_behaviour() {
     let peer_1_id = "peer_1_id";
     let peer_2_id = "peer_2_id";
 
@@ -262,7 +262,11 @@ fn canon_with_join_behaviour() {
 
     let result = checked_call_vm!(peer_2, <_>::default(), script, "", "");
     let actual_trace = trace_from_result(&result);
-    let expected_trace = vec![executed_state::par(1, 0), executed_state::request_sent_by(peer_2_id)];
+    let expected_trace = vec![
+        executed_state::par(1, 1),
+        executed_state::request_sent_by(peer_2_id),
+        executed_state::canon(vec![]),
+    ];
 
     assert_eq!(actual_trace, expected_trace);
 }

air/tests/test_module/instructions/canon.rs

@@ -233,3 +233,38 @@ fn canon_empty_stream() {
     let expected_trace = vec![executed_state::canon(vec![]), executed_state::scalar(json!([]))];
     assert_eq!(actual_trace, expected_trace);
 }
+
+#[test]
+fn canon_over_later_defined_stream() {
+    let vm_peer_id_1 = "vm_peer_id_1";
+    let mut peer_vm_1 = create_avm(echo_call_service(), vm_peer_id_1);
+
+    let vm_peer_id_2 = "vm_peer_id_2";
+    let mut peer_vm_2 = create_avm(echo_call_service(), vm_peer_id_2);
+
+    let vm_peer_id_3 = "vm_peer_id_3";
+    let mut peer_vm_3 = create_avm(echo_call_service(), vm_peer_id_3);
+
+    let script = f!(r#"
+        (par
+            (call "{vm_peer_id_2}" ("" "") [1] $stream)
+            (seq
+                (canon "{vm_peer_id_1}" $stream #canon_stream) ; it returns a catchable error
+                (call "{vm_peer_id_3}" ("" "") [#canon_stream])
+            )
+        )
+    "#);
+
+    let result = checked_call_vm!(peer_vm_1, <_>::default(), &script, "", "");
+    let result = checked_call_vm!(peer_vm_2, <_>::default(), &script, "", result.data);
+    let result = checked_call_vm!(peer_vm_3, <_>::default(), &script, "", result.data);
+    let actual_trace = trace_from_result(&result);
+
+    let expected_trace = vec![
+        executed_state::par(1, 2),
+        executed_state::stream_number(1, 0),
+        executed_state::canon(vec![]),
+        executed_state::scalar(json!([])),
+    ];
+    assert_eq!(actual_trace, expected_trace);
+}

air/tests/test_module/issues/issue_346.rs

@@ -0,0 +1,38 @@
+/*
+ * 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/346
+fn issue_346() {
+    let vm_peer_id = "peer_id_1";
+    let mut peer_vm = create_avm(echo_call_service(), vm_peer_id);
+
+    let script = f!(r#"
+        (par
+            (call "unknown_peer" ("" "") [] $stream) ; to make validator happy
+            (xor
+                (canon "{vm_peer_id}" $stream #canon_stream) ; it returns a catchable error
+                (call "{vm_peer_id}" ("" "") [""])
+            )
+        )
+    "#);
+
+    let result = call_vm!(peer_vm, <_>::default(), &script, "", "");
+    assert_eq!(result.ret_code, INTERPRETER_SUCCESS);
+}

air/tests/test_module/issues/mod.rs

@@ -31,3 +31,4 @@ mod issue_300;
 mod issue_304;
 mod issue_306;
 mod issue_331;
+mod issue_346;