fix(engine): a gather in a loop counts only the current pass's lanes

CHANGELOG.md

@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Gather in a looping scatter
+
+- **Fixed** a `gather` inside a loop body collecting an earlier pass's lanes: a
+  narrower pass (3 lanes, then 2) released the stale third lane too. Lane slots
+  now record the scatter and the super-step of the activation that opened them
+  (`_scatter`, `_scatter_step`), and a gather counts only the current
+  activation's lanes against its own scatter's `lane_count`. Slots from a
+  checkpoint written before this change carry no stamp and are counted as before
+  when no stamped slot exists, so an in-flight run still completes.
+- **Added** `LaneContext::scatter_step` (`Option<usize>`); code that builds a
+  `LaneContext` literal must set it.
+- **Fixed** the validation reason for a gather with no upstream `scatter`: it
+  releases with no items, silently dropping its predecessors' output, rather
+  than waiting out its poll budget.
+
 ### Run event outcomes
 
 - **Added** the engine output to `Runs.PollRun` parked and finished event

crates/tinyflows/src/engine.rs

@@ -267,6 +267,7 @@ mod state;
 pub(crate) use state::replace;
 #[cfg(test)]
 pub(crate) use state::{LANE_KEY, REPLACE};
+pub(crate) use state::{LANE_SCATTER_KEY, LANE_SCATTER_STEP_KEY};
 use state::{
     MergeReducer, collect_input, collect_input_since, lane_context, lane_envelope, lane_input,
     lane_items_update, merge, stamp_activation_step,

crates/tinyflows/src/engine/build/activation/routing.rs

@@ -37,9 +37,14 @@
                 if gather_nodes.contains(&target) {
                     RouteTarget::Node(target.into())
                 } else {
-                    let envelope =
-                        lane_envelope(&lane.origin, lane.index, lane.count, routed_items)
-                            .unwrap_or(Value::Null);
+                    let envelope = lane_envelope(
+                        &lane.origin,
+                        lane.index,
+                        lane.count,
+                        lane.scatter_step,
+                        routed_items,
+                    )
+                    .unwrap_or(Value::Null);
                     RouteTarget::Send(crate::graph::Send::new(target, envelope))
                 }
             })

crates/tinyflows/src/engine/build/outcome.rs

@@ -118,8 +118,14 @@ where
                     let mut routed: Vec<RouteTarget> = Vec::new();
                     for (index, lane_items) in lanes.iter().enumerate() {
                         for target in plain_targets {
-                            let envelope = lane_envelope(&node.id, index, count, lane_items)
-                                .unwrap_or(Value::Null);
+                            let envelope = lane_envelope(
+                                &node.id,
+                                index,
+                                count,
+                                Some(activation_step),
+                                lane_items,
+                            )
+                            .unwrap_or(Value::Null);
                             routed.push(RouteTarget::Send(crate::graph::Send::new(
                                 target.clone(),
                                 envelope,

crates/tinyflows/src/engine/state.rs

@@ -96,19 +96,30 @@ pub(super) fn lane_items_update(
         "status": status,
         "index": lane.index,
     });
-    if let (Some(Value::Object(extra)), Some(map)) = (meta, slot.as_object_mut()) {
-        for (key, value) in extra {
-            map.insert(key.clone(), value.clone());
+    if let Some(map) = slot.as_object_mut() {
+        if let Some(Value::Object(extra)) = meta {
+            for (key, value) in extra {
+                map.insert(key.clone(), value.clone());
+            }
+        }
+        // After `meta`, so a node's own meta cannot claim another activation.
+        map.insert(LANE_SCATTER_KEY.to_string(), json!(lane.origin));
+        if let Some(step) = lane.scatter_step {
+            map.insert(LANE_SCATTER_STEP_KEY.to_string(), json!(step));
         }
     }
     Ok(json!({ "nodes": { node_id: { "lanes": { lane.id.clone(): slot } } } }))
 }
 
 /// The lane envelope a fan-out schedules one activation with.
+///
+/// `scatter_step` is the super-step of the scatter activation that opened the
+/// lane; every node along the lane passes it on unchanged.
 pub(super) fn lane_envelope(
     origin: &str,
     index: usize,
     count: usize,
+    scatter_step: Option<usize>,
     items: &[Item],
 ) -> crate::graph::Result<Value> {
     Ok(json!({
@@ -117,6 +128,7 @@ pub(super) fn lane_envelope(
             "origin": origin,
             "index": index,
             "count": count,
+            "scatter_step": scatter_step,
         },
         "items": serde_json::to_value(items)?,
     }))
@@ -147,6 +159,15 @@ pub(super) fn lane_input(send_arg: Option<&Value>) -> Vec<Item> {
 /// and must never be mistaken for user data.
 pub(crate) const LANE_KEY: &str = "_lane";
 
+/// The lane-slot key naming the scatter that opened the lane.
+pub(crate) const LANE_SCATTER_KEY: &str = "_scatter";
+
+/// The lane-slot key recording the super-step of the scatter activation that
+/// opened the lane. Lane ids repeat across loop passes (`<scatter>#<index>`)
+/// and slots are never cleared, so this is what lets a gather count only the
+/// current pass. Absent on slots written from a pre-stamp envelope.
+pub(crate) const LANE_SCATTER_STEP_KEY: &str = "_scatter_step";
+
 /// Decodes the lane identity from an activation's `send_arg`.
 ///
 /// `None` for an ordinary activation — one scheduled by a plain route rather
@@ -160,6 +181,12 @@ pub(super) fn lane_context(send_arg: Option<&Value>) -> Option<crate::nodes::Lan
         origin: lane.get("origin")?.as_str()?.to_string(),
         index: usize::try_from(lane.get("index")?.as_u64()?).ok()?,
         count: usize::try_from(lane.get("count")?.as_u64()?).ok()?,
+        // Optional: an envelope from an older binary has none, and must still
+        // decode as a lane.
+        scatter_step: lane
+            .get("scatter_step")
+            .and_then(Value::as_u64)
+            .and_then(|n| usize::try_from(n).ok()),
     })
 }
 

crates/tinyflows/src/lane_context_tests.rs

@@ -14,6 +14,26 @@ fn a_well_formed_envelope_decodes() {
     assert_eq!(lane.origin, "fan");
     assert_eq!(lane.index, 2);
     assert_eq!(lane.count, 5);
+    assert_eq!(
+        lane.scatter_step, None,
+        "an envelope from before the stamp still decodes, unstamped"
+    );
+}
+
+/// The scatter's activation step survives the envelope and lands on the lane
+/// slot, where a node's own meta cannot overwrite it.
+#[test]
+fn the_scatter_step_travels_from_envelope_to_lane_slot() {
+    let packet = lane_envelope("fan", 1, 2, Some(7), &[]).expect("envelope");
+    let lane = lane_context(Some(&packet)).expect("decodes");
+    assert_eq!(lane.scatter_step, Some(7));
+
+    let forged = json!({ "_scatter_step": 1, "_scatter": "other" });
+    let meta = Some(&forged);
+    let update = lane_items_update("work", &lane, &[], None, "ok", meta).expect("update");
+    let slot = &update["nodes"]["work"]["lanes"]["fan#1"];
+    assert_eq!(slot["_scatter_step"], 7);
+    assert_eq!(slot["_scatter"], "fan");
 }
 
 /// The ordinary case, and the only one that occurs until a fan-out node

crates/tinyflows/src/nodes/control_flow/gather.rs

@@ -6,7 +6,9 @@
 //! graph. A gather cannot: how many lanes exist is decided at run time by the
 //! scatter, from data. So its barrier is **data-driven**: it counts arrivals in
 //! `nodes.<lane terminal>.lanes.*` against the `lane_count` the scatter
-//! recorded, and asks to be re-run until its release policy is satisfied.
+//! recorded, and asks to be re-run until its release policy is satisfied. Only
+//! lanes opened by the scatter's current activation count, so a loop pass never
+//! gathers an earlier pass's lanes.
 //!
 //! That is also why it supports the same policies as a [`gate`]: once the wait
 //! is a decision rather than a topological fact, "proceed on a quorum" and
@@ -19,6 +21,7 @@ use async_trait::async_trait;
 use serde_json::{Value, json};
 
 use crate::data::Item;
+use crate::engine::{LANE_SCATTER_KEY, LANE_SCATTER_STEP_KEY};
 use crate::error::{EngineError, Result};
 use crate::nodes::release::{Release, ReleasePolicy};
 use crate::nodes::{NodeContext, NodeExecutor, NodeOutput};
@@ -64,23 +67,61 @@ struct Arrived {
     failed: Option<String>,
 }
 
-/// Reads every lane slot recorded by this gather's lane-terminal predecessors.
+/// Every lane slot recorded by this gather's lane-terminal predecessors.
 ///
 /// Lanes are keyed by id under `nodes.<pred>.lanes`, which is what keeps N
 /// concurrent activations of one node from clobbering each other — the reducer
 /// merges objects key-by-key, so distinct lane keys never collide.
-fn arrivals(ctx: &NodeContext<'_>, predecessors: &[String]) -> Vec<Arrived> {
-    let mut arrived = Vec::new();
-    for pred in predecessors {
-        let Some(lanes) = ctx
-            .nodes
-            .get(pred)
-            .and_then(|slot| slot.get("lanes"))
-            .and_then(Value::as_object)
-        else {
-            continue;
-        };
-        for slot in lanes.values() {
+fn lane_slots<'a>(nodes: &'a Value, predecessors: &[String]) -> Vec<&'a Value> {
+    predecessors
+        .iter()
+        .filter_map(|pred| nodes.get(pred)?.get("lanes")?.as_object())
+        .flat_map(|lanes| lanes.values())
+        .collect()
+}
+
+/// The scatter that opened these lanes, as the lane slots record it.
+fn scatter_of<'a>(slots: &[&'a Value]) -> Option<&'a str> {
+    slots
+        .iter()
+        .copied()
+        .find_map(|slot| slot.get(LANE_SCATTER_KEY)?.as_str())
+}
+
+/// Keeps only the slots opened by the scatter's **current** activation.
+///
+/// Slots are never cleared, so in a loop whose body scatters, an earlier pass's
+/// lanes are still in state when a later pass gathers; counting them would
+/// release a narrower pass with a stale lane in it. The current activation is
+/// the super-step the scatter's own slot was last stamped with.
+///
+/// A slot without the stamp was written from an envelope that predates it (a
+/// run checkpointed by an older binary). Such slots are all counted, as before,
+/// when no stamped slot exists, so a run resumed across an upgrade still
+/// completes; once any stamped slot exists, unstamped ones are stale and ignored.
+fn current_activation<'a>(
+    nodes: &Value,
+    slots: Vec<&'a Value>,
+    scatter: Option<&str>,
+) -> Vec<&'a Value> {
+    let step_of = |slot: &Value| slot.get(LANE_SCATTER_STEP_KEY).and_then(Value::as_u64);
+    if !slots.iter().any(|slot| step_of(slot).is_some()) {
+        return slots;
+    }
+    let current = scatter
+        .and_then(|id| nodes.get(id)?.get("_activation_step")?.as_u64())
+        .or_else(|| slots.iter().filter_map(|slot| step_of(slot)).max());
+    slots
+        .into_iter()
+        .filter(|slot| step_of(slot) == current)
+        .collect()
+}
+
+/// Decodes the lane slots that count.
+fn arrivals(slots: Vec<&Value>) -> Vec<Arrived> {
+    slots
+        .into_iter()
+        .map(|slot| {
             let index = slot
                 .get("index")
                 .and_then(Value::as_u64)
@@ -106,27 +147,31 @@ fn arrivals(ctx: &NodeContext<'_>, predecessors: &[String]) -> Vec<Arrived> {
                         .unwrap_or("lane failed")
                         .to_string()
                 });
-            arrived.push(Arrived {
+            Arrived {
                 index,
                 items,
                 failed,
-            });
-        }
-    }
-    arrived
+            }
+        })
+        .collect()
 }
 
 /// How many lanes the scatter upstream of this gather opened.
 ///
 /// Read from the scatter's own slot rather than inferred from arrivals: a
 /// gather that guessed "however many turned up" would release immediately on
-/// the first one, which is the bug this exists to prevent.
-fn expected_lanes(ctx: &NodeContext<'_>) -> Option<usize> {
-    ctx.nodes.as_object()?.values().find_map(|slot| {
+/// the first one, which is the bug this exists to prevent. Slots that do not
+/// name their scatter (pre-stamp) fall back to the first `lane_count` in state.
+fn expected_lanes(nodes: &Value, scatter: Option<&str>) -> Option<usize> {
+    let lane_count = |slot: &Value| {
         slot.get("lane_count")
             .and_then(Value::as_u64)
             .and_then(|n| usize::try_from(n).ok())
-    })
+    };
+    match scatter {
+        Some(id) => nodes.get(id).and_then(lane_count),
+        None => nodes.as_object()?.values().find_map(lane_count),
+    }
 }
 
 fn positive_u64(config: &Value, key: &str, default: u64) -> u64 {
@@ -157,8 +202,10 @@ impl NodeExecutor for GatherNode {
             })
             .unwrap_or_default();
 
-        let arrived = arrivals(&ctx, &predecessors);
-        let expected = expected_lanes(&ctx).unwrap_or(arrived.len());
+        let slots = lane_slots(ctx.nodes, &predecessors);
+        let scatter = scatter_of(&slots);
+        let arrived = arrivals(current_activation(ctx.nodes, slots, scatter));
+        let expected = expected_lanes(ctx.nodes, scatter).unwrap_or(arrived.len());
 
         let polls = ctx
             .nodes

crates/tinyflows/src/nodes/control_flow/gather_tests.rs

@@ -65,3 +65,87 @@ fn on_lane_error_defaults_to_collect() {
         "an unrecognised policy must not silently fail the run"
     );
 }
+
+/// The lane indices a gather over `evaluate` would count, and its expected total.
+fn counted(nodes: &Value) -> (Vec<usize>, Option<usize>) {
+    let slots = lane_slots(nodes, &["evaluate".to_string()]);
+    let scatter = scatter_of(&slots);
+    let mut indices: Vec<usize> = arrivals(current_activation(nodes, slots, scatter))
+        .iter()
+        .map(|lane| lane.index)
+        .collect();
+    indices.sort_unstable();
+    (indices, expected_lanes(nodes, scatter))
+}
+
+fn stamped(index: usize, step: u64) -> Value {
+    json!({
+        "items": [], "status": "ok", "index": index,
+        "_scatter": "scatter", "_scatter_step": step,
+    })
+}
+
+/// A second loop pass opened 2 lanes at step 9; pass 1's lane #2 (step 3) is
+/// still in state and must not be counted.
+#[test]
+fn lanes_from_an_earlier_scatter_activation_are_ignored() {
+    let nodes = json!({
+        "scatter": { "lane_count": 2, "_activation_step": 9 },
+        "evaluate": { "lanes": {
+            "scatter#0": stamped(0, 9),
+            "scatter#1": stamped(1, 9),
+            "scatter#2": stamped(2, 3),
+        } },
+    });
+    assert_eq!(counted(&nodes), (vec![0, 1], Some(2)));
+}
+
+/// Before any of this pass's lanes land, the earlier pass's lanes count for
+/// nothing — the gather waits rather than releasing stale results.
+#[test]
+fn an_activation_with_no_lanes_landed_yet_counts_none() {
+    let nodes = json!({
+        "scatter": { "lane_count": 2, "_activation_step": 9 },
+        "evaluate": { "lanes": { "scatter#0": stamped(0, 3), "scatter#1": stamped(1, 3) } },
+    });
+    assert_eq!(counted(&nodes), (vec![], Some(2)));
+}
+
+/// The expected total is this gather's own scatter's, not the first
+/// `lane_count` anywhere in state.
+#[test]
+fn the_expected_count_comes_from_the_scatter_that_opened_the_lanes() {
+    let nodes = json!({
+        "an_earlier_scatter": { "lane_count": 7, "_activation_step": 1 },
+        "scatter": { "lane_count": 1, "_activation_step": 4 },
+        "evaluate": { "lanes": { "scatter#0": stamped(0, 4) } },
+    });
+    assert_eq!(counted(&nodes), (vec![0], Some(1)));
+}
+
+/// Slots written from a pre-stamp envelope (a checkpoint from an older binary)
+/// all count when nothing is stamped, so an in-flight run still completes.
+#[test]
+fn unstamped_slots_count_when_no_slot_is_stamped() {
+    let nodes = json!({
+        "scatter": { "lane_count": 2, "_activation_step": 9 },
+        "evaluate": { "lanes": {
+            "scatter#0": { "items": [], "status": "ok", "index": 0 },
+            "scatter#1": { "items": [], "status": "ok", "index": 1 },
+        } },
+    });
+    assert_eq!(counted(&nodes), (vec![0, 1], Some(2)));
+}
+
+/// Once this pass's lanes are stamped, unstamped slots are an earlier pass's.
+#[test]
+fn unstamped_slots_are_stale_once_a_stamped_slot_exists() {
+    let nodes = json!({
+        "scatter": { "lane_count": 1, "_activation_step": 9 },
+        "evaluate": { "lanes": {
+            "scatter#0": stamped(0, 9),
+            "scatter#1": { "items": [], "status": "ok", "index": 1 },
+        } },
+    });
+    assert_eq!(counted(&nodes), (vec![0], Some(1)));
+}