Fix sub-orchestration child ID collision across loop generations

pinodeca · pinodeca · commit b26327a576d6 · 2026-06-27T13:50:12.000Z
Nested loops and loop bodies containing JOIN/RACE branches hung forever
because duroxide auto-generated child instance IDs ({parent}::sub::{event_id})
reset their event counter on each continue_as_new generation, so every loop
generation re-derived the same child ID and collided with the previous
(now terminal) child.

Use schedule_sub_orchestration_with_id with a deterministic,
generation-qualified child ID ({instance}::e{execution_id}::{tag}::{node_id})
at all loop/join/race spawn sites.

Add nested-loop, loop-in-JOIN-branch, and non-root while-loop regression
tests, and a USER_GUIDE note that each df.loop runs as its own child
orchestration.
diff --git a/USER_GUIDE.md b/USER_GUIDE.md
@@ -948,6 +948,12 @@ SELECT df.start(
 
 Use `@>` operator or `df.loop()` to create functions that run forever. Each iteration creates a new execution with fresh state (via continue-as-new).
 
+> **Note:** Each `df.loop()` runs as its own child orchestration. The loop's iterations
+> (the continue-as-new generations) are scoped to that child, so any nodes *before* the
+> loop in the graph run exactly once and any nodes *after* the loop run once the loop
+> exits. When inspecting `df.instances`, a looping function therefore shows a parent
+> instance plus a separate child instance for the loop.
+
 ```sql
 -- Simple heartbeat every 30 seconds (using @> operator)
 SELECT df.start(
diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs
@@ -790,6 +790,23 @@ pub async fn execute_loop(ctx: OrchestrationContext, input_json: String) -> Resu
         .map_err(|e| format!("continue_as_new failed: {e:?}"))
 }
 
+/// Build a deterministic, generation-qualified child instance ID for a sub-orchestration.
+///
+/// Duroxide's auto-generated child IDs (`{parent}::sub::{event_id}`) reset their event
+/// counter across `continue_as_new` generations. When a loop body itself spawns
+/// sub-orchestrations (a nested `df.loop`, or a parallel/race branch), each loop
+/// generation would otherwise re-derive the *same* child ID, colliding with the previous
+/// (now terminal) generation's child and stalling forever. Embedding the current
+/// execution (generation) ID plus the spawning node ID makes the child ID unique per
+/// generation while staying deterministic across replays of the same generation.
+fn child_instance_id(ctx: &OrchestrationContext, tag: &str, node_id: &str) -> String {
+    format!(
+        "{}::e{}::{tag}::{node_id}",
+        ctx.instance_id(),
+        ctx.execution_id()
+    )
+}
+
 async fn execute_loop_node(
     ctx: &OrchestrationContext,
     graph: &FunctionGraph,
@@ -822,8 +839,9 @@ async fn execute_loop_node(
         "Spawning loop sub-orchestration for node {node_id}"
     ));
 
+    let child_id = child_instance_id(ctx, "loop", node_id);
     let raw = ctx
-        .schedule_sub_orchestration(LOOP_NAME, loop_input)
+        .schedule_sub_orchestration_with_id(LOOP_NAME, child_id, loop_input)
         .await
         .map_err(|e| format!("Loop sub-orchestration failed: {e}"))?;
 
@@ -1057,8 +1075,11 @@ async fn execute_join_node(
     })
     .to_string();
 
-    // Build list of branch inputs
-    let mut branch_inputs = vec![left_input, right_input];
+    // Build list of (branch node id, branch input) pairs.
+    let mut branch_inputs: Vec<(String, String)> = vec![
+        (left_id.clone(), left_input),
+        (right_id.clone(), right_input),
+    ];
 
     // Check for extra nodes (join3)
     if let Some(config_str) = &node.query {
@@ -1074,17 +1095,20 @@ async fn execute_join_node(
                             "label": exec_ctx.label
                         })
                         .to_string();
-                        branch_inputs.push(extra_input);
+                        branch_inputs.push((extra_id.to_string(), extra_input));
                     }
                 }
             }
         }
     }
 
-    // Schedule sub-orchestrations and collect DurableFutures
+    // Schedule sub-orchestrations and collect DurableFutures. Use explicit,
+    // generation-qualified child IDs so a JOIN inside a loop body does not collide
+    // across continue_as_new generations.
     let mut durable_futures = Vec::new();
-    for input in branch_inputs {
-        let fut = ctx.schedule_sub_orchestration(SUBTREE_NAME, input);
+    for (branch_node_id, input) in branch_inputs {
+        let child_id = child_instance_id(ctx, "subtree", &branch_node_id);
+        let fut = ctx.schedule_sub_orchestration_with_id(SUBTREE_NAME, child_id, input);
         durable_futures.push(fut);
     }
 
@@ -1176,9 +1200,18 @@ async fn execute_race_node(
     })
     .to_string();
 
-    // Schedule sub-orchestrations
-    let left_fut = ctx.schedule_sub_orchestration(SUBTREE_NAME, left_input);
-    let right_fut = ctx.schedule_sub_orchestration(SUBTREE_NAME, right_input);
+    // Schedule sub-orchestrations with explicit, generation-qualified child IDs so a
+    // RACE inside a loop body does not collide across continue_as_new generations.
+    let left_fut = ctx.schedule_sub_orchestration_with_id(
+        SUBTREE_NAME,
+        child_instance_id(ctx, "subtree", left_id),
+        left_input,
+    );
+    let right_fut = ctx.schedule_sub_orchestration_with_id(
+        SUBTREE_NAME,
+        child_instance_id(ctx, "subtree", right_id),
+        right_input,
+    );
 
     // Use ctx.select2() - first to complete wins
     // select2 now returns Either2<Left, Right> instead of (winner_idx, DurableOutput)
diff --git a/tests/e2e/sql/24_nonroot_loop.sql b/tests/e2e/sql/24_nonroot_loop.sql
@@ -204,5 +204,206 @@ END $$;
 DROP TABLE _t3;
 DROP TABLE test_nonroot3_log;
 
+-- === Test 4: Nested loop — a loop body that itself contains a loop ===
+--
+-- Each df.loop() spawns an execute_loop sub-orchestration, so a nested loop spawns a
+-- child execute_loop from *within* another execute_loop generation.  This verifies that
+-- continue_as_new in the outer loop does not disturb the inner loop and vice versa.
+--
+-- Outer loop runs 2 iterations (break when outer_marker has 2 rows); each outer iteration
+-- runs an inner loop that inserts exactly one row and breaks immediately.
+-- Expected: outer_marker = 2 rows, inner_table = 2 rows.
+
+DROP TABLE IF EXISTS test_nested_outer;
+DROP TABLE IF EXISTS test_nested_inner;
+CREATE TABLE test_nested_outer (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+CREATE TABLE test_nested_inner (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+
+CREATE TEMP TABLE _t4 AS
+SELECT df.start(
+    df.loop(
+        'INSERT INTO test_nested_outer DEFAULT VALUES'
+        ~> df.loop(
+            'INSERT INTO test_nested_inner DEFAULT VALUES'
+            ~> df.break()
+        )
+        ~> (
+            'SELECT COUNT(*) >= 2 FROM test_nested_outer'
+                ?> df.break()
+                !> df.sleep(1)
+        )
+    ),
+    'test-nested-loop'
+) AS instance_id;
+
+DO $$
+DECLARE
+    v_id    TEXT;
+    v_status TEXT;
+    v_outer INT;
+    v_inner INT;
+BEGIN
+    SELECT instance_id INTO v_id FROM _t4;
+    RAISE NOTICE 'Test 4 - nested loop: instance %', v_id;
+
+    SELECT df.wait_for_completion(v_id, 90) INTO v_status;
+
+    IF v_status != 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED [nested]: expected completed, got %', v_status;
+    END IF;
+
+    SELECT COUNT(*) INTO v_outer FROM test_nested_outer;
+    SELECT COUNT(*) INTO v_inner FROM test_nested_inner;
+
+    IF v_outer != 2 THEN
+        RAISE EXCEPTION 'TEST FAILED [nested]: outer ran % time(s) (expected 2)', v_outer;
+    END IF;
+
+    IF v_inner != 2 THEN
+        RAISE EXCEPTION 'TEST FAILED [nested]: inner ran % time(s) (expected 2)', v_inner;
+    END IF;
+
+    RAISE NOTICE 'PASSED: nested loop — outer ran twice, inner ran once per outer iteration';
+END $$;
+
+DROP TABLE _t4;
+DROP TABLE test_nested_outer;
+DROP TABLE test_nested_inner;
+
+-- === Test 5: Loop inside a JOIN branch ===
+--
+-- JOIN branches execute as execute_subtree sub-orchestrations, so a loop in a branch
+-- spawns an execute_loop child from *within* execute_subtree.  This verifies the loop
+-- sub-orchestration nests correctly under a parallel branch and that the JOIN still
+-- completes once both branches finish.
+--
+-- Left branch inserts one row; right branch loops until join_loop has 2 rows.
+-- Expected: join_left = 1 row, join_loop = 2 rows.
+
+DROP TABLE IF EXISTS test_join_left;
+DROP TABLE IF EXISTS test_join_loop;
+CREATE TABLE test_join_left (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+CREATE TABLE test_join_loop (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+
+CREATE TEMP TABLE _t5 AS
+SELECT df.start(
+    'INSERT INTO test_join_left DEFAULT VALUES'
+    & df.loop(
+        'INSERT INTO test_join_loop DEFAULT VALUES'
+        ~> (
+            'SELECT COUNT(*) >= 2 FROM test_join_loop'
+                ?> df.break()
+                !> df.sleep(1)
+        )
+    ),
+    'test-loop-in-join-branch'
+) AS instance_id;
+
+DO $$
+DECLARE
+    v_id    TEXT;
+    v_status TEXT;
+    v_left  INT;
+    v_loop  INT;
+BEGIN
+    SELECT instance_id INTO v_id FROM _t5;
+    RAISE NOTICE 'Test 5 - loop in JOIN branch: instance %', v_id;
+
+    SELECT df.wait_for_completion(v_id, 90) INTO v_status;
+
+    IF v_status != 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED [loop-in-join]: expected completed, got %', v_status;
+    END IF;
+
+    SELECT COUNT(*) INTO v_left FROM test_join_left;
+    SELECT COUNT(*) INTO v_loop FROM test_join_loop;
+
+    IF v_left != 1 THEN
+        RAISE EXCEPTION 'TEST FAILED [loop-in-join]: left branch ran % time(s) (expected 1)', v_left;
+    END IF;
+
+    IF v_loop != 2 THEN
+        RAISE EXCEPTION 'TEST FAILED [loop-in-join]: loop branch body ran % time(s) (expected 2)', v_loop;
+    END IF;
+
+    RAISE NOTICE 'PASSED: loop in JOIN branch — left ran once, loop body ran twice';
+END $$;
+
+DROP TABLE _t5;
+DROP TABLE test_join_left;
+DROP TABLE test_join_loop;
+
+-- === Test 6: Non-root while-loop — prefix once, while-condition exit, suffix once ===
+--
+-- The earlier tests exit via df.break(); this one exits via a false while-condition
+-- (df.loop(body, condition)).  The condition node also runs inside the loop
+-- sub-orchestration, so this exercises the while-false exit path across generations.
+--
+-- Graph: INSERT prefix ~> df.loop(body, 'COUNT < 3') ~> INSERT suffix
+-- Expected: prefix = 1 row, body = 3 rows (loop stops when count reaches 3), suffix = 1 row.
+
+DROP TABLE IF EXISTS test_while_prefix;
+DROP TABLE IF EXISTS test_while_body;
+DROP TABLE IF EXISTS test_while_suffix;
+CREATE TABLE test_while_prefix (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+CREATE TABLE test_while_body   (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+CREATE TABLE test_while_suffix (id SERIAL, ts TIMESTAMPTZ DEFAULT clock_timestamp());
+
+CREATE TEMP TABLE _t6 AS
+SELECT df.start(
+    df.seq(
+        'INSERT INTO test_while_prefix DEFAULT VALUES',
+        df.seq(
+            df.loop(
+                'INSERT INTO test_while_body DEFAULT VALUES' ~> df.sleep(1),
+                'SELECT COUNT(*) < 3 FROM test_while_body'
+            ),
+            'INSERT INTO test_while_suffix DEFAULT VALUES'
+        )
+    ),
+    'test-nonroot-while-loop'
+) AS instance_id;
+
+DO $$
+DECLARE
+    v_id     TEXT;
+    v_status TEXT;
+    v_prefix INT;
+    v_body   INT;
+    v_suffix INT;
+BEGIN
+    SELECT instance_id INTO v_id FROM _t6;
+    RAISE NOTICE 'Test 6 - non-root while loop: instance %', v_id;
+
+    SELECT df.wait_for_completion(v_id, 90) INTO v_status;
+
+    IF v_status != 'completed' THEN
+        RAISE EXCEPTION 'TEST FAILED [nonroot-while]: expected completed, got %', v_status;
+    END IF;
+
+    SELECT COUNT(*) INTO v_prefix FROM test_while_prefix;
+    SELECT COUNT(*) INTO v_body   FROM test_while_body;
+    SELECT COUNT(*) INTO v_suffix FROM test_while_suffix;
+
+    IF v_prefix != 1 THEN
+        RAISE EXCEPTION 'TEST FAILED [nonroot-while]: prefix ran % time(s) (expected 1)', v_prefix;
+    END IF;
+
+    IF v_body != 3 THEN
+        RAISE EXCEPTION 'TEST FAILED [nonroot-while]: body ran % time(s) (expected 3)', v_body;
+    END IF;
+
+    IF v_suffix != 1 THEN
+        RAISE EXCEPTION 'TEST FAILED [nonroot-while]: suffix ran % time(s) (expected 1)', v_suffix;
+    END IF;
+
+    RAISE NOTICE 'PASSED: non-root while loop — prefix once, body 3x via while-condition, suffix once';
+END $$;
+
+DROP TABLE _t6;
+DROP TABLE test_while_prefix;
+DROP TABLE test_while_body;
+DROP TABLE test_while_suffix;
+
 RESET SESSION AUTHORIZATION;
 SELECT 'TEST PASSED' AS result;
