fix unstable split

Signed-off-by: Teo Koon Peng <[email protected]>

Author: koonpeng

src/diagram.rs

@@ -262,7 +262,7 @@ pub enum DiagramOperation {
     ///         },
     ///         "join": {
     ///             "type": "join",
-    ///             "order": ["op1", "op2"],
+    ///             "inputs": ["op1", "op2"],
     ///             "next": { "builtin": "terminate" }
     ///         }
     ///     }

src/diagram/join.rs

@@ -18,7 +18,7 @@ pub struct JoinOp {
 
     /// Controls the order of the resulting join. Each item must be an operation id of one of the
     /// incoming outputs.
-    pub(super) order: Vec<SourceOperation>,
+    pub(super) inputs: Vec<SourceOperation>,
 
     /// Do not serialize before performing the join. If true, joins can only be done
     /// on outputs of the same type.
@@ -176,7 +176,7 @@ mod tests {
                 },
                 "join": {
                     "type": "join",
-                    "order": ["op1", "op2"],
+                    "inputs": ["op1", "op2"],
                     "next": "serialize_join_output",
                     "no_serialize": true,
                 },
@@ -197,6 +197,14 @@ mod tests {
         assert_eq!(result[1], 6);
     }
 
+    /// This test is to ensure that the order of split and join operations are stable.
+    #[test]
+    fn test_join_stress() {
+        for _ in 1..20 {
+            test_join();
+        }
+    }
+
     #[test]
     fn test_empty_join() {
         let mut fixture = DiagramTestFixture::new();
@@ -241,7 +249,7 @@ mod tests {
                 },
                 "join": {
                     "type": "join",
-                    "order": [],
+                    "inputs": [],
                     "next": { "builtin": "terminate" },
                     "no_serialize": true,
                 },
@@ -297,7 +305,7 @@ mod tests {
                 },
                 "join": {
                     "type": "join",
-                    "order": ["op1", "op2"],
+                    "inputs": ["op1", "op2"],
                     "next": { "builtin": "terminate" },
                 },
             }

src/diagram/workflow_builder.rs

@@ -276,7 +276,7 @@ fn connect_vertex<'a>(
                 .collect();
 
             let mut ordered_outputs: Vec<DynOutput> = Vec::with_capacity(target.in_edges.len());
-            for source_id in join_op.order.iter() {
+            for source_id in join_op.inputs.iter() {
                 let o = outputs
                     .remove(source_id)
                     .ok_or(DiagramError::OperationNotFound(source_id.to_string()))?;
@@ -415,7 +415,7 @@ fn connect_edge<'a>(
             }
         }
         DiagramOperation::Split(split_op) => {
-            let outputs = if output.type_id == TypeId::of::<serde_json::Value>() {
+            let mut outputs = if output.type_id == TypeId::of::<serde_json::Value>() {
                 let chain = output.into_output::<serde_json::Value>()?.chain(builder);
                 split_chain(chain, split_op)
             } else {
@@ -428,14 +428,11 @@ fn connect_edge<'a>(
                 let reg = registry.get_registration(&origin.builder)?;
                 reg.split(builder, output, split_op)
             }?;
-            outputs
-                .outputs
-                .into_iter()
-                .zip(target.out_edges.iter())
-                .for_each(|((_, o), e)| {
-                    let out_edge = edges.get_mut(e).unwrap();
-                    out_edge.state = EdgeState::Ready { output: o, origin };
-                });
+            for e in &target.out_edges {
+                let out_edge = edges.get_mut(e).unwrap();
+                let output = outputs.outputs.remove(out_edge.target).unwrap();
+                out_edge.state = EdgeState::Ready { output, origin };
+            }
             if let Some(_) = &split_op.remaining {
                 let out_edge = edges.get_mut(target.out_edges.last().unwrap()).unwrap();
                 out_edge.state = EdgeState::Ready {