Re-implement df.loop via sub-orchestration to fix non-root continue_as_new

df.loop called continue_as_new inline in the main orchestration, so every
new generation restarted from graph.root_node_id, re-executing prefix nodes
on every iteration. Each df.loop() node now spawns a dedicated child
sub-orchestration (execute_loop) that owns continue_as_new; the parent awaits
it and runs any suffix nodes exactly once.

Relies on duroxide PR #31 (parent link preserved across continue_as_new),
pulled in as a git dependency until merged and released.

Co-authored-by: copilot-swe-agent <copilot@github.com>
Co-authored-by: pinodeca <32303022+pinodeca@users.noreply.github.com>

Author: pinodeca

Cargo.lock

@@ -34,7 +34,14 @@ serde_json = { version = "1.0", features = ["preserve_order"] }
 uuid = { version = "1.0", features = ["v4", "serde"] }
 
 # duroxide integration
-duroxide = "=0.1.29"
+# Using git dependency on pinodeca/continue-parent-link branch which preserves the
+# parent link when a sub-orchestration calls continue_as_new, unblocking the
+# sub-orchestration approach for df.loop.
+# Compatibility: duroxide-pg 0.1.34 has been verified to compile and run correctly
+# against this branch (same public API as 0.1.29 — PR #31 is a runtime-only change).
+# Once the branch is merged and a new duroxide release is published, revert both
+# entries back to crates.io version pins as a compatible pair.
+duroxide = { git = "https://github.com/microsoft/duroxide.git", branch = "pinodeca/continue-parent-link" }
 duroxide-pg = "=0.1.34"
 tokio = { version = "1", features = ["rt-multi-thread", "sync", "time"] }
 
@@ -55,6 +62,11 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 [dev-dependencies]
 pgrx-tests = "=0.16.1"
 
+# Override the crates.io duroxide with the git branch for both the direct dependency
+# and duroxide-pg's transitive dependency on duroxide.
+[patch.crates-io]
+duroxide = { git = "https://github.com/microsoft/duroxide.git", branch = "pinodeca/continue-parent-link" }
+
 [profile.dev]
 panic = "unwind"
 

Cargo.toml

@@ -526,81 +526,156 @@ async fn execute_wait_schedule_node(
 /// deficit so an empty-bodied loop can't busy-spin via continue_as_new.
 const LOOP_MIN_ITER_DURATION: Duration = Duration::from_secs(1);
 
-async fn execute_loop_node(
-    ctx: &OrchestrationContext,
-    graph: &FunctionGraph,
-    node: &FunctionNode,
-    node_id: &str,
-    results: &mut HashMap<String, String>,
-    exec_ctx: &ExecutionContext,
-) -> NodeResult {
+/// Orchestration name for the loop sub-orchestration.
+///
+/// Each `df.loop()` node spawns a child orchestration under this name.  The
+/// child handles all iterations via `continue_as_new`; when the loop exits it
+/// returns a `SubtreeEnvelope` to the parent.  The parent link is preserved
+/// across `continue_as_new` generations by duroxide (see duroxide PR #31), so
+/// the parent orchestration is notified when the loop finally completes.
+pub const LOOP_NAME: &str = "pg_durable::orchestration::execute-loop";
+
+/// Build the `SubtreeEnvelope` a loop returns to its parent on exit.
+///
+/// A loop always exits with a *normal* result: a `df.break()` inside the body is the loop's
+/// own terminator (caught here as `NodeError::Break`), not a break that should unwind past
+/// the loop, so the envelope is always tagged `Normal`. `execute_loop_node` merges `results`
+/// back into the parent map via `parse_subtree_envelope`.
+fn loop_exit_envelope(
+    result: String,
+    results: HashMap<String, String>,
+) -> Result<String, String> {
+    let envelope = SubtreeEnvelope {
+        control: Some(SubtreeControl::Normal),
+        result,
+        results,
+    };
+    serde_json::to_string(&envelope).map_err(|e| format!("Failed to serialize loop envelope: {e}"))
+}
+
+/// Sub-orchestration that runs a single loop iteration and either returns or
+/// calls `continue_as_new` for the next iteration.
+///
+/// Input JSON:
+/// ```json
+/// { "instance_id": "...", "loop_node_id": "...",
+///   "results": "<results_json>", "vars": "<vars_json>", "label": "..." }
+/// ```
+///
+/// `load_function_graph` is called at the start of **every** generation
+/// (including after `continue_as_new`) so that cross-iteration security
+/// tampering is caught and the instance is failed — the same guarantee the
+/// main `execute()` orchestration provides at its generation boundary.
+///
+/// On loop exit the function returns a `SubtreeEnvelope` containing the final
+/// result and any named results accumulated during the loop.
+pub async fn execute_loop(ctx: OrchestrationContext, input_json: String) -> Result<String, String> {
+    let input: serde_json::Value = serde_json::from_str(&input_json)
+        .map_err(|e| format!("Failed to parse ExecuteLoop input: {e}"))?;
+
+    let instance_id = input["instance_id"]
+        .as_str()
+        .ok_or("Missing instance_id in ExecuteLoop input")?
+        .to_string();
+    let loop_node_id = input["loop_node_id"]
+        .as_str()
+        .ok_or("Missing loop_node_id in ExecuteLoop input")?
+        .to_string();
+    let results_json = input["results"]
+        .as_str()
+        .ok_or("Missing results in ExecuteLoop input")?;
+
+    // re-load the graph from the database on every generation — this re-validates
+    // submitted_by and catches cross-iteration security tampering.
+    let graph_json = ctx
+        .schedule_activity(activities::load_function_graph::NAME, instance_id.clone())
+        .await?;
+    let graph: FunctionGraph = serde_json::from_str(&graph_json)
+        .map_err(|e| format!("Failed to parse graph in ExecuteLoop: {e}"))?;
+
+    let mut results: HashMap<String, String> = serde_json::from_str(results_json)
+        .map_err(|e| format!("Failed to parse results in ExecuteLoop: {e}"))?;
+
+    let vars: HashMap<String, String> = if let Some(vars_str) = input["vars"].as_str() {
+        serde_json::from_str(vars_str)
+            .map_err(|e| format!("Failed to parse vars in ExecuteLoop: {e}"))?
+    } else {
+        HashMap::new()
+    };
+    let label: Option<String> = input["label"].as_str().map(|s| s.to_string());
+
+    let exec_ctx = ExecutionContext { vars, label };
+
+    let node = graph
+        .nodes
+        .get(&loop_node_id)
+        .ok_or_else(|| format!("Loop node not found: {loop_node_id}"))?;
+
     let body_id = node
         .left_node
         .as_ref()
-        .ok_or_else(|| format!("LOOP node {node_id} has no body"))?;
+        .ok_or_else(|| format!("LOOP node {loop_node_id} has no body"))?
+        .clone();
 
-    // Capture the iteration start time so we can rate-limit `continue_as_new`
-    // below.  `utc_now()` is duroxide's deterministic clock (recorded in
-    // history and replayed verbatim), so this remains replay-safe.
+    // Capture iteration start time for rate-limiting continue_as_new.
     let iter_started = ctx.utc_now().await.ok();
 
     ctx.trace_info("Executing loop iteration");
 
-    // The loop is the only place that catches `NodeError::Break`: a break unwinds through
-    // every compound node in the body via `?` and is converted here into a normal loop exit.
-    // A `Failure` still propagates out of the loop unchanged.
-    let body_result = match Box::pin(execute_function_node_with_vars(
-        ctx, graph, body_id, results, exec_ctx,
-    ))
-    .await
-    {
-        Ok(v) => v,
-        Err(NodeError::Break(break_value)) => {
-            ctx.trace_info(format!(
-                "Loop terminated by break with value: {break_value}"
-            ));
-            store_named_result(ctx, node, &break_value, results, "LOOP");
-            return Ok(break_value);
-        }
-        Err(e @ NodeError::Failure(_)) => return Err(e),
-    };
+    // The loop is where `NodeError::Break` is caught: a break unwinds through the body via
+    // `?` and is converted here into the loop's normal exit value.  A `Failure` propagates
+    // out of the sub-orchestration unchanged.
+    let body_result =
+        match execute_function_node_with_vars(&ctx, &graph, &body_id, &mut results, &exec_ctx).await
+        {
+            Ok(v) => v,
+            Err(NodeError::Break(break_value)) => {
+                ctx.trace_info(format!(
+                    "Loop terminated by break with value: {break_value}"
+                ));
+                store_named_result(&ctx, node, &break_value, &mut results, "LOOP");
+                return loop_exit_envelope(break_value, results);
+            }
+            Err(NodeError::Failure(e)) => return Err(e),
+        };
 
-    // Check while-condition if present
+    // While-condition: if present and false, exit the loop.
     if let Some(ref config_str) = node.query {
         if let Ok(config) = serde_json::from_str::<serde_json::Value>(config_str) {
             if let Some(condition_node_id) = config["condition_node"].as_str() {
                 ctx.trace_info("Evaluating loop condition");
-                let condition_result = Box::pin(execute_function_node_with_vars(
-                    ctx,
-                    graph,
+                let condition_result = match execute_function_node_with_vars(
+                    &ctx,
+                    &graph,
                     condition_node_id,
-                    results,
-                    exec_ctx,
-                ))
-                .await?;
+                    &mut results,
+                    &exec_ctx,
+                )
+                .await
+                {
+                    Ok(v) => v,
+                    Err(NodeError::Break(break_value)) => {
+                        store_named_result(&ctx, node, &break_value, &mut results, "LOOP");
+                        return loop_exit_envelope(break_value, results);
+                    }
+                    Err(NodeError::Failure(e)) => return Err(e),
+                };
 
-                // Parse condition result to check truthiness (uses evaluate_condition to extract boolean from SQL result)
                 let should_continue = evaluate_condition(&condition_result).unwrap_or(false);
                 ctx.trace_info(format!(
                     "Loop condition evaluated to: {condition_result} (continue={should_continue})"
                 ));
 
                 if !should_continue {
                     ctx.trace_info("Loop condition false, exiting loop");
-                    store_named_result(ctx, node, &body_result, results, "LOOP");
-                    return Ok(body_result);
+                    store_named_result(&ctx, node, &body_result, &mut results, "LOOP");
+                    return loop_exit_envelope(body_result, results);
                 }
             }
         }
     }
 
-    ctx.trace_info("Continuing as new for next loop iteration");
-
-    // Enforce a minimum per-iteration wall-clock duration to prevent
-    // busy-looping (e.g. `df.loop(df.sleep(0))`).  Compute the elapsed time
-    // from the deterministic clock; if the iteration finished faster than
-    // LOOP_MIN_ITER_DURATION, schedule a timer for the deficit so the next
-    // continue_as_new is gated by at least that much real-clock time.
+    // Enforce a minimum per-iteration wall-clock duration to prevent busy-looping.
     if let Some(started) = iter_started {
         if let Ok(now) = ctx.utc_now().await {
             let elapsed = now.duration_since(started).unwrap_or(Duration::ZERO);
@@ -615,19 +690,65 @@ async fn execute_loop_node(
         }
     }
 
-    // Preserve vars in continue_as_new input
-    let new_input = FunctionInput {
-        instance_id: graph.instance_id.clone(),
-        label: exec_ctx.label.clone(),
-        vars: exec_ctx.vars.clone(),
-    };
+    // Another iteration needed: continue_as_new within this sub-orchestration.
+    // The parent orchestration keeps its awaiting handle because duroxide preserves
+    // the parent link across continue_as_new (duroxide PR #31).
+    ctx.trace_info(format!(
+        "Loop continuing with continue_as_new at node {loop_node_id}"
+    ));
+    let new_results_json = serde_json::to_string(&results)
+        .map_err(|e| format!("Failed to serialize updated results: {e}"))?;
+    let mut new_input = input.clone();
+    new_input["results"] = serde_json::Value::String(new_results_json);
+    let new_input_json = serde_json::to_string(&new_input)
+        .map_err(|e| format!("Failed to serialize loop input: {e}"))?;
+    ctx.continue_as_new(new_input_json)
+        .await
+        .map(|_| String::new())
+        .map_err(|e| format!("continue_as_new failed: {e:?}"))
+}
+
+async fn execute_loop_node(
+    ctx: &OrchestrationContext,
+    graph: &FunctionGraph,
+    node: &FunctionNode,
+    node_id: &str,
+    results: &mut HashMap<String, String>,
+    exec_ctx: &ExecutionContext,
+) -> NodeResult {
+    // Validate that the loop has a body before spawning the sub-orchestration.
+    node.left_node
+        .as_ref()
+        .ok_or_else(|| format!("LOOP node {node_id} has no body"))?;
+
+    let results_json =
+        serde_json::to_string(results).map_err(|e| format!("Failed to serialize results: {e}"))?;
+    let vars_json = serde_json::to_string(&exec_ctx.vars)
+        .map_err(|e| format!("Failed to serialize vars: {e}"))?;
 
-    // duroxide 0.1.1: continue_as_new returns an awaitable future - return it directly
-    return ctx
-        .continue_as_new(serde_json::to_string(&new_input).unwrap_or(graph.instance_id.clone()))
+    let loop_input = serde_json::json!({
+        "instance_id": graph.instance_id,
+        "loop_node_id": node_id,
+        "results": results_json,
+        "vars": vars_json,
+        "label": exec_ctx.label,
+    })
+    .to_string();
+
+    ctx.trace_info(format!(
+        "Spawning loop sub-orchestration for node {node_id}"
+    ));
+
+    let raw = ctx
+        .schedule_sub_orchestration(LOOP_NAME, loop_input)
         .await
-        .map(|_| body_result)
-        .map_err(|e| NodeError::Failure(format!("continue_as_new failed: {e:?}")));
+        .map_err(|e| format!("Loop sub-orchestration failed: {e}"))?;
+
+    // Merge named results from the loop sub-orchestration back into the parent map and
+    // return the loop's final result.  The loop always returns a `Normal` envelope (a break
+    // inside the body is the loop's own terminator), so `parse_subtree_envelope` will not
+    // re-raise a `NodeError::Break` here.
+    parse_subtree_envelope(&raw, "LOOP", results)
 }
 
 async fn execute_break_node(

src/orchestrations/execute_function_graph.rs

@@ -55,5 +55,9 @@ pub fn create_orchestration_registry() -> OrchestrationRegistry {
             orchestrations::execute_function_graph::SUBTREE_NAME,
             orchestrations::execute_function_graph::execute_subtree,
         )
+        .register(
+            orchestrations::execute_function_graph::LOOP_NAME,
+            orchestrations::execute_function_graph::execute_loop,
+        )
         .build()
 }