df.loop continue_as_new restarts from root when loop is not root

[pinodeca]

Summary

df.loop uses continue_as_new after each iteration, but the continuation input only preserves the instance metadata (instance_id, label, vars). When the loop node is not the root of the function graph, the next orchestration generation reloads the graph and starts again from the instance root, re-executing any prefix nodes before the loop.

This causes side effects before the loop to run once per loop iteration instead of once per function instance.

Repro

Run this in a database where pg_durable is installed and the background worker is running:

DROP TABLE IF EXISTS table_outside_loop;
DROP TABLE IF EXISTS table_inside_loop;

CREATE TABLE table_outside_loop (
    id SERIAL PRIMARY KEY,
    inserted_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

CREATE TABLE table_inside_loop (
    id SERIAL PRIMARY KEY,
    inserted_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp()
);

SELECT df.start(
    df.seq(
        'INSERT INTO table_outside_loop DEFAULT VALUES RETURNING id, inserted_at',
        df.loop(
            'INSERT INTO table_inside_loop DEFAULT VALUES RETURNING id, inserted_at'
            ~> df.sleep(10)
        )
    ),
    'manual-nonroot-loop-prefix',
    NULL
) AS instance_id;

After two loop iterations, inspect the tables:

SELECT COUNT(*) AS outside_loop_rows FROM table_outside_loop;
SELECT COUNT(*) AS inside_loop_rows FROM table_inside_loop;

TABLE table_outside_loop ORDER BY id;
TABLE table_inside_loop ORDER BY id;

Expected behavior

table_outside_loop should contain exactly one row for the function instance. The outer/prefix SQL step should run once before entering the loop.

table_inside_loop should grow by one row per loop iteration.

Actual behavior

table_outside_loop gets one new row per loop iteration/generation. In a local run, after the second loop generation:

outside_loop_rows = 2
inside_loop_rows  = 2

The timestamped rows showed the outside-loop insert running again immediately before the second inside-loop insert.

Likely cause

execute_loop_node calls continue_as_new with only:

FunctionInput {
    instance_id: graph.instance_id.clone(),
    label: exec_ctx.label.clone(),
    vars: exec_ctx.vars.clone(),
}

The new orchestration generation then runs the top-level execute path, calls load_function_graph, and starts from graph.root_node_id again. It does not preserve the loop node id or the current execution context as the continuation start point.

This is especially problematic for finite loops when setup state appears before the loop. For example, a prefix SQL node that resets a counter can cause a loop with df.break after N iterations to never reach N, because the counter is reset on every continue_as_new generation.

Notes

This affects loop surfaces built on the LOOP node type, including:

• df.loop(body)
• df.loop(body, condition)
• @> body
• recurring cron-style patterns that wrap df.wait_for_schedule(...) in df.loop / @>

[pinodeca]

Additional observed output from running the repro:

postgres=# TABLE table_outside_loop ORDER BY id;
TABLE table_inside_loop ORDER BY id;
 id |          inserted_at          
----+-------------------------------
  1 | 2026-06-11 16:53:04.062588-05
  2 | 2026-06-11 16:53:15.932074-05
  3 | 2026-06-11 16:53:27.803196-05
  4 | 2026-06-11 16:53:39.655268-05
(4 rows)

 id |          inserted_at          
----+-------------------------------
  1 | 2026-06-11 16:53:04.751788-05
  2 | 2026-06-11 16:53:16.633941-05
  3 | 2026-06-11 16:53:28.483375-05
  4 | 2026-06-11 16:53:40.341201-05
(4 rows)

This shows table_outside_loop growing in lockstep with table_inside_loop; the outside-loop insert is re-executed once per loop generation.

[pinodeca]

Fix sketch

At a high level, continue_as_new needs to resume the loop node, not restart the whole function root.

Today loop continuation input is only:

FunctionInput {
    instance_id,
    label,
    vars,
}

Then the next generation enters the top-level execution path and does:

execute_function_node_with_vars(&ctx, &graph, &graph.root_node_id, ...)

So it always restarts from root_node_id.

A simple-looking fix would be to extend FunctionInput with continuation state:

pub struct FunctionInput {
    pub instance_id: String,
    pub label: Option<String>,
    pub vars: HashMap<String, String>,

    // new
    pub start_node_id: Option<String>,
    pub results: HashMap<String, String>,
}

Then top-level execute() would choose:

let start_node_id = input
    .start_node_id
    .as_deref()
    .unwrap_or(&graph.root_node_id);

and execute from that node instead of always from the root.

For loops, execute_loop_node would call continue_as_new with:

FunctionInput {
    instance_id: graph.instance_id.clone(),
    label: exec_ctx.label.clone(),
    vars: exec_ctx.vars.clone(),
    start_node_id: Some(node_id.to_string()),
    results: results.clone(),
}

That would let the next generation start at the same LOOP node, preserving named results produced before the loop. The loop body could then run the next iteration without replaying prefix nodes.

However, start_node_id alone is probably not sufficient for the general case.

Important details:

1. Preserve named results

   Without carrying results, a loop that depends on a named result computed before the loop would break after the first continue_as_new.

2. Status updates

   The top-level execute() currently marks the whole instance running each generation. That can stay idempotent. Node status may need care because the same loop/body nodes are reused every iteration.

3. Completion after df.break or while-condition false

   If the resumed start node is the loop itself and it exits, what should happen next is tricky. If the original graph was:

   prefix ~> loop ~> suffix
   

   resuming only at loop loses the suffix after the loop.

So a complete fix likely needs a continuation stack, not just start_node_id.

For a correct general solution, the continuation should preserve “where to continue after this loop exits.” Conceptually:

struct FunctionInput {
    instance_id: String,
    label: Option<String>,
    vars: HashMap<String, String>,
    continuation: Option<Continuation>,
    results: HashMap<String, String>,
}

struct Continuation {
    start_node_id: String,
    after_node_id: Option<String>, // or a stack/path
}

But because the graph is binary and currently executed recursively, “after this node” is not directly encoded in child nodes. We would need either:

• build a parent/continuation map from the graph at load time, or
• change execution to an explicit interpreter/stack instead of recursive calls, so the stack can be serialized into continue_as_new.

The robust design is the second one: represent execution progress as explicit frames and serialize that frame stack through continue_as_new. Then a loop iteration can resume at the loop body/loop node, and when the loop eventually breaks or its while-condition turns false, execution can continue with the frame that was after the loop.

A pragmatic staged approach could be:

1. Short term: reject or document/validate that LOOP must be the root node. This prevents silent bad behavior.
2. Medium term: add start_node_id + results continuation, good for root loops and maybe simple loop-only continuations.
3. Full fix: make the orchestration carry an explicit continuation stack so non-root loops, break, while-loops, and suffix nodes all behave correctly across continue_as_new.

I would avoid stopping at only start_node_id: it fixes “prefix reruns” but risks breaking prefix ~> loop ~> suffix by never running the suffix after the loop exits.

[pinodeca]

Design note: execute loops as sub-orchestrations

A promising fix is to make LOOP execution a dedicated sub-orchestration rather than having the main execute-function-graph orchestration call continue_as_new directly.

Today a non-root loop is executed inline by the root orchestration. When the loop reaches the next iteration, it calls continue_as_new with only the top-level FunctionInput:

FunctionInput {
    instance_id,
    label,
    vars,
}

The next generation then re-enters the root orchestration and starts from graph.root_node_id, so any prefix before the loop runs again.

If loops were child orchestrations, the parent would execute the graph around the loop and schedule the loop node as a durable child:

prefix
~> schedule_sub_orchestration("execute-loop", { loop_node_id, graph, results, vars, label })
~> suffix

Then continue_as_new would apply only to the child loop orchestration. The parent would remain parked waiting for the child result. This gives the expected behavior for:

prefix ~> df.loop(body) ~> suffix

where prefix runs once, body runs once per loop generation, and suffix runs once after the loop exits.

The loop child input should carry at least:

struct LoopInput {
    graph: String,
    loop_node_id: String,
    results: HashMap<String, String>,
    vars: HashMap<String, String>,
    label: Option<String>,
}

On every continue_as_new, the child must preserve the loop node id, named results, vars, and label. On normal exit (df.break or a false while condition), it should return an envelope similar to execute_subtree:

struct LoopEnvelope {
    result: String,
    results: HashMap<String, String>,
}

The parent then parses the envelope, merges returned named results into its local results map, stores any loop result name, and continues executing the rest of the graph.

This is cleaner than making the root orchestration continuation-aware with only start_node_id, because it preserves composition for prefix ~> loop ~> suffix without needing the root generation to know how to resume in the middle of a THEN chain. It also matches the existing JOIN/RACE pattern where sub-orchestrations return both their result and updated named-results map.

Implementation considerations:

• Register a new orchestration name, for example pg_durable::orchestration::execute-loop.
• Change the LOOP branch in the parent executor to schedule the loop child and merge its envelope, instead of executing execute_loop_node inline.
• Move the current loop-body/condition/rate-limit logic into the loop child orchestration.
• Ensure nested loops work: a loop body containing another loop should schedule another loop child, and each loop's continue_as_new should stay scoped to its own child instance.
• Ensure break propagation remains local: a df.break exits the nearest loop child and returns its break value to the parent, rather than leaking as a sentinel through unrelated parents.
• Preserve named results across loop generations by serializing the updated results map in the child continue_as_new input.
• Add regression coverage for a non-root loop with both a prefix and suffix: assert the prefix runs once, the loop body runs multiple times, and the suffix runs once after loop exit.