test: expand complex workflow fuzz coverage

proptest-regressions/engine_merge_tests.txt

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc f78576800bd471035e4f5ddf75018f7f19417f718915c85213fdefb25ca72003 # shrinks to base = Null, replacement = Null, path = ["a"]

src/engine.rs

@@ -278,13 +278,68 @@ fn merge(base: &mut Value, update: Value) {
         *base = value.clone();
         return;
     }
-    match (base, update) {
-        (Value::Object(base), Value::Object(update)) => {
+    match update {
+        Value::Object(update) => {
+            // Recurse even when this subtree does not exist yet. Assigning the
+            // incoming object wholesale would leave any nested `$replace`
+            // sentinel as literal state on its first write.
+            if !base.is_object() {
+                *base = Value::Object(Map::new());
+            }
+            let base = base.as_object_mut().expect("object created above");
             for (key, value) in update {
                 merge(base.entry(key).or_insert(Value::Null), value);
             }
         }
-        (base, update) => *base = update,
+        update => *base = update,
+    }
+}
+
+/// Cancels background tasks that a cancelled run started.
+///
+/// A `spawn` hands task ownership to the host runner. Merely stopping graph
+/// scheduling does not stop that work, so without this pass a cancellation
+/// between `spawn` and `gate` leaks every issued ticket. The runner contract
+/// makes cancellation a no-op for already-settled work, so every issued ticket
+/// can be handed back without racing an extra poll against task completion.
+async fn cancel_spawned_tasks(
+    workflow: &CompiledWorkflow,
+    state: &Value,
+    capabilities: &Capabilities,
+) {
+    let Some(runner) = capabilities.tasks.as_ref() else {
+        return;
+    };
+    for spawn in workflow
+        .graph
+        .nodes
+        .iter()
+        .filter(|node| node.kind == NodeKind::Spawn)
+    {
+        let Some(slot) = state.get("nodes").and_then(|nodes| nodes.get(&spawn.id)) else {
+            continue;
+        };
+        let mut item_lists: Vec<&Vec<Value>> = slot
+            .get("items")
+            .and_then(Value::as_array)
+            .into_iter()
+            .collect();
+        item_lists.extend(
+            slot.get("lanes")
+                .and_then(Value::as_object)
+                .into_iter()
+                .flat_map(|lanes| lanes.values())
+                .filter_map(|lane| lane.get("items").and_then(Value::as_array)),
+        );
+        for ticket in item_lists.into_iter().flatten().filter_map(|item| {
+            item.get("json")
+                .and_then(|json| json.get("ticket"))
+                .and_then(Value::as_str)
+        }) {
+            if let Err(error) = runner.cancel(ticket).await {
+                tracing::warn!(%ticket, %error, "failed to cancel spawned task");
+            }
+        }
     }
 }
 
@@ -1205,11 +1260,10 @@ fn build_graph(
         // Successors on the emitted port, needed only inside a lane: `Plain`
         // routing normally rides static edges, but a lane has to re-schedule
         // every successor as a `Send`, so it needs the target list explicitly.
-        let plain_targets: Vec<String> = graph
-            .edges
+        let plain_targets_by_port = outgoing_by_port(graph, &node.id);
+        let plain_targets: Vec<String> = plain_targets_by_port
             .iter()
-            .filter(|e| e.from_node == node.id)
-            .map(|e| e.to_node.clone())
+            .flat_map(|(_, targets)| targets.iter().cloned())
             .collect();
         // Which successors end a lane. Routing to one of these is a plain
         // activation, so the lanes converge on it instead of each running their
@@ -1240,6 +1294,7 @@ fn build_graph(
             let token = token.clone();
             let routing = routing.clone();
             let plain_targets = plain_targets.clone();
+            let plain_targets_by_port = plain_targets_by_port.clone();
             let gather_nodes = gather_nodes.clone();
             // The resume value delivered to this node on a checkpointed resume, if
             // any. A bare `true` means "approve the interrupted gate"; a structured
@@ -1302,7 +1357,11 @@ fn build_graph(
                     if let Some(lane) = lane.as_ref() {
                         let emitted = port.unwrap_or("main");
                         let targets: Vec<String> = match &routing {
-                            HandlerRouting::Plain => plain_targets.clone(),
+                            HandlerRouting::Plain => plain_targets_by_port
+                                .iter()
+                                .find(|(port, _)| port == emitted)
+                                .map(|(_, targets)| targets.clone())
+                                .unwrap_or_default(),
                             HandlerRouting::FanOut(targets) => targets.clone(),
                             HandlerRouting::PortCommand(groups) => groups
                                 .iter()
@@ -1831,18 +1890,46 @@ fn build_graph(
                         };
                         match on_error {
                             // Turn the failure into data on the default port.
-                            "continue" => Ok(emit(
-                                items_update(&node.id, &[error_item(&node.id, &err)], None)?,
-                                None,
-                                &[error_item(&node.id, &err)],
-                            )),
+                            "continue" => {
+                                let item = error_item(&node.id, &err);
+                                let update = match lane.as_ref() {
+                                    Some(lane) => lane_items_update(
+                                        &node.id,
+                                        lane,
+                                        std::slice::from_ref(&item),
+                                        None,
+                                        "ok",
+                                        None,
+                                    )?,
+                                    None => items_update(
+                                        &node.id,
+                                        std::slice::from_ref(&item),
+                                        None,
+                                    )?,
+                                };
+                                Ok(emit(update, None, std::slice::from_ref(&item)))
+                            }
                             // Turn the failure into data on the `error` port so the
                             // graph can route it to a recovery sub-graph.
-                            "route" => Ok(emit(
-                                items_update(&node.id, &[error_item(&node.id, &err)], Some("error"))?,
-                                Some("error"),
-                                &[error_item(&node.id, &err)],
-                            )),
+                            "route" => {
+                                let item = error_item(&node.id, &err);
+                                let update = match lane.as_ref() {
+                                    Some(lane) => lane_items_update(
+                                        &node.id,
+                                        lane,
+                                        std::slice::from_ref(&item),
+                                        Some("error"),
+                                        "ok",
+                                        None,
+                                    )?,
+                                    None => items_update(
+                                        &node.id,
+                                        std::slice::from_ref(&item),
+                                        Some("error"),
+                                    )?,
+                                };
+                                Ok(emit(update, Some("error"), std::slice::from_ref(&item)))
+                            }
                             // "stop" (default) and any unknown policy fail the run.
                             //
                             // Stash the structured error before handing
@@ -1857,6 +1944,23 @@ fn build_graph(
                             // may produce several.
                             _ => {
                                 let message = err.to_string();
+                                // A lane failure belongs to its gather, whose
+                                // `on_lane_error` policy decides whether to
+                                // collect, skip, or fail the overall run. Record
+                                // it in the disjoint lane slot and carry the lane
+                                // to the gather instead of aborting here.
+                                if let Some(lane) = lane.as_ref() {
+                                    let meta = json!({ "error": message });
+                                    let update = lane_items_update(
+                                        &node.id,
+                                        lane,
+                                        &[],
+                                        None,
+                                        "failed",
+                                        Some(&meta),
+                                    )?;
+                                    return Ok(emit(update, None, &[]));
+                                }
                                 let mut slot =
                                     terminal_error.lock().expect("terminal error mutex poisoned");
                                 if slot.is_none() {
@@ -2270,6 +2374,10 @@ async fn build_and_run(
         .map(|interrupt| interrupt.id.clone())
         .collect();
 
+    if token.is_cancelled() {
+        cancel_spawned_tasks(workflow, &execution.state, capabilities).await;
+    }
+
     tracing::info!(
         steps = execution.steps,
         visited = execution.visited.len(),
@@ -2958,6 +3066,10 @@ mod merge_tests {
     }
 }
 
+#[cfg(test)]
+#[path = "engine_merge_tests.rs"]
+mod merge_property_tests;
+
 #[cfg(test)]
 mod lane_context_tests {
     use super::*;

src/engine_merge_tests.rs

@@ -0,0 +1,100 @@
+//! Property tests for the workflow-state reducer.
+
+use proptest::prelude::*;
+use serde_json::{Map, Value, json};
+
+use super::{merge, replace};
+
+fn leaf() -> impl Strategy<Value = Value> {
+    prop_oneof![
+        Just(Value::Null),
+        any::<bool>().prop_map(Value::Bool),
+        any::<i32>().prop_map(|value| json!(value)),
+        ".{0,16}".prop_map(Value::String),
+        prop::collection::vec(any::<i16>(), 0..8).prop_map(|values| json!(values)),
+    ]
+}
+
+fn arbitrary_json() -> impl Strategy<Value = Value> {
+    leaf().prop_recursive(4, 96, 8, |inner| {
+        prop_oneof![
+            prop::collection::vec(inner.clone(), 0..6).prop_map(Value::Array),
+            prop::collection::btree_map("[a-z]{1,5}", inner, 0..6)
+                .prop_map(|entries| Value::Object(entries.into_iter().collect())),
+        ]
+    })
+}
+
+/// Real engine updates have a stable object shape down to the leaf being
+/// assigned. Generating that shape avoids meaningless type-conflict examples
+/// (`scalar` followed by `object`) that the engine never emits.
+fn compatible_update() -> impl Strategy<Value = Value> {
+    (0usize..5, 0usize..5, leaf()).prop_map(|(node, field, value)| {
+        json!({ "nodes": { format!("n{node}"): { format!("k{field}"): value } } })
+    })
+}
+
+fn folded(initial: Value, updates: impl IntoIterator<Item = Value>) -> Value {
+    let mut state = initial;
+    for update in updates {
+        merge(&mut state, update);
+    }
+    state
+}
+
+fn nested_update(path: &[String], value: Value) -> Value {
+    path.iter()
+        .rev()
+        .fold(replace(value), |child, key| json!({ key: child }))
+}
+
+fn assign_at_path(root: &mut Value, path: &[String], value: Value) {
+    if path.is_empty() {
+        *root = value;
+        return;
+    }
+    if !root.is_object() {
+        *root = Value::Object(Map::new());
+    }
+    let child = root
+        .as_object_mut()
+        .expect("object created above")
+        .entry(path[0].clone())
+        .or_insert(Value::Null);
+    assign_at_path(child, &path[1..], value);
+}
+
+proptest! {
+    #![proptest_config(ProptestConfig { cases: 256, ..ProptestConfig::default() })]
+
+    #[test]
+    fn compatible_update_sequences_are_associative(
+        run_state in arbitrary_json(),
+        updates in prop::collection::vec(compatible_update(), 0..40),
+        split in 0usize..40,
+    ) {
+        let initial = json!({ "run": run_state, "nodes": {} });
+        let split = split.min(updates.len());
+        let sequential = folded(initial.clone(), updates.clone());
+
+        let left = folded(Value::Object(Map::new()), updates[..split].iter().cloned());
+        let right = folded(Value::Object(Map::new()), updates[split..].iter().cloned());
+        let grouped = folded(initial, [left, right]);
+
+        prop_assert_eq!(sequential, grouped);
+    }
+
+    #[test]
+    fn replace_round_trips_at_arbitrary_nesting(
+        base in arbitrary_json(),
+        replacement in arbitrary_json(),
+        path in prop::collection::vec("[a-z]{1,5}", 0..7),
+    ) {
+        let mut actual = base.clone();
+        merge(&mut actual, nested_update(&path, replacement.clone()));
+
+        let mut expected = base;
+        assign_at_path(&mut expected, &path, replacement);
+        prop_assert_eq!(actual, expected);
+    }
+}