Merge pull request #216 from code0-tech/#207-merge-execution-fix

fix: handled duration for nodes

Author: raphael-goetz

crates/taurus-core/src/runtime/engine.rs

@@ -265,8 +265,12 @@ impl ExecutionEngine {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::handler::argument::Argument;
+    use crate::handler::registry::{FunctionRegistration, FunctionStore, ThunkRunner};
+    use crate::runtime::execution::value_store::ValueStore;
     use crate::types::exit_reason::ExitReason;
     use std::cell::RefCell;
+    use std::time::Duration;
     use tucana::shared::{
         InputType, ListValue, NodeParameter, NodeValue, ReferenceValue, Struct, SubFlow,
         SubFlowSetting, Value, node_execution_result, node_value, reference_value,
@@ -405,6 +409,15 @@ mod tests {
         }
     }
 
+    fn sleep_handler(
+        _args: &[Argument],
+        _ctx: &mut ValueStore,
+        _run: &mut ThunkRunner<'_>,
+    ) -> Signal {
+        std::thread::sleep(Duration::from_micros(2_000));
+        Signal::Success(null_value())
+    }
+
     fn input_type_ref_param(
         database_id: i64,
         runtime_parameter_id: &str,
@@ -925,6 +938,103 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn node_execution_result_tracks_actual_node_duration() {
+        let mut handlers = FunctionStore::new();
+        handlers.populate(&[FunctionRegistration::eager("test::sleep", sleep_handler, 0)]);
+        let engine = ExecutionEngine { handlers };
+        let sleep_node = node(1, "test::sleep", vec![], None);
+
+        let report = engine.execute_graph_report(1, vec![sleep_node], None, None, None, false);
+
+        assert_eq!(report.exit_reason, ExitReason::Success);
+        assert_eq!(report.node_execution_results.len(), 1);
+
+        let node_result = &report.node_execution_results[0];
+        assert_eq!(node_result.node_id, 1);
+        assert!(node_result.started_at >= 1_000_000_000_000_000);
+        assert!(node_result.finished_at > node_result.started_at);
+        assert!(node_result.finished_at - node_result.started_at >= 1_000);
+    }
+
+    #[test]
+    fn execution_report_keeps_every_for_each_callback_execution() {
+        let engine = ExecutionEngine::new();
+        let for_each_node = node(
+            1,
+            "std::list::for_each",
+            vec![
+                literal_param(
+                    100,
+                    "list",
+                    list_value(vec![int_value(1), int_value(2), int_value(3)]),
+                ),
+                thunk_param(101, "consumer", 2),
+            ],
+            None,
+        );
+        let callback_node = node(
+            2,
+            "std::number::add",
+            vec![
+                input_type_ref_param(200, "first", 1, 1, 0),
+                literal_param(201, "second", int_value(2)),
+            ],
+            None,
+        );
+
+        let report = engine.execute_graph_report(
+            1,
+            vec![for_each_node, callback_node],
+            None,
+            None,
+            None,
+            false,
+        );
+
+        assert_eq!(report.exit_reason, ExitReason::Success);
+        assert_eq!(report.node_execution_results.len(), 4);
+
+        let callback_results: Vec<_> = report
+            .node_execution_results
+            .iter()
+            .filter(|result| result.node_id == 2)
+            .collect();
+        assert_eq!(callback_results.len(), 3);
+
+        let callback_values: Vec<_> = callback_results
+            .iter()
+            .map(|result| match result.result.as_ref() {
+                Some(node_execution_result::Result::Success(value)) => value.clone(),
+                other => panic!("expected callback success result, got {:?}", other),
+            })
+            .collect();
+
+        assert_eq!(
+            callback_values,
+            vec![int_value(3), int_value(4), int_value(5)]
+        );
+        let callback_parameters: Vec<_> = callback_results
+            .iter()
+            .map(|result| {
+                result
+                    .parameter_results
+                    .iter()
+                    .map(|parameter| parameter.value.clone())
+                    .collect::<Vec<_>>()
+            })
+            .collect();
+        assert_eq!(
+            callback_parameters,
+            vec![
+                vec![Some(int_value(1)), Some(int_value(2))],
+                vec![Some(int_value(2)), Some(int_value(2))],
+                vec![Some(int_value(3)), Some(int_value(2))],
+            ]
+        );
+        assert_eq!(report.node_execution_results[3].node_id, 1);
+    }
+
     #[test]
     fn emitter_emits_start_and_finish_for_successful_execution() {
         let engine = ExecutionEngine::new();

crates/taurus-core/src/runtime/engine/executor.rs

@@ -26,6 +26,7 @@ use crate::runtime::execution::trace::{
 use crate::runtime::execution::tracer::{ExecutionTracer, Tracer};
 use crate::runtime::execution::value_store::{ValueStore, ValueStoreResult};
 use crate::runtime::remote::{RemoteExecution, RemoteRuntime};
+use crate::time::now_unix_micros;
 use crate::types::errors::runtime_error::RuntimeError;
 use crate::types::signal::Signal;
 
@@ -66,6 +67,8 @@ struct NodeResult {
     signal: Signal,
     frame_id: Option<u64>,
     parameter_results: Vec<NodeParameterNodeExecutionResult>,
+    started_at: i64,
+    finished_at: i64,
 }
 
 struct ExecutedNode {
@@ -125,10 +128,12 @@ impl<'a> EngineExecutor<'a> {
                         emitter.emit(self.execution_id, EmitType::OngoingExec, value.clone());
                     }
 
-                    value_store.insert_success_with_parameters(
+                    value_store.insert_success_with_timing(
                         node_id,
                         value.clone(),
                         result.parameter_results,
+                        result.started_at,
+                        result.finished_at,
                     );
                     match next_idx {
                         Some(next) => current_idx = next,
@@ -243,25 +248,38 @@ impl<'a> EngineExecutor<'a> {
         let frame_id = self.trace_enter(node, value_store);
         let result = match &node.execution_target {
             NodeExecutionTarget::Local => {
+                let started_at = now_unix_micros();
                 let executed = self.execute_local_node(node, value_store, frame_id);
+                let finished_at = now_unix_micros();
                 let parameter_results = executed.parameter_results;
                 let signal = self.commit_result(
                     node.id,
                     executed.signal,
                     parameter_results.clone(),
+                    started_at,
+                    finished_at,
                     value_store,
                 );
                 NodeResult {
                     signal,
                     frame_id,
                     parameter_results,
+                    started_at,
+                    finished_at,
+                }
+            }
+            NodeExecutionTarget::Remote { service } => {
+                let started_at = now_unix_micros();
+                let signal = self.execute_remote_node(node, service, value_store, frame_id);
+                let finished_at = now_unix_micros();
+                NodeResult {
+                    signal,
+                    frame_id,
+                    parameter_results: Vec::new(),
+                    started_at,
+                    finished_at,
                 }
             }
-            NodeExecutionTarget::Remote { service } => NodeResult {
-                signal: self.execute_remote_node(node, service, value_store, frame_id),
-                frame_id,
-                parameter_results: Vec::new(),
-            },
         };
         self.trace_exit(frame_id, &result.signal, value_store);
 
@@ -331,6 +349,7 @@ impl<'a> EngineExecutor<'a> {
         value_store: &mut ValueStore,
         frame_id: Option<u64>,
     ) -> Signal {
+        let started_at = now_unix_micros();
         let remote_runtime = match self.remote {
             Some(remote) => remote,
             None => {
@@ -342,6 +361,8 @@ impl<'a> EngineExecutor<'a> {
                         "Remote runtime not configured",
                     )),
                     Vec::new(),
+                    started_at,
+                    now_unix_micros(),
                     value_store,
                 );
             }
@@ -350,7 +371,14 @@ impl<'a> EngineExecutor<'a> {
         let mut args = match self.build_args(node, value_store, frame_id) {
             Ok(args) => args,
             Err(err) => {
-                return self.commit_result(node.id, Signal::Failure(err), Vec::new(), value_store);
+                return self.commit_result(
+                    node.id,
+                    Signal::Failure(err),
+                    Vec::new(),
+                    started_at,
+                    now_unix_micros(),
+                    value_store,
+                );
             }
         };
 
@@ -361,6 +389,8 @@ impl<'a> EngineExecutor<'a> {
                     node.id,
                     signal,
                     parameter_results_from_args(&args),
+                    started_at,
+                    now_unix_micros(),
                     value_store,
                 );
             }
@@ -374,6 +404,8 @@ impl<'a> EngineExecutor<'a> {
                     node.id,
                     Signal::Failure(err),
                     parameter_results,
+                    started_at,
+                    now_unix_micros(),
                     value_store,
                 );
             }
@@ -390,6 +422,8 @@ impl<'a> EngineExecutor<'a> {
                 node.id,
                 Signal::Failure(err),
                 parameter_results,
+                started_at,
+                now_unix_micros(),
                 value_store,
             ),
         }
@@ -678,19 +712,29 @@ impl<'a> EngineExecutor<'a> {
         node_id: i64,
         signal: Signal,
         parameter_results: Vec<NodeParameterNodeExecutionResult>,
+        started_at: i64,
+        finished_at: i64,
         value_store: &mut ValueStore,
     ) -> Signal {
         match signal {
             Signal::Success(value) => {
-                value_store.insert_success_with_parameters(
+                value_store.insert_success_with_timing(
                     node_id,
                     value.clone(),
                     parameter_results,
+                    started_at,
+                    finished_at,
                 );
                 Signal::Success(value)
             }
             Signal::Failure(err) => {
-                value_store.insert_error_with_parameters(node_id, err.clone(), parameter_results);
+                value_store.insert_error_with_timing(
+                    node_id,
+                    err.clone(),
+                    parameter_results,
+                    started_at,
+                    finished_at,
+                );
                 Signal::Failure(err)
             }
             // Control signals are transient and should not be cached as node outputs.