feat: Add pipe parameter validation and fix VM call frame

Adds a check within the `PipeHook` to ensure the number of input streams
provided to a `pipe` operation matches the number of parameters expected
by the associated lambda function. This prevents runtime errors and
provides a clearer compile-time diagnostic.

Additionally, this commit corrects the order in which a `CallFrame` is
pushed onto the VM's frame stack. By pushing the frame earlier, it
ensures that slow-path `unpack` operations have access to a valid frame,
preventing "No call frame" errors.
This commit is contained in:
2026-03-30 09:43:58 +02:00
parent e82d48d1cb
commit 9b13546609
4 changed files with 85 additions and 10 deletions
+19 -4
View File
@@ -1,8 +1,23 @@
;; Skip: data output to stdout ;; Skip: data output to stdout
(do (do
(def ohlc (create-m1-stream "EURUSD")) (def EURUSD (create-m1-stream "EURUSD" (date "2020-01-03 09:30:00") (date "2020-01-03 17:30:00")))
(def GER40 (create-m1-stream "GER40" (date "2020-01-03 09:30:00") (date "2020-01-03 17:30:00")))
;; Sink: print returns void, so nothing is propagated downstream ;; Sink: print returns void, so nothing is propagated downstream
(pipe ohlc
(fn [bar] (def cnt 1)
(print (.close bar))))) (def gs (series 2))
;IDEE (pipe lookback [EURUSD acc] --> series acc
(pipe [EURUSD GER40]
(fn [eu ge]
(do
(def dax (.close ge))
(push gs dax)
(def eur (- dax (gs 1)))
(print cnt ": dax=" eur " (" (/ eur (.close eu)) "$)" )
(assign cnt (+ cnt 1))
)
)
)
)
+39 -1
View File
@@ -1,4 +1,5 @@
use crate::ast::compiler::call_hooks::RtlCompilerHook; use crate::ast::compiler::call_hooks::{InferenceAccess, RtlCompilerHook};
use crate::ast::diagnostics::Diagnostics;
use crate::ast::nodes::{Node, NodeKind, TypedNode, TypedPhase}; use crate::ast::nodes::{Node, NodeKind, TypedNode, TypedPhase};
use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember}; use crate::ast::rtl::series::{RingBuffer, ScalarValue, SeriesMember};
use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SourceLocation, StreamStorage, Value}; use crate::ast::types::{NativeFunction, NodeIdentity, PipeFn, SourceLocation, StreamStorage, Value};
@@ -403,6 +404,43 @@ pub struct PipeHook {
} }
impl RtlCompilerHook for PipeHook { impl RtlCompilerHook for PipeHook {
fn post_call(
&self,
args: &TypedNode,
ret_ty: StaticType,
_ctx: &dyn InferenceAccess,
diag: &mut Diagnostics,
) -> StaticType {
// Validate: input stream count must match lambda parameter count.
if let NodeKind::Tuple { elements } = &args.kind
&& elements.len() == 2
{
let expected = match &elements[0].ty {
StaticType::Stream(_) | StaticType::Series(_) => 1,
StaticType::Vector(_, count) => *count,
StaticType::Tuple(elems) => elems.len(),
_ => return ret_ty,
};
if let NodeKind::Lambda { params, .. } = &elements[1].kind {
let actual = match &params.kind {
NodeKind::Tuple { elements } => elements.len(),
_ => 1,
};
if actual != expected {
diag.push_error(
format!(
"pipe: lambda expects {} parameter(s) but {} input stream(s) provided",
actual, expected
),
Some(elements[1].identity.clone()),
);
}
}
}
ret_ty
}
fn finalize( fn finalize(
&self, &self,
_callee: Rc<TypedNode>, _callee: Rc<TypedNode>,
+7 -5
View File
@@ -239,6 +239,13 @@ impl VM {
self.stack.clear(); self.stack.clear();
self.frames.clear(); self.frames.clear();
// Push the call frame early so that `unpack` (slow path) can use
// `set_value` / `get_value`, which require a valid call frame.
self.frames.push(CallFrame {
stack_base: 0,
closure: Some(closure_rc.clone()),
});
let closure = closure_rc.as_ref(); let closure = closure_rc.as_ref();
if let Some(count) = closure.positional_count if let Some(count) = closure.positional_count
&& args.len() == count as usize && args.len() == count as usize
@@ -254,11 +261,6 @@ impl VM {
self.stack.resize(closure.stack_size as usize, Value::Void); self.stack.resize(closure.stack_size as usize, Value::Void);
} }
self.frames.push(CallFrame {
stack_base: 0,
closure: Some(closure_rc.clone()),
});
let result = self.eval_internal(observer, &closure.exec_node); let result = self.eval_internal(observer, &closure.exec_node);
self.frames.pop(); self.frames.pop();
self.resolve_tail_calls(observer, result) self.resolve_tail_calls(observer, result)
+20
View File
@@ -93,6 +93,26 @@ fn test_pipe_chain_preserves_concrete_types() {
assert!(env.run_script(source).is_ok(), "Chained pipe failed at runtime"); assert!(env.run_script(source).is_ok(), "Chained pipe failed at runtime");
} }
/// Pipe with 2 input streams but lambda expecting 1 parameter must be a compile error.
/// Regression: previously crashed at runtime with "No call frame" instead of a
/// clean compile-time diagnostic.
#[test]
fn test_pipe_param_count_mismatch_is_compile_error() {
let env = Environment::new();
let source = "(do
(def src (create-random-ohlc 42 5))
(pipe [src src] (fn [bar] (.close bar)))
)";
let res = env.run_script(source);
assert!(res.is_err(), "Pipe with 2 inputs but 1-param lambda must be a compile error");
let err = res.unwrap_err();
assert!(
err.contains("pipe") && err.contains("parameter"),
"Error should mention parameter mismatch, got: {err}"
);
}
/// Verifies that pipe with Optional return (filter pattern) still produces Stream(Float). /// Verifies that pipe with Optional return (filter pattern) still produces Stream(Float).
#[test] #[test]
fn test_pipeline_optional_type() { fn test_pipeline_optional_type() {