Fix a bug on json path on empty stream (#133)

.cargo/config.toml

@@ -0,0 +1,6 @@
+[http]
+timeout = 30                # timeout for each HTTP request, in seconds
+multiplexing = false        # HTTP/2 multiplexing
+
+[net]
+retry = 50                  # network retries

air/src/execution_step/air/call.rs

@@ -26,7 +26,7 @@ use super::ExecutionError;
 use super::ExecutionResult;
 use super::LastErrorDescriptor;
 use super::TraceHandler;
-use crate::execution_step::joinable::Joinable;
+use crate::execution_step::Joinable;
 use crate::execution_step::RSecurityTetraplet;
 use crate::joinable_call;
 use crate::log_instruction;

air/src/execution_step/air/match_.rs

@@ -19,7 +19,7 @@ use super::ExecutionCtx;
 use super::ExecutionError;
 use super::ExecutionResult;
 use super::TraceHandler;
-use crate::execution_step::joinable::Joinable;
+use crate::execution_step::Joinable;
 use crate::joinable;
 use crate::log_instruction;
 

air/src/execution_step/air/mismatch.rs

@@ -19,7 +19,7 @@ use super::ExecutionCtx;
 use super::ExecutionError;
 use super::ExecutionResult;
 use super::TraceHandler;
-use crate::execution_step::joinable::Joinable;
+use crate::execution_step::Joinable;
 use crate::joinable;
 use crate::log_instruction;
 

air/src/execution_step/boxed_value/jvaluable.rs

@@ -15,7 +15,7 @@
  */
 
 mod cell_vec_resolved_call_result;
-mod empty;
+mod empty_stream;
 mod iterable_item;
 mod resolved_call_result;
 mod stream;

air/src/execution_step/boxed_value/jvaluable/empty_stream.rs

@@ -14,20 +14,24 @@
  * limitations under the License.
  */
 
+use super::ExecutionError;
 use super::ExecutionResult;
 use super::JValuable;
+use crate::exec_err;
 use crate::execution_step::SecurityTetraplets;
 use crate::JValue;
 
 use std::borrow::Cow;
 
 impl JValuable for () {
-    fn apply_json_path(&self, _json_path: &str) -> ExecutionResult<Vec<&JValue>> {
-        Ok(vec![])
+    fn apply_json_path(&self, json_path: &str) -> ExecutionResult<Vec<&JValue>> {
+        // applying json path to an empty stream will produce a join behaviour
+        exec_err!(ExecutionError::EmptyStreamJsonPathError(json_path.to_string()))
     }
 
-    fn apply_json_path_with_tetraplets(&self, _json_path: &str) -> ExecutionResult<(Vec<&JValue>, SecurityTetraplets)> {
-        Ok((vec![], vec![]))
+    fn apply_json_path_with_tetraplets(&self, json_path: &str) -> ExecutionResult<(Vec<&JValue>, SecurityTetraplets)> {
+        // applying json path to an empty stream will produce a join behaviour
+        exec_err!(ExecutionError::EmptyStreamJsonPathError(json_path.to_string()))
     }
 
     fn as_jvalue(&self) -> Cow<'_, JValue> {

air/src/execution_step/errors.rs

@@ -15,12 +15,13 @@
  */
 
 mod catchable;
+mod joinable;
 
 pub(crate) use catchable::Catchable;
+pub(crate) use joinable::Joinable;
 
 use super::trace_handler::MergerApResult;
 use super::trace_handler::TraceHandlerError;
-use super::Joinable;
 use super::ResolvedCallResult;
 use super::Stream;
 use crate::build_targets::CallServiceResult;
@@ -67,6 +68,10 @@ pub(crate) enum ExecutionError {
     #[error("variable with path '{1}' not found in '{0:?}' with error: '{2}'")]
     StreamJsonPathError(Stream, String, JsonPathError),
 
+    /// An error occurred while trying to apply json path to an empty stream.
+    #[error("json path {0} is applied to an empty stream")]
+    EmptyStreamJsonPathError(String),
+
     /// Provided JValue has incompatible with target type.
     #[error("expected JValue type '{1}', but got '{0}' JValue")]
     IncompatibleJValueType(JValue, &'static str),
@@ -155,7 +160,8 @@ impl ExecutionError {
             StreamJsonPathError(..) => 18,
             StreamDontHaveSuchGeneration(..) => 19,
             ApResultNotCorrespondToInstr(_) => 20,
-            TraceError(_) => 21,
+            EmptyStreamJsonPathError(_) => 21,
+            TraceError(_) => 22,
         }
     }
 }
@@ -182,6 +188,10 @@ impl Joinable for ExecutionError {
                 log_join!("  waiting for an argument with path '{}' on stream '{:?}'", json_path, stream);
                 true
             }
+            EmptyStreamJsonPathError(json_path) => {
+                log_join!("  waiting on empty stream for path '{}'", json_path);
+                true
+            }
 
             _ => false,
         }

air/src/execution_step/mod.rs

@@ -18,7 +18,6 @@ mod air;
 mod boxed_value;
 mod errors;
 pub(crate) mod execution_context;
-mod joinable;
 mod trace_handler;
 mod utils;
 
@@ -30,8 +29,8 @@ pub(super) use boxed_value::Scalar;
 pub(super) use boxed_value::Stream;
 pub(crate) use errors::Catchable;
 pub(super) use errors::ExecutionError;
+pub(crate) use errors::Joinable;
 pub(crate) use execution_context::ExecutionCtx;
-use joinable::Joinable;
 pub(crate) use trace_handler::TraceHandler;
 
 use std::cell::RefCell;

air/tests/test_module/integration/join_behaviour.rs

@@ -14,12 +14,12 @@
  * limitations under the License.
  */
 
-use air_test_utils::call_vm;
 use air_test_utils::checked_call_vm;
 use air_test_utils::create_avm;
 use air_test_utils::set_variables_call_service;
 use air_test_utils::trace_from_result;
 use air_test_utils::unit_call_service;
+use air_test_utils::{call_vm, echo_call_service};
 
 use serde_json::json;
 
@@ -108,6 +108,35 @@ fn wait_on_stream_json_path_by_id() {
     assert_eq!(actual_trace.len(), 2); // par and the first call emit traces, second call doesn't
 }
 
+#[test]
+fn wait_on_empty_stream_json_path() {
+    let local_peer_id = "local_peer_id";
+    let mut local_vm = create_avm(echo_call_service(), local_peer_id);
+
+    let join_stream_script = format!(
+        r#"
+
+    (seq
+        (seq
+            (call "{0}" ("" "") [[]] nodes)
+            (fold nodes n
+                (par
+                    (call n ("" "") [n] $ns)
+                    (next n)
+                )
+            )
+        )
+        (call "{0}" ("" "") [$ns.$.[0] $ns.$.[1] $ns])
+     )"#,
+        local_peer_id
+    );
+
+    let result = checked_call_vm!(local_vm, "", join_stream_script, "", "");
+    let actual_trace = trace_from_result(&result);
+
+    assert_eq!(actual_trace.len(), 1); // only the first call should produce a trace
+}
+
 #[test]
 fn dont_wait_on_json_path_on_scalars() {
     let array = json!([1, 2, 3, 4, 5]);

air/tests/test_module/integration/streams_early_exit.rs

@@ -104,7 +104,7 @@ fn par_early_exit() {
     ];
     let setter_3_malicious_data = raw_data_from_trace(setter_3_malicious_trace);
     let init_result_5 = call_vm!(init, "", &script, init_result_3.data.clone(), setter_3_malicious_data);
-    assert_eq!(init_result_5.ret_code, 1021);
+    assert_eq!(init_result_5.ret_code, 1022);
 
     let actual_trace = trace_from_result(&init_result_5);
     let expected_trace = trace_from_result(&init_result_3);