Skip to content
Merged
40 changes: 35 additions & 5 deletions .agents/skills/pg-durable-sql/SKILL.md
Original file line number Diff line number Diff line change
Expand Up @@ -93,9 +93,18 @@ df.if(condition TEXT, then_branch TEXT, else_branch TEXT) → TEXT
-- result_name is a capture from |=> earlier in the graph.
df.if_rows(result_name TEXT, then_branch TEXT, else_branch TEXT) → TEXT

-- Loop — infinite or while-condition
df.loop(body TEXT) → TEXT -- Infinite loop
df.loop(body TEXT, condition TEXT) → TEXT -- While-loop: repeats while condition is truthy
-- Loop — one unified signature
df.loop(
body TEXT,
condition TEXT DEFAULT NULL,
continue_on_failure BOOLEAN DEFAULT false
) → TEXT
Comment thread
tjgreen42 marked this conversation as resolved.
-- Experimental: continue_on_failure syntax may change in future releases.
-- NULL condition: infinite loop.
-- Non-NULL condition: do-while semantics; evaluate it after each successful body.
-- With continue_on_failure => true, a consumed body activity failure skips the
-- condition and starts the next iteration. The loop's condition and
-- orchestration/runtime failures remain fatal.

-- Break from enclosing loop
df.break() → TEXT -- Exit with NULL
Expand Down Expand Up @@ -227,7 +236,7 @@ Automatically available during execution:

## Condition Evaluation (Truthiness)

Used by: `?>`, `!>`, `df.if()`, `df.loop(body, condition)`
Used by: `?>`, `!>`, `df.if()`, and the optional condition in `df.loop()`

The first column of the first row is evaluated:

Expand Down Expand Up @@ -345,7 +354,7 @@ SELECT df.start(
-- Cancel with: SELECT df.cancel('instance_id', 'Stopping heartbeat');
```

### While-Loop with Break
### Loop with Break

```sql
SELECT df.start(
Expand All @@ -361,6 +370,27 @@ SELECT df.start(
);
```

### Conditional Loop with Failure Continuation

```sql
SELECT df.start(
df.loop(
'SELECT process_next_item()',
'SELECT EXISTS (
SELECT 1 FROM work_queue WHERE status = ''pending''
)',
continue_on_failure => true
),
'resilient-worker'
);
```

This is a do-while loop: after a successful body execution, the condition is
evaluated and the loop continues while it is truthy. After a consumed body
activity failure, the condition is skipped and the body starts again.
Errors returned by body SQL, HTTP, and multipart activities are consumable.
Condition and orchestration/runtime failures remain fatal.

### Cron Scheduled Job

```sql
Expand Down
19 changes: 19 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,25 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc

## [0.2.8] - Unreleased

### Added

- **Failure-isolated loops:** the unified
`df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)`
signature supports resilient infinite and conditional loops. With
`continue_on_failure => true`, a consumed typed body activity failure skips
the condition and starts the next iteration; after a successful body, the
condition is evaluated normally. All errors returned by body SQL, HTTP, and
multipart activities are consumable, including query, authorization,
connection, and network errors. Condition, graph, protocol, unknown child,
and orchestration/runtime failures remain fatal. The
`continue_on_failure` syntax is experimental and may change in future
releases.

### Changed

- **Loop lifetime:** raises the loop-iteration backstop from 100,000 to
8,388,608 (`2^23`), approximately 80 years at five-minute ticks.

### Fixed

- **Caller-transaction handoff:** `df.start()` now tracks the originating transaction until it commits or aborts, so legal caller transactions lasting more than five seconds no longer leave a `pending` `df.instances` row paired with a failed engine execution. Graph admission uses durable backoff and bounded-history compaction rather than holding a worker connection while it waits.
Expand Down
94 changes: 87 additions & 7 deletions USER_GUIDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -264,8 +264,7 @@ df.sql('SELECT 1') ~> df.sql('SELECT 2')
| `df.join3(a, b, c)` | Three in parallel | `df.join3(a, b, c)` |
| `df.race(a, b)` | Execute in parallel, first wins | `df.race(fast_query, slow_query)` |
| `df.if(cond, then, else)` | Conditional branch | `df.if('SELECT true', a, b)` |
| `df.loop(body)` | Repeat forever | `df.loop(body)` |
| `df.loop(body, cond)` | Repeat while condition is true | `df.loop(body, 'SELECT count(*) > 0 FROM q')` |
| `df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)` | Repeat forever or while a condition is true, optionally continuing after body activity failures | `df.loop(body, 'SELECT count(*) > 0 FROM q', continue_on_failure => true)` |
| `df.break()` | Exit enclosing loop | `df.break()` |
| `df.break(value)` | Exit loop with **literal** return value (not auto-wrapped as SQL) | `df.break('{"done": true}')` |
| `df.start(func, label, database)` | Start function (optionally in another database) | `df.start('SELECT 1', 'job')` |
Expand Down Expand Up @@ -531,7 +530,8 @@ SELECT df.start(

### Loop Condition Example

For `df.loop(body, condition)`, the condition is evaluated after each iteration:
For `df.loop(body, condition)`, the condition is evaluated after each
successful body execution:

```sql
-- Loop while there are pending items
Expand Down Expand Up @@ -1180,6 +1180,75 @@ SELECT df.start(
);
```

The unified signature is
`df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)`.

> **Experimental:** The `continue_on_failure` syntax is subject to change in
> future releases.

It supports four call shapes:

```sql
df.loop(body)
df.loop(body, condition)
df.loop(body, continue_on_failure => true)
df.loop(body, condition, continue_on_failure => true)
```

By default, it and `@>` are fail-fast: an iteration failure fails the loop.
Passing `continue_on_failure => false` is equivalent to the default and does
not change where the loop is hosted. When it is `true`, each body iteration
runs as a child orchestration. After a successful body iteration, the child's
results are merged into the parent result map before the optional condition is
evaluated. After a consumed typed application failure from a body activity,
the condition is skipped and the parent starts the next iteration. This
includes any error returned by a body SQL, HTTP, or multipart activity, such as
a query, authorization, connection, or network error. Condition failures,
malformed graph or child data, unrecognized child errors, child-ID collisions,
and orchestration/runtime failures remain fatal.

This is useful for scheduled maintenance such as a pg_textsearch-style
per-index compactor. A simplified periodic backstop repeatedly runs one
compaction step until the index has no more reducible debt:

```sql
SELECT df.start(
df.loop(
df.wait_for_schedule('*/5 * * * *')
~> df.loop(
$$SELECT public.bm25_compact_step(
'documents_idx'::regclass
) AS ran$$
|=> 'step'
~> df.if(
'SELECT $step.ran',
'SELECT true',
df.break()
)
),
continue_on_failure => true
),
'documents-index-compaction'
);
```

The managed pg_textsearch workflow also validates the physical index identity
before each cascade, so replacing or dropping the index stops stale work.

The condition and continuation policy can be combined:

```sql
df.loop(
'SELECT process_next_item()',
'SELECT count(*) > 0 FROM task_queue',
continue_on_failure => true
)
```

The next schedule is computed only after the prior iteration finishes. Slow or
failed iterations therefore do not overlap, queue missed ticks, or backfill
earlier schedule times.

### Cron-Style Scheduling

Use `df.wait_for_schedule()` with a cron expression:
Expand Down Expand Up @@ -1285,10 +1354,21 @@ SELECT df.start(

Each loop iteration advances via *continue-as-new*, which restarts the loop with fresh state while preserving durability. Where that restart happens depends on whether the loop is the **root** of the function:

- A **root loop** (the function graph's outermost node, e.g. `df.start(df.loop(...))` or `df.start(@> 'SELECT work()')`) runs inline on the function's own orchestration. There is no surrounding work to preserve, so each iteration simply restarts the function. The `@>` operator does not make a loop root by itself: `df.seq('SELECT setup()', @> 'SELECT work()')` is non-root because the sequence is the graph root.
- A **non-root loop** (a loop with prefix/suffix nodes, or one nested inside a `df.if()`, JOIN (`&`), or RACE (`|`) branch) runs as its own **child sub-orchestration**. Only the loop body restarts on each iteration — any work *before* the loop runs exactly once and is never re-executed, and a loop nested in a parallel branch gets its own durable instance.

This is transparent to your workflow; it only affects observability. The child sub-orchestration is an internal durable instance: it does **not** appear in `df.list_instances()` (which lists only the instances you started with `df.start()`). Instead, the loop node's status in `df.instance_nodes()` / `df.explain()` reflects the child's progress, so you observe the loop through its parent instance as usual.
- A fail-fast **root loop** (the function graph's outermost node, e.g. `df.start(df.loop(...))` or `df.start(@> 'SELECT work()')`) runs inline on the function's own orchestration. There is no surrounding work to preserve, so each iteration simply restarts the function. The `@>` operator does not make a loop root by itself: `df.seq('SELECT setup()', @> 'SELECT work()')` is non-root because the sequence is the graph root.
- A fail-fast **non-root loop** (a loop with prefix/suffix nodes, or one nested inside a `df.if()`, JOIN (`&`), or RACE (`|`) branch) runs as its own **child sub-orchestration**. Only the loop body restarts on each iteration — any work *before* the loop runs exactly once and is never re-executed, and a loop nested in a parallel branch gets its own durable instance.
- Fail-fast iterations execute inline within whichever orchestration hosts the loop. Explicit `continue_on_failure => false` keeps this placement; it does not force a non-root loop into the function's root orchestration.
- A loop with `continue_on_failure => true` runs every body iteration in a child orchestration. After a successful body iteration, the child's results are merged into the parent result map before the optional condition is evaluated. After any error returned by a body SQL, HTTP, or multipart activity, the condition is skipped and the loop parent proceeds to the next iteration. Condition failures and unrecognized child, graph, protocol, orchestration, or runtime failures remain fatal.

These child sub-orchestrations are internal durable instances and do **not**
appear in `df.list_instances()`, which lists only instances started with
`df.start()`. `df.instance_nodes()` / `df.explain()` show the loop parent as
running and may show a failed iteration on its body-root or descendant nodes.
Those node statuses are not permanent per-iteration history: a later iteration
can supersede them with newer execution status.

All loops have a finite backstop of 8,388,608 (`2^23`) iterations: about 80
years at five-minute intervals, or a little over 97 days at the enforced
one-second minimum.

### Stopping a Loop Externally

Expand Down
94 changes: 55 additions & 39 deletions docs/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -477,31 +477,36 @@ of every handler having to recognise an in-band JSON break sentinel:
pub enum NodeError {
/// df.break() fired. Carries the break value. Caught only by execute_loop_node.
Break(String),
/// A genuine failure. Surfaces as a failed instance.
/// An expected workflow activity failure.
Application(String),
/// A graph, protocol, configuration, or runtime failure.
Failure(String),
}

pub type NodeResult = Result<String, NodeError>;

// Any `?` on an existing Result<_, String> auto-converts the error to Failure, so
// activity calls and helpers need no per-call changes.
// Structural/configuration helpers still convert String errors to Failure.
impl From<String> for NodeError {
fn from(e: String) -> Self {
NodeError::Failure(e)
}
}
```

`execute_loop_node` is the only handler that catches `NodeError::Break` (turning it into the
loop's `Ok` result); `NodeError::Failure` keeps propagating. The orchestration boundary
functions (`execute` / `execute_subtree`) still return `Result<String, String>` because they
are registered with duroxide:
SQL, HTTP, and multipart activity scheduling boundaries explicitly map activity
errors to `NodeError::Application`; structural/configuration helper errors use
`NodeError::Failure`. `execute_loop_node` is the only handler that catches
`NodeError::Break` (turning it into the loop's `Ok` result). The orchestration
boundary functions (`execute` / `execute_subtree`) still return
`Result<String, String>` because they are registered with duroxide:

- `execute`: an uncaught top-level `Break` becomes a clear `Err` ("df.break() was called
outside of a loop"), so the instance fails instead of completing with a sentinel value.
- `execute_subtree` (used by JOIN/RACE branches): a `Break` is carried out-of-band in the
subtree envelope's `control` field and re-raised as `NodeError::Break` by
`parse_subtree_envelope` in the parent orchestration.
- `execute_subtree`: a `Break` is carried out-of-band in the subtree envelope's
`control` field. `NodeError::Application` is encoded as a namespaced,
serde-tagged subtree failure so application classification survives nested
JOIN, RACE, and LOOP boundaries. `NodeError::Failure` remains an unrecognized
child error and propagates fatally.

The orchestration walks the graph recursively:

Expand Down Expand Up @@ -696,34 +701,25 @@ pub fn is_truthy(value: &serde_json::Value) -> bool {

### Parallel Execution (JOIN/RACE)

JOIN and RACE use duroxide's sub-orchestration support:

```rust
async fn execute_join_node(...) -> Result<String, String> {
let left_id = node.left_node.as_ref().ok_or("JOIN missing left")?;
let right_id = node.right_node.as_ref().ok_or("JOIN missing right")?;

// Create sub-orchestration inputs
let left_input = create_subtree_input(graph, left_id, results);
let right_input = create_subtree_input(graph, right_id, results);

// Schedule parallel sub-orchestrations
let left_handle = ctx.schedule_orchestration(SUBTREE_NAME, &left_id, left_input);
let right_handle = ctx.schedule_orchestration(SUBTREE_NAME, &right_id, right_input);

// Wait for all to complete (duroxide handles parallelism)
let (left_result, right_result) = tokio::join!(
left_handle.into_orchestration(),
right_handle.into_orchestration()
);

// Combine results
let results = vec![left_result?, right_result?];
Ok(serde_json::to_string(&results)?)
}
```

For RACE, duroxide's `select` is used to return the first completed result.
Each JOIN or RACE branch runs as an explicitly named `execute_subtree` child
whose input contains the validated graph snapshot, variables, label, and
canonically serialized named results. The child returns a `SubtreeEnvelope`
containing its result, named-result updates, and optional `df.break()` control
flow.

JOIN schedules its two branches, plus any ordered `join3` extras, then waits
with `ctx.join()`. Successful envelopes are processed in branch order and
their named results are merged into the parent. Historically, JOIN returned
the first branch error in that same order. That behavior remains unchanged
outside failure-isolated loop bodies for replay compatibility. Inside a
failure-isolated body, every settled outcome is inspected so a fatal sibling
cannot be hidden by a recoverable activity failure or `df.break()`. The
deterministic priority is fatal failure, then break, then recoverable activity
failure; equal-priority outcomes keep the first branch.

RACE uses `ctx.select2()` and returns the first completed branch. The losing
branch is cancelled, and a losing loop branch receives a terminal fallback
node-status stamp because cancellation may stop it before it can stamp itself.

### Loops and Continue-As-New

Expand All @@ -734,7 +730,27 @@ Loops use duroxide's `continue_as_new` to avoid unbounded history growth. Their

`execute_subtree` is therefore structurally identical to `execute_function_graph`: both root an execution context at their own node and host an inline root loop. They differ only in the input envelope they re-enter with on `continue_as_new` (`FunctionInput` vs `SubtreeInput`), in the fact that only the root orchestration touches instance-level status, and in where their graph comes from — `execute_function_graph` loads it from `df.nodes` on its first generation, while `execute_subtree` receives it inline from its parent. The graph is loaded exactly once per instance and then carried inline through every child input and every `continue_as_new` generation, so `submitted_by` is fixed for the instance's lifetime and a post-start `df.nodes` tamper is never read. Role deletion and privilege revocation are still enforced per node execution, by connecting *as* `submitted_by` for SQL and by re-checking `EXECUTE` privilege per HTTP request.

Both paths call `run_loop_iteration`, which executes the body, catches `NodeError::Break`, evaluates the optional post-body condition, and propagates `NodeError::Failure`. A child stamps its LOOP node `running` on each generation and `completed` or `failed` on exit; because `continue_as_new` returns a future that never resolves, a continuing generation never stamps a terminal status. If a live loop loses a RACE, the parent records the loop node as terminal `failed` with a cancellation reason because duroxide cancellation stops the child before it can run its own terminal stamp.
Fail-fast loops call `run_loop_iteration`, which executes the body inline,
catches `NodeError::Break`, evaluates the optional post-body condition, and
propagates both application and non-application failures. A loop configured
with `continue_on_failure => true` instead calls
`run_failure_isolated_body`: each body iteration runs as a fresh
`execute_subtree` child. On success, the subtree envelope is parsed and its
named results are merged into the parent result map before the parent evaluates
the optional condition. On a structurally encoded body application failure,
the parent consumes the failure, skips the condition because body results may
be absent, and advances to the next generation. Every error returned by a body
SQL, HTTP, or multipart activity uses this application-failure path, including
query, authorization, connection, and network errors. Condition failures,
malformed envelopes or graph data, child-ID collisions, and unrecognized
orchestration/runtime failures remain fatal.

A child stamps its LOOP node `running` on each generation and `completed` or
`failed` on exit; because `continue_as_new` returns a future that never
resolves, a continuing generation never stamps a terminal status. If a live
loop loses a RACE, the parent records the loop node as terminal `failed` with a
cancellation reason because duroxide cancellation stops the child before it
can run its own terminal stamp.

Node status stamps contain the full composed orchestration lineage: `{root_instance}::{generation}::{child_node}::{generation}...`. Read-time inference and the write fence walk that lineage so stale writes and superseded nested branches are evaluated at every ancestor generation.

Expand Down
Loading