diff --git a/.agents/skills/pg-durable-sql/SKILL.md b/.agents/skills/pg-durable-sql/SKILL.md index cd9218d4..e082f23f 100644 --- a/.agents/skills/pg-durable-sql/SKILL.md +++ b/.agents/skills/pg-durable-sql/SKILL.md @@ -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 +-- 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 @@ -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: @@ -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( @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 6967c191..483db67c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 5d8dc608..634d3caa 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -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')` | @@ -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 @@ -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: @@ -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 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 74b8168d..0e8e146a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -477,14 +477,15 @@ 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; -// 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 for NodeError { fn from(e: String) -> Self { NodeError::Failure(e) @@ -492,16 +493,20 @@ impl From for NodeError { } ``` -`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` 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` 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: @@ -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 { - 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 @@ -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. diff --git a/docs/api-reference.md b/docs/api-reference.md index e74aad1e..282c886a 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -160,22 +160,63 @@ df.if_rows('data', 'SELECT $data.id', 'SELECT ''no data''') --- -### df.loop(body [, condition]) / `@>` operator +### df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false) -Repeats body (forever or while condition is true). +> **Experimental:** The `continue_on_failure` syntax is subject to change in +> future releases. + +Repeats `body` forever or while `condition` is true. The supported call shapes +are: + +```sql +df.loop(body) +df.loop(body, condition) +df.loop(body, continue_on_failure => true) +df.loop(body, condition, continue_on_failure => true) +``` | Parameter | Type | Auto-wrap | Description | |-----------|------|-----------|-------------| | `body` | TEXT | ✅ Auto-wrap | Node to repeat | -| `condition` | TEXT | ✅ Auto-wrap | (Optional) Continue while truthy | +| `condition` | TEXT | ✅ Auto-wrap | (Optional) Evaluate after a successful body; continue while truthy | +| `continue_on_failure` | BOOLEAN | ❌ Literal | (Optional) Continue after a body activity failure; default `false` | ```sql -- Infinite loop df.loop('SELECT process_item()' ~> df.sleep(1)) -@> ('SELECT process_item()' ~> df.sleep(1)) -- operator (infinite only) --- While loop (function only, no operator) +-- While loop df.loop('SELECT process_item()', 'SELECT count(*) > 0 FROM queue') + +-- Infinite loop that continues after body activity failures +df.loop( + 'SELECT process_item()' ~> df.sleep(1), + continue_on_failure => true +) + +-- Conditional loop that continues after body activity failures +df.loop( + 'SELECT process_item()', + 'SELECT count(*) > 0 FROM queue', + continue_on_failure => true +) +``` + +By default, and when `continue_on_failure` is `false`, loop execution is +fail-fast. When it is `true`, each body iteration runs in a child +orchestration. After a successful body iteration, the child's results are +merged into the parent result map before condition evaluation. A consumed +typed application failure from a body activity skips condition evaluation and +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. + +The `@>` operator remains an infinite, fail-fast loop: + +```sql +@> ('SELECT process_item()' ~> df.sleep(1)) ``` --- @@ -640,7 +681,7 @@ SELECT df.clearvars(); | `df.join3(a, b, c)` | `a`, `b`, `c` | | `df.race(a, b)` | `a`, `b` | | `df.if(cond, then, else)` | `cond`, `then`, `else` | -| `df.loop(body, cond)` | `body`, `cond` | +| `df.loop(body, condition, continue_on_failure)` | `body`, `condition` | | `df.start(fut, label)` | `fut` | | All others | No auto-wrap (literals only) | diff --git a/docs/grammar.md b/docs/grammar.md index 20411881..10cb5ce2 100644 --- a/docs/grammar.md +++ b/docs/grammar.md @@ -61,11 +61,18 @@ node_function ::= df.sql( QUERY ) | df.race( expression, expression ) | df.seq( expression, expression ) | df.if( condition, then_expr, else_expr ) - | df.loop( expression [, condition] ) + | df.loop( + expression + [, expression] + [, continue_on_failure => BOOLEAN] + ) | df.break( [value] ) | df.as( expression, NAME ) ``` +> **Experimental:** The `continue_on_failure` syntax is subject to change in +> future releases. + ### Terminals ```ebnf @@ -78,6 +85,7 @@ METHOD ::= 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH' BODY ::= JSON string (supports $variable substitution) HEADERS ::= JSONB object TIMEOUT ::= positive integer (seconds) +BOOLEAN ::= true | false STRING ::= single-quoted SQL string ``` @@ -166,6 +174,20 @@ df.loop( 'SELECT count(*) > 0 FROM queue' -- while queue has items ) +-- Infinite loop with failure-isolated iterations +df.loop( + df.wait_for_schedule('*/5 * * * *') + ~> 'CALL refresh_search_index()', + continue_on_failure => true +) + +-- Conditional loop with failure-isolated iterations +df.loop( + 'SELECT process_item()', + 'SELECT count(*) > 0 FROM queue', + continue_on_failure => true +) + -- Loop with df.break() to exit df.loop( 'SELECT process_batch()' |=> 'batch' @@ -300,4 +322,3 @@ compensate_expr ::= atom_expr [ '<->' atom_expr ] ``` See [spec-compensation.md](spec-compensation.md) for details. - diff --git a/docs/loop_rework_problems.md b/docs/loop_rework_problems.md index 44ed7681..7953e605 100644 --- a/docs/loop_rework_problems.md +++ b/docs/loop_rework_problems.md @@ -54,9 +54,10 @@ let new_input = SubtreeInput { The graph snapshot is also copied into every JOIN/RACE child input. A large graph or named result is therefore persisted repeatedly across loop generations -and child histories. `MAX_LOOP_ITERATIONS` limits generations to 100,000, but -that is not a meaningful byte bound: a 1 MB carried result can still produce -storage measured in tens of gigabytes before the iteration guard trips. +and child histories. `MAX_LOOP_ITERATIONS` limits generations to 8,388,608 +(`2^23`), but that is not a meaningful byte bound: a 1 MB carried result can +still produce roughly 8.4 TB of payload copies (8 TiB for a 1 MiB payload) +before the iteration guard trips, excluding engine-record and graph overhead. Child engine records also remain until the root instance is retired. A nested loop can create a child under each outer generation, so the retained child @@ -258,8 +259,9 @@ between parent and descendant stamps at equal generations. `MAX_LOOP_ITERATIONS` is enforced per orchestration instance. A non-root inner loop starts in its own child and receives its own counter, while an outer loop can spawn a new inner child under every outer generation. Two loops that each -stay under the 100,000-iteration cap can therefore produce a combinatorial -amount of work and retained child state. +stay under the 8,388,608 (`2^23`)-iteration cap can therefore produce up to +`2^46` (70,368,744,177,664) inner iterations, plus outer-loop work, and retain +a correspondingly large child-state population. The one-second floor limits the rate of each individual loop but does not place a workflow-wide bound on nested work. The existing nested-loop E2E test proves diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index a06efd5a..116d5c90 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -203,6 +203,36 @@ gate, so they never need to be added to the exclude list. Each schema-changing PR should add a section here documenting what changed, what the upgrade script handles, and any backward compatibility considerations. +### 0.2.8 + +- `sql/pg_durable--0.2.7--0.2.8.sql` renames `df.loop(text, text)` to + `df._loop_legacy(text, text)`, preserving its function OID and dependent + objects, then creates the single public + `df.loop(text, text DEFAULT NULL, boolean DEFAULT false)` signature. + `_loop_legacy` is an internal upgrade-compatibility object, not a user-facing + alternative. +- For Scenario B1, the new `.so` retains `loop_fn_wrapper`, so every supported + schema that has not run `ALTER EXTENSION UPDATE` can continue calling its + cataloged `df.loop(text, text)` function safely. +- For Scenario A, fresh installs and upgraded schemas both contain the unified + public signature and the internal `_loop_legacy` object, so their schema + snapshots remain equal while upgrade dependencies stay attached to the + renamed function OID. +- Existing LOOP graphs have no `continue_on_failure` key and remain inline and + fail-fast. Only newly constructed opted-in loops schedule an iteration child. +- For opted-in conditional loops, a successful body is followed by condition + evaluation; a consumed typed body activity failure skips the condition and + starts the next iteration. All errors returned by body SQL, HTTP, and + multipart activities are consumable. Condition failures, malformed + graph/protocol data, unrecognized child errors, child-ID collisions, and + orchestration/runtime failures remain fatal. +- Raises the loop backstop for all loops from 100,000 to 8,388,608 (`2^23`), + which is about 80 years at five-minute ticks. +- Replay is unchanged before iteration 100,000. At the old boundary, a history + that recorded the previous terminal-failure path cannot replay under the new + binary, which continues toward the higher backstop instead. Drain such + long-running loops before upgrade when continuity is required. + ### v0.2.6 → v0.2.7 #### Transaction-aware graph admission diff --git a/scripts/test-upgrade.sh b/scripts/test-upgrade.sh index 6975e314..bd04f63f 100755 --- a/scripts/test-upgrade.sh +++ b/scripts/test-upgrade.sh @@ -774,6 +774,10 @@ test_b1_dsl_construction() { assert_sql_contains "SELECT df.sql('SELECT 1');" '"node_type":"SQL"' } +test_b1_conditional_loop() { + assert_sql_contains "SELECT df.loop('SELECT 1', 'SELECT false');" '"node_type":"LOOP"' +} + test_b1_dsl_chain() { assert_sql_contains "SELECT df.sql('SELECT 1') ~> df.sql('SELECT 2');" '"node_type":"THEN"' } @@ -904,6 +908,7 @@ else run_test "B1 [v${B1_VERSION}]: df.getvar()" test_b1_getvar run_test "B1 [v${B1_VERSION}]: df.version()" test_b1_version run_test "B1 [v${B1_VERSION}]: df.sql() construction" test_b1_dsl_construction + run_test "B1 [v${B1_VERSION}]: df.loop(body, condition)" test_b1_conditional_loop run_test "B1 [v${B1_VERSION}]: DSL chain (~>)" test_b1_dsl_chain run_test "B1 [v${B1_VERSION}]: conditional operators (?>/!>)" test_b1_conditional_operators run_test "B1 [v${B1_VERSION}]: df.start()/wait_for_completion()" test_b1_start_and_complete @@ -942,6 +947,7 @@ test_b2_data_survives_upgrade() { assert_sql_equals "SELECT df.clearvars();" "OK" || return 1 assert_sql_equals "SELECT df.setvar('b2_key', 'b2_value');" "OK" || return 1 + run_sql_capture "CREATE VIEW test_loop_upgrade_dependency AS SELECT df.loop('SELECT 1', NULL::text) AS graph;" >/dev/null || return 1 B2_PRE_INSTANCE_ID=$(run_sql_capture "SELECT df.start('INSERT INTO test_upgrade_b2_log (kind, msg) VALUES (''pre'', ''{b2_key}'') RETURNING msg', 'b2-pre-upgrade');") || return 1 B2_INFLIGHT_INSTANCE_ID=$(run_sql_capture "SELECT df.start(df.sleep(2) ~> 'SELECT ''b2-running'' AS value', 'b2-inflight');") || return 1 @@ -977,6 +983,13 @@ test_b2_inflight_work_after_upgrade() { assert_sql_contains "SELECT df.result('${B2_INFLIGHT_INSTANCE_ID}');" "b2-running" } +test_b2_loop_dependency_survives_upgrade() { + assert_sql_equals "SELECT graph LIKE '%\"node_type\":\"LOOP\"%' FROM test_loop_upgrade_dependency;" "t" && + assert_sql_equals "SELECT to_regprocedure('df._loop_legacy(text,text)') IS NOT NULL;" "t" && + assert_sql_equals "SELECT to_regprocedure('df.loop(text,text,boolean)') IS NOT NULL;" "t" && + assert_sql_equals "SELECT (df.loop('SELECT 1', 'SELECT false', continue_on_failure => true)::jsonb ->> 'query') LIKE '%\"continue_on_failure\":true%';" "t" +} + test_b2_new_data_after_upgrade() { assert_sql_equals "SELECT df.setvar('b2_new_key', 'new_value');" "OK" || return 1 assert_sql_equals "SELECT df.getvar('b2_new_key');" "new_value" || return 1 @@ -1021,6 +1034,7 @@ if [ "$HAS_COMPAT_PREV" = true ]; then run_test "B2: Pre-upgrade data survives ALTER EXTENSION UPDATE" test_b2_data_survives_upgrade run_test "B2: Pre-upgrade instance remains queryable" test_b2_pre_upgrade_instance_after_upgrade run_test "B2: In-flight work completes after upgrade" test_b2_inflight_work_after_upgrade + run_test "B2: Loop dependency and unified API survive upgrade" test_b2_loop_dependency_survives_upgrade run_test "B2: New data and execution after upgrade" test_b2_new_data_after_upgrade run_test "B2: df.grant_usage() works and df.debug_connection() is gone after upgrade" test_b2_grant_usage_after_upgrade fi diff --git a/sql/pg_durable--0.2.7--0.2.8.sql b/sql/pg_durable--0.2.7--0.2.8.sql index 471f893a..e23a62b6 100644 --- a/sql/pg_durable--0.2.7--0.2.8.sql +++ b/sql/pg_durable--0.2.7--0.2.8.sql @@ -6,5 +6,12 @@ -- See docs/upgrade-testing.md for the upgrade-script and backward-compatibility -- requirements (Scenario A / B1 / B2). -- --- No schema changes yet for 0.2.8. Add DDL below as the 0.2.8 cycle lands --- extension-schema changes. +ALTER FUNCTION df."loop"(TEXT, TEXT) RENAME TO "_loop_legacy"; + +CREATE FUNCTION df."loop"( + "body" TEXT, + "condition" TEXT DEFAULT NULL, + "continue_on_failure" bool DEFAULT false +) RETURNS TEXT +LANGUAGE c +AS 'MODULE_PATHNAME', 'loop_with_policy_wrapper'; diff --git a/src/dsl.rs b/src/dsl.rs index 7e399bef..13ce8851 100644 --- a/src/dsl.rs +++ b/src/dsl.rs @@ -15,7 +15,7 @@ use crate::client::start_durable_function; use crate::types::{ flatten_graph, get_max_new_transaction_starts, get_new_transaction_start_timeout, mark_non_future_helper_call, short_id, validate_result_name, Durofut, FunctionInput, - MaterializedNode, + LoopConfig, MaterializedNode, }; /// Check if we're running inside a workflow context (background worker connection). @@ -309,7 +309,8 @@ pub fn wait_for_schedule(cron_expr: &str) -> String { /// Creates a loop node. /// /// With one argument: repeats the body indefinitely (infinite loop). -/// With two arguments: repeats while the condition is true (while loop). +/// With a text condition: repeats while the condition is true (while loop). +/// With `continue_on_failure => true`: continues after body activity failures. /// /// The body and condition can be either Durofut JSON or plain SQL strings (auto-wrapped). /// The condition is evaluated after each iteration (do-while semantics). @@ -321,21 +322,48 @@ pub fn wait_for_schedule(cron_expr: &str) -> String { /// /// -- While loop - continues while condition is true /// df.loop('SELECT process_item()', 'SELECT count(*) > 0 FROM queue') +/// +/// -- Infinite loop that continues after a body failure +/// df.loop( +/// df.wait_for_schedule('*/5 * * * *') ~> 'SELECT process_batch()', +/// continue_on_failure => true +/// ) /// ``` -#[pg_extern(name = "loop", schema = "df")] -pub fn loop_fn(body: &str, condition: default!(Option<&str>, "NULL")) -> String { +fn build_loop(body: &str, condition: Option<&str>, continue_on_failure: bool) -> String { let body_fut = Durofut::ensure(body); let condition_node = condition.map(|cond| Durofut::ensure(cond).into_raw()); + let query = continue_on_failure.then(|| { + serde_json::to_string(&LoopConfig { + continue_on_failure: true, + condition_node: None, + }) + .expect("LoopConfig serialization cannot fail") + }); Durofut { node_type: "LOOP".to_string(), left_node: Some(body_fut.into_raw()), condition_node, + query, ..Default::default() } .to_json() } +#[pg_extern(name = "_loop_legacy", schema = "df")] +pub fn loop_fn(body: &str, condition: default!(Option<&str>, "NULL")) -> String { + build_loop(body, condition, false) +} + +#[pg_extern(name = "loop", schema = "df")] +pub fn loop_with_policy( + body: &str, + condition: default!(Option<&str>, "NULL"), + continue_on_failure: default!(bool, "false"), +) -> String { + build_loop(body, condition, continue_on_failure) +} + /// Creates a break node that exits the enclosing loop. /// /// When executed, the loop terminates and returns the provided value (or null). diff --git a/src/explain.rs b/src/explain.rs index ae1e4d44..3f828515 100644 --- a/src/explain.rs +++ b/src/explain.rs @@ -6,7 +6,7 @@ use pgrx::prelude::*; use std::collections::HashMap; -use crate::types::{flatten_graph, Durofut, MaterializedNode}; +use crate::types::{flatten_graph, Durofut, LoopConfig, MaterializedNode}; /// Represents a node for visualization #[derive(Debug, Clone)] @@ -511,12 +511,11 @@ fn render_children( ) { match node.node_type.as_str() { "LOOP" => { - // Check for while-condition let condition_id = node .query .as_ref() - .and_then(|q| serde_json::from_str::(q).ok()) - .and_then(|v| v["condition_node"].as_str().map(|s| s.to_string())); + .and_then(|query| serde_json::from_str::(query).ok()) + .and_then(|config| config.condition_node); if let Some(ref body_id) = node.left_node { output.push_str(&format!("{prefix}↻ body:\n")); @@ -702,18 +701,18 @@ fn format_node_display(node: &ExplainNode) -> String { format!("SIGNAL '{signal_name}'{timeout_str}{name_suffix}") } "LOOP" => { - // Check if it has a while condition - let has_condition = node + let config = node .query .as_ref() - .and_then(|q| serde_json::from_str::(q).ok()) - .map(|cfg| cfg["condition_node"].is_string()) - .unwrap_or(false); - if has_condition { - format!("LOOP (while){name_suffix}") - } else { - format!("LOOP (infinite){name_suffix}") - } + .and_then(|query| serde_json::from_str::(query).ok()) + .unwrap_or_default(); + let kind = match (config.condition_node.is_some(), config.continue_on_failure) { + (false, false) => "infinite", + (true, false) => "while", + (false, true) => "infinite, continue on failure", + (true, true) => "while, continue on failure", + }; + format!("LOOP ({kind}){name_suffix}") } "BREAK" => { // Parse config to get break value diff --git a/src/lib.rs b/src/lib.rs index 985ce7c0..90dfd9a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -818,7 +818,7 @@ CREATE OPERATOR @> ( dsl::join, dsl::race, dsl::if_fn, - dsl::loop_fn + dsl::loop_with_policy ] ); @@ -1123,6 +1123,89 @@ mod tests { assert!(fut.query.is_none()); } + #[pg_test] + fn loop_call_forms() { + use crate::types::LoopConfig; + + let infinite = Spi::get_one::("SELECT df.loop('SELECT work()')") + .unwrap() + .unwrap(); + let conditional = + Spi::get_one::("SELECT df.loop('SELECT work()', 'SELECT keep_going()')") + .unwrap() + .unwrap(); + let conditional_disabled = Spi::get_one::( + "SELECT df.loop( + 'SELECT work()', + 'SELECT keep_going()', + continue_on_failure => false + )", + ) + .unwrap() + .unwrap(); + let legacy_conditional = Spi::get_one::( + "SELECT df._loop_legacy('SELECT work()', 'SELECT keep_going()')", + ) + .unwrap() + .unwrap(); + let resilient = + Spi::get_one::("SELECT df.loop('SELECT work()', continue_on_failure => true)") + .unwrap() + .unwrap(); + let disabled = + Spi::get_one::("SELECT df.loop('SELECT work()', continue_on_failure => false)") + .unwrap() + .unwrap(); + let resilient_conditional = Spi::get_one::( + "SELECT df.loop( + 'SELECT work()', + 'SELECT keep_going()', + continue_on_failure => true + )", + ) + .unwrap() + .unwrap(); + + let public_count = Spi::get_one::( + "SELECT count(*) + FROM pg_proc p + JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = 'df' AND p.proname = 'loop'", + ) + .unwrap() + .unwrap(); + assert_eq!(public_count, 1); + + let legacy_exists = Spi::get_one::( + "SELECT to_regprocedure('df._loop_legacy(text,text)') IS NOT NULL", + ) + .unwrap() + .unwrap(); + assert!(legacy_exists); + + let infinite_fut = Durofut::from_json(&infinite); + let conditional_fut = Durofut::from_json(&conditional); + let resilient_fut = Durofut::from_json(&resilient); + let resilient_conditional_fut = Durofut::from_json(&resilient_conditional); + + assert_eq!(disabled, infinite); + assert_eq!(conditional, legacy_conditional); + assert_eq!(conditional_disabled, legacy_conditional); + assert!(infinite_fut.query.is_none()); + assert!(conditional_fut.query.is_none()); + assert!(conditional_fut.condition_node.is_some()); + + let resilient_config: LoopConfig = + serde_json::from_str(resilient_fut.query.as_deref().unwrap()).unwrap(); + assert!(resilient_config.continue_on_failure); + assert!(resilient_fut.condition_node.is_none()); + + let combined_config: LoopConfig = + serde_json::from_str(resilient_conditional_fut.query.as_deref().unwrap()).unwrap(); + assert!(combined_config.continue_on_failure); + assert!(resilient_conditional_fut.condition_node.is_some()); + } + #[pg_test] fn test_break_creates_break_node() { let json = crate::dsl::break_fn(None); @@ -2092,6 +2175,41 @@ mod tests { assert!(result.contains("body"), "Expected body section: {result}"); } + #[pg_test] + fn test_explain_expression_loop_modes_via_sql() { + let cases = [ + ("df.loop(df.sql('SELECT 1'))", "LOOP (infinite)"), + ( + "df.loop(df.sql('SELECT 1'), df.sql('SELECT false'))", + "LOOP (while)", + ), + ( + "df.loop(df.sql('SELECT 1'), continue_on_failure => true)", + "LOOP (infinite, continue on failure)", + ), + ( + "df.loop( + df.sql('SELECT 1'), + df.sql('SELECT false'), + continue_on_failure => true + )", + "LOOP (while, continue on failure)", + ), + ]; + + for (expression, expected) in cases { + let result = + Spi::get_one::(&format!("SELECT df.explain($dsl${expression}$dsl$)")) + .unwrap() + .unwrap(); + assert!( + result.contains(expected), + "Expected {expected} for {expression}: {result}" + ); + assert!(result.contains("body"), "Expected body section: {result}"); + } + } + #[pg_test] fn test_explain_expression_if() { let result = crate::explain::explain( diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index 89af983b..b276cec3 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -20,7 +20,7 @@ use crate::activities; use crate::activities::load_function_graph::{TransactionAwareLoadInput, TransactionGraphProbe}; use crate::types::{ evaluate_condition, string_map_to_json, substitute_all, substitute_all_raw, FunctionGraph, - FunctionInput, FunctionNode, SystemVars, + FunctionInput, FunctionNode, LoopConfig, SystemVars, }; /// Orchestration name for ExecuteFunctionGraph @@ -34,6 +34,8 @@ pub const SUBTREE_NAME: &str = "pg_durable::orchestration::execute-subtree"; struct ExecutionContext { vars: HashMap, label: Option, + /// Whether an enclosing loop may consume application failures from this subtree. + failure_isolated: bool, /// Loop iteration counter (persisted across continue_as_new generations). loop_iteration: u64, /// Node id at the root of the *current* orchestration's node tree: `graph.root_node_id` @@ -85,26 +87,45 @@ struct SubtreeInput { label: Option, #[serde(default)] iteration: u64, + #[serde(default, skip_serializing_if = "is_false")] + failure_isolated: bool, +} + +fn is_false(value: &bool) -> bool { + !*value } /// Control-flow-aware error type returned by every node handler. /// /// `Break` is **not** a failure: it unwinds through compound nodes (THEN, IF, JOIN, /// RACE, and the subtree boundary) via the `?` operator until the nearest enclosing -/// `execute_loop_node` catches it. `Failure` is a genuine error that propagates to the -/// orchestration result. Encoding break this way means forgetting to propagate it is a -/// compile error rather than a silently-ignored value (see issue #148 / #132). +/// `execute_loop_node` catches it. `Application` is an expected body-activity failure; +/// `Failure` is a graph, protocol, or runtime fault. Encoding these distinctions in the +/// type prevents a broad catch from consuming failures it does not recognize. #[derive(Debug)] enum NodeError { /// A `df.break()` signal carrying its (already-stringified) value, caught by the loop. Break(String), + /// An expected application failure returned by a workflow activity. + Application(String), /// A real failure; propagates to the orchestration's `Err` result. Failure(String), } +impl NodeError { + fn message(&self) -> &str { + match self { + NodeError::Break(value) | NodeError::Application(value) | NodeError::Failure(value) => { + value + } + } + } +} + /// All helper functions (`substitute_all`, `evaluate_condition`) and activity scheduling -/// return `Result<_, String>`. This conversion lets `?` turn those `String` errors into -/// `NodeError::Failure` automatically, so only genuine control flow needs explicit handling. +/// return `Result<_, String>`. This conversion lets `?` turn structural/configuration errors +/// into `NodeError::Failure` automatically. Activity failures are mapped explicitly to +/// `NodeError::Application` at their scheduling boundaries. impl From for NodeError { fn from(e: String) -> Self { NodeError::Failure(e) @@ -293,6 +314,86 @@ enum SubtreeControl { Break, } +#[derive(Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde( + tag = "pg_durable_subtree_failure", + content = "message", + rename_all = "snake_case" +)] +enum SubtreeFailure { + Application(String), +} + +fn encode_subtree_application_failure(message: &str) -> String { + serde_json::to_string(&SubtreeFailure::Application(message.to_string())) + .expect("SubtreeFailure serialization cannot fail") +} + +fn decode_subtree_application_failure(error: &str) -> Option { + let failure: SubtreeFailure = serde_json::from_str(error).ok()?; + match failure { + SubtreeFailure::Application(message) => Some(message), + } +} + +fn classify_subtree_failure(error: String) -> NodeError { + match decode_subtree_application_failure(&error) { + Some(message) => NodeError::Application(message), + None => NodeError::Failure(error), + } +} + +fn contextualize_subtree_failure(context: &str, error: String) -> NodeError { + match classify_subtree_failure(error) { + NodeError::Application(message) => NodeError::Application(format!("{context}: {message}")), + NodeError::Failure(message) => NodeError::Failure(format!("{context}: {message}")), + NodeError::Break(_) => unreachable!("subtree failures cannot carry break control flow"), + } +} + +fn retain_higher_priority_join_error( + selected: &mut Option<(NodeError, Option)>, + candidate: NodeError, + loop_node_id: Option, +) { + let priority = |error: &NodeError| match error { + NodeError::Application(_) => 0, + NodeError::Break(_) => 1, + NodeError::Failure(_) => 2, + }; + let should_replace = selected + .as_ref() + .map(|(current, _)| priority(&candidate) > priority(current)) + .unwrap_or(true); + if should_replace { + *selected = Some((candidate, loop_node_id)); + } +} + +#[derive(Debug, PartialEq, Eq)] +enum FailureIsolatedBodyOutcome { + Succeeded(String), + ApplicationFailed, + Break(String), +} + +fn classify_isolated_body_result( + result: Result, + results: &mut HashMap, +) -> Result { + match result { + Ok(raw) => match parse_subtree_envelope(&raw, "LOOP iteration", results) { + Ok(body_result) => Ok(FailureIsolatedBodyOutcome::Succeeded(body_result)), + Err(NodeError::Break(value)) => Ok(FailureIsolatedBodyOutcome::Break(value)), + Err(error) => Err(error), + }, + Err(error) => match decode_subtree_application_failure(&error) { + Some(_) => Ok(FailureIsolatedBodyOutcome::ApplicationFailed), + None => Err(NodeError::Failure(error)), + }, + } +} + /// Envelope returned by `execute_subtree` containing the SQL result and the updated /// named-results map so the parent orchestration can merge any new entries after join/race. /// `control` carries a `df.break()` signal back across the sub-orchestration boundary so the @@ -367,11 +468,11 @@ async fn finalize_instance_status(ctx: &OrchestrationContext, instance_id: &str, /// Internally every node handler returns `NodeResult`, where `NodeError::Break` is /// **intentional control flow** (a `df.break()` signal), not a failure. Break unwinds /// through compound nodes via `?` and is caught by the nearest enclosing -/// `execute_loop_node`; only `NodeError::Failure` represents a genuine error. This -/// boundary collapses the typed result back to `Result`: a `Break` -/// that reaches here was used outside `df.loop()`, so it is surfaced as a clear failure -/// rather than completing with a control-flow value. Callers should treat the returned -/// `Err` strictly as a failure and must not add retry/recovery logic for break. +/// `execute_loop_node`; `NodeError::Application` distinguishes activity failures that an +/// opted-in loop may consume, while `NodeError::Failure` carries graph, protocol, and runtime +/// faults. This boundary collapses both failure variants back to `Result`: +/// a `Break` that reaches here was used outside `df.loop()`, so it is surfaced as a clear +/// failure rather than completing with a control-flow value. pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result { let input: FunctionInput = serde_json::from_str(&input_json) .map_err(|e| format!("Invalid orchestration input: {e}"))?; @@ -444,6 +545,7 @@ pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result Result = match function_outcome { Ok(result) => Ok(result), - Err(NodeError::Failure(err)) => Err(err), + Err(NodeError::Application(err)) | Err(NodeError::Failure(err)) => Err(err), Err(NodeError::Break(_)) => Err( "df.break() was called outside of a loop. df.break() may only be used inside df.loop()." .to_string(), @@ -548,6 +650,7 @@ pub async fn execute_subtree( let exec_ctx = ExecutionContext { vars, label: input.label.clone(), + failure_isolated: input.failure_isolated, loop_iteration: input.iteration, subtree_root: input.node_id.clone(), continuation: Continuation::Subtree, @@ -556,7 +659,9 @@ pub async fn execute_subtree( // Build the envelope carrying the result, the updated named-results map, and a typed // control signal. A `Break` inside the subtree is re-encoded as `control: Break` (not a // sentinel smuggled inside `result`) so the parent can re-raise it as `NodeError::Break`. - // A genuine `Failure` propagates as `Err` across the sub-orchestration boundary. + // An expected activity failure is encoded as a typed application-failure payload before + // crossing the sub-orchestration boundary. Other failures remain unrecognized strings so + // callers cannot accidentally consume graph, protocol, collision, or runtime faults. // // When the root node is a loop that needs another iteration it calls `continue_as_new`, // whose future never resolves — so none of the arms below run for a continuing @@ -589,7 +694,10 @@ pub async fn execute_subtree( results, } } - Err(NodeError::Failure(e)) => return Err(e), + Err(NodeError::Application(error)) => { + return Err(encode_subtree_application_failure(&error)); + } + Err(NodeError::Failure(error)) => return Err(error), }; serde_json::to_string(&envelope) @@ -619,6 +727,7 @@ fn build_subtree_input( ), label: exec_ctx.label.clone(), iteration: 0, + failure_isolated: exec_ctx.failure_isolated, }; serde_json::to_string(&input).map_err(|e| format!("Failed to serialize subtree input: {e}")) } @@ -630,12 +739,21 @@ fn build_subtree_input( /// a complete parent-to-child lineage and per-generation uniqueness: the parent execution id /// advances on every loop `continue_as_new`, while the child root node id distinguishes sibling /// branches. df.instance_nodes() and the write fence walk the full composed lineage. +fn compose_subtree_instance_id( + parent_instance_id: &str, + parent_execution_id: &str, + child_root_node_id: &str, +) -> String { + format!("{parent_instance_id}::{parent_execution_id}::{child_root_node_id}") +} + fn subtree_instance_id(ctx: &OrchestrationContext, child_root_node_id: &str) -> String { - format!( - "{}::{}::{}", - ctx.instance_id(), - ctx.execution_id(), - child_root_node_id + let parent_instance_id = ctx.instance_id(); + let parent_execution_id = ctx.execution_id().to_string(); + compose_subtree_instance_id( + &parent_instance_id, + &parent_execution_id, + child_root_node_id, ) } @@ -700,13 +818,13 @@ async fn execute_function_node_with_vars( // Update node with final status and result. A `Break` is control flow rather than a // failure: record the node as completed (carrying the break value) so observability is - // unchanged from when break travelled as a normal `Ok` sentinel. Only `Failure` marks - // the node failed. All three arms schedule exactly one `update_node_status`, so collapse + // unchanged from when break travelled as a normal `Ok` sentinel. Both failure variants + // mark the node failed. All arms schedule exactly one `update_node_status`, so collapse // them to a single (status, result) pair to keep the recorded history identical. let (status, status_result) = match &execute_result { Ok(result) => ("completed", result.as_str()), Err(NodeError::Break(value)) => ("completed", value.as_str()), - Err(NodeError::Failure(err)) => ("failed", err.as_str()), + Err(NodeError::Application(err)) | Err(NodeError::Failure(err)) => ("failed", err.as_str()), }; let status_input = serde_json::json!({ "node_id": node_id, @@ -787,7 +905,8 @@ async fn execute_sql_node( let result = ctx .schedule_activity(activities::execute_sql::NAME, input.to_string()) - .await?; + .await + .map_err(NodeError::Application)?; if let Some(name) = &node.result_name { ctx.trace_info(format!("Storing result as ${name}")); @@ -930,8 +1049,8 @@ const LOOP_MIN_ITER_DURATION: Duration = Duration::from_secs(1); /// Maximum loop iterations before the orchestration is forcibly terminated. /// This prevents runaway infinite loops from consuming resources indefinitely. -/// At the minimum 1-second rate limit, this allows ~27 hours of looping. -const MAX_LOOP_ITERATIONS: u64 = 100_000; +/// At the minimum 1-second rate limit, this allows over 97 days of looping. +const MAX_LOOP_ITERATIONS: u64 = 1 << 23; /// Stamp a loop node's status from its *parent* orchestration. /// @@ -985,7 +1104,7 @@ async fn fail_loop_child_future( ctx: &OrchestrationContext, graph: &FunctionGraph, loop_node_id: &str, - error: String, + error: NodeError, ) -> NodeResult { let child_stamp = format!("{}::1", subtree_instance_id(ctx, loop_node_id)); stamp_loop_node( @@ -993,11 +1112,19 @@ async fn fail_loop_child_future( &graph.instance_id, loop_node_id, "failed", - Some(&error), + Some(error.message()), &child_stamp, ) .await; - Err(NodeError::Failure(error)) + Err(error) +} + +fn parse_loop_config(node: &FunctionNode, node_id: &str) -> Result { + match node.query.as_deref() { + Some(query) => serde_json::from_str(query) + .map_err(|e| format!("LOOP node {node_id}: failed to parse config: {e}")), + None => Ok(LoopConfig::default()), + } } /// Run one iteration of a loop body (and its optional while-condition). @@ -1012,11 +1139,11 @@ async fn run_loop_iteration( ctx: &OrchestrationContext, graph: &FunctionGraph, node: &FunctionNode, - loop_node_id: &str, body_id: &str, + condition_node: Option<&str>, results: &mut HashMap, exec_ctx: &ExecutionContext, -) -> Result, String> { +) -> Result, NodeError> { // 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. @@ -1030,52 +1157,92 @@ async fn run_loop_iteration( store_named_result(ctx, node, &break_value, results, "LOOP"); return Ok(Some(break_value)); } - Err(NodeError::Failure(e)) => return Err(e), + Err(NodeError::Application(error)) => return Err(NodeError::Application(error)), + Err(NodeError::Failure(error)) => return Err(NodeError::Failure(error)), }; - // While-condition: if present and false, exit the loop. - if let Some(ref config_str) = node.query { - let config: serde_json::Value = serde_json::from_str(config_str).map_err(|e| { - // M8: Malformed condition config should fail the loop rather than - // silently creating an infinite loop without exit condition. - format!("LOOP node {loop_node_id}: failed to parse condition config: {e}") - })?; - if let Some(condition_node_id) = config["condition_node"].as_str() { - ctx.trace_info("Evaluating loop condition"); - let condition_result = match execute_function_node_with_vars( - ctx, - graph, - condition_node_id, - results, - exec_ctx, - ) - .await + Box::pin(evaluate_loop_condition( + ctx, + graph, + node, + condition_node, + body_result, + results, + exec_ctx, + )) + .await +} + +async fn evaluate_loop_condition( + ctx: &OrchestrationContext, + graph: &FunctionGraph, + node: &FunctionNode, + condition_node: Option<&str>, + body_result: String, + results: &mut HashMap, + exec_ctx: &ExecutionContext, +) -> Result, NodeError> { + if let Some(condition_node_id) = condition_node { + ctx.trace_info("Evaluating loop condition"); + let condition_result = + match execute_function_node_with_vars(ctx, graph, condition_node_id, results, exec_ctx) + .await { - Ok(v) => v, + Ok(value) => value, Err(NodeError::Break(break_value)) => { store_named_result(ctx, node, &break_value, results, "LOOP"); return Ok(Some(break_value)); } - Err(NodeError::Failure(e)) => return Err(e), + // A condition controls loop execution; no enclosing resilient loop may + // consume its failure as an ordinary body activity error. + Err(NodeError::Application(error)) => return Err(NodeError::Failure(error)), + Err(NodeError::Failure(error)) => return Err(NodeError::Failure(error)), }; - // 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(Some(body_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(Some(body_result)); } } Ok(None) } +async fn run_failure_isolated_body( + ctx: &OrchestrationContext, + graph: &FunctionGraph, + loop_node_id: &str, + body_id: &str, + results: &mut HashMap, + exec_ctx: &ExecutionContext, +) -> Result { + let mut child_exec_ctx = exec_ctx.clone(); + child_exec_ctx.failure_isolated = true; + let child_input = build_subtree_input(graph, body_id, results, &child_exec_ctx)?; + let child_id = subtree_instance_id(ctx, body_id); + + ctx.trace_info(format!( + "Starting failure-isolated iteration for LOOP node {loop_node_id} as child {child_id}" + )); + + let child_result = ctx + .schedule_sub_orchestration_with_id(SUBTREE_NAME, child_id, child_input) + .await; + + let outcome = classify_isolated_body_result(child_result, results)?; + if outcome == FailureIsolatedBodyOutcome::ApplicationFailed { + ctx.trace_warn(format!( + "LOOP node {loop_node_id} iteration application failure; continuing" + )); + } + Ok(outcome) +} + /// Execute a loop node inline, driving the *current* orchestration's `continue_as_new`. /// /// Only a loop sitting at `exec_ctx.subtree_root` reaches this function; a deeper loop is @@ -1107,6 +1274,7 @@ async fn execute_loop_node( .left_node .as_ref() .ok_or_else(|| format!("LOOP node {node_id} has no body"))?; + let config = parse_loop_config(node, node_id).map_err(NodeError::Failure)?; // Capture the iteration start time so we can rate-limit `continue_as_new` // below. `utc_now()` is duroxide's deterministic clock (recorded in @@ -1115,12 +1283,44 @@ async fn execute_loop_node( ctx.trace_info("Executing loop iteration"); - if let Some(final_result) = Box::pin(run_loop_iteration( - ctx, graph, node, node_id, body_id, results, exec_ctx, - )) - .await - .map_err(NodeError::Failure)? - { + let final_result = if config.continue_on_failure { + match Box::pin(run_failure_isolated_body( + ctx, graph, node_id, body_id, results, exec_ctx, + )) + .await? + { + FailureIsolatedBodyOutcome::Succeeded(body_result) => { + Box::pin(evaluate_loop_condition( + ctx, + graph, + node, + config.condition_node.as_deref(), + body_result, + results, + exec_ctx, + )) + .await? + } + FailureIsolatedBodyOutcome::ApplicationFailed => None, + FailureIsolatedBodyOutcome::Break(value) => { + store_named_result(ctx, node, &value, results, "LOOP"); + Some(value) + } + } + } else { + Box::pin(run_loop_iteration( + ctx, + graph, + node, + body_id, + config.condition_node.as_deref(), + results, + exec_ctx, + )) + .await? + }; + + if let Some(final_result) = final_result { return Ok(final_result); } @@ -1193,6 +1393,7 @@ async fn execute_loop_node( ), label: exec_ctx.label.clone(), iteration: next_iteration, + failure_isolated: exec_ctx.failure_isolated, }; serde_json::to_string(&new_input) .map_err(|e| format!("Failed to serialize loop input: {e}"))? @@ -1265,7 +1466,7 @@ async fn execute_loop_suborchestration( ctx, graph, node_id, - format!("Loop sub-orchestration failed: {e}"), + contextualize_subtree_failure("Loop sub-orchestration failed", e), ) .await; } @@ -1544,41 +1745,63 @@ async fn execute_join_node( // Each Ok value is a JSON envelope {"result": "...", "results": {...}} produced by // execute_subtree; unwrap it and merge the branch's named results into the parent map. let mut join_results: Vec = Vec::new(); + let mut selected_error: Option<(NodeError, Option)> = None; for (i, result) in results_vec.into_iter().enumerate() { match result { Ok(r) => { let context = format!("JOIN branch {}", i + 1); - // A break in any branch surfaces as `NodeError::Break` from - // `parse_subtree_envelope` and unwinds via `?` to the enclosing loop. - let branch_result = parse_subtree_envelope(&r, &context, results)?; - let parsed = serde_json::from_str::(&branch_result) - .map_err(|e| format!("JOIN branch {} result parse error: {}", i + 1, e))?; - join_results.push(parsed); + match parse_subtree_envelope(&r, &context, results) { + Ok(branch_result) => { + match serde_json::from_str::(&branch_result) { + Ok(parsed) => join_results.push(parsed), + Err(e) => { + let error = NodeError::Failure(format!( + "JOIN branch {} result parse error: {}", + i + 1, + e + )); + if !exec_ctx.failure_isolated { + return Err(error); + } + retain_higher_priority_join_error(&mut selected_error, error, None); + } + } + } + Err(error) => { + if !exec_ctx.failure_isolated { + return Err(error); + } + retain_higher_priority_join_error(&mut selected_error, error, None); + } + } } Err(e) => { - if graph + let error = + contextualize_subtree_failure(&format!("JOIN branch {} failed", i + 1), e); + let loop_node_id = graph .nodes .get(&branch_ids[i]) .map(|branch| branch.node_type.eq_ignore_ascii_case("loop")) .unwrap_or(false) - { - return fail_loop_child_future( - ctx, - graph, - &branch_ids[i], - format!("JOIN branch {} failed: {}", i + 1, e), - ) - .await; + .then(|| branch_ids[i].clone()); + if !exec_ctx.failure_isolated { + if let Some(loop_node_id) = loop_node_id { + return fail_loop_child_future(ctx, graph, &loop_node_id, error).await; + } + return Err(error); } - return Err(NodeError::Failure(format!( - "JOIN branch {} failed: {}", - i + 1, - e - ))); + retain_higher_priority_join_error(&mut selected_error, error, loop_node_id); } } } + if let Some((error, loop_node_id)) = selected_error { + if let Some(loop_node_id) = loop_node_id { + return fail_loop_child_future(ctx, graph, &loop_node_id, error).await; + } + return Err(error); + } + ctx.trace_info(format!( "JOIN completed with {} results", join_results.len() @@ -1641,21 +1864,16 @@ async fn execute_race_node( } duroxide::Either2::First(Err(e)) => { cancel_losing_loop_branch(ctx, graph, right_id).await; + let error = contextualize_subtree_failure("RACE left branch failed", e); if graph .nodes .get(left_id) .map(|branch| branch.node_type.eq_ignore_ascii_case("loop")) .unwrap_or(false) { - return fail_loop_child_future( - ctx, - graph, - left_id, - format!("RACE left branch failed: {e}"), - ) - .await; + return fail_loop_child_future(ctx, graph, left_id, error).await; } - Err(format!("RACE left branch failed: {e}")) + Err(error) } duroxide::Either2::Second(Ok(r)) => { ctx.trace_info("RACE completed - right branch won"); @@ -1664,21 +1882,16 @@ async fn execute_race_node( } duroxide::Either2::Second(Err(e)) => { cancel_losing_loop_branch(ctx, graph, left_id).await; + let error = contextualize_subtree_failure("RACE right branch failed", e); if graph .nodes .get(right_id) .map(|branch| branch.node_type.eq_ignore_ascii_case("loop")) .unwrap_or(false) { - return fail_loop_child_future( - ctx, - graph, - right_id, - format!("RACE right branch failed: {e}"), - ) - .await; + return fail_loop_child_future(ctx, graph, right_id, error).await; } - Err(format!("RACE right branch failed: {e}")) + Err(error) } }?; @@ -1755,7 +1968,8 @@ async fn execute_http_node( let result = ctx .schedule_activity(activities::execute_http::NAME, final_config) - .await?; + .await + .map_err(NodeError::Application)?; // Store result if named if let Some(name) = &node.result_name { @@ -1854,7 +2068,8 @@ async fn execute_http_multipart_node( let result = ctx .schedule_activity(activities::execute_multipart::NAME, final_config) - .await?; + .await + .map_err(NodeError::Application)?; // Store result if named if let Some(name) = &node.result_name { @@ -2037,6 +2252,182 @@ mod tests { assert!(graph_retry_budget_exceeded(u32::MAX)); } + #[test] + fn subtree_instance_ids_are_stable_and_generation_scoped() { + assert_eq!( + compose_subtree_instance_id("parent", "1", "deadbeef"), + "parent::1::deadbeef" + ); + assert_ne!( + compose_subtree_instance_id("parent", "1", "deadbeef"), + compose_subtree_instance_id("parent", "2", "deadbeef") + ); + } + + #[test] + fn isolated_body_success_returns_result_and_merges_results() { + let raw = envelope_json( + Some("Normal"), + "42", + serde_json::json!({"body_result": "stored"}), + ); + let mut results = HashMap::new(); + + assert_eq!( + classify_isolated_body_result(Ok(raw), &mut results).unwrap(), + FailureIsolatedBodyOutcome::Succeeded("42".to_string()) + ); + assert_eq!( + results.get("body_result").map(String::as_str), + Some("stored") + ); + } + + #[test] + fn isolated_body_consumes_typed_application_failure() { + let encoded = encode_subtree_application_failure("boom"); + assert_eq!( + encoded, + r#"{"pg_durable_subtree_failure":"application","message":"boom"}"# + ); + let mut results = HashMap::new(); + + assert_eq!( + classify_isolated_body_result(Err(encoded), &mut results).unwrap(), + FailureIsolatedBodyOutcome::ApplicationFailed + ); + } + + #[test] + fn isolated_body_propagates_unrecognized_runtime_failure() { + let mut results = HashMap::new(); + + match classify_isolated_body_result(Err("instance id collision".to_string()), &mut results) + { + Err(NodeError::Failure(error)) => assert_eq!(error, "instance id collision"), + other => panic!("expected runtime failure, got {other:?}"), + } + } + + #[test] + fn isolated_body_preserves_break_value() { + let raw = envelope_json(Some("Break"), r#"{"status":"done"}"#, serde_json::json!({})); + let mut results = HashMap::new(); + + assert_eq!( + classify_isolated_body_result(Ok(raw), &mut results).unwrap(), + FailureIsolatedBodyOutcome::Break(r#"{"status":"done"}"#.to_string()) + ); + } + + #[test] + fn failure_isolated_join_prioritizes_errors_deterministically() { + let mut selected = None; + retain_higher_priority_join_error( + &mut selected, + NodeError::Application("recoverable".to_string()), + None, + ); + retain_higher_priority_join_error( + &mut selected, + NodeError::Break("break".to_string()), + None, + ); + retain_higher_priority_join_error( + &mut selected, + NodeError::Failure("first fatal".to_string()), + None, + ); + retain_higher_priority_join_error( + &mut selected, + NodeError::Failure("second fatal".to_string()), + None, + ); + + match selected { + Some((NodeError::Failure(error), None)) => assert_eq!(error, "first fatal"), + other => panic!("expected first fatal error, got {other:?}"), + } + } + + #[test] + fn legacy_subtree_input_preserves_first_error_behavior() { + let input: SubtreeInput = serde_json::from_str( + r#"{"instance_id":"deadbeef","node_id":"cafebabe","graph":"{}","results":"{}"}"#, + ) + .unwrap(); + + assert!(!input.failure_isolated); + } + + #[test] + fn subtree_input_is_byte_stable_across_result_insertion_order() { + let graph = FunctionGraph { + instance_id: "deadbeef".to_string(), + root_node_id: "cafebabe".to_string(), + nodes: std::collections::BTreeMap::new(), + }; + let exec_ctx = ExecutionContext { + vars: HashMap::new(), + label: Some("stable".to_string()), + failure_isolated: false, + loop_iteration: 7, + subtree_root: "cafebabe".to_string(), + continuation: Continuation::Root, + }; + let first = HashMap::from([ + ("alpha".to_string(), "1".to_string()), + ("beta".to_string(), "2".to_string()), + ]); + let second = HashMap::from([ + ("beta".to_string(), "2".to_string()), + ("alpha".to_string(), "1".to_string()), + ]); + + let first_input = build_subtree_input(&graph, "deadbeef", &first, &exec_ctx).unwrap(); + let second_input = build_subtree_input(&graph, "deadbeef", &second, &exec_ctx).unwrap(); + let expected = r#"{"instance_id":"deadbeef","node_id":"deadbeef","graph":"{\"instance_id\":\"deadbeef\",\"root_node_id\":\"cafebabe\",\"nodes\":{}}","results":"{\"alpha\":\"1\",\"beta\":\"2\"}","vars":"{}","label":"stable","iteration":0}"#; + + assert_eq!(first_input, expected); + assert_eq!(second_input, expected); + } + + #[test] + fn loop_config_parser_defaults_legacy_node_to_fail_fast() { + let node = FunctionNode { + id: "aaaaaaaa".to_string(), + node_type: "LOOP".to_string(), + query: None, + result_name: None, + left_node: Some("bbbbbbbb".to_string()), + right_node: None, + submitted_by: "postgres".to_string(), + database: None, + }; + + let config = parse_loop_config(&node, &node.id).unwrap(); + assert!(!config.continue_on_failure); + assert_eq!(config.condition_node, None); + } + + #[test] + fn loop_config_parser_accepts_continue_on_failure_with_condition() { + let node = FunctionNode { + id: "aaaaaaaa".to_string(), + node_type: "LOOP".to_string(), + query: Some(r#"{"continue_on_failure":true,"condition_node":"bbbbbbbb"}"#.to_string()), + result_name: None, + left_node: Some("cccccccc".to_string()), + right_node: None, + submitted_by: "postgres".to_string(), + database: None, + }; + + let config = parse_loop_config(&node, &node.id).unwrap(); + assert!(config.continue_on_failure); + assert_eq!(config.condition_node.as_deref(), Some("bbbbbbbb")); + } + #[test] fn parse_legacy_break_sentinel_decodes_string_value() { // A JSON string value round-trips as the quoted JSON string, matching the old @@ -2091,6 +2482,17 @@ mod tests { ); } + #[test] + fn malformed_subtree_envelope_remains_fatal() { + let mut parent = HashMap::new(); + match parse_subtree_envelope("not-json", "LOOP iteration", &mut parent) { + Err(NodeError::Failure(error)) => { + assert!(error.contains("LOOP iteration envelope parse error")); + } + other => panic!("expected fatal envelope error, got {other:?}"), + } + } + #[test] fn envelope_new_format_normal_with_sentinel_shaped_result_is_not_reraised() { // Regression guard for the #229 review finding: a new-binary `Normal` envelope whose diff --git a/src/types.rs b/src/types.rs index 58d43906..0ef96b06 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1482,6 +1482,18 @@ where Ok(value) } +fn is_false(value: &bool) -> bool { + !*value +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct LoopConfig { + #[serde(default, skip_serializing_if = "is_false")] + pub continue_on_failure: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub condition_node: Option, +} + /// The Durofut type represents a "durable future" - a reference to a node in the function graph. /// Children are embedded as opaque JSON objects, not stored as ID references. Keeping them as /// `RawValue` lets each graph level deserialize independently without serde_json's recursion limit. @@ -1721,8 +1733,7 @@ impl Durofut { )); } - self.reject_embedded_config_children()?; - Ok(()) + self.reject_embedded_config_children() } fn materialized_query( @@ -1769,6 +1780,62 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn loop_config_defaults_to_fail_fast() { + let config: LoopConfig = serde_json::from_str("{}").unwrap(); + assert!(!config.continue_on_failure); + assert_eq!(config.condition_node, None); + } + + #[test] + fn legacy_loop_config_without_query_is_fail_fast() { + assert!(!LoopConfig::default().continue_on_failure); + } + + #[test] + fn conditional_loop_config_without_continue_flag_is_fail_fast() { + let config: LoopConfig = serde_json::from_str(r#"{"condition_node":"deadbeef"}"#).unwrap(); + assert!(!config.continue_on_failure); + assert_eq!(config.condition_node.as_deref(), Some("deadbeef")); + } + + #[test] + fn loop_config_round_trips_continue_on_failure() { + let config = LoopConfig { + continue_on_failure: true, + condition_node: None, + }; + let json = serde_json::to_string(&config).unwrap(); + assert_eq!(json, r#"{"continue_on_failure":true}"#); + assert_eq!(serde_json::from_str::(&json).unwrap(), config); + } + + #[test] + fn flatten_graph_materializes_continue_on_failure_with_condition() { + let sql = Durofut { + node_type: "SQL".to_string(), + query: Some("SELECT true".to_string()), + ..Default::default() + }; + let loop_node = Durofut { + node_type: "LOOP".to_string(), + left_node: Some(sql.clone().into_raw()), + condition_node: Some(sql.into_raw()), + query: Some(r#"{"continue_on_failure":true}"#.to_string()), + ..Default::default() + }; + let mut ids = ["root", "body-id", "condition-id"] + .into_iter() + .map(str::to_string); + let (_, nodes) = flatten_graph(&loop_node, &mut || Ok(ids.next().unwrap())).unwrap(); + let materialized_loop = &nodes[0]; + + assert_eq!( + materialized_loop.query.as_deref(), + Some(r#"{"continue_on_failure":true,"condition_node":"condition-id"}"#) + ); + } + #[test] fn durofut_raw_children_preserve_wire_format() { let json = r#"{"node_type":"THEN","left_node":{"node_type":"SQL","query":"SELECT 1"},"right_node":{"node_type":"SQL","query":"SELECT 2"}}"#; diff --git a/tests/e2e/sql/70_loop_continue_on_failure.sql b/tests/e2e/sql/70_loop_continue_on_failure.sql new file mode 100644 index 00000000..dcb6fef5 --- /dev/null +++ b/tests/e2e/sql/70_loop_continue_on_failure.sql @@ -0,0 +1,546 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +CREATE OR REPLACE FUNCTION pg_temp.duroxide_instance_status(p_instance_id TEXT) +RETURNS TEXT +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog +AS $$ +DECLARE + provider_schema TEXT := df.duroxide_schema(); + engine_status TEXT; +BEGIN + EXECUTE format( + 'SELECT i.status FROM %I.get_instance_info($1) i', + provider_schema + ) + INTO engine_status + USING p_instance_id; + RETURN engine_status; +END +$$; + +CREATE OR REPLACE FUNCTION pg_temp.duroxide_child_count(p_parent_instance_id TEXT) +RETURNS BIGINT +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog +AS $$ +DECLARE + provider_schema TEXT := df.duroxide_schema(); + child_count BIGINT; +BEGIN + EXECUTE format( + 'SELECT count(*) FROM %I.instances WHERE parent_instance_id = $1', + provider_schema + ) + INTO child_count + USING p_parent_instance_id; + RETURN child_count; +END +$$; + +SET SESSION AUTHORIZATION df_e2e_user; + +DROP TABLE IF EXISTS test_loop_continue_attempts; +CREATE TABLE test_loop_continue_attempts ( + id SERIAL PRIMARY KEY, + scenario TEXT NOT NULL +); + +CREATE TEMP TABLE _loop_continue_instances ( + scenario TEXT PRIMARY KEY, + instance_id TEXT NOT NULL +); + +INSERT INTO _loop_continue_instances +SELECT 'continue', + df.start( + df.loop( + $$INSERT INTO test_loop_continue_attempts(scenario) + VALUES ('continue')$$ + ~> df.if( + $$SELECT count(*) = 1 + FROM test_loop_continue_attempts + WHERE scenario = 'continue'$$, + 'SELECT 1 / 0', + df.break('"completed-after-failure"') + ), + continue_on_failure => true + ), + 'test-loop-continue-on-failure' + ); + +INSERT INTO _loop_continue_instances +SELECT 'fail-fast', + df.start( + df.loop('SELECT 1 / 0', continue_on_failure => false), + 'test-loop-explicit-fail-fast' + ); + +DO $$ +DECLARE + continued_status TEXT; + failed_status TEXT; + attempts INT; + continued_explain TEXT; +BEGIN + SELECT df.await_instance(instance_id, 30) + INTO continued_status + FROM _loop_continue_instances + WHERE scenario = 'continue'; + + SELECT df.await_instance(instance_id, 30) + INTO failed_status + FROM _loop_continue_instances + WHERE scenario = 'fail-fast'; + + SELECT count(*) INTO attempts + FROM test_loop_continue_attempts + WHERE scenario = 'continue'; + + SELECT df.explain(instance_id) INTO continued_explain + FROM _loop_continue_instances + WHERE scenario = 'continue'; + + IF continued_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION + 'TEST FAILED [continue]: expected completed, got %', + continued_status; + END IF; + IF failed_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [fail-fast]: expected failed, got %', + failed_status; + END IF; + IF attempts <> 2 THEN + RAISE EXCEPTION + 'TEST FAILED [continue]: expected exactly two isolated iterations, got %', + attempts; + END IF; + IF continued_explain NOT LIKE '%LOOP (infinite, continue on failure)%' THEN + RAISE EXCEPTION + 'TEST FAILED [continue]: df.explain() omitted loop policy: %', + continued_explain; + END IF; +END $$; + +DROP TABLE _loop_continue_instances; +DROP TABLE test_loop_continue_attempts; + +DROP TABLE IF EXISTS test_nested_continue; +CREATE TABLE test_nested_continue (stage TEXT NOT NULL); + +CREATE TEMP TABLE _nested_continue_instance AS +SELECT df.start( + $$INSERT INTO test_nested_continue VALUES ('prefix')$$ + ~> df.loop( + $$INSERT INTO test_nested_continue VALUES ('iteration')$$ + ~> df.if( + $$SELECT count(*) = 1 + FROM test_nested_continue + WHERE stage = 'iteration'$$, + 'SELECT 1 / 0', + df.break() + ), + continue_on_failure => true + ) + ~> $$INSERT INTO test_nested_continue VALUES ('suffix')$$, + 'test-nested-loop-continue' +) AS instance_id; + +DO $$ +DECLARE + final_status TEXT; + prefix_count INT; + iteration_count INT; + suffix_count INT; +BEGIN + SELECT df.await_instance(instance_id, 30) + INTO final_status + FROM _nested_continue_instance; + + SELECT + count(*) FILTER (WHERE stage = 'prefix'), + count(*) FILTER (WHERE stage = 'iteration'), + count(*) FILTER (WHERE stage = 'suffix') + INTO prefix_count, iteration_count, suffix_count + FROM test_nested_continue; + + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION + 'TEST FAILED [nested]: expected completed, got %', + final_status; + END IF; + IF prefix_count <> 1 OR iteration_count <> 2 OR suffix_count <> 1 THEN + RAISE EXCEPTION + 'TEST FAILED [nested]: expected prefix/iteration/suffix = 1/2/1, got %/%/%', + prefix_count, iteration_count, suffix_count; + END IF; +END $$; + +DROP TABLE _nested_continue_instance; +DROP TABLE test_nested_continue; + +DROP FUNCTION IF EXISTS test_conditional_continue_body(); +DROP FUNCTION IF EXISTS test_conditional_continue_condition(); +DROP FUNCTION IF EXISTS test_conditional_continue_failing_condition(); +DROP SEQUENCE IF EXISTS test_conditional_continue_attempt_seq; +DROP TABLE IF EXISTS test_conditional_continue_state; +DROP TABLE IF EXISTS test_nested_condition_failure_attempts; + +CREATE TABLE test_conditional_continue_state ( + body_attempts INT NOT NULL DEFAULT 0, + condition_checks INT NOT NULL DEFAULT 0 +); +INSERT INTO test_conditional_continue_state DEFAULT VALUES; +CREATE TABLE test_nested_condition_failure_attempts (id SERIAL PRIMARY KEY); + +-- Sequence values are not rolled back when the first function call raises, allowing the +-- second successful call to persist the total attempt count in the state table. +CREATE SEQUENCE test_conditional_continue_attempt_seq; + +CREATE FUNCTION test_conditional_continue_body() RETURNS INT +LANGUAGE plpgsql AS $$ +DECLARE + attempts INT; +BEGIN + attempts := nextval('test_conditional_continue_attempt_seq'); + UPDATE test_conditional_continue_state + SET body_attempts = attempts; + + IF attempts = 1 THEN + RAISE EXCEPTION 'transient body failure'; + END IF; + RETURN attempts; +END +$$; + +CREATE FUNCTION test_conditional_continue_condition() RETURNS BOOLEAN +LANGUAGE plpgsql AS $$ +BEGIN + UPDATE test_conditional_continue_state + SET condition_checks = condition_checks + 1; + RETURN false; +END +$$; + +CREATE FUNCTION test_conditional_continue_failing_condition() RETURNS BOOLEAN +LANGUAGE plpgsql AS $$ +BEGIN + RAISE EXCEPTION 'fatal condition failure'; +END +$$; + +CREATE TEMP TABLE _conditional_continue_instances ( + scenario TEXT PRIMARY KEY, + instance_id TEXT NOT NULL +); + +INSERT INTO _conditional_continue_instances +SELECT 'body-recovery', + df.start( + df.loop( + 'SELECT test_conditional_continue_body()', + 'SELECT test_conditional_continue_condition()', + continue_on_failure => true + ), + 'test-conditional-loop-continues-after-body-failure' + ); + +INSERT INTO _conditional_continue_instances +SELECT 'condition-failure', + df.start( + df.loop( + 'SELECT 42', + 'SELECT test_conditional_continue_failing_condition()', + continue_on_failure => true + ), + 'test-conditional-loop-condition-failure-is-fatal' + ); + +INSERT INTO _conditional_continue_instances +SELECT 'named-result-condition', + df.start( + df.loop( + df.sql('SELECT 42 AS value') |=> 'body_value', + 'SELECT $body_value.value = 0', + continue_on_failure => true + ), + 'test-conditional-loop-body-result-visible-to-condition' + ); + +INSERT INTO _conditional_continue_instances +SELECT 'nested-condition-failure', + df.start( + df.loop( + 'INSERT INTO test_nested_condition_failure_attempts DEFAULT VALUES' + ~> df.if( + 'SELECT count(*) = 1 FROM test_nested_condition_failure_attempts', + df.loop( + 'SELECT 42', + 'SELECT test_conditional_continue_failing_condition()', + continue_on_failure => true + ), + df.break('"nested-condition-was-consumed"') + ), + continue_on_failure => true + ), + 'test-nested-condition-failure-is-fatal' + ); + +DO $$ +DECLARE + recovered_status TEXT; + condition_failure_status TEXT; + named_result_condition_status TEXT; + nested_condition_failure_status TEXT; + body_attempts INT; + condition_checks INT; + nested_attempts INT; + conditional_explain TEXT; +BEGIN + SELECT df.await_instance(instance_id, 30) + INTO recovered_status + FROM _conditional_continue_instances + WHERE scenario = 'body-recovery'; + + SELECT df.await_instance(instance_id, 30) + INTO condition_failure_status + FROM _conditional_continue_instances + WHERE scenario = 'condition-failure'; + + SELECT df.await_instance(instance_id, 30) + INTO named_result_condition_status + FROM _conditional_continue_instances + WHERE scenario = 'named-result-condition'; + + SELECT df.await_instance(instance_id, 30) + INTO nested_condition_failure_status + FROM _conditional_continue_instances + WHERE scenario = 'nested-condition-failure'; + + SELECT s.body_attempts, s.condition_checks + INTO body_attempts, condition_checks + FROM test_conditional_continue_state s; + + SELECT count(*) INTO nested_attempts + FROM test_nested_condition_failure_attempts; + + SELECT df.explain(instance_id) INTO conditional_explain + FROM _conditional_continue_instances + WHERE scenario = 'body-recovery'; + + IF recovered_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION + 'TEST FAILED [conditional continue]: expected completed, got %', + recovered_status; + END IF; + IF body_attempts <> 2 OR condition_checks <> 1 THEN + RAISE EXCEPTION + 'TEST FAILED [conditional continue]: expected body/condition = 2/1, got %/%', + body_attempts, condition_checks; + END IF; + IF condition_failure_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [condition failure]: expected failed, got %', + condition_failure_status; + END IF; + IF named_result_condition_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION + 'TEST FAILED [named result condition]: expected completed, got %', + named_result_condition_status; + END IF; + IF nested_condition_failure_status IS DISTINCT FROM 'failed' + OR nested_attempts <> 1 THEN + RAISE EXCEPTION + 'TEST FAILED [nested condition failure]: expected failed after one outer iteration, got % / %', + nested_condition_failure_status, nested_attempts; + END IF; + IF conditional_explain NOT LIKE '%LOOP (while, continue on failure)%' THEN + RAISE EXCEPTION + 'TEST FAILED [conditional continue]: df.explain() omitted combined loop mode: %', + conditional_explain; + END IF; +END $$; + +DROP TABLE _conditional_continue_instances; +DROP FUNCTION test_conditional_continue_body(); +DROP FUNCTION test_conditional_continue_condition(); +DROP FUNCTION test_conditional_continue_failing_condition(); +DROP SEQUENCE test_conditional_continue_attempt_seq; +DROP TABLE test_conditional_continue_state; +DROP TABLE test_nested_condition_failure_attempts; + +CREATE TEMP TABLE _cancel_continue_instance (instance_id TEXT); + +INSERT INTO _cancel_continue_instance(instance_id) +SELECT df.start( + df.loop( + df.sleep(30), + continue_on_failure => true + ), + 'test-cancel-failure-isolated-loop' +); + +DO $$ +DECLARE + parent_id TEXT; + child_stamp TEXT; + child_id TEXT; + parent_status TEXT; + child_status TEXT; + child_count BIGINT; + attempts INT := 0; +BEGIN + SELECT c.instance_id INTO parent_id FROM _cancel_continue_instance c; + + LOOP + SELECT n.status_details::jsonb->>'execution_id' + INTO child_stamp + FROM df.instance_nodes(parent_id) n + WHERE n.node_type = 'SLEEP' + AND n.status = 'running'; + EXIT WHEN child_stamp IS NOT NULL OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF child_stamp IS NULL THEN + RAISE EXCEPTION + 'TEST FAILED [cancel]: active iteration child was not observed'; + END IF; + + child_id := array_to_string( + (string_to_array(child_stamp, '::'))[ + 1:array_length(string_to_array(child_stamp, '::'), 1) - 1 + ], + '::' + ); + child_status := pg_temp.duroxide_instance_status(child_id); + IF lower(COALESCE(child_status, '')) IS DISTINCT FROM 'running' THEN + RAISE EXCEPTION + 'TEST FAILED [cancel]: expected active child %, got status %', + child_id, child_status; + END IF; + + PERFORM df.cancel(parent_id, 'test cancellation propagation'); + SELECT df.await_instance(parent_id, 30) INTO parent_status; + + IF parent_status IS DISTINCT FROM 'cancelled' THEN + RAISE EXCEPTION + 'TEST FAILED [cancel]: expected cancelled parent, got %', + parent_status; + END IF; + + attempts := 0; + LOOP + child_status := pg_temp.duroxide_instance_status(child_id); + EXIT WHEN lower(COALESCE(child_status, '')) = 'failed' OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF lower(COALESCE(child_status, '')) IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [cancel]: active child % was not cancelled; engine status %', + child_id, child_status; + END IF; + + child_count := pg_temp.duroxide_child_count(parent_id); + IF child_count <> 1 THEN + RAISE EXCEPTION + 'TEST FAILED [cancel]: expected one cancelled iteration child and no continuation, got % children', + child_count; + END IF; +END $$; + +DROP TABLE _cancel_continue_instance; + +DROP TABLE IF EXISTS test_scheduled_loop_ticks; +CREATE TABLE test_scheduled_loop_ticks ( + id SERIAL PRIMARY KEY, + fired_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp() +); + +CREATE TEMP TABLE _scheduled_loop_instance AS +SELECT df.start( + df.loop( + df.wait_for_schedule('* * * * *') + ~> 'INSERT INTO test_scheduled_loop_ticks DEFAULT VALUES' + ~> df.if( + 'SELECT count(*) = 1 FROM test_scheduled_loop_ticks', + 'SELECT 1 / 0', + df.break('"scheduled-loop-recovered"') + ), + continue_on_failure => true + ), + 'test-scheduled-loop-continues' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + attempts INT := 0; + current_status TEXT; + failed_nodes INT; +BEGIN + SELECT s.instance_id INTO instance_id FROM _scheduled_loop_instance s; + LOOP + SELECT count(*) INTO failed_nodes + FROM df.instance_nodes(instance_id) + WHERE status = 'failed'; + EXIT WHEN failed_nodes > 0 OR attempts >= 900; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + SELECT s INTO current_status FROM df.status(instance_id) s; + IF failed_nodes = 0 THEN + RAISE EXCEPTION + 'TEST FAILED [scheduled]: failed child node was not recorded'; + END IF; + IF current_status IS DISTINCT FROM 'running' THEN + RAISE EXCEPTION + 'TEST FAILED [scheduled]: parent stopped after child failure: %', + current_status; + END IF; +END $$; + +DO $$ +DECLARE + final_status TEXT; + tick_count INT; + first_tick TIMESTAMPTZ; + second_tick TIMESTAMPTZ; +BEGIN + SELECT df.await_instance(instance_id, 180) + INTO final_status + FROM _scheduled_loop_instance; + + SELECT count(*), min(fired_at), max(fired_at) + INTO tick_count, first_tick, second_tick + FROM test_scheduled_loop_ticks; + + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION + 'TEST FAILED [scheduled]: expected completed, got %', + final_status; + END IF; + IF tick_count <> 2 THEN + RAISE EXCEPTION + 'TEST FAILED [scheduled]: expected two ticks, got %', + tick_count; + END IF; + IF second_tick - first_tick < interval '50 seconds' THEN + RAISE EXCEPTION + 'TEST FAILED [scheduled]: next iteration did not wait for the next tick: % / %', + first_tick, second_tick; + END IF; +END $$; + +DROP TABLE _scheduled_loop_instance; +DROP TABLE test_scheduled_loop_ticks; +RESET SESSION AUTHORIZATION; +SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/71_join_failure_priority.sql b/tests/e2e/sql/71_join_failure_priority.sql new file mode 100644 index 00000000..383a6d51 --- /dev/null +++ b/tests/e2e/sql/71_join_failure_priority.sql @@ -0,0 +1,167 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- A fatal JOIN branch failure must take precedence over a recoverable activity +-- failure in an earlier branch. Otherwise a failure-isolated loop consumes the +-- recoverable error without inspecting the fatal sibling and starts another +-- iteration. +SET SESSION AUTHORIZATION df_e2e_user; + +DROP SEQUENCE IF EXISTS test_join_failure_priority_attempt_seq; +CREATE SEQUENCE test_join_failure_priority_attempt_seq CACHE 1; + +CREATE TEMP TABLE _join_failure_priority_instance AS +SELECT df.start( + df.loop( + $$SELECT 1 / ( + CASE + WHEN nextval('test_join_failure_priority_attempt_seq') >= 2 THEN 1 + ELSE 0 + END + )$$ + & ( + df.sql('SELECT NULL::int AS value') |=> 'nullable' + ~> 'SELECT $nullable' + ), + continue_on_failure => true + ), + 'test-join-fatal-failure-priority' +) AS instance_id; + +DO $$ +DECLARE + final_status TEXT; + attempts BIGINT; + instance_output TEXT; + waited INT; +BEGIN + SELECT df.await_instance(i.instance_id, 30) + INTO final_status + FROM _join_failure_priority_instance i; + + SELECT last_value + INTO attempts + FROM test_join_failure_priority_attempt_seq; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [join failure priority]: expected failed, got %', + final_status; + END IF; + IF attempts IS DISTINCT FROM 1 THEN + RAISE EXCEPTION + 'TEST FAILED [join failure priority]: fatal sibling was hidden for % iterations', + attempts; + END IF; + + FOR waited IN 1..100 LOOP + SELECT info.output + INTO instance_output + FROM _join_failure_priority_instance i + CROSS JOIN LATERAL df.instance_info(i.instance_id) info; + EXIT WHEN instance_output IS NOT NULL; + PERFORM pg_sleep(0.1); + END LOOP; + + IF COALESCE(instance_output, '') NOT LIKE '%$nullable is NULL%' THEN + RAISE EXCEPTION + 'TEST FAILED [join failure priority]: fatal sibling error did not surface: %', + instance_output; + END IF; +END $$; + +DROP TABLE _join_failure_priority_instance; +DROP SEQUENCE test_join_failure_priority_attempt_seq; + +-- A fatal sibling also takes precedence over intentional break control flow. +CREATE TEMP TABLE _join_break_failure_priority_instance AS +SELECT df.start( + df.loop( + df.break('"break-must-not-win"') + & ( + df.sql('SELECT NULL::int AS value') |=> 'break_nullable' + ~> 'SELECT $break_nullable' + ), + continue_on_failure => true + ), + 'test-join-fatal-over-break-priority' +) AS instance_id; + +DO $$ +DECLARE + final_status TEXT; + instance_output TEXT; + waited INT; +BEGIN + SELECT df.await_instance(i.instance_id, 30) + INTO final_status + FROM _join_break_failure_priority_instance i; + + FOR waited IN 1..100 LOOP + SELECT info.output + INTO instance_output + FROM _join_break_failure_priority_instance i + CROSS JOIN LATERAL df.instance_info(i.instance_id) info; + EXIT WHEN instance_output IS NOT NULL; + PERFORM pg_sleep(0.1); + END LOOP; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [join break priority]: expected failed, got %', + final_status; + END IF; + IF COALESCE(instance_output, '') NOT LIKE '%$break_nullable is NULL%' THEN + RAISE EXCEPTION + 'TEST FAILED [join break priority]: fatal sibling error did not surface: %', + instance_output; + END IF; +END $$; + +DROP TABLE _join_break_failure_priority_instance; + +-- Outside a failure-isolated loop, retain the historical first-branch error. +CREATE TEMP TABLE _join_legacy_failure_priority_instance AS +SELECT df.start( + 'SELECT 1 / 0' + & ( + df.sql('SELECT NULL::int AS value') |=> 'legacy_nullable' + ~> 'SELECT $legacy_nullable' + ), + 'test-join-legacy-first-error' +) AS instance_id; + +DO $$ +DECLARE + final_status TEXT; + instance_output TEXT; + waited INT; +BEGIN + SELECT df.await_instance(i.instance_id, 30) + INTO final_status + FROM _join_legacy_failure_priority_instance i; + + FOR waited IN 1..100 LOOP + SELECT info.output + INTO instance_output + FROM _join_legacy_failure_priority_instance i + CROSS JOIN LATERAL df.instance_info(i.instance_id) info; + EXIT WHEN instance_output IS NOT NULL; + PERFORM pg_sleep(0.1); + END LOOP; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION + 'TEST FAILED [join legacy priority]: expected failed, got %', + final_status; + END IF; + IF COALESCE(instance_output, '') NOT LIKE '%division by zero%' THEN + RAISE EXCEPTION + 'TEST FAILED [join legacy priority]: first branch error changed: %', + instance_output; + END IF; +END $$; + +DROP TABLE _join_legacy_failure_priority_instance; +RESET SESSION AUTHORIZATION; +SELECT 'TEST PASSED' AS result;