chore(execution-engine): Some stream-related LambdaError are unjoinab…

…le b/c: canon stream replaces normal stream, when canon stream is used, it is materialized and its size is known

air/src/execution_step/errors/catchable_errors.rs

@@ -141,15 +141,6 @@ impl Joinable for CatchableError {
                 log_join!("  waiting for an argument with name '{}'", var_name);
                 true
             }
-            LambdaApplierError(LambdaError::StreamNotHaveEnoughValues { stream_size, idx }) => {
-                log_join!("  waiting for an argument with idx '{}' on stream with size '{}'", idx, stream_size);
-                true
-            }
-            LambdaApplierError(LambdaError::EmptyStream) => {
-                log_join!("  waiting on empty stream for path ");
-                true
-            }
-
             _ => false,
         }
     }

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

@@ -86,11 +86,12 @@ fn wait_on_empty_stream_json_path() {
         )
      )"#);
 
-    let result = checked_call_vm!(local_vm, <_>::default(), join_stream_script, "", "");
-    print_trace(&result, "");
+    let result = local_vm.call(&join_stream_script, "", "", <_>::default()).unwrap();
     let actual_trace = trace_from_result(&result);
-
     assert_eq!(actual_trace.len(), 2); // only the first call and canon should produce a trace
+    let expected_error =
+        CatchableError::LambdaApplierError(LambdaError::StreamNotHaveEnoughValues { stream_size: 0, idx: 0 });
+    assert!(check_error(&result, expected_error));
 }
 
 #[test]

air/tests/test_module/features/lambda/flattening.rs

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+use air::CatchableError;
 use air_test_utils::prelude::*;
 
 use std::cell::RefCell;
@@ -155,11 +156,14 @@ fn flattening_empty_values() {
         )
         "#);
 
-    let result = checked_call_vm!(set_variable_vm, <_>::default(), script.clone(), "", "");
-    let result = checked_call_vm!(local_vm, <_>::default(), script, "", result.data);
+    let result = set_variable_vm.call(script.clone(), "", "", <_>::default()).unwrap();
+    let result = local_vm.call(script.clone(), "", result.data, <_>::default()).unwrap();
 
-    assert!(is_interpreter_succeded(&result));
+    assert!(!is_interpreter_succeded(&result));
     assert_eq!(closure_call_args.args_var, Rc::new(RefCell::new(vec![])));
+    let expected_error =
+        CatchableError::LambdaApplierError(air::LambdaError::StreamNotHaveEnoughValues { stream_size: 1, idx: 1 });
+    assert!(check_error(&result, expected_error));
 }
 
 #[test]