diff --git a/CHANGELOG.md b/CHANGELOG.md index 5db06943..1eb0d965 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to this project are documented in this file. The format is b Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may include breaking changes. +## [0.2.7] - Unreleased + +### Added + +- **Node failure policy on `df.start()`:** three new arguments — `max_attempts`, `max_backoff`, and `on_failure` — control what happens when a `df.sql()`, `df.http()`, or `df.http_multipart()` node fails. The node is retried with exponential backoff starting at 1 second and doubling up to `max_backoff`; the wait is a durable timer, so it holds no connection and survives a restart. Once the attempts are spent, `on_failure => 'continue'` abandons the rest of the current loop iteration and starts the next one, while `on_failure => 'fail'` fails the instance. Outside a loop there is no next iteration, so both settings fail the instance. **The defaults (`max_attempts => 1`, `on_failure => 'fail'`) are the pre-0.2.7 behaviour, so the feature is entirely opt-in and upgrading changes nothing for an existing workflow.** Graph-level errors (malformed graph, unknown node type, failure to start a sub-orchestration) are not transient and still fail immediately, as do statement-level errors that a retry cannot fix — SQLSTATE classes 42, 23, 28, 3D and 3F fail on the first attempt however many attempts are allowed. +- **`df.instance_activity([idle_for])`:** reports every non-terminal instance with the time of its last node transition, how long it has been idle, how many of its nodes are running or failed, and its most recent error. `status` alone cannot distinguish a healthy eternal loop from a wedged one — both read `'running'` — which matters more now that `on_failure => 'continue'` lets a loop keep running through failures. Row-level-security filtered to the calling user. + +### Changed + +- **The 100,000-iteration `df.loop()` cap was removed.** It was never a meaningful storage bound — a large carried result exhausts storage long before the count trips — and it gave a workflow meant to run indefinitely, such as a compactor on a five-minute schedule, an arbitrary expiry date. Use `df.instance_activity()` to spot a loop that is running without making progress, and `df.cancel()` to stop it. +- **`df.start()` signature:** the four-argument `df.start(text, text, text, text)` is replaced by `df.start(text, text, text, text, int, interval, text)`. Behaviour is unchanged, because the new arguments default to the previous semantics. Un-upgraded schemas keep resolving to the previous function; see [docs/upgrade-testing.md](docs/upgrade-testing.md). + ## [0.2.6] - 2026-08-23 ### Added diff --git a/Cargo.lock b/Cargo.lock index a6ed0123..6cf0019f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1909,7 +1909,7 @@ dependencies = [ [[package]] name = "pg_durable" -version = "0.2.6" +version = "0.2.7" dependencies = [ "base64", "bigdecimal", diff --git a/Cargo.toml b/Cargo.toml index 599be32c..afbb5048 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pg_durable" -version = "0.2.6" +version = "0.2.7" edition = "2021" license = "PostgreSQL" repository = "https://github.com/microsoft/pg_durable" diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 053be94f..e19d4176 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -227,6 +227,112 @@ executed as a single statement on an autocommit connection, so a plain `df.start()` there is already independent of any caller transaction; `'new'` would buy nothing and consume an extra backend. +### Node Failure Policy + +Nodes that reach outside the workflow — `df.sql()`, `df.http()`, and +`df.http_multipart()` — fail for transient reasons: a deadlock, a dropped +connection, a rate-limited endpoint. `df.start()` takes three arguments that +decide how hard to try and what to do when trying is over: + +| Argument | Default | Meaning | +|---|---|---| +| `max_attempts` | `1` | Total attempts for a failing node, including the first. `1` disables retrying. | +| `max_backoff` | `'16 seconds'` | Upper bound on the wait between attempts. | +| `on_failure` | `'fail'` | What to do once the attempts are spent: `'continue'` or `'fail'`. | + +**The defaults are the behaviour pg_durable has always had**: one attempt, and +a failing node fails the instance. The policy is entirely opt-in, so upgrading +to 0.2.7 cannot change how an existing workflow handles a failure. Ask for +retries when you want them: + +```sql +-- Retry a flaky endpoint five times, then fail the instance as usual. +SELECT df.start( + df.http('POST', 'https://api.example.com/sync'), + 'sync', + max_attempts => 5 +); +``` + +Retries back off exponentially, starting at 1 second and doubling until they +reach `max_backoff`. With `max_attempts => 5` a node is tried at 0s, 1s, 3s, +7s, and 15s before its policy decides. The delay is a durable timer, so it +costs no connection and survives a restart. Note that `max_backoff` only binds +once `max_attempts` exceeds 5: five attempts use waits of 1s, 2s, 4s, and 8s, +none of which reach the 16-second default cap. + +A failure that is a property of the statement rather than of the moment is +**not** retried, however many attempts you allow. An undefined table, a syntax +error, a permission denied, or a constraint violation (SQLSTATE classes 42, 23, +28, 3D and 3F) fails immediately, because the next attempt would produce the +identical error. Deadlocks and serialization failures (class 40), connection +errors (08), resource limits (53), operator intervention (57), and anything +without a SQLSTATE — including every `df.http()` failure — are retried. Division +by zero and other data exceptions (class 22) are retried on purpose: the data +they complain about can change between attempts. + +Once the attempts are spent, `on_failure` chooses between two outcomes: + +- `'fail'` (the default) fails the whole instance, which is what pg_durable has + always done. +- `'continue'` abandons the **rest of the current loop iteration** and starts + the next one. Nodes downstream of the failure are skipped — a failed extract + does not run its load — and the loop's `while` condition is skipped too, + since it usually reads results the abandoned iteration never produced. + +Outside a loop there is no next iteration to continue into, so both settings +fail the instance. `'continue'` is a statement about *recurring* work. + +> **Write loop bodies to be idempotent.** Each `df.sql()` node commits on its +> own autocommit connection, so "abandon the rest of the iteration" does not +> roll anything back. If node A inserts and node B then fails, A's insert is +> already durable and the next iteration starts over on top of it. The same +> applies within a single node: a retry re-runs the statement, and a statement +> that committed server-side but lost its connection before returning will run +> twice. Guard side effects with `ON CONFLICT`, an idempotency key, or a check +> against the state the previous attempt would have written — or set +> `on_failure => 'fail'` so a human reconciles instead. + +```sql +-- A compactor that must survive a bad batch: skip it and pick up the next tick. +SELECT df.start( + df.loop( + 'CALL compact_next_partition()' ~> df.wait_for_schedule('*/5 * * * *') + ), + 'compactor', + max_attempts => 5, + on_failure => 'continue' +); + +-- A one-shot migration that must not be retried and must surface its error. +-- This is the default, so it is also what a plain df.start() does. +SELECT df.start( + 'CALL migrate_tenant(42)', + 'migrate-42' +); +``` + +The policy covers node execution only. A malformed graph, an unknown node type, +or a failure to start a sub-orchestration is not a transient condition and fails +the instance immediately, whatever the policy says. + +A loop running under `on_failure => 'continue'` never reaches a terminal status +on its own: it keeps starting iterations, so its instance stays `running` even +when every iteration fails. Use `df.instance_activity()` (below) to tell that +apart from healthy work, and `df.cancel()` to stop it. + +Attempts are not individually visible through `df.instance_nodes()` or +`df.explain()`, which report a node's status once its attempts have settled. A +node being retried reads as still running; the individual failures are in the +worker log. + +> **Note for workflows written before 0.2.7:** the defaults changed. A workflow +> that used to fail on its first node error now retries five times and, inside a +> loop, keeps running afterwards. Pass `max_attempts => 1, on_failure => 'fail'` +> to restore the old behaviour. In particular, a `while` loop whose body always +> fails under `'continue'` never re-evaluates its condition and so never ends; +> use `'fail'`, or bound the work with `df.break()`. + --- ## DSL Reference @@ -258,6 +364,7 @@ df.sql('SELECT 1') ~> df.sql('SELECT 2') | `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')` | | `df.start(func, label, database, transaction_mode)` | Start function in its own transaction when `transaction_mode => 'new'` (survives caller rollback) | `df.start('INSERT INTO audit ...', 'audit', transaction_mode => 'new')` | +| `df.start(..., max_attempts, max_backoff, on_failure)` | Retry a failing node and choose what happens once the attempts are spent | `df.start('CALL sync()', 'sync', max_attempts => 1, on_failure => 'fail')` | | `df.cancel(id, reason)` | Cancel function | `df.cancel('a1b2c3d4', 'Done')` | | `df.status(id)` | Get status by instance_id (not label) | `df.status('a1b2c3d4')` | | `df.result(id)` | Get result by instance_id (not label) | `df.result('a1b2c3d4')` | @@ -1823,6 +1930,51 @@ This is useful for dashboards and operational queries that need to understand wh --- +### Is Anything Actually Happening? + +`status` alone cannot distinguish a workflow that is working from one that is +stuck: a healthy eternal loop, an instance blocked on a signal that will never +arrive, and a loop retrying a broken node all read `'running'`. +`df.instance_activity()` answers the operational question directly — when did +this instance last transition a node, and is anything failing right now? + +```sql +-- Every non-terminal instance, quietest first. +SELECT * FROM df.instance_activity() ORDER BY idle_for_seconds DESC; + +-- Only instances that have done nothing for ten minutes. +SELECT instance_id, label, idle_for_seconds, last_error +FROM df.instance_activity('10 minutes'); +``` + +| Column | Meaning | +|---|---| +| `instance_id` / `label` / `status` | Identity, as in `df.list_instances()`. | +| `last_activity_at` | When a node of this instance last changed state. | +| `idle_for_seconds` | Seconds since then. A large value on a workflow that should be busy is the signal to investigate. | +| `running_node_count` | Nodes currently executing. Zero with a large `idle_for_seconds` means nothing is in flight. | +| `failed_node_count` | Nodes that ended in failure. Non-zero on a `running` instance means a loop is failing and continuing. | +| `last_error` | The most recent node error, so you do not have to join to `df.instance_nodes()` to see why. | + +The argument filters to instances idle for at least that long; it defaults to +zero, which reports them all. Terminal instances (`completed`, `failed`, +`cancelled`) are never reported — the question only applies to work in +progress. Results are row-level-security filtered, so you see only your own +instances. + +A loop under `on_failure => 'continue'` is the case this exists for. It stays +`running` indefinitely by design, so a rising `failed_node_count` with a +repeating `last_error` is how you spot one that will never succeed: + +```sql +-- Loops that are running but only producing failures. +SELECT instance_id, label, failed_node_count, last_error +FROM df.instance_activity() +WHERE failed_node_count > 0 AND running_node_count = 0; +``` + +--- + ### System Metrics (Explicit Grant Required) ```sql diff --git a/docs/api-reference.md b/docs/api-reference.md index e74aad1e..48a760ae 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -328,7 +328,7 @@ Returns the same envelope as `df.http()`. ## Control Functions -### df.start(fut [, label] [, database] [, transaction_mode]) +### df.start(fut [, label] [, database] [, transaction_mode] [, max_attempts] [, max_backoff] [, on_failure]) Starts a durable function. @@ -338,6 +338,9 @@ Starts a durable function. | `label` | TEXT | ❌ Literal | (Optional) Human-readable label | | `database` | TEXT | ❌ Literal | (Optional) Target database on the cluster | | `transaction_mode` | TEXT | ❌ Literal | (Optional) `'caller'` (default) or `'new'` | +| `max_attempts` | INT | ❌ Literal | (Optional) Attempts per failing node, including the first (default `1`, minimum `1`) | +| `max_backoff` | INTERVAL | ❌ Literal | (Optional) Upper bound on the wait between attempts (default `'16 seconds'`) | +| `on_failure` | TEXT | ❌ Literal | (Optional) `'fail'` (default) or `'continue'` | ```sql df.start('SELECT 1') -- auto-wrapped @@ -383,6 +386,68 @@ An unrecognised value raises an error rather than falling back to the default. > idempotent because a connection failure can make launch outcome uncertain. See > the Transaction Semantics section of `USER_GUIDE.md` for details. +#### max_attempts, max_backoff, on_failure + +Control what happens when a node that reaches outside the workflow — +`df.sql()`, `df.http()`, or `df.http_multipart()` — fails. + +The defaults (`max_attempts => 1`, `on_failure => 'fail'`) are the behaviour +pg_durable had before 0.2.7: one attempt, and a failing node fails the instance. +The policy is opt-in, so upgrading changes nothing for an existing workflow. + +A failing node is retried up to `max_attempts` times in total, waiting 1 second +before the second attempt and doubling thereafter until the wait reaches +`max_backoff`. With `max_attempts => 5` a node is attempted at 0s, 1s, 3s, 7s, +and 15s. The wait is a durable timer: it holds no connection and survives a +restart. `max_backoff` is only reached once `max_attempts` exceeds 5 — five +attempts use waits of 1s, 2s, 4s, and 8s. + +Once the attempts are spent, `on_failure` decides: + +- `'fail'` (default) — fail the instance. +- `'continue'` — abandon the rest of the current loop iteration and start the + next one. Nodes downstream of the failure are skipped, and so is the loop's + `while` condition, which usually reads results the abandoned iteration never + produced. + +Outside a loop there is no next iteration, so both settings fail the instance. + +```sql +-- One-shot work that must not be retried and must surface its error (default). +df.start('CALL migrate_tenant(42)', 'migrate') + +-- Recurring work that should survive a bad batch. +df.start( + df.loop('CALL compact()' ~> df.wait_for_schedule('*/5 * * * *')), + 'compactor', + max_attempts => 5, + on_failure => 'continue' +) +``` + +`max_attempts < 1`, a negative `max_backoff`, and an unrecognised `on_failure` +each raise an error. + +The policy covers node execution only. A malformed graph, an unknown node type, +or a failure to start a sub-orchestration fails the instance immediately. + +A failure that is a property of the statement rather than of the moment is also +not retried, whatever `max_attempts` says. SQLSTATE classes 42 (syntax or access +rule violation), 23 (integrity constraint violation), 28 (invalid authorization), +3D (invalid catalog name) and 3F (invalid schema name) fail on the first attempt, +because a retry would reproduce the identical error. Classes 40 (serialization +failure, deadlock), 08 (connection exception), 53 (insufficient resources) and +57 (operator intervention) are retried, as is any failure without a SQLSTATE — +which includes every `df.http()` and `df.http_multipart()` error. Class 22 (data +exception, e.g. division by zero) is retried deliberately: it describes the data, +which another node can change between attempts. + +> **Note:** attempts are not individually visible through `df.instance_nodes()` +> or `df.explain()`, which report a node's status once its attempts have +> settled. A node being retried reads as still running; the individual failures +> are in the worker log. Use `df.instance_activity()` to see whether an instance +> is making progress at all. + --- ### df.signal(instance_id, signal_name [, signal_data]) @@ -499,6 +564,46 @@ SELECT df.result('a1b2c3d4'); --- +### df.instance_activity([idle_for]) + +Returns one row per **non-terminal** instance with a measure of whether it is +making progress. `status` alone cannot separate a healthy eternal loop, an +instance blocked on a signal that will never arrive, and a loop retrying a +broken node — all three read `'running'`. + +| Parameter | Type | Auto-wrap | Description | +|-----------|------|-----------|-------------| +| `idle_for` | INTERVAL | ❌ Literal | (Optional) Report only instances idle at least this long (default `'0 seconds'`, i.e. all) | + +Return columns: + +| Column | Type | Description | +|--------|------|-------------| +| `instance_id` | VARCHAR(8) | Instance id | +| `label` | TEXT | Label given at `df.start()`, or `NULL` | +| `status` | TEXT | Stored instance status | +| `last_activity_at` | TIMESTAMPTZ | When a node of this instance last changed state | +| `idle_for_seconds` | DOUBLE PRECISION | Seconds since `last_activity_at` | +| `running_node_count` | BIGINT | Nodes currently executing | +| `failed_node_count` | BIGINT | Nodes that ended in failure | +| `last_error` | TEXT | Most recent node error, or `NULL` | + +Completed, failed, and cancelled instances are never returned — the question +only applies to work in progress. Results are row-level-security filtered to the +calling user's own instances. + +```sql +-- Everything in flight, quietest first. +SELECT * FROM df.instance_activity() ORDER BY idle_for_seconds DESC; + +-- Loops that are running but only producing failures. +SELECT instance_id, label, failed_node_count, last_error +FROM df.instance_activity() +WHERE failed_node_count > 0 AND running_node_count = 0; +``` + +--- + ### df.instance_nodes(instance_id) Returns one row per node in an instance's graph, with each node's stored physical diff --git a/docs/loop_rework_problems.md b/docs/loop_rework_problems.md index 44ed7681..1675bb35 100644 --- a/docs/loop_rework_problems.md +++ b/docs/loop_rework_problems.md @@ -24,7 +24,7 @@ Ordered by severity. | 5 | [Parent fallback stamps assume child generation 1](#5-parent-fallback-stamps-assume-child-generation-1) | cancelled or failed loop children | 🟡 Medium | Open | | 6 | [Composed child ids block orphan reclamation](#6-composed-child-ids-block-orphan-reclamation) | loop and parallel children | 🟡 Medium | Resolved in #323 | | 7 | [Stamp validation fails open](#7-stamp-validation-fails-open) | status write fence and inference | 🟡 Medium | Open | -| 8 | [Nested loops have no cumulative iteration budget](#8-nested-loops-have-no-cumulative-iteration-budget) | nested loops | 🟡 Medium | Open | +| 8 | [Nested loops have no cumulative iteration budget](#8-nested-loops-have-no-cumulative-iteration-budget) | nested loops | 🟡 Medium | Open — the per-instance cap was removed in 0.2.7 | | 9 | [Signals inherit the sub-orchestration startup race](#9-signals-inherit-the-sub-orchestration-startup-race) | non-root loops | 🟡 Medium | Open | | 10 | [The 0.2.5 drain guidance is not an executable runbook](#10-the-025-drain-guidance-is-not-an-executable-runbook) | 0.2.4 → 0.2.5 upgrade | 🟡 Medium | Open | | 11 | [The status-write fence holds a row lock across round trips](#11-the-status-write-fence-holds-a-row-lock-across-round-trips) | schemas with `status_details` | 🟢 Low | Open — measure before changing | @@ -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. Nothing bounds this: the 100,000-generation cap was never a +meaningful byte bound (a 1 MB carried result could still produce storage measured +in tens of gigabytes before it tripped), and 0.2.7 removed it outright so that a +workflow meant to run indefinitely is not given an expiry date. 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 @@ -255,11 +256,12 @@ between parent and descendant stamps at equal generations. **Severity: 🟡 Medium — applies to nested loops.** -`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. +There is no iteration budget at all as of 0.2.7, which removed the per-instance +100,000-generation cap (see the node failure policy design). Even while it +existed it was per orchestration instance: a non-root inner loop starts in its +own child with its own counter, and an outer loop can spawn a new inner child +under every outer generation, so two loops that each stayed under the cap could +still produce a combinatorial amount of work and retained child state. 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/spec-failure-policy.md b/docs/spec-failure-policy.md new file mode 100644 index 00000000..5c4b5ad4 --- /dev/null +++ b/docs/spec-failure-policy.md @@ -0,0 +1,536 @@ +# Node Failure Policy Specification + +**Status:** Implemented in 0.2.7 +**Date:** 2026-08-19 (revised 2026-08-23 against the code) +**Target version:** 0.2.7 (unreleased). The design was written against an +unreleased 0.2.6; that version shipped on 2026-08-23 (tag `v0.2.6`), so the +work lands in 0.2.7 and every "0.2.6" boundary below reads 0.2.7. +**Related:** #155 (resilience gap); the wait-for-condition design + +## Overview + +Nothing is retried today, and one failed node fails the whole instance. Any +workflow that runs long enough will hit a deadlock, a lock timeout, or a +dropped connection. A background compactor for a bm25 index, meant to run +indefinitely, gets marked `failed` on the first bad night, and nothing runs +again until a human notices. + +This adds retries, and a policy for what happens when retrying doesn't help. +The policy only matters inside a loop, since a loop is the only case where the +workflow has somewhere to go other than `failed`. + +## API + +Three new arguments on `df.start()`. The example is that compactor: every five +minutes it merges the index's segments and records what it merged. + +```sql +SELECT df.start( + @> ( + df.wait_for_schedule('*/5 * * * *') + ~> df.sql($$SELECT tp_force_merge('docs_idx')$$) |=> 'merged' + ~> df.sql($$UPDATE compaction_log SET last_run = now(), segs = $merged$$) + ), + 'compactor', + max_attempts => 5, + max_backoff => '16s'::interval, + on_failure => 'continue' +); +``` + +**Parameters:** +- `max_attempts` - Tries per node, including the first. Defaults to `1`. +- `max_backoff` - Cap on the delay between tries. Defaults to `'16s'`. Taken as + an `interval` and converted with pgrx's `Interval::as_micros()`, which counts + a month as 30 days, so the conversion is deterministic. +- `on_failure` - What to do once the tries are spent. `'continue'` or + `'fail'`. Defaults to `'fail'`. + +The defaults reproduce the pre-0.2.7 behavior — one try, then fail the instance +— so the feature is opt-in and upgrading changes nothing. The example above +therefore has to name all three. + +## Alternatives: where the policy attaches + +This design puts the policy on `df.start()`, which is the coarsest of several +options. That choice was inherited from the first draft rather than argued for, +so it is set out here against the alternatives. + +The DSL builds a tree: leaves are operations (`df.sql()`, `df.http()`, +`df.http_multipart()`), interior nodes are composition (`~>`, `df.loop()`, +`df.join()`), and the root is the instance. A policy can attach at any of the +three. + +**A. Root — the instance. What this design implements.** + +One policy set once and inherited by every node, propagated through +`ExecutionContext.retry` and `SubtreeInput.retry`. There is one place to look +and nothing to thread through the DSL. The cost is a single lever for the whole +graph: a flaky vendor API and a local `INSERT` are treated identically, and +nested loops cannot differ from each other. + +**B. Leaf — the individual node.** + +```sql +df.http('https://vendor.example/api', max_attempts => 10) + ~> df.sql($$INSERT INTO local ...$$, max_attempts => 1) +``` + +The policy sits on the thing that actually fails, which is where the knowledge +lives. Two costs. Auto-wrap means the idiomatic form of a SQL node is a bare +string, so configuring one forces the explicit `df.sql(...)` call and gives up +the terser form. And a graph with thirty SQL nodes that should all retry has to +say so thirty times. Only the three node types that perform I/O would take the +arguments; the other nine are control flow, sleep, and signal, and a +`max_attempts` on `df.sleep()` would mean nothing. + +**C. Interior — a scoped region.** + +A wrapper setting the policy for everything beneath it: + +```sql +df.loop( + df.retry($$CALL fetch_from_vendor()$$ ~> $$CALL parse_response()$$, + max_attempts => 10, max_backoff => '1 minute') + ~> $$CALL store_locally()$$, -- outside the region, not retried + 'SELECT more_work()' +) +``` + +This is the most native to a composition DSL. It reads in the same direction as +`~>`, has the shape of a `try`/`with` block, and expresses "retry this *phase*" +without annotating every leaf. It is also cheap to build: the policy is already +threaded down the tree in `ExecutionContext.retry`, so a scope node replaces +that value for its subtree and changes nothing else. Its cost is a new node type +and a third place a reader must look to know what policy a node runs under. + +**D. No knobs — classify and fix the schedule.** + +Retry on transient errors, never on permanent ones, with a single built-in +backoff. The SQLSTATE rules under "What does not retry" are already half of +this. The argument for going further is that most callers cannot pick good +numbers, and every knob is a way to get it wrong. + +### These compose + +A, B, and C are a cascade, not a contest: the instance sets a default, a region +overrides it, a node overrides that — the relationship a GUC has with a +per-statement `SET`. **A can therefore ship first without foreclosing the +others.** Adding C or B later is additive, because a narrower scope only +overrides a default that already exists, and a graph that names neither behaves +exactly as it does today. + +One caveat if either is added later: a per-node or per-region policy lives in +the graph, and the graph is serialized into `FunctionInput` and `SubtreeInput`. +That is the same replay break class described under "Upgrade & Migration" — the +new fields must be omitted from the serialized form when they hold the inherited +value, or in-flight instances fail with a nondeterminism error. + +### `on_failure` is a separate question + +`max_attempts` and `max_backoff` answer "how hard do I try this operation", +which is a property of the operation. `on_failure` answers "what happens to the +enclosing iteration when it gives up", which is a property of the **loop**. +Fusing them onto one object is why `on_failure` has no effect outside a loop, +where the instance fails either way — recorded under "Behavior" below as a +consequence, but really a sign that the argument is attached to the wrong +thing. `df.loop(body, condition, +on_failure => 'continue')` would put it where it applies, and would let an outer +loop over batches fail while an inner loop over items skips a bad one, which A +cannot express at all. + +Unlike the retry scope, this one does not compose. Moving `on_failure` from +`df.start()` to `df.loop()` after release is a breaking change to a shipped +argument, so it is worth settling before 0.2.7 rather than after. + +## Behavior + +A failing `df.sql()`, `df.http()`, or `df.http_multipart()` node is retried +with exponential backoff, capped at `max_backoff`, up to `max_attempts` tries. +With `max_attempts => 5` that's four delays (1s, 2s, 4s, 8s), and the cap +doesn't bind until `max_attempts` goes past 5. Succeed on any try and the +workflow proceeds normally. + +When the tries run out, `'continue'` abandons the rest of the current loop +iteration and starts the next one. That's `continue` in the ordinary +loop-control sense, and the counterpart to the `df.break()` the DSL already +has. + +The next iteration begins at the top of the body, so what happens next is +whatever the body starts with. In the compactor that's +`df.wait_for_schedule()`, which computes the next tick strictly after the +current time, so the failed run's tick is skipped: the `UPDATE compaction_log` +never runs, the workflow parks until the next tick, and the log correctly +records no run for that tick. A body that doesn't open with a wait re-enters +right away, held to one iteration per second by `LOOP_MIN_ITER_DURATION`. + +Why abandon the whole iteration rather than skip just the failed node? In the +compactor, `tp_force_merge()` binds `$merged` and the next step writes it to +`compaction_log`. Skipping only the failed node would run that `UPDATE` with +`$merged` unbound, recording a compaction that never happened. Unwinding to the +loop means no step ever runs on a result its producer failed to compute. + +With no enclosing loop there is no next iteration, so `on_failure` has no +effect: the instance fails once the tries run out either way. A one-shot +workflow behaves as it does today, with retries added. + +`'fail'` moves the instance to `failed` as soon as the tries run out, +preserving the node's error. `'fail'` with `max_attempts => 1` is exactly the +pre-0.2.6 behavior. + +The two work in sequence: `max_attempts` and `max_backoff` govern the retries, +and `on_failure` takes over once they're spent. + +| | `'continue'` | `'fail'` | +|---|---|---| +| A retry succeeds | workflow proceeds | workflow proceeds | +| Tries exhausted, inside a loop | next iteration | instance fails | +| Tries exhausted, no loop | instance fails | instance fails | + +### Removing the iteration cap + +A `'continue'` policy is pointless if the loop has a deadline, and today it +does. `execute_function_graph.rs` fails any loop once `loop_iteration` reaches +`MAX_LOOP_ITERATIONS` (100,000), which at a five-minute cron is 347 days. The +constant and its check are removed. + +The comment on it says it prevents runaway loops from consuming resources +indefinitely, but `LOOP_MIN_ITER_DURATION` already does that, holding every +iteration to a second of wall clock with a compensating timer. The cap runs +after that guard, so it doesn't bound the rate of anything; it just sets an +expiry. A loop that busy-spins is already rate-limited, and a loop that +legitimately runs forever gets killed with a message telling the operator to +call `df.break()`. A genuinely runaway workflow is better served by +`df.cancel()` and by watching activity, both of which work from the first +iteration rather than 27 hours in. + +Watching `df.instances` alone isn't enough, though. Its `status` reads +`'running'` for a healthy eternal loop, for one blocked on a signal that never +arrives, and for one retrying a broken node forever — and `'continue'` makes +that last case reachable. So this design also adds `df.instance_activity()`, a +`LANGUAGE SQL STABLE` function over `df.instances` and `df.nodes` returning, for +each non-terminal instance, the timestamp of its last node transition, the +seconds since, its running and failed node counts, and its most recent node +error. `df.instance_activity('10 minutes')` is then "show me what has stopped +moving". It is a function rather than a view because view-level RLS +pass-through needs `security_invoker`, which is PostgreSQL 15+, while this +extension supports 13; an ordinary SECURITY INVOKER function inherits the +existing policies on both tables. + +`loop_iteration` stays in `FunctionInput` and `SubtreeInput`. It still +increments and is still carried across `continue_as_new`, and it remains useful +in traces; nothing reads it for control flow any more. + +### What does not retry + +Retry and the `on_failure` policy cover `df.sql()`, `df.http()`, and +`df.http_multipart()`. Graph errors (unknown node type, undeserializable child, +failed graph load) fail the instance immediately under both policies. Retrying +a malformed graph can't succeed, and looping on it forever would hide the +defect. + +Within those three node types the retry is *nearly* unconditional. Errors that +are a property of the statement rather than of the moment are not retried at +all, because a retry reproduces them byte for byte: SQLSTATE classes 42 (syntax +or access rule violation), 23 (integrity constraint violation), 28 (invalid +authorization), 3D (invalid catalog name), and 3F (invalid schema name). These +fail on the first try however high `max_attempts` is, and `on_failure` then +applies as usual. + +Everything else is retried: classes 40, 08, 53, 57, anything unclassified, and +every `df.http()` / `df.http_multipart()` error, which carries no SQLSTATE at +all. Class 22 (data exception, e.g. division by zero) is retried deliberately — +it describes the *data*, which another node can change between tries. + +Classification is by SQLSTATE, not by error text. `execute_sql` reads +`sqlx::Error::as_database_error()?.code()` and stamps `[SQLSTATE xxxxx]` into +the message before it is stringified into the activity error; the orchestration +matches on the two-character class. duroxide's +`schedule_activity_with_retry` retries every error with no predicate hook, so +the retry loop is hand-rolled — it emits the identical sequence of durable +operations (one `schedule_activity` per try, a timer between) but can break out +early. That equivalence is what preserves replay compatibility. + +### Relation to df.break() + +Both unwind to the nearest enclosing loop: `df.break()` exits it, a continue +starts the next iteration. The unwind passes through compound nodes, so a +failure in one branch of `df.join()` continues the iteration containing the +join. + +Two consequences of routing a continue through a parallel node are worth +stating, since both are deterministic and neither is a replay hazard: + +- **`df.join()` honours the first branch that carries a continue.** Branch + results are inspected in fixed input order, so if an earlier branch returns a + continue and a later one returns a genuine failure (a malformed graph, say), + the iteration is abandoned and the failure is not surfaced that iteration. It + reappears the moment the earlier branch stops continuing. +- **`df.race()` treats an exhausted branch as a completed one.** A branch whose + retries ran out under `'continue'` returns *successfully* with a continue + marker, so it can win the race and abandon the iteration even though the + other branch might have succeeded; the loser is cancelled as usual. Under + `'continue'` this is benign — the next iteration retries both — but it means + a fast-failing branch can starve a slow-succeeding one. + +### Validation + +`df.start()` rejects `max_attempts < 1`, a non-positive `max_backoff`, and any +`on_failure` other than `'continue'` or `'fail'`. + +## Observability + +An abandoned node is recorded `failed` with its error. The instance stays +`running`. + +A retried node is stamped once, after the retries settle: node status is +written by `execute_function_node_with_vars` around the whole handler, so +`df.instance_nodes()` shows `completed` for a node that failed twice and +succeeded on the third try. The individual attempts are visible in the +duroxide history and in the worker log (`~/.pgrx/17.log`), where duroxide +traces `Activity '' attempt N/M failed: ... Retrying...`. + +```sql +SELECT node_id, node_type, status, error +FROM df.instance_nodes('a1b2c3d4') +WHERE status = 'failed'; +``` + +This is the cost of `'continue'`. A compactor whose target table has been +dropped keeps looping and keeps failing without ever changing status. +Monitoring has to watch for failed nodes under running instances, because a +healthy instance status no longer means a healthy workflow. + +## Implementation + +The retry policy travels in `FunctionInput` (`src/types.rs`, built in +`src/dsl.rs`), threaded into `SubtreeInput` so sub-orchestrations inherit it, +and carried across `continue_as_new` with the rest of the input. Inside the +orchestration it rides on `ExecutionContext` alongside `vars` and +`loop_iteration`, which is what puts it in reach of the three activity call +sites and of `build_subtree_input`. + +One `RetryPolicySpec { max_attempts: u32, max_backoff_micros: i64, on_failure: +OnFailure }` type carries all three arguments, so `df.start()`, +`FunctionInput`, `SubtreeInput`, and `ExecutionContext` all thread a single +field rather than three parallel ones. It stores microseconds rather than a +`Duration` so the serialized form is a plain integer that round-trips through +history exactly. + +It must not come from a GUC. Orchestration code is replayed, and a GUC read at +execution time would produce different durable operations on replay. Carrying +the policy in the recorded input is why these are per-instance arguments and +not server settings. + +The three activity call sites in `src/orchestrations/execute_function_graph.rs` +move from `ctx.schedule_activity()` to `ctx.schedule_activity_with_retry()`. +duroxide 0.1.30 spells the policy as a `RetryPolicy` struct, not the +`Backoff`/`initial`/`coefficient` names this design first used: + +```rust +duroxide::RetryPolicy { + max_attempts, // u32, from FunctionInput + backoff: duroxide::BackoffStrategy::Exponential { + base: Duration::from_secs(1), + multiplier: 2.0, + max: max_backoff, + }, + timeout: None, +} +``` + +`RetryPolicy::new()` asserts `max_attempts >= 1` and would panic inside the +orchestration, so the struct is built literally and the bound is enforced at +`df.start()`. `delay_for_attempt(n)` is `base * multiplier^(n-1)` capped at +`max`, which is the 1s/2s/4s/8s sequence above. duroxide schedules each backoff +as a durable timer, so a workflow waiting to retry survives a restart, and it +retries on any activity error (`timeout: None` means the timeout path, which is +deliberately not retried, never engages). + +A new `NodeError::Continue(String)` joins `Break` and `Failure`, produced at +those three call sites when the tries run out under `'continue'`. Compound +nodes propagate it through `?` with no new code, the same way they already +propagate `Break`. Six sites match on `NodeError` explicitly and need a new +arm, in four groups: + +| Site | Behavior | +|---|---| +| Loop body (`run_loop_iteration`, both the body and the while-condition arms) | Warn, discard the iteration, run the next one | +| Node status (`execute_function_node_with_vars`) | Record the node `failed` with its error | +| Subtree envelope encode (`execute_subtree`) / decode (`parse_subtree_envelope`) | New `SubtreeControl::Continue`, so it crosses the sub-orchestration boundary | +| Top level (`execute`) | Instance `failed`, original error preserved | + +The unit test at the bottom of `execute_function_graph.rs` that asserts on +`NodeError::Break` matches with a catch-all `other =>` arm and compiles +unchanged; it is not one of the six. + +`run_loop_iteration` needs no signature change. It already returns +`Result>`, where `Ok(None)` means "run the next iteration", so a +`Continue` is handled by tracing a warning and returning `Ok(None)` from both +arms. The while-condition is deliberately *skipped* when the body continues, +rather than being evaluated against a half-finished iteration: the condition +typically reads named results (`$extracted.count`) that the abandoned iteration +never produced, so evaluating it would replace a clear "this iteration failed" +with an unrelated substitution error. The consequence, which the Observability +section states outright, is that a `while` loop whose body fails on every +iteration never terminates. + +`MAX_LOOP_ITERATIONS` and the `next_iteration >= MAX_LOOP_ITERATIONS` check in +`execute_loop_node` are deleted. The constant has no other reader, so it goes +with the check rather than being left behind unused. The doc comment on +`FunctionInput::loop_iteration` in `src/types.rs` says "Used to enforce a +maximum iteration safeguard" and needs updating with it. + +## Upgrade & Migration + +No `df` table changes. The policy lives in duroxide history, not in +`df.instances`. + +`df.start()` gains three defaulted arguments, which changes its signature. +Follow the `transaction_mode` precedent in `sql/pg_durable--0.2.4--0.2.5.sql`: + +1. Bump `Cargo.toml` to `0.2.7` and create `sql/pg_durable--0.2.6--0.2.7.sql` + (the 0.2.6 release is already tagged, so 0.2.6's own upgrade script is + immutable). +2. Add a Rust `start_v3` bound to `start_v3_wrapper`. Keep `start_v2` as + `#[pg_extern(sql = false)]` so it emits no DDL but keeps its symbol — + dropping its `name`/`schema` attributes and its `default!()` wrappers, the + way `start()` was reduced when `start_v2` superseded it. +3. In `sql/pg_durable--0.2.6--0.2.7.sql`, drop + `df.start(text, text, text, text)` and create the seven-argument function + from the pgrx-generated DDL, copied verbatim so Scenario A sees identical + fresh-install and upgrade schemas. +4. Add a "v0.2.6 → v0.2.7" section to `docs/upgrade-testing.md`. + +The overloads can't coexist. With defaults on both, a four-argument call +matches both signatures and PostgreSQL raises "function is not unique". + +**B1 (new `.so`, un-upgraded schema):** a 0.2.5/0.2.6 schema declares +`df.start(text, text, text, text)` against `start_v2_wrapper` and keeps +resolving to it (and a pre-0.2.5 schema its three-argument `start_wrapper`). +Those instances run `'fail'` and don't expose the new arguments. + +**`transaction_mode => 'new'`:** that path re-issues the start on a loopback +session, and `client::start_on_new_session` deliberately calls +`df.start($1, $2, $3)` with three positional arguments so it also resolves on +pre-0.2.5 schemas. The retry arguments have to reach the inner start, but +adding them unconditionally would break that resolution under B1, where +`start_v2` is still the entry point. So the policy is passed as an `Option`: +`None` (the `start_v2` entry point) keeps the existing three-argument +statement verbatim, and `Some(policy)` — only reachable through `start_v3`, +which only exists on a 0.2.7+ schema — issues +`df.start($1, $2, $3, 'caller', $4, $5, $6)`. + +**Instances started after an upgrade:** a caller who upgrades and keeps using +the four-argument `df.start()` resolves to the new function but picks up +defaults of `max_attempts => 1, on_failure => 'fail'`, which is exactly what +the four-argument function did. Nothing changes without an explicit opt-in. +This is deliberate: silently converting every existing workflow's fail-fast +semantics into retry-and-continue is not a change a caller should discover in +production. It is also why the existing E2E suite needed no edits — the five +tests that assert a node failure still see one. + +**Replay of in-flight instances:** the new `FunctionInput` and `SubtreeInput` +fields are `#[serde(default)]`, defaulting to `'fail'` with +`max_attempts = 1`, so instances started by the old binary keep their old +behavior. The first attempt of `schedule_activity_with_retry` records the same +history operation as `schedule_activity` (duroxide's retry helper simply calls +`schedule_activity` in a loop, adding a timer only between attempts), so +existing activity histories replay unchanged. + +A serde default alone is *not* sufficient, and the first draft of this design +got that wrong. A default only governs deserialization; the field is still +written on the way out, so the `execute-subtree` envelope a 0.2.7 parent emits +would no longer be byte-equal to the one a 0.2.6 parent recorded. duroxide +matches a sub-orchestration schedule with `name == en && input == ei` +(`replay_engine::action_matches_event_kind`), so that mismatch would have +failed every in-flight JOIN branch, RACE branch, and non-root loop child with a +nondeterminism error — precisely the break `docs/upgrade-testing.md` records for +v0.2.4 → v0.2.5, where adding `instance_id`, `vars`, `label`, and `iteration` to +the same envelope broke in-flight parallel branches unconditionally. + +Both fields are therefore also declared +`skip_serializing_if = "RetryPolicySpec::is_legacy"`, so an instance carrying +the legacy policy serializes to exactly the pre-0.2.7 shape. The predicate keys +off the legacy value rather than `Default`, because an instance started under +0.2.7 with a real policy must still carry it into its subtrees. Two unit tests +pin both directions: a legacy envelope must contain no `retry` key, and a +non-legacy one must round-trip its policy. + +Removing the iteration cap is safe on replay because it only ever turned a +continuation into a failure. An in-flight loop past 100,000 iterations would +already have failed under the old binary, so there is no history in which the +old code continued and the new code doesn't. + +## Testing + +**Unit** (`cargo test --features pg17 --no-default-features --lib`, and +`./scripts/test-unit.sh` for anything needing a live backend): +- Argument validation: `max_attempts < 1`, non-positive `max_backoff`, unknown + `on_failure`. Validation is factored into a plain `parse_retry_policy()` that + returns `Result`, so these are ordinary `#[test]`s; + `df.start()` only turns the `Err` into `pgrx::error!`. This follows the + repository's existing split — `pgrx::error!` paths themselves are asserted + from SQL (see `55_start_transaction_mode.sql`), not with `should_panic`. +- Backoff sequence derivation, including the `max_backoff` cap. This asserts on + duroxide's `BackoffStrategy::delay_for_attempt`, pinning the sequence this + design promises (1s, 2s, 4s, 8s) to the policy we actually construct. +- `FunctionInput` and `SubtreeInput` deserialize from pre-0.2.7 JSON with no + retry field, yielding `'fail'` and `max_attempts = 1`. `SubtreeInput` is + private to `execute_function_graph.rs`, so its test lives in that file's + existing `#[cfg(test)] mod tests`, next to the break-propagation test. +- Round-trip: a `RetryPolicySpec` survives serialization into `FunctionInput` + and back, so `continue_as_new` cannot silently reset the policy. + +**E2E** (`tests/e2e/sql/68_failure_policy.sql` — 67 was taken by a branch in +flight): +1. **Transient recovery.** A node fails twice, then succeeds; the instance + completes and the attempt count is exactly 3. +2. **Continue.** A loop whose body fails through the first iteration and + succeeds in the second. The instance completes, the attempt count shows the + abandoned iteration's tries, and the node downstream of the failure ran only + once — proving the rest of the failing iteration was skipped. +3. **No enclosing loop.** A one-shot under `'continue'` runs out of tries and + fails, the original SQL error surfaces in the instance output, and + `df.instance_nodes()` reports the node failed. +4. **Fail.** `on_failure => 'fail', max_attempts => 1` inside a loop fails the + instance after exactly one attempt. +5. **Defaults are legacy.** A plain `df.start()` with no policy arguments, on a + loop whose body always fails, fails the instance after exactly one attempt — + pinning "upgrading changes nothing" as a test rather than a claim. +6. **Validation.** `max_attempts => 0`, a negative `max_backoff`, and an + unrecognised `on_failure` are each rejected by `df.start()`. + +Plus `tests/e2e/sql/69_instance_activity.sql` for the monitoring function: a +busy instance is listed with a live `last_activity_at`; the `idle_for` argument +filters; a loop wedged under `'continue'` surfaces a non-zero +`failed_node_count` and its `last_error` while still reporting `'running'`; +terminal instances are excluded; and a second role sees none of it. + +Attempts are counted with a **sequence**, not a counter table: the failing +attempt's transaction rolls back, taking any row it inserted with it, whereas +`nextval()` is non-transactional and survives. `SELECT 1 / (CASE WHEN +nextval('s') >= 3 THEN 1 ELSE 0 END)` both counts the attempt and decides +whether it fails, and `last_value` afterwards is an exact attempt count. + +Retries cost wall clock and the E2E await helper gives up well before the +defaults would finish, so every case pins `max_attempts` (1, 2, or 5) and +`max_backoff => '1 second'`. + +Two cases from the original plan are **not** covered by E2E. A graph-level +error (unknown node type) cannot be constructed through the DSL — the +`nodes_node_type_chk` constraint rejects it — so the "not retried" guarantee +for graph errors rests on where `NodeError::Continue` is produced (only the +three activity call sites) rather than on a test. The old iteration cap +likewise has no E2E: `loop_iteration` is an orchestration-input field with no +SQL surface, and its removal is a deletion with nothing left to assert against. + +**Existing tests that assert a failure must opt out of the new default.** Five +E2E tests (`13_user_isolation`, `14_database`, `45_connection_limit_timeout`, +`61_loop_child_start_failure`, `64_loop_branch_failure`) were written when the +first node error was fatal. Under the new defaults they either time out waiting +for a failure that is still being retried, loop forever under `'continue'`, or +— in the connection-limit case — succeed on the retry. Each now passes +`max_attempts => 1, on_failure => 'fail'`, which is both the correct expression +of what they test and the first real use of the escape hatch the release notes +recommend. diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 294fcae0..31eff0b5 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -203,6 +203,21 @@ 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. +### v0.2.6 → v0.2.7 + +#### Add the node failure policy to `df.start()` +- **DDL change (df schema):** Replaces `df.start(text, text, text, text)` with `df.start(text, text, text, text, int, interval, text)`, bound to the new C symbol `start_v3_wrapper`. The three new trailing arguments are `max_attempts int DEFAULT 1`, `max_backoff interval DEFAULT '16 seconds'`, and `on_failure text DEFAULT 'fail'`. A failing `df.sql()`, `df.http()`, or `df.http_multipart()` node is retried with exponential backoff (1s, doubling, capped at `max_backoff`) up to `max_attempts` tries; once those are spent, `on_failure` chooses between abandoning the rest of the current loop iteration (`'continue'`) and failing the instance (`'fail'`). With no enclosing loop there is no next iteration, so both settings fail the instance. The defaults reproduce the pre-0.2.7 behaviour exactly. +- **DDL change (df schema):** Adds `df.instance_activity(interval)`, a `LANGUAGE SQL STABLE` function reporting non-terminal instances with their last node transition, idle time, running/failed node counts, and last error. It is a function rather than a view because view-level RLS pass-through (`security_invoker`) requires PostgreSQL 15 and this extension supports 13; as an ordinary SECURITY INVOKER function the existing policies on `df.instances` and `df.nodes` filter it to the caller's own rows. The upgrade script's definition is kept byte-identical to the fresh-install block in `src/lib.rs` so the Scenario A comparison matches. +- **Upgrade script:** `sql/pg_durable--0.2.6--0.2.7.sql` runs `DROP FUNCTION IF EXISTS df.start(text, text, text, text)` followed by the `CREATE FUNCTION df.start(...)` seven-argument DDL copied verbatim from the pgrx-generated fresh-install output. The drop is required, not cosmetic: with defaults on the trailing arguments of both, a four-argument call such as `df.start(fut, label, database, transaction_mode)` would match both and PostgreSQL would raise `function ... is not unique`. New `df.*` functions retain PostgreSQL's default PUBLIC `EXECUTE`, gated by `USAGE ON SCHEMA df`, so no explicit `GRANT` is needed. +- **No behaviour change for existing callers:** On an upgraded schema, an unchanged four-argument (or one-argument) `df.start()` call resolves to the new function, but its new arguments default to `max_attempts => 1, on_failure => 'fail'` — one attempt, then fail — which is exactly what the four-argument function did. Retrying is opt-in. This is why no existing E2E test needed changing: the five tests that assert a node failure still observe it. The one deliberate exception to "retry everything" is that SQLSTATE classes 42, 23, 28, 3D and 3F are never retried even when `max_attempts` is raised, since a retry reproduces the identical error. +- **Scenario A considerations:** `df.instance_activity(interval)` must appear identically on both paths, which is why the upgrade script carries a verbatim copy of the fresh-install definition. A fresh install exposes exactly one `df.start`, the seven-argument one — `src/dsl.rs` keeps the four-argument Rust `start_v2()` for binary compatibility but marks it `#[pg_extern(sql = false)]`, so it contributes no DDL. The upgrade script's drop-then-create reaches the same single-overload end state, so the Scenario A snapshot matches. +- **Scenario B1 considerations:** The `start_v2_wrapper` symbol is deliberately preserved in the binary by that `sql = false` Rust function, which still takes exactly four arguments and delegates with the legacy policy (one attempt, then fail — the pre-0.2.7 behaviour). Pre-0.2.7 schemas (0.2.2 through 0.2.6) declare `df.start(text, text, text, text)` against `start_v2_wrapper` and keep resolving to it unchanged; they simply do not expose the new arguments. `transaction_mode => 'new'` starts the durable function on a separate session by calling `df.start()` there: it uses the original three-positional-argument form whenever the policy is the legacy one, which resolves on every shipped schema, and only issues the seven-argument form when a caller actually supplied a policy (which can only happen on an upgraded schema). +- **Scenario B2 considerations:** No data migration. Both structures that carry the policy default it to the legacy one-attempt-then-fail behaviour when the field is absent, so orchestrations enqueued before the upgrade replay with their original semantics. Keeping in-flight histories replayable takes two separate guarantees, because duroxide matches a recorded schedule by exact equality on its serialized input: + - *Activities.* The retry helper schedules its first attempt with the same activity name and the same input the previous binary recorded, and only creates a backoff timer after an attempt has already failed — something no successfully recorded pre-0.2.7 history contains. In-flight activity schedules therefore match unchanged. + - *The `execute-subtree` envelope.* `SubtreeInput` gained a `retry` field, and a JOIN (`&`) branch, RACE (`|`) branch, or non-root `df.loop()` child scheduled before the upgrade recorded that envelope without it. Serialising the field unconditionally would have re-emitted a different input string on replay and failed every such in-flight branch with a nondeterminism error — exactly the break documented under v0.2.4 → v0.2.5. Both `retry` fields are therefore declared `skip_serializing_if = "RetryPolicySpec::is_legacy"`, so an instance carrying the legacy policy serializes byte-identically to the pre-0.2.7 envelope and in-flight parallel branches and loop children replay unchanged. The suppression keys off the legacy value specifically, not `Default`, so an instance started under 0.2.7 with a real policy still propagates it into its subtrees. +- **Downgrade considerations:** Orchestration input is downgrade-safe: neither `FunctionInput` nor `SubtreeInput` uses `#[serde(deny_unknown_fields)]`, so a `retry` field written by 0.2.7 is ignored rather than rejected by a 0.2.6 binary, and the instance resumes with the old one-attempt-then-fail behaviour. The *schema* is not downgradable, though, and this line ships no downgrade scripts: once `ALTER EXTENSION UPDATE TO '0.2.7'` has run, `df.start` is bound to `start_v3_wrapper`, a symbol a 0.2.6 `.so` does not export, so reverting the binary alone breaks every `df.start()` call. Revert the binary only on a schema that was never upgraded. +- **Runtime change (no DDL):** The 100,000-iteration `df.loop()` cap was removed. It was never a meaningful storage bound (a large carried result exhausts storage long before the count trips), and it gave a compactor-style workflow meant to run indefinitely an arbitrary expiry date. + ### v0.2.5 → v0.2.6 #### Remove `df.ensure_durofut()` diff --git a/scripts/test-upgrade.sh b/scripts/test-upgrade.sh index 6975e314..11e4711a 100755 --- a/scripts/test-upgrade.sh +++ b/scripts/test-upgrade.sh @@ -1011,7 +1011,7 @@ test_b2_grant_usage_after_upgrade() { out=$(run_sql_capture "SELECT df.grant_usage('${probe_role}');") || { echo "$out"; return 1; } # ... and a representative privilege was actually granted. - assert_sql_equals "SELECT has_function_privilege('${probe_role}', 'df.start(text, text, text, text)', 'EXECUTE');" "t" || return 1 + assert_sql_equals "SELECT has_function_privilege('${probe_role}', 'df.start(text, text, text, text, int, interval, text)', 'EXECUTE');" "t" || return 1 # Clean up the probe role. run_sql_capture "DROP OWNED BY ${probe_role}; DROP ROLE IF EXISTS ${probe_role};" >/dev/null 2>&1 || true diff --git a/sql/pg_durable--0.2.6--0.2.7.sql b/sql/pg_durable--0.2.6--0.2.7.sql new file mode 100644 index 00000000..adad92ad --- /dev/null +++ b/sql/pg_durable--0.2.6--0.2.7.sql @@ -0,0 +1,145 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- pg_durable upgrade: 0.2.6 -> 0.2.7 +-- +-- See docs/upgrade-testing.md for the upgrade-script and backward-compatibility +-- requirements (Scenario A / B1 / B2). + +-- ============================================================================ +-- Add the df.start() node failure policy: max_attempts, max_backoff, on_failure. +-- +-- A failing df.sql(), df.http(), or df.http_multipart() node is now retried with +-- exponential backoff (1s, doubling, capped at max_backoff) up to max_attempts +-- tries. Once those are spent, on_failure decides between abandoning the rest of +-- the current loop iteration ('continue', the default) and failing the instance +-- ('fail'). With no enclosing loop there is no next iteration, so both settings +-- fail the instance. +-- +-- The four-argument df.start() is dropped and replaced by a seven-argument one +-- bound to a new C symbol (start_v3_wrapper). Both cannot coexist: with defaults +-- on the trailing arguments of each, a four-argument call such as +-- df.start(fut, label, database, transaction_mode) matches both and PostgreSQL +-- raises "function is not unique". Dropping leaves exactly one df.start at every +-- arity. +-- +-- Scenario B1 (new .so, un-upgraded schema) is preserved without the overload: +-- src/dsl.rs keeps the four-argument Rust fn start_v2() and therefore the +-- start_v2_wrapper symbol, marked #[pg_extern(sql = false)] so it contributes no +-- DDL. Pre-0.2.7 schemas declare df.start(text, text, text, text) against that +-- symbol and keep resolving to it; those instances run the pre-0.2.7 behaviour +-- (one attempt, then fail) and simply do not expose the new arguments. Because +-- sql = false emits nothing, a fresh 0.2.7 install has only the seven-argument +-- df.start, which is what this script produces too (Scenario A). +-- +-- Note for existing callers: a four-argument df.start() on an upgraded schema +-- resolves to the new function and therefore picks up the new defaults, so a +-- workflow that used to fail on its first node error now retries five times and, +-- inside a loop, keeps running afterwards. Pass +-- max_attempts => 1, on_failure => 'fail' to restore the previous behaviour. +-- +-- The CREATE FUNCTION block is the pgrx-generated fresh-install DDL for +-- src/dsl.rs::start_v3 copied verbatim, so the Scenario A snapshot matches a +-- fresh 0.2.7 install. New df.* functions retain PostgreSQL's default PUBLIC +-- EXECUTE (gated by USAGE ON SCHEMA df), so no explicit GRANT is needed. +-- ============================================================================ +DROP FUNCTION IF EXISTS df.start(text, text, text, text); + +-- pg_durable::dsl::start +CREATE FUNCTION df."start"( + "fut" TEXT, /* &str */ + "label" TEXT DEFAULT NULL, /* core::option::Option<&str> */ + "database" TEXT DEFAULT NULL, /* core::option::Option<&str> */ + "transaction_mode" TEXT DEFAULT 'caller', /* &str */ + "max_attempts" INT DEFAULT 1, /* i32 */ + "max_backoff" interval DEFAULT '16 seconds', /* pgrx::datum::interval::Interval */ + "on_failure" TEXT DEFAULT 'fail' /* &str */ +) RETURNS TEXT /* alloc::string::String */ + +LANGUAGE c /* Rust */ +AS 'MODULE_PATHNAME', 'start_v3_wrapper'; + +-- ============================================================================ +-- df.instance_activity(): report whether an instance is making progress. +-- +-- New in 0.2.7. Definition kept byte-identical to the fresh-install block in +-- src/lib.rs (extension_sql! name = "instance_activity") so the Scenario A +-- schema comparison matches. +-- ============================================================================ + +-- df.instance_activity(): is this workflow actually doing anything? +-- +-- A long-running instance and a wedged one both read 'running', so status alone +-- cannot tell them apart. This reports when each non-terminal instance last +-- transitioned a node, how long it has been quiet since, and whether any node is +-- currently failing -- which is what a retrying-and-abandoning loop looks like +-- from the outside. +-- +-- SECURITY INVOKER (the default) so the row-level security policies on +-- df.instances and df.nodes apply: a caller sees only their own instances. +-- Deliberately a function rather than a view, because view-level RLS +-- pass-through (security_invoker) requires PostgreSQL 15 and this extension +-- supports 13. +-- +-- p_idle_for filters to instances quiet for at least that long. The default of +-- zero reports every non-terminal instance, so the common call is simply +-- SELECT * FROM df.instance_activity() ORDER BY idle_for_seconds DESC; +-- +-- Idle time is measured against clock_timestamp(), not now(). now() is the +-- calling transaction's start time, and the worker keeps writing node +-- timestamps after that, so a busy instance's last activity can be *later* +-- than now() -- which would make its idle time negative and drop it below any +-- threshold, including zero. Reading a moving clock makes the function +-- VOLATILE, which is honest: two calls in one transaction can legitimately +-- differ. +-- +-- GREATEST and EXTRACT are parser constructs and cannot be schema-qualified; +-- the pinned search_path resolves the remaining names to pg_catalog. +CREATE OR REPLACE FUNCTION df.instance_activity(p_idle_for interval DEFAULT '0 seconds') +RETURNS TABLE ( + instance_id VARCHAR(8), + label TEXT, + status TEXT, + last_activity_at TIMESTAMPTZ, + idle_for_seconds DOUBLE PRECISION, + running_node_count BIGINT, + failed_node_count BIGINT, + last_error TEXT +) +LANGUAGE SQL +VOLATILE +SET search_path = pg_catalog, df, pg_temp +AS $fn$ + SELECT + i.id, + i.label, + i.status, + activity.last_activity_at, + GREATEST(0, EXTRACT(EPOCH FROM pg_catalog.clock_timestamp() - activity.last_activity_at))::double precision, + activity.running_node_count, + activity.failed_node_count, + activity.last_error + FROM df.instances i + CROSS JOIN LATERAL ( + SELECT + -- GREATEST ignores NULLs, so an instance whose nodes have never + -- transitioned still reports its own updated_at. + GREATEST(i.updated_at, max(n.updated_at)) AS last_activity_at, + count(*) FILTER (WHERE n.status = 'running') AS running_node_count, + count(*) FILTER (WHERE n.status = 'failed') AS failed_node_count, + -- A failed node's message is written to df.nodes.result, not to + -- df.nodes.error, which nothing writes. #>> '{}' unwraps the JSONB + -- scalar so the caller gets the message, not a quoted JSON string. + (array_agg(n.result #>> '{}' ORDER BY n.updated_at DESC) + FILTER (WHERE n.status = 'failed' AND n.result IS NOT NULL))[1] AS last_error + FROM df.nodes n + WHERE n.instance_id = i.id + ) AS activity + WHERE (i.status IS NULL OR i.status NOT IN ('completed', 'failed', 'cancelled')) + AND pg_catalog.clock_timestamp() - activity.last_activity_at >= p_idle_for; +$fn$; + +COMMENT ON FUNCTION df.instance_activity(interval) IS + 'Non-terminal instances with their last node transition, how long they have been idle, ' + 'and any current node failure. Use to tell a working workflow from a wedged one, which ' + 'status alone cannot show. RLS-filtered to the calling user.'; diff --git a/src/activities/execute_sql.rs b/src/activities/execute_sql.rs index 29e7edeb..2995431b 100644 --- a/src/activities/execute_sql.rs +++ b/src/activities/execute_sql.rs @@ -254,7 +254,15 @@ pub async fn execute( Ok(result.to_string()) } Err(e) => { - let err_msg = format!("SQL execution failed: {e}"); + // Stamp the SQLSTATE so the orchestration can tell a statement that is simply + // wrong (undefined table, syntax error, constraint violation) from one that + // failed because of the moment (deadlock, dropped connection, resource limit) + // and skip the retries that cannot help. The code has to travel as text because + // the activity boundary is `Result`. + let err_msg = match e.as_database_error().and_then(|db| db.code()) { + Some(code) => format!("SQL execution failed [SQLSTATE {code}]: {e}"), + None => format!("SQL execution failed: {e}"), + }; ctx.trace_info(&err_msg); Err(err_msg) } diff --git a/src/client.rs b/src/client.rs index a5eaba48..5f63e5a8 100644 --- a/src/client.rs +++ b/src/client.rs @@ -266,6 +266,7 @@ async fn start_on_new_session( fut: &str, label: &Option, database: &Option, + retry: Option, ) -> Result { use sqlx::Row; @@ -295,16 +296,40 @@ async fn start_on_new_session( // Three positional arguments on purpose: this must also resolve on schemas // predating `transaction_mode`, where `df.start` takes exactly three. On - // current schemas it resolves to the four-argument `df.start` and defaults + // current schemas it resolves to the newest `df.start` and defaults // to 'caller', which is what we want — the separate session is already the // new transaction, so it must not recurse. - let row = sqlx::query("SELECT df.start($1, $2, $3) AS id") - .bind(fut) - .bind(label) - .bind(database) - .fetch_one(&mut *conn) - .await - .map_err(classify_new_transaction_start_error)?; + // + // `retry` is `Some` only when the caller reached df.start() through a schema that + // declares the failure-policy arguments, so forwarding them cannot break resolution on + // an older schema (which would have entered through the four-argument symbol and passed + // `None`). The interval is bound as a microsecond string rather than an interval type so + // no extra sqlx type mapping is needed. + let row = match retry { + Some(spec) => { + sqlx::query("SELECT df.start($1, $2, $3, 'caller', $4, $5::interval, $6) AS id") + .bind(fut) + .bind(label) + .bind(database) + .bind(spec.max_attempts as i32) + .bind(format!("{} microseconds", spec.max_backoff_micros)) + .bind(match spec.on_failure { + crate::types::OnFailure::Continue => "continue", + crate::types::OnFailure::Fail => "fail", + }) + .fetch_one(&mut *conn) + .await + } + None => { + sqlx::query("SELECT df.start($1, $2, $3) AS id") + .bind(fut) + .bind(label) + .bind(database) + .fetch_one(&mut *conn) + .await + } + } + .map_err(classify_new_transaction_start_error)?; row.try_get("id") .map_err(|e| format!("df.start() on new transaction returned no instance id: {e}")) @@ -333,6 +358,7 @@ pub fn start_in_new_transaction( label: Option<&str>, database: Option<&str>, user: &str, + retry: Option, ) -> Result { use sqlx::Connection; @@ -351,7 +377,7 @@ pub fn start_in_new_transaction( // instance property for the worker to execute against. let mut conn = connect_as_user_for_new_transaction(&user).await?; - let result = start_on_new_session(&mut conn, &fut, &label, &database).await; + let result = start_on_new_session(&mut conn, &fut, &label, &database, retry).await; // Close explicitly so the extra backend exits promptly rather than // lingering until the server notices a dropped socket. diff --git a/src/dsl.rs b/src/dsl.rs index 7070e52b..ba9e6a55 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, + MaterializedNode, OnFailure, RetryPolicySpec, }; /// Check if we're running inside a workflow context (background worker connection). @@ -941,6 +941,50 @@ fn acquire_new_transaction_start_slot() -> Result Result { + if max_attempts < 1 { + return Err(format!( + "invalid max_attempts {max_attempts} for df.start(): must be at least 1 \ + (1 disables retries)" + )); + } + + // as_micros() counts a month as 30 days, so the conversion is a pure function of the + // interval and cannot vary with the calendar between replays. + let micros = max_backoff.as_micros(); + if micros <= 0 { + return Err("invalid max_backoff for df.start(): must be a positive interval".to_string()); + } + let max_backoff_micros = i64::try_from(micros) + .map_err(|_| "invalid max_backoff for df.start(): interval is too large".to_string())?; + + let on_failure = if on_failure.eq_ignore_ascii_case("continue") { + OnFailure::Continue + } else if on_failure.eq_ignore_ascii_case("fail") { + OnFailure::Fail + } else { + return Err(format!( + "invalid on_failure \"{on_failure}\" for df.start(): expected 'continue' or 'fail'" + )); + }; + + Ok(RetryPolicySpec { + max_attempts: max_attempts as u32, + max_backoff_micros, + on_failure, + }) +} + /// Starts a durable SQL function. /// /// The fut argument can be either Durofut JSON or plain SQL string (auto-wrapped). @@ -963,20 +1007,74 @@ fn acquire_new_transaction_start_slot() -> Result, "NULL"), database: default!(Option<&str>, "NULL"), transaction_mode: default!(&str, "'caller'"), + max_attempts: default!(i32, "1"), + max_backoff: default!(pgrx::datum::Interval, "'16 seconds'"), + on_failure: default!(&str, "'fail'"), +) -> String { + let retry = match parse_retry_policy(max_attempts, max_backoff, on_failure) { + Ok(spec) => spec, + Err(e) => pgrx::error!("{}", e), + }; + start_dispatch(fut, label, database, transaction_mode, retry, Some(retry)) +} + +/// Legacy four-argument `df.start()`, retained for binary compatibility only. +/// +/// Schemas from before the failure policy existed declare `df.start(text, text, text, text)` +/// against this symbol, so it must keep taking exactly four arguments. Instances started +/// through it run the pre-policy behavior (one attempt, then fail). It also passes `None` for +/// the loopback policy, so `transaction_mode => 'new'` keeps issuing the three-positional +/// inner `df.start()` that resolves on those older schemas. +#[pg_extern(sql = false)] +pub fn start_v2( + fut: &str, + label: Option<&str>, + database: Option<&str>, + transaction_mode: &str, +) -> String { + start_dispatch( + fut, + label, + database, + transaction_mode, + RetryPolicySpec::legacy(), + None, + ) +} + +/// Shared body of every `df.start()` entry point: validate the transaction mode and branch. +/// +/// `loopback_retry` is `Some` only when the caller reached us through a schema that declares +/// the failure-policy arguments, which is what lets `transaction_mode => 'new'` forward them +/// without breaking argument resolution on older schemas. +fn start_dispatch( + fut: &str, + label: Option<&str>, + database: Option<&str>, + transaction_mode: &str, + retry: RetryPolicySpec, + loopback_retry: Option, ) -> String { // Reject anything we do not recognise. Silently treating a typo as the // default would hand back an instance id for a start the caller believes // survives their rollback, and which quietly does not. if transaction_mode.eq_ignore_ascii_case(TXN_MODE_CALLER) { - start_in_caller_transaction(fut, label, database) + start_in_caller_transaction(fut, label, database, retry) } else if transaction_mode.eq_ignore_ascii_case(TXN_MODE_NEW) { - start_in_new_transaction(fut, label, database) + start_in_new_transaction(fut, label, database, loopback_retry) } else { pgrx::error!( "invalid transaction_mode \"{}\" for df.start(): expected '{}' or '{}'", @@ -997,11 +1095,16 @@ pub fn start_v2( /// four-argument `df.start()` above and no ambiguous overload exists. #[pg_extern(sql = false)] pub fn start(fut: &str, label: Option<&str>, database: Option<&str>) -> String { - start_in_caller_transaction(fut, label, database) + start_in_caller_transaction(fut, label, database, RetryPolicySpec::legacy()) } /// `df.start()` under `transaction_mode => 'new'`. -fn start_in_new_transaction(fut: &str, label: Option<&str>, database: Option<&str>) -> String { +fn start_in_new_transaction( + fut: &str, + label: Option<&str>, + database: Option<&str>, + retry: Option, +) -> String { // Inside a workflow this mode is pure cost. A df.sql() node is executed as // a single statement on an autocommit connection, so a plain df.start() // there already commits on its own and cannot be rolled back by the @@ -1032,7 +1135,7 @@ fn start_in_new_transaction(fut: &str, label: Option<&str>, database: Option<&st Err(e) => pgrx::error!("{e}"), }; - match crate::client::start_in_new_transaction(fut, label, database, &user_name) { + match crate::client::start_in_new_transaction(fut, label, database, &user_name, retry) { Ok(id) => id, Err(e) => pgrx::error!("{}", e), } @@ -1040,7 +1143,12 @@ fn start_in_new_transaction(fut: &str, label: Option<&str>, database: Option<&st /// `df.start()` under `transaction_mode => 'caller'`: build and persist the /// graph through SPI, so it lives or dies with the caller's transaction. -fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option<&str>) -> String { +fn start_in_caller_transaction( + fut: &str, + label: Option<&str>, + database: Option<&str>, + retry: RetryPolicySpec, +) -> String { let durofut = match Durofut::ensure_strict(fut) { Ok(d) => d, Err(e) => pgrx::error!("Invalid durable function: {}", e), @@ -1280,6 +1388,7 @@ fn start_in_caller_transaction(fut: &str, label: Option<&str>, database: Option< // Generation 0 loads the graph from df.nodes; only a root loop continuing as new // carries it inline. graph: None, + retry, }; let input_json = serde_json::to_string(&input).unwrap_or(instance_id.clone()); @@ -1488,7 +1597,67 @@ pub fn wait_for_completion( #[cfg(test)] mod tests { - use super::{node_insert_sql, parse_semver, pick_id_with_retry}; + use super::{node_insert_sql, parse_retry_policy, parse_semver, pick_id_with_retry}; + use crate::types::OnFailure; + use pgrx::datum::Interval; + + fn seconds(secs: i64) -> Interval { + Interval::new(0, 0, secs * 1_000_000).unwrap() + } + + #[test] + fn parse_retry_policy_accepts_the_documented_defaults() { + let spec = parse_retry_policy(5, seconds(16), "continue").unwrap(); + + assert_eq!(spec.max_attempts, 5); + assert_eq!(spec.max_backoff(), std::time::Duration::from_secs(16)); + assert_eq!(spec.on_failure, OnFailure::Continue); + } + + #[test] + fn parse_retry_policy_accepts_fail_case_insensitively() { + let spec = parse_retry_policy(1, seconds(1), "FAIL").unwrap(); + + assert_eq!(spec.on_failure, OnFailure::Fail); + } + + #[test] + fn parse_retry_policy_rejects_max_attempts_below_one() { + let err = parse_retry_policy(0, seconds(16), "continue").unwrap_err(); + + assert!(err.contains("max_attempts"), "unhelpful message: {err}"); + } + + #[test] + fn parse_retry_policy_rejects_non_positive_max_backoff() { + for interval in [seconds(0), seconds(-5)] { + let err = parse_retry_policy(5, interval, "continue").unwrap_err(); + assert!(err.contains("max_backoff"), "unhelpful message: {err}"); + } + } + + #[test] + fn parse_retry_policy_rejects_unknown_on_failure() { + let err = parse_retry_policy(5, seconds(16), "explode").unwrap_err(); + + assert!(err.contains("on_failure"), "unhelpful message: {err}"); + assert!( + err.contains("explode"), + "message should quote the input: {err}" + ); + } + + #[test] + fn parse_retry_policy_accepts_a_month_valued_backoff_as_thirty_days() { + // Interval::as_micros() counts a month as 30 days, which keeps the conversion + // deterministic (a replayed orchestration cannot land on a different month length). + let spec = parse_retry_policy(2, Interval::new(1, 0, 0).unwrap(), "fail").unwrap(); + + assert_eq!( + spec.max_backoff(), + std::time::Duration::from_secs(30 * 24 * 60 * 60) + ); + } #[test] fn test_node_insert_sql_uses_current_schema_columns_and_numbered_parameters() { diff --git a/src/lib.rs b/src/lib.rs index 11736117..21c164cb 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -697,6 +697,89 @@ END $$; requires = [df] ); +extension_sql!( + r#" +-- df.instance_activity(): is this workflow actually doing anything? +-- +-- A long-running instance and a wedged one both read 'running', so status alone +-- cannot tell them apart. This reports when each non-terminal instance last +-- transitioned a node, how long it has been quiet since, and whether any node is +-- currently failing -- which is what a retrying-and-abandoning loop looks like +-- from the outside. +-- +-- SECURITY INVOKER (the default) so the row-level security policies on +-- df.instances and df.nodes apply: a caller sees only their own instances. +-- Deliberately a function rather than a view, because view-level RLS +-- pass-through (security_invoker) requires PostgreSQL 15 and this extension +-- supports 13. +-- +-- p_idle_for filters to instances quiet for at least that long. The default of +-- zero reports every non-terminal instance, so the common call is simply +-- SELECT * FROM df.instance_activity() ORDER BY idle_for_seconds DESC; +-- +-- Idle time is measured against clock_timestamp(), not now(). now() is the +-- calling transaction's start time, and the worker keeps writing node +-- timestamps after that, so a busy instance's last activity can be *later* +-- than now() -- which would make its idle time negative and drop it below any +-- threshold, including zero. Reading a moving clock makes the function +-- VOLATILE, which is honest: two calls in one transaction can legitimately +-- differ. +-- +-- GREATEST and EXTRACT are parser constructs and cannot be schema-qualified; +-- the pinned search_path resolves the remaining names to pg_catalog. +CREATE OR REPLACE FUNCTION df.instance_activity(p_idle_for interval DEFAULT '0 seconds') +RETURNS TABLE ( + instance_id VARCHAR(8), + label TEXT, + status TEXT, + last_activity_at TIMESTAMPTZ, + idle_for_seconds DOUBLE PRECISION, + running_node_count BIGINT, + failed_node_count BIGINT, + last_error TEXT +) +LANGUAGE SQL +VOLATILE +SET search_path = pg_catalog, df, pg_temp +AS $fn$ + SELECT + i.id, + i.label, + i.status, + activity.last_activity_at, + GREATEST(0, EXTRACT(EPOCH FROM pg_catalog.clock_timestamp() - activity.last_activity_at))::double precision, + activity.running_node_count, + activity.failed_node_count, + activity.last_error + FROM df.instances i + CROSS JOIN LATERAL ( + SELECT + -- GREATEST ignores NULLs, so an instance whose nodes have never + -- transitioned still reports its own updated_at. + GREATEST(i.updated_at, max(n.updated_at)) AS last_activity_at, + count(*) FILTER (WHERE n.status = 'running') AS running_node_count, + count(*) FILTER (WHERE n.status = 'failed') AS failed_node_count, + -- A failed node's message is written to df.nodes.result, not to + -- df.nodes.error, which nothing writes. #>> '{}' unwraps the JSONB + -- scalar so the caller gets the message, not a quoted JSON string. + (array_agg(n.result #>> '{}' ORDER BY n.updated_at DESC) + FILTER (WHERE n.status = 'failed' AND n.result IS NOT NULL))[1] AS last_error + FROM df.nodes n + WHERE n.instance_id = i.id + ) AS activity + WHERE (i.status IS NULL OR i.status NOT IN ('completed', 'failed', 'cancelled')) + AND pg_catalog.clock_timestamp() - activity.last_activity_at >= p_idle_for; +$fn$; + +COMMENT ON FUNCTION df.instance_activity(interval) IS + 'Non-terminal instances with their last node transition, how long they have been idle, ' + 'and any current node failure. Use to tell a working workflow from a wedged one, which ' + 'status alone cannot show. RLS-filtered to the calling user.'; +"#, + name = "instance_activity", + requires = ["create_tables"] +); + // ============================================================================ // SQL Operators // ============================================================================ @@ -3306,6 +3389,7 @@ mod tests { vars: std::collections::HashMap::new(), loop_iteration: 42, graph: None, + retry: crate::types::RetryPolicySpec::legacy(), }; let json = serde_json::to_string(&input).unwrap(); let deserialized: FunctionInput = serde_json::from_str(&json).unwrap(); diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index d2e42dbb..d040ec97 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -19,7 +19,7 @@ use duroxide::OrchestrationContext; use crate::activities; use crate::types::{ evaluate_condition, string_map_to_json, substitute_all, substitute_all_raw, FunctionGraph, - FunctionInput, FunctionNode, SystemVars, + FunctionInput, FunctionNode, RetryPolicySpec, SystemVars, }; /// Orchestration name for ExecuteFunctionGraph @@ -43,6 +43,9 @@ struct ExecutionContext { subtree_root: String, /// Shape of the input this orchestration re-enters itself with on loop `continue_as_new`. continuation: Continuation, + /// Retry and failure policy recorded at `df.start()`, inherited by every node this + /// orchestration executes and by every subtree it spawns. + retry: RetryPolicySpec, } /// Which input envelope an inline loop must rebuild when it calls `continue_as_new`. @@ -84,6 +87,15 @@ struct SubtreeInput { label: Option, #[serde(default)] iteration: u64, + /// Inherited failure policy. Defaults to the pre-0.2.7 behavior so a subtree spawned by + /// an older binary replays as it was recorded, and is omitted from the wire entirely + /// when it holds that value so the envelope stays byte-identical to what that binary + /// recorded -- duroxide matches a sub-orchestration schedule on the exact input string. + #[serde( + default = "RetryPolicySpec::legacy", + skip_serializing_if = "RetryPolicySpec::is_legacy" + )] + retry: RetryPolicySpec, } /// Control-flow-aware error type returned by every node handler. @@ -97,6 +109,14 @@ struct SubtreeInput { enum NodeError { /// A `df.break()` signal carrying its (already-stringified) value, caught by the loop. Break(String), + /// A node whose retries were exhausted under `on_failure => 'continue'`, carrying the + /// node's last error. Like `Break` this is control flow, not a failure: it unwinds to + /// the nearest enclosing `execute_loop_node`, which abandons the rest of the iteration + /// and starts the next one. Unwinding the whole iteration (rather than skipping just the + /// failed node) is what keeps a downstream node from running on a named result its + /// producer never computed. With no enclosing loop there is nowhere to continue to, so + /// the top level turns it back into a failure. + Continue(String), /// A real failure; propagates to the orchestration's `Err` result. Failure(String), } @@ -132,6 +152,9 @@ type NodeResult = Result; enum SubtreeControl { Normal, Break, + /// A `NodeError::Continue` raised inside the subtree, so the parent can re-raise it and + /// let it unwind to the loop that contains the join/race/loop node. + Continue, } /// Envelope returned by `execute_subtree` containing the SQL result and the updated @@ -150,6 +173,130 @@ struct SubtreeEnvelope { results: HashMap, } +/// Marker `execute_sql` stamps onto a failure that carries a PostgreSQL SQLSTATE. +/// +/// The activity boundary is `Result`, so the code has to survive as text. +/// It is read back by `error_is_retryable` from a string that was *recorded in history*, +/// which is what keeps the retry decision deterministic across replay. +const SQLSTATE_MARKER: &str = "[SQLSTATE "; + +/// SQLSTATE classes whose failures are a property of the statement, not of the moment. +/// +/// 42 (syntax error or access rule violation), 23 (integrity constraint violation), +/// 28 (invalid authorization specification), 3D (invalid catalog name) and 3F (invalid +/// schema name) all describe a statement that is simply wrong for the database it was sent +/// to. Retrying one spends the whole backoff budget to arrive at the identical error. +/// +/// Class 22 (data exception, e.g. division by zero) is deliberately *absent*: it is a +/// statement about the data, and another node or an outside writer can change that between +/// attempts. Everything not listed here -- notably 40 (serialization/deadlock), 08 +/// (connection), 53 (insufficient resources) and 57 (operator intervention) -- is retried. +const PERMANENT_SQLSTATE_CLASSES: [&str; 5] = ["42", "23", "28", "3D", "3F"]; + +/// Whether a failed activity is worth another attempt. +/// +/// Anything without a well-formed marker is retryable. That is the conservative default and +/// it is also what keeps HTTP nodes, pg_durable's own pre-execution failures (connection +/// limit, role resolution), and histories recorded before the marker existed behaving as +/// they did before. +fn error_is_retryable(error: &str) -> bool { + let Some((_, rest)) = error.split_once(SQLSTATE_MARKER) else { + return true; + }; + let Some((code, _)) = rest.split_once(']') else { + return true; + }; + // A SQLSTATE is exactly five characters. `get` rather than slicing: the code is parsed + // from a recorded string and must not be trusted to be ASCII, and a char-boundary panic + // inside orchestration code would abort a replay. + if code.len() != 5 { + return true; + } + match code.get(..2) { + Some(class) => !PERMANENT_SQLSTATE_CLASSES.contains(&class), + None => true, + } +} + +/// Translate the per-instance policy into duroxide's retry policy. +/// +/// `RetryPolicy::new()` asserts `max_attempts >= 1` and would panic inside the +/// orchestration, so the struct is built literally and clamped; `df.start()` already rejects +/// zero, and a history recorded by some future binary must not be able to abort a replay. +fn build_retry_policy(spec: &RetryPolicySpec) -> duroxide::RetryPolicy { + duroxide::RetryPolicy { + max_attempts: spec.max_attempts.max(1), + backoff: duroxide::BackoffStrategy::Exponential { + base: Duration::from_secs(1), + multiplier: 2.0, + max: spec.max_backoff(), + }, + // Per-attempt timeouts are not retried by duroxide, so enabling one here would turn + // a slow node into an un-retried failure. Node-level time limits are the activity's + // own concern (statement_timeout, HTTP timeouts). + timeout: None, + } +} + +/// Decide what an exhausted node's error means for the workflow. +fn classify_exhausted_activity(error: String, spec: &RetryPolicySpec) -> NodeError { + match spec.on_failure { + crate::types::OnFailure::Continue => NodeError::Continue(error), + crate::types::OnFailure::Fail => NodeError::Failure(error), + } +} + +/// Schedule a retryable node activity (SQL / HTTP / multipart) under the instance's policy. +/// +/// This is the only place the policy is applied, so all three node types retry identically +/// and an exhausted node lands on the same control-flow decision. +/// +/// The loop deliberately mirrors `duroxide::schedule_activity_with_retry` -- one +/// `schedule_activity` per attempt with a timer in between -- rather than calling it, because +/// that helper retries every error and offers no hook to opt out of one. Recording the same +/// durable operations in the same order is what lets a history written before the policy +/// existed replay unchanged. `ctx.trace_*` is log-only (duroxide suppresses it while +/// replaying), so the extra tracing here adds nothing to history. +async fn schedule_node_activity( + ctx: &OrchestrationContext, + name: &str, + input: String, + exec_ctx: &ExecutionContext, +) -> NodeResult { + let policy = build_retry_policy(&exec_ctx.retry); + let mut last_error = String::new(); + + for attempt in 1..=policy.max_attempts { + match ctx.schedule_activity(name, input.as_str()).await { + Ok(result) => return Ok(result), + Err(e) => { + let retryable = error_is_retryable(&e); + last_error = e; + + if !retryable { + ctx.trace_info(format!( + "Activity '{name}' failed with a non-retryable error, not retrying: {last_error}" + )); + break; + } + + if attempt < policy.max_attempts { + ctx.trace_warn(format!( + "Activity '{name}' attempt {attempt}/{} failed: {last_error}. Retrying...", + policy.max_attempts + )); + let delay = policy.backoff.delay_for_attempt(attempt); + if !delay.is_zero() { + ctx.schedule_timer(delay).await; + } + } + } + } + } + + Err(classify_exhausted_activity(last_error, &exec_ctx.retry)) +} + /// Execute a complete function graph — the entry point for a durable function. /// /// # Control flow @@ -248,6 +395,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), + // A continue that reaches the top level had no enclosing loop to unwind to, so the + // instance fails with the node's own error — the documented "no loop, no continue". + Err(NodeError::Continue(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(), @@ -370,6 +521,7 @@ pub async fn execute_subtree( loop_iteration: input.iteration, subtree_root: input.node_id.clone(), continuation: Continuation::Subtree, + retry: input.retry, }; // Build the envelope carrying the result, the updated named-results map, and a typed @@ -408,6 +560,17 @@ pub async fn execute_subtree( results, } } + Err(NodeError::Continue(error)) => { + ctx.trace_info(format!( + "ExecuteSubtree: node {} exhausted its retries (continuing)", + input.node_id + )); + SubtreeEnvelope { + control: Some(SubtreeControl::Continue), + result: error, + results, + } + } Err(NodeError::Failure(e)) => return Err(e), }; @@ -438,6 +601,7 @@ fn build_subtree_input( ), label: exec_ctx.label.clone(), iteration: 0, + retry: exec_ctx.retry, }; serde_json::to_string(&input).map_err(|e| format!("Failed to serialize subtree input: {e}")) } @@ -525,6 +689,9 @@ async fn execute_function_node_with_vars( let (status, status_result) = match &execute_result { Ok(result) => ("completed", result.as_str()), Err(NodeError::Break(value)) => ("completed", value.as_str()), + // An abandoned node is a failed node even though the instance keeps running: this + // is the only record that the iteration lost work, so monitoring can find it. + Err(NodeError::Continue(err)) => ("failed", err.as_str()), Err(NodeError::Failure(err)) => ("failed", err.as_str()), }; let status_input = serde_json::json!({ @@ -604,9 +771,13 @@ async fn execute_sql_node( "database": node.database, }); - let result = ctx - .schedule_activity(activities::execute_sql::NAME, input.to_string()) - .await?; + let result = schedule_node_activity( + ctx, + activities::execute_sql::NAME, + input.to_string(), + exec_ctx, + ) + .await?; if let Some(name) = &node.result_name { ctx.trace_info(format!("Storing result as ${name}")); @@ -747,11 +918,6 @@ async fn execute_wait_schedule_node( /// deficit so an empty-bodied loop can't busy-spin via continue_as_new. const LOOP_MIN_ITER_DURATION: Duration = Duration::from_secs(1); -/// 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; - /// Stamp a loop node's status from its *parent* orchestration. /// /// A non-root loop node is the root of its own child instance, so the child normally owns @@ -827,6 +993,12 @@ async fn fail_loop_child_future( /// Returns `Ok(Some(final_result))` when the loop should exit (a `df.break()` in the body, /// or the while-condition evaluating false), `Ok(None)` when another iteration is needed, /// and `Err` when the body or condition fails. +/// +/// This is also where `NodeError::Continue` is caught. An abandoned iteration is reported as +/// `Ok(None)`: the rest of the body — including the while-condition — is skipped, and the +/// next iteration starts at the top of the body. Skipping the condition is deliberate. It is +/// evaluated from the same named results the body populates, so running it after a failed +/// body would decide the loop's fate from a result its producer never computed. async fn run_loop_iteration( ctx: &OrchestrationContext, graph: &FunctionGraph, @@ -849,6 +1021,12 @@ async fn run_loop_iteration( store_named_result(ctx, node, &break_value, results, "LOOP"); return Ok(Some(break_value)); } + Err(NodeError::Continue(error)) => { + ctx.trace_warn(format!( + "Loop iteration abandoned after a node exhausted its retries: {error}" + )); + return Ok(None); + } Err(NodeError::Failure(e)) => return Err(e), }; @@ -875,6 +1053,12 @@ async fn run_loop_iteration( store_named_result(ctx, node, &break_value, results, "LOOP"); return Ok(Some(break_value)); } + Err(NodeError::Continue(error)) => { + ctx.trace_warn(format!( + "Loop condition abandoned after a node exhausted its retries: {error}" + )); + return Ok(None); + } Err(NodeError::Failure(e)) => return Err(e), }; @@ -945,14 +1129,12 @@ async fn execute_loop_node( ctx.trace_info("Continuing as new for next loop iteration"); - // M7: Enforce maximum iteration count to prevent runaway infinite loops + // There is deliberately no iteration ceiling. A workflow that legitimately runs + // indefinitely (a scheduled compactor, a poller) must not acquire an expiry date, and + // LOOP_MIN_ITER_DURATION below already bounds the *rate* of a runaway loop. An actually + // runaway instance is handled by df.cancel() and by watching df.instances, both of which + // work from the first iteration rather than 27 hours in. let next_iteration = exec_ctx.loop_iteration + 1; - if next_iteration >= MAX_LOOP_ITERATIONS { - return Err(NodeError::Failure(format!( - "Loop exceeded maximum iteration count of {MAX_LOOP_ITERATIONS}. \ - Use df.break() to exit the loop or restructure the workflow." - ))); - } // Enforce a minimum per-iteration wall-clock duration to prevent // busy-looping (e.g. `df.loop(df.sleep(0))`). Compute the elapsed time @@ -992,6 +1174,7 @@ async fn execute_loop_node( vars: exec_ctx.vars.clone(), loop_iteration: next_iteration, graph: Some(graph_json), + retry: exec_ctx.retry, }; serde_json::to_string(&new_input) .map_err(|e| format!("Failed to serialize loop input: {e}"))? @@ -1009,6 +1192,7 @@ async fn execute_loop_node( ), label: exec_ctx.label.clone(), iteration: next_iteration, + retry: exec_ctx.retry, }; serde_json::to_string(&new_input) .map_err(|e| format!("Failed to serialize loop input: {e}"))? @@ -1254,6 +1438,9 @@ fn parse_subtree_envelope( parent_results.extend(envelope.results); match envelope.control { Some(SubtreeControl::Break) => Err(NodeError::Break(envelope.result)), + // Re-raise the subtree's continue so it keeps unwinding toward the enclosing loop, + // exactly as a break does. + Some(SubtreeControl::Continue) => Err(NodeError::Continue(envelope.result)), // A new binary always writes an explicit `control`, so `Some(Normal)` is a genuine // normal result and must NOT be run through the legacy sentinel check: otherwise a // branch whose real SQL result happens to be shaped like `{"__break__": true, ...}` @@ -1569,9 +1756,8 @@ async fn execute_http_node( let method = config["method"].as_str().unwrap_or("POST"); ctx.trace_info(format!("Executing HTTP {method} {url}")); - let result = ctx - .schedule_activity(activities::execute_http::NAME, final_config) - .await?; + let result = + schedule_node_activity(ctx, activities::execute_http::NAME, final_config, exec_ctx).await?; // Store result if named if let Some(name) = &node.result_name { @@ -1668,9 +1854,13 @@ async fn execute_http_multipart_node( let method = config["method"].as_str().unwrap_or("POST"); ctx.trace_info(format!("Executing HTTP_MULTIPART {method} {url}")); - let result = ctx - .schedule_activity(activities::execute_multipart::NAME, final_config) - .await?; + let result = schedule_node_activity( + ctx, + activities::execute_multipart::NAME, + final_config, + exec_ctx, + ) + .await?; // Store result if named if let Some(name) = &node.result_name { @@ -1764,6 +1954,255 @@ async fn execute_signal_node( mod tests { use super::*; + #[test] + fn subtree_input_without_retry_defaults_to_legacy() { + // A JOIN/RACE branch or non-root loop spawned by a pre-0.2.7 binary recorded no + // retry field; replaying it under this binary must keep the old semantics. + let json = r#"{"instance_id":"i","node_id":"n","graph":"{}","results":"{}"}"#; + + let input: SubtreeInput = serde_json::from_str(json).unwrap(); + + assert_eq!(input.retry.max_attempts, 1); + assert_eq!(input.retry.on_failure, crate::types::OnFailure::Fail); + } + + #[test] + fn legacy_subtree_input_serializes_to_the_pre_0_2_7_envelope() { + // duroxide matches a sub-orchestration schedule by exact equality on the serialized + // input (replay_engine::action_matches_event_kind). A JOIN/RACE branch or non-root + // loop recorded by a pre-0.2.7 binary carries no retry field, so re-emitting one + // during replay is a nondeterminism error that bricks the instance -- the same break + // documented for v0.2.4 -> v0.2.5. A legacy policy must therefore serialize away. + let input = SubtreeInput { + instance_id: "i".to_string(), + node_id: "n".to_string(), + graph: "{}".to_string(), + results: "{}".to_string(), + vars: None, + label: None, + iteration: 0, + retry: crate::types::RetryPolicySpec::legacy(), + }; + + let json = serde_json::to_string(&input).unwrap(); + + assert!( + !json.contains("retry"), + "legacy envelope must not emit retry: {json}" + ); + } + + #[test] + fn non_legacy_subtree_input_still_serializes_its_policy() { + // The suppression above must key off the legacy value specifically: an instance + // started under 0.2.7 with a real policy has to propagate it into its subtrees. + let input = SubtreeInput { + instance_id: "i".to_string(), + node_id: "n".to_string(), + graph: "{}".to_string(), + results: "{}".to_string(), + vars: None, + label: None, + iteration: 0, + retry: crate::types::RetryPolicySpec { + max_attempts: 5, + max_backoff_micros: 16_000_000, + on_failure: crate::types::OnFailure::Continue, + }, + }; + + let json = serde_json::to_string(&input).unwrap(); + let round_tripped: SubtreeInput = serde_json::from_str(&json).unwrap(); + + assert!( + json.contains("retry"), + "a real policy must be carried: {json}" + ); + assert_eq!(round_tripped.retry.max_attempts, 5); + assert_eq!( + round_tripped.retry.on_failure, + crate::types::OnFailure::Continue + ); + } + + #[test] + fn subtree_input_carries_the_parents_policy() { + // A sub-orchestration must inherit the instance's policy: a node inside a JOIN + // branch is no less retryable than the same node in the trunk. + let graph = FunctionGraph { + instance_id: "inst1234".to_string(), + root_node_id: "root".to_string(), + nodes: std::collections::BTreeMap::new(), + }; + let retry = crate::types::RetryPolicySpec { + max_attempts: 4, + max_backoff_micros: 2_000_000, + on_failure: crate::types::OnFailure::Continue, + }; + let exec_ctx = ExecutionContext { + vars: HashMap::new(), + label: None, + loop_iteration: 0, + subtree_root: "root".to_string(), + continuation: Continuation::Root, + retry, + }; + + let json = + build_subtree_input(&graph, "child", &HashMap::new(), &exec_ctx).expect("built input"); + let input: SubtreeInput = serde_json::from_str(&json).unwrap(); + + assert_eq!(input.retry, retry); + } + + #[test] + fn deterministic_sqlstate_classes_are_not_retried() { + // A statement that failed because it is wrong will fail identically on every + // attempt. Retrying it just spends the caller's backoff budget before reporting the + // same error, and for an integrity violation it can actively mislead: a unique + // violation on attempt two is what a committed-but-disconnected insert looks like. + for code in [ + "42P01", // undefined_table + "42601", // syntax_error + "42501", // insufficient_privilege + "23505", // unique_violation + "23503", // foreign_key_violation + "28P01", // invalid_password + "3D000", // invalid_catalog_name + "3F000", // invalid_schema_name + ] { + let error = format!("SQL execution failed [SQLSTATE {code}]: boom"); + assert!(!error_is_retryable(&error), "{code} should not be retried"); + } + } + + #[test] + fn transient_and_unclassified_sqlstate_classes_are_retried() { + // Class 40/08/53/57 are the textbook transient failures. Class 22 (data exception, + // e.g. division by zero) stays retryable on purpose: it is a statement about the + // *data*, which another node or an outside writer can change between attempts. + for code in [ + "40001", // serialization_failure + "40P01", // deadlock_detected + "08006", // connection_failure + "53300", // too_many_connections + "57P01", // admin_shutdown + "22012", // division_by_zero + "XX000", // internal_error + ] { + let error = format!("SQL execution failed [SQLSTATE {code}]: boom"); + assert!(error_is_retryable(&error), "{code} should be retried"); + } + } + + #[test] + fn errors_without_a_sqlstate_are_retried() { + // HTTP and multipart nodes carry no SQLSTATE, and neither do pg_durable's own + // pre-execution failures (connection limit, role resolution). Retrying is the + // conservative default, and it is what histories recorded before this marker + // existed replay as. + assert!(error_is_retryable("connection limit reached")); + assert!(error_is_retryable( + "HTTP request failed: 503 Service Unavailable" + )); + assert!(error_is_retryable( + "SQL execution failed: some older message" + )); + } + + #[test] + fn a_malformed_sqlstate_marker_is_retried() { + // The marker is parsed from a recorded string, so it must not be trusted to be + // well-formed; anything unparseable falls back to the retryable default. + assert!(error_is_retryable("SQL execution failed [SQLSTATE ]: boom")); + assert!(error_is_retryable( + "SQL execution failed [SQLSTATE 4]: boom" + )); + assert!(error_is_retryable("[SQLSTATE")); + } + + #[test] + fn retry_policy_backoff_is_one_second_doubling_capped_at_max_backoff() { + // The documented sequence at the df.start() defaults: 1s, 2s, 4s, 8s between five + // tries, with the cap only binding beyond that. + let spec = crate::types::RetryPolicySpec { + max_attempts: 5, + max_backoff_micros: 16_000_000, + on_failure: crate::types::OnFailure::Continue, + }; + + let policy = build_retry_policy(&spec); + + assert_eq!(policy.max_attempts, 5); + assert!( + policy.timeout.is_none(), + "per-attempt timeouts are not retried" + ); + let delays: Vec = (1..=6) + .map(|attempt| policy.backoff.delay_for_attempt(attempt).as_secs()) + .collect(); + assert_eq!(delays, vec![1, 2, 4, 8, 16, 16]); + } + + #[test] + fn retry_policy_honors_a_backoff_cap_below_the_first_delay() { + let spec = crate::types::RetryPolicySpec { + max_attempts: 3, + max_backoff_micros: 500_000, + on_failure: crate::types::OnFailure::Fail, + }; + + let policy = build_retry_policy(&spec); + + assert_eq!(policy.backoff.delay_for_attempt(1).as_millis(), 500); + } + + #[test] + fn exhausted_node_under_continue_becomes_a_continue_error() { + let spec = crate::types::RetryPolicySpec { + max_attempts: 2, + max_backoff_micros: 1_000_000, + on_failure: crate::types::OnFailure::Continue, + }; + + match classify_exhausted_activity("relation does not exist".to_string(), &spec) { + NodeError::Continue(e) => assert_eq!(e, "relation does not exist"), + other => panic!("expected NodeError::Continue, got {other:?}"), + } + } + + #[test] + fn exhausted_node_under_fail_becomes_a_failure() { + let spec = crate::types::RetryPolicySpec { + max_attempts: 2, + max_backoff_micros: 1_000_000, + on_failure: crate::types::OnFailure::Fail, + }; + + match classify_exhausted_activity("boom".to_string(), &spec) { + NodeError::Failure(e) => assert_eq!(e, "boom"), + other => panic!("expected NodeError::Failure, got {other:?}"), + } + } + + #[test] + fn continue_crosses_the_subtree_boundary() { + // A branch that exhausts its tries under 'continue' must unwind the loop containing + // the join, not complete the branch with an error string as its value. + let envelope = envelope_json( + Some("Continue"), + "relation does not exist", + serde_json::json!({}), + ); + + let result = parse_subtree_envelope(&envelope, "JOIN branch", &mut HashMap::new()); + + match result { + Err(NodeError::Continue(e)) => assert_eq!(e, "relation does not exist"), + other => panic!("expected NodeError::Continue, got {other:?}"), + } + } + /// Build an envelope JSON string the way `execute_subtree` serializes a `SubtreeEnvelope`. /// When `control` is `None` the field is omitted entirely, reproducing an envelope recorded /// by a pre-#148 binary (<= v0.2.2) that had no `control` field. diff --git a/src/types.rs b/src/types.rs index 00ab16c3..16f3c95f 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1277,9 +1277,20 @@ pub struct FunctionInput { #[serde(default, serialize_with = "serialize_string_map")] pub vars: std::collections::HashMap, /// Loop iteration counter, incremented on each `continue_as_new`. - /// Used to enforce a maximum iteration safeguard. + /// Carried across generations for tracing; nothing reads it for control flow. #[serde(default)] pub loop_iteration: u64, + /// Retry and failure policy chosen at `df.start()`. + /// + /// Missing in histories recorded before the policy existed, which is why the default is + /// the pre-policy behavior rather than the `df.start()` default. Omitted from the wire + /// when it holds that legacy value, so a `continue_as_new` generation of a pre-0.2.7 + /// instance re-emits the input shape that instance was started with. + #[serde( + default = "RetryPolicySpec::legacy", + skip_serializing_if = "RetryPolicySpec::is_legacy" + )] + pub retry: RetryPolicySpec, /// Serialized `FunctionGraph`, carried across `continue_as_new` generations. /// /// `df.start()` leaves this `None`, so generation 0 loads the graph from the database. @@ -1289,6 +1300,68 @@ pub struct FunctionInput { pub graph: Option, } +/// What a workflow does with a node whose retries are exhausted. +/// +/// Serialized in orchestration input, so the spellings must stay stable: they are the same +/// strings `df.start(on_failure => ...)` accepts. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum OnFailure { + /// Abandon the rest of the current loop iteration and start the next one. With no + /// enclosing loop there is nowhere to continue to, so the instance fails. + Continue, + /// Fail the instance as soon as the tries are spent. + Fail, +} + +/// Per-instance retry and failure policy. +/// +/// This travels in the orchestration input rather than being read from a GUC: orchestration +/// code is replayed, and a setting read at execution time would produce different durable +/// operations on replay. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub struct RetryPolicySpec { + /// Tries per node, including the first. Always >= 1 (validated at `df.start()`). + pub max_attempts: u32, + /// Cap on the delay between tries, in microseconds. Always > 0. + pub max_backoff_micros: i64, + pub on_failure: OnFailure, +} + +impl RetryPolicySpec { + /// The behavior of instances started before the policy existed: one try, then fail. + /// + /// This is the serde default for the input fields, so an in-flight instance whose + /// history predates this feature replays with exactly the semantics it started under. + pub fn legacy() -> Self { + Self { + max_attempts: 1, + max_backoff_micros: DEFAULT_MAX_BACKOFF_MICROS, + on_failure: OnFailure::Fail, + } + } + + /// Whether this is the pre-0.2.7 policy, in which case it is omitted from serialized + /// input entirely. + /// + /// duroxide matches a recorded sub-orchestration schedule by exact equality on the + /// serialized input, so emitting a field a pre-0.2.7 binary never wrote turns every + /// in-flight JOIN branch, RACE branch, and non-root loop into a nondeterminism error. + /// Suppressing the legacy value keeps those envelopes byte-identical. This deliberately + /// keys off the legacy value rather than `Default`: an instance started under 0.2.7 with + /// a real policy must still carry it. + pub fn is_legacy(&self) -> bool { + *self == Self::legacy() + } + + pub fn max_backoff(&self) -> std::time::Duration { + std::time::Duration::from_micros(self.max_backoff_micros.max(0) as u64) + } +} + +/// Default `max_backoff` for `df.start()`: 16 seconds. +pub const DEFAULT_MAX_BACKOFF_MICROS: i64 = 16_000_000; + pub(crate) fn serialize_string_map( map: &std::collections::HashMap, serializer: S, @@ -1741,6 +1814,98 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn function_input_without_retry_defaults_to_legacy() { + // An instance started by a pre-0.2.7 binary recorded no retry field. It must keep + // the old behavior on replay: one attempt, then fail. + let json = r#"{"instance_id":"abc12345","label":"test","vars":{},"loop_iteration":0}"#; + let input: FunctionInput = serde_json::from_str(json).unwrap(); + + assert_eq!(input.retry.max_attempts, 1); + assert_eq!(input.retry.on_failure, OnFailure::Fail); + } + + #[test] + fn retry_policy_survives_function_input_round_trip() { + // continue_as_new re-serializes FunctionInput every generation, so a loop must not + // silently lose the policy it started with. + let input = FunctionInput { + instance_id: "abc12345".to_string(), + label: None, + vars: std::collections::HashMap::new(), + loop_iteration: 7, + graph: None, + retry: RetryPolicySpec { + max_attempts: 5, + max_backoff_micros: 16_000_000, + on_failure: OnFailure::Continue, + }, + }; + + let encoded = serde_json::to_string(&input).unwrap(); + let decoded: FunctionInput = serde_json::from_str(&encoded).unwrap(); + + assert_eq!(decoded.retry, input.retry); + } + + #[test] + fn legacy_function_input_serializes_without_the_retry_field() { + // A pre-0.2.7 instance that reaches a continue_as_new under this binary must re-emit + // the input shape it was started with, so its later generations stay consistent with + // the history the old binary recorded. + let input = FunctionInput { + instance_id: "abc12345".to_string(), + label: None, + vars: std::collections::HashMap::new(), + loop_iteration: 7, + graph: None, + retry: RetryPolicySpec::legacy(), + }; + + let encoded = serde_json::to_string(&input).unwrap(); + let decoded: FunctionInput = serde_json::from_str(&encoded).unwrap(); + + assert!( + !encoded.contains("retry"), + "legacy input must not emit retry: {encoded}" + ); + assert!(decoded.retry.is_legacy()); + } + + #[test] + fn retry_policy_max_backoff_converts_micros_to_duration() { + let spec = RetryPolicySpec { + max_attempts: 5, + max_backoff_micros: 16_000_000, + on_failure: OnFailure::Fail, + }; + + assert_eq!(spec.max_backoff(), std::time::Duration::from_secs(16)); + } + + #[test] + fn retry_policy_defaults_match_the_documented_start_defaults() { + // df.start() defaults to the pre-0.2.7 behavior: the policy is entirely opt-in, so + // upgrading cannot change how an existing workflow handles a failing node. + let spec = RetryPolicySpec::legacy(); + + assert_eq!(spec.max_attempts, 1); + assert_eq!(spec.max_backoff(), std::time::Duration::from_secs(16)); + assert_eq!(spec.on_failure, OnFailure::Fail); + } + + #[test] + fn on_failure_serializes_as_lowercase_sql_spelling() { + assert_eq!( + serde_json::to_string(&OnFailure::Continue).unwrap(), + r#""continue""# + ); + assert_eq!( + serde_json::to_string(&OnFailure::Fail).unwrap(), + r#""fail""# + ); + } + #[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"}}"#; @@ -2171,6 +2336,7 @@ mod tests { vars: forward, loop_iteration: 0, graph: None, + retry: RetryPolicySpec::legacy(), }; let reverse_input = FunctionInput { instance_id: "instance".to_string(), @@ -2178,6 +2344,7 @@ mod tests { vars: reverse, loop_iteration: 0, graph: None, + retry: RetryPolicySpec::legacy(), }; assert_eq!( serde_json::to_string(&forward_input).unwrap(), diff --git a/tests/e2e/sql/18_delegated_grants.sql b/tests/e2e/sql/18_delegated_grants.sql index 296b016e..0fc34509 100644 --- a/tests/e2e/sql/18_delegated_grants.sql +++ b/tests/e2e/sql/18_delegated_grants.sql @@ -239,7 +239,7 @@ DECLARE can_http BOOLEAN; BEGIN SELECT has_schema_privilege('dg_app', 'df', 'USAGE') INTO has_usage; - SELECT has_function_privilege('dg_app', 'df.start(text, text, text, text)', 'EXECUTE') INTO can_start; + SELECT has_function_privilege('dg_app', 'df.start(text, text, text, text, int, interval, text)', 'EXECUTE') INTO can_start; SELECT has_function_privilege('dg_app', 'df.http(text, text, text, jsonb, integer)', 'EXECUTE') INTO can_http; IF NOT has_usage THEN diff --git a/tests/e2e/sql/68_failure_policy.sql b/tests/e2e/sql/68_failure_policy.sql new file mode 100644 index 00000000..19f06471 --- /dev/null +++ b/tests/e2e/sql/68_failure_policy.sql @@ -0,0 +1,283 @@ +-- Copyright (c) Microsoft Corporation. +-- Licensed under the PostgreSQL License. + +-- df.start()'s node failure policy: max_attempts / max_backoff retry a failing +-- node, and on_failure decides what happens once the attempts are spent. +-- +-- Attempts are counted with sequences rather than tables: nextval() is +-- non-transactional, so the count survives the rollback of the failed attempt +-- that produced it. Every case pins max_backoff to 1 second so the whole file +-- stays well inside the e2e time budget. +SET SESSION AUTHORIZATION df_e2e_user; + +DROP TABLE IF EXISTS test_fp_loop_log; +DROP SEQUENCE IF EXISTS test_fp_transient_seq; +DROP SEQUENCE IF EXISTS test_fp_loop_seq; +DROP SEQUENCE IF EXISTS test_fp_no_loop_seq; +DROP SEQUENCE IF EXISTS test_fp_fail_seq; +DROP SEQUENCE IF EXISTS test_fp_default_seq; + +CREATE TABLE test_fp_loop_log (id SERIAL PRIMARY KEY, note TEXT); +-- Attempts are counted with sequences rather than tables: a failed attempt's transaction +-- rolls back and would take an inserted row with it, whereas nextval() is non-transactional +-- and survives. CACHE 1 is the default but is stated explicitly, because a larger cache +-- would let last_value run ahead of the number of nextval() calls and break the count. +CREATE SEQUENCE test_fp_transient_seq CACHE 1; +CREATE SEQUENCE test_fp_loop_seq CACHE 1; +CREATE SEQUENCE test_fp_no_loop_seq CACHE 1; +CREATE SEQUENCE test_fp_fail_seq CACHE 1; +CREATE SEQUENCE test_fp_default_seq CACHE 1; + +-- --------------------------------------------------------------------------- +-- Case 1: a transient failure is retried until it succeeds. +-- The node divides by zero on its first two attempts and succeeds on the third. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _fp_transient AS +SELECT df.start( + $$SELECT 1 / (CASE WHEN nextval('test_fp_transient_seq') >= 3 THEN 1 ELSE 0 END)$$, + 'test-failure-policy-transient', + max_attempts => 5, + max_backoff => '1 second' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + final_status TEXT; + attempts BIGINT; +BEGIN + SELECT i.instance_id INTO instance_id FROM _fp_transient i; + SELECT df.await_instance(instance_id, 60) INTO final_status; + + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED [transient]: expected completed, got % (%)', + final_status, (SELECT output FROM df.instance_info(instance_id)); + END IF; + + SELECT last_value INTO attempts FROM test_fp_transient_seq; + IF attempts IS DISTINCT FROM 3 THEN + RAISE EXCEPTION 'TEST FAILED [transient]: expected exactly 3 attempts, got %', attempts; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 2: on_failure => 'continue' abandons the rest of the failing iteration +-- and runs the next one. The body fails through iteration 1 (attempts 1-2) and +-- the first attempt of iteration 2, then succeeds. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _fp_loop AS +SELECT df.start( + df.loop( + $$SELECT 1 / (CASE WHEN nextval('test_fp_loop_seq') >= 4 THEN 1 ELSE 0 END)$$ + ~> $$INSERT INTO test_fp_loop_log (note) VALUES ('body-completed')$$, + 'SELECT count(*) < 1 FROM test_fp_loop_log' + ), + 'test-failure-policy-loop-continue', + max_attempts => 2, + max_backoff => '1 second', + on_failure => 'continue' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + final_status TEXT; + attempts BIGINT; + completed_bodies INT; +BEGIN + SELECT i.instance_id INTO instance_id FROM _fp_loop i; + SELECT df.await_instance(instance_id, 60) INTO final_status; + + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED [loop-continue]: expected completed, got % (%)', + final_status, (SELECT output FROM df.instance_info(instance_id)); + END IF; + + -- 2 attempts in the abandoned first iteration, 2 more in the second. + SELECT last_value INTO attempts FROM test_fp_loop_seq; + IF attempts IS DISTINCT FROM 4 THEN + RAISE EXCEPTION 'TEST FAILED [loop-continue]: expected 4 attempts across 2 iterations, got %', attempts; + END IF; + + -- The node after the failure is skipped in the abandoned iteration, so it + -- runs exactly once even though the loop ran twice. + SELECT count(*) INTO completed_bodies FROM test_fp_loop_log; + IF completed_bodies IS DISTINCT FROM 1 THEN + RAISE EXCEPTION 'TEST FAILED [loop-continue]: expected 1 completed body, got %', completed_bodies; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 3: 'continue' with no enclosing loop has no next iteration to continue +-- into, so the instance fails once the attempts are spent. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _fp_no_loop AS +SELECT df.start( + $$SELECT 1 / (CASE WHEN nextval('test_fp_no_loop_seq') < 0 THEN 1 ELSE 0 END)$$, + 'test-failure-policy-no-loop', + max_attempts => 2, + max_backoff => '1 second', + on_failure => 'continue' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + final_status TEXT; + instance_output TEXT; + attempts BIGINT; + failed_nodes INT; + waited INT; +BEGIN + SELECT i.instance_id INTO instance_id FROM _fp_no_loop i; + SELECT df.await_instance(instance_id, 60) INTO final_status; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION 'TEST FAILED [no-loop]: expected failed, got %', final_status; + END IF; + + SELECT last_value INTO attempts FROM test_fp_no_loop_seq; + IF attempts IS DISTINCT FROM 2 THEN + RAISE EXCEPTION 'TEST FAILED [no-loop]: expected 2 attempts, got %', attempts; + END IF; + + -- df.instances.status can be visible a moment before the failure output is + -- written, so give the output a bounded window to appear. + FOR waited IN 1..100 LOOP + SELECT output INTO instance_output FROM df.instance_info(instance_id); + EXIT WHEN instance_output IS NOT NULL; + PERFORM pg_sleep(0.1); + END LOOP; + + IF COALESCE(instance_output, '') NOT LIKE '%division by zero%' THEN + RAISE EXCEPTION 'TEST FAILED [no-loop]: original SQL error did not surface: %', instance_output; + END IF; + + -- The node itself is stamped failed once the attempts settle. + SELECT count(*) INTO failed_nodes + FROM df.instance_nodes(instance_id) n + WHERE n.inferred_status = 'failed'; + IF failed_nodes < 1 THEN + RAISE EXCEPTION 'TEST FAILED [no-loop]: no failed node reported by df.instance_nodes()'; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 4: on_failure => 'fail' with max_attempts => 1 restores the pre-0.2.7 +-- behaviour — a single attempt, then the instance fails even inside a loop. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _fp_fail AS +SELECT df.start( + df.loop( + $$SELECT 1 / (CASE WHEN nextval('test_fp_fail_seq') < 0 THEN 1 ELSE 0 END)$$, + 'SELECT true' + ), + 'test-failure-policy-fail', + max_attempts => 1, + on_failure => 'fail' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + final_status TEXT; + attempts BIGINT; +BEGIN + SELECT i.instance_id INTO instance_id FROM _fp_fail i; + SELECT df.await_instance(instance_id, 60) INTO final_status; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION 'TEST FAILED [fail]: expected failed, got %', final_status; + END IF; + + SELECT last_value INTO attempts FROM test_fp_fail_seq; + IF attempts IS DISTINCT FROM 1 THEN + RAISE EXCEPTION 'TEST FAILED [fail]: expected exactly 1 attempt, got %', attempts; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 5: the defaults are the pre-0.2.7 behaviour. A df.start() that passes no +-- policy arguments must still fail on the first node error, even inside a loop, +-- so upgrading to 0.2.7 cannot silently change an existing workflow. This is the +-- same shape as case 4 with the arguments omitted rather than spelled out. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _fp_default AS +SELECT df.start( + df.loop( + $$SELECT 1 / (CASE WHEN nextval('test_fp_default_seq') < 0 THEN 1 ELSE 0 END)$$, + 'SELECT true' + ), + 'test-failure-policy-default' +) AS instance_id; + +DO $$ +DECLARE + instance_id TEXT; + final_status TEXT; + attempts BIGINT; +BEGIN + SELECT i.instance_id INTO instance_id FROM _fp_default i; + SELECT df.await_instance(instance_id, 60) INTO final_status; + + IF final_status IS DISTINCT FROM 'failed' THEN + RAISE EXCEPTION 'TEST FAILED [default]: expected failed, got %', final_status; + END IF; + + SELECT last_value INTO attempts FROM test_fp_default_seq; + IF attempts IS DISTINCT FROM 1 THEN + RAISE EXCEPTION 'TEST FAILED [default]: expected exactly 1 attempt under the defaults, got %', attempts; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 6: argument validation is rejected at df.start() time. +-- --------------------------------------------------------------------------- +DO $$ +DECLARE + err TEXT; +BEGIN + BEGIN + PERFORM df.start('SELECT 1', 'test-failure-policy-bad-attempts', max_attempts => 0); + RAISE EXCEPTION 'TEST FAILED [validation]: max_attempts => 0 was accepted'; + EXCEPTION WHEN OTHERS THEN + GET STACKED DIAGNOSTICS err = MESSAGE_TEXT; + IF err NOT LIKE '%max_attempts%' THEN + RAISE EXCEPTION 'TEST FAILED [validation]: unexpected error for max_attempts => 0: %', err; + END IF; + END; + + BEGIN + PERFORM df.start('SELECT 1', 'test-failure-policy-bad-backoff', max_backoff => '-1 second'); + RAISE EXCEPTION 'TEST FAILED [validation]: negative max_backoff was accepted'; + EXCEPTION WHEN OTHERS THEN + GET STACKED DIAGNOSTICS err = MESSAGE_TEXT; + IF err NOT LIKE '%max_backoff%' THEN + RAISE EXCEPTION 'TEST FAILED [validation]: unexpected error for negative max_backoff: %', err; + END IF; + END; + + BEGIN + PERFORM df.start('SELECT 1', 'test-failure-policy-bad-on-failure', on_failure => 'explode'); + RAISE EXCEPTION 'TEST FAILED [validation]: on_failure => ''explode'' was accepted'; + EXCEPTION WHEN OTHERS THEN + GET STACKED DIAGNOSTICS err = MESSAGE_TEXT; + IF err NOT LIKE '%on_failure%' THEN + RAISE EXCEPTION 'TEST FAILED [validation]: unexpected error for on_failure => ''explode'': %', err; + END IF; + END; +END $$; + +DROP TABLE _fp_transient; +DROP TABLE _fp_loop; +DROP TABLE _fp_no_loop; +DROP TABLE _fp_fail; +DROP TABLE _fp_default; +DROP TABLE test_fp_loop_log; +DROP SEQUENCE test_fp_transient_seq; +DROP SEQUENCE test_fp_loop_seq; +DROP SEQUENCE test_fp_no_loop_seq; +DROP SEQUENCE test_fp_fail_seq; +DROP SEQUENCE test_fp_default_seq; +RESET SESSION AUTHORIZATION; +SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/69_instance_activity.sql b/tests/e2e/sql/69_instance_activity.sql new file mode 100644 index 00000000..d410fb74 --- /dev/null +++ b/tests/e2e/sql/69_instance_activity.sql @@ -0,0 +1,230 @@ +-- ============================================================================ +-- E2E Test: df.instance_activity() +-- +-- Answers "is this workflow actually doing anything?". A long-running instance +-- is indistinguishable from a wedged one by status alone -- both read +-- 'running' -- so this reports when each instance last transitioned a node and +-- how long it has been quiet since. +-- +-- The report is about work *in progress*, so every assertion about a listed +-- instance has to be made while that instance is still non-terminal. +-- ============================================================================ + +DROP TABLE IF EXISTS test_activity_log; + +CREATE TABLE test_activity_log (id SERIAL PRIMARY KEY, note TEXT); + +-- --------------------------------------------------------------------------- +-- Case 1: a working instance reports recent activity and a small idle time. +-- The instance sleeps between two nodes, so there is a wide window in which it +-- is unambiguously alive. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _ia_busy AS +SELECT df.start( + $$INSERT INTO test_activity_log (note) VALUES ('first')$$ + ~> df.sleep(8) + ~> $$INSERT INTO test_activity_log (note) VALUES ('second')$$, + 'test-activity-busy' +) AS instance_id; + +DO $$ +DECLARE + inst TEXT; + idle_seconds DOUBLE PRECISION; + last_activity TIMESTAMPTZ; + row_count INT; + attempts INT := 0; +BEGIN + SELECT i.instance_id INTO inst FROM _ia_busy i; + + -- Wait until the worker has actually picked the instance up, so the + -- assertions below describe a running instance rather than a queued one. + LOOP + EXIT WHEN lower(df.status(inst)) = 'running' OR attempts > 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + SELECT count(*) INTO row_count + FROM df.instance_activity() a + WHERE a.instance_id = inst; + + IF row_count <> 1 THEN + RAISE EXCEPTION 'TEST FAILED [busy]: expected the running instance to be listed once, got % rows', row_count; + END IF; + + SELECT a.last_activity_at, a.idle_for_seconds + INTO last_activity, idle_seconds + FROM df.instance_activity() a + WHERE a.instance_id = inst; + + IF last_activity IS NULL THEN + RAISE EXCEPTION 'TEST FAILED [busy]: last_activity_at must not be NULL'; + END IF; + + IF idle_seconds IS NULL OR idle_seconds < 0 THEN + RAISE EXCEPTION 'TEST FAILED [busy]: idle_for_seconds must be non-negative, got %', idle_seconds; + END IF; + + -- --------------------------------------------------------------------- + -- Case 2: the idle-threshold filter, asserted on this same live instance. + -- It has been active seconds ago, so an hour-long threshold must exclude + -- it while a zero threshold must include it. + -- --------------------------------------------------------------------- + SELECT count(*) INTO row_count + FROM df.instance_activity('1 hour') a + WHERE a.instance_id = inst; + + IF row_count <> 0 THEN + RAISE EXCEPTION 'TEST FAILED [threshold]: an instance active seconds ago must not be reported idle for 1 hour, got % rows', row_count; + END IF; + + SELECT count(*) INTO row_count + FROM df.instance_activity('0 seconds') a + WHERE a.instance_id = inst; + + IF row_count <> 1 THEN + RAISE EXCEPTION 'TEST FAILED [threshold]: a zero threshold must report a running instance, got % rows', row_count; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 3: the report distinguishes a live instance from a wedged one. A loop +-- whose body always fails under on_failure => 'continue' keeps its status at +-- 'running' forever, which is exactly the case status alone cannot diagnose. +-- It must be listed with its failing node's error. +-- --------------------------------------------------------------------------- +CREATE TEMP TABLE _ia_wedged AS +SELECT df.start( + df.loop( + $$SELECT 1 / 0$$, + 'SELECT true' + ), + 'test-activity-wedged', + max_attempts => 1, + on_failure => 'continue' +) AS instance_id; + +DO $$ +DECLARE + inst TEXT; + status TEXT; + failed_nodes BIGINT; + last_error TEXT; + attempts INT := 0; +BEGIN + SELECT i.instance_id INTO inst FROM _ia_wedged i; + + -- Wait for the loop to abandon at least one iteration. + LOOP + SELECT a.failed_node_count, a.last_error + INTO failed_nodes, last_error + FROM df.instance_activity('0 seconds') a + WHERE a.instance_id = inst; + + EXIT WHEN COALESCE(failed_nodes, 0) > 0 OR attempts > 600; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + + IF COALESCE(failed_nodes, 0) = 0 THEN + RAISE EXCEPTION 'TEST FAILED [wedged]: expected a failed node to be reported, got %', failed_nodes; + END IF; + + IF last_error IS NULL OR last_error NOT LIKE '%division by zero%' THEN + RAISE EXCEPTION 'TEST FAILED [wedged]: expected the division error to be surfaced, got %', last_error; + END IF; + + -- The instance is still 'running', which is the whole point: status alone + -- says healthy, the activity report says otherwise. + SELECT df.status(inst) INTO status; + IF lower(status) NOT IN ('running', 'pending') THEN + RAISE EXCEPTION 'TEST FAILED [wedged]: expected the wedged loop to still be running, got %', status; + END IF; +END $$; + +-- --------------------------------------------------------------------------- +-- Case 4: the report respects row-level security. Asserted while both +-- instances are still non-terminal, so a clean result means RLS filtered them +-- and not that they had already dropped out of the report. +-- --------------------------------------------------------------------------- +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'ia_other_user') THEN + CREATE ROLE ia_other_user LOGIN; + END IF; +END $$; + +SELECT df.grant_usage('ia_other_user'); + +DO $$ +DECLARE + visible INT; + leaked INT; +BEGIN + -- Precondition: as the owner, both instances are in the report right now. + SELECT count(*) INTO visible + FROM df.instance_activity('0 seconds') a + WHERE a.label IN ('test-activity-busy', 'test-activity-wedged'); + + IF visible <> 2 THEN + RAISE EXCEPTION 'TEST FAILED [rls]: expected the owner to see both live instances, got %', visible; + END IF; + + SET LOCAL ROLE ia_other_user; + + SELECT count(*) INTO leaked + FROM df.instance_activity('0 seconds') a + WHERE a.label IN ('test-activity-busy', 'test-activity-wedged'); + + IF leaked <> 0 THEN + RAISE EXCEPTION 'TEST FAILED [rls]: another user saw % of our instances', leaked; + END IF; +END $$; + +RESET ROLE; + +-- --------------------------------------------------------------------------- +-- Case 5: a terminal instance is not reported. The report is about work in +-- progress, so once the busy instance completes it must drop out entirely, +-- even under a zero threshold. +-- --------------------------------------------------------------------------- +DO $$ +DECLARE + inst TEXT; + final_status TEXT; + found INT; +BEGIN + SELECT i.instance_id INTO inst FROM _ia_busy i; + + SELECT df.await_instance(inst, 60) INTO final_status; + IF final_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED [terminal]: expected completed, got %', final_status; + END IF; + + SELECT count(*) INTO found + FROM df.instance_activity('0 seconds') a + WHERE a.instance_id = inst; + + IF found <> 0 THEN + RAISE EXCEPTION 'TEST FAILED [terminal]: a completed instance must not be reported, got % rows', found; + END IF; +END $$; + +-- Cleanup +DO $$ +DECLARE + inst TEXT; +BEGIN + SELECT i.instance_id INTO inst FROM _ia_wedged i; + PERFORM df.cancel(inst); + PERFORM df.await_instance(inst, 60); +END $$; + +DROP TABLE _ia_busy; +DROP TABLE _ia_wedged; +DROP TABLE test_activity_log; +SELECT df.revoke_usage('ia_other_user'); +DROP ROLE IF EXISTS ia_other_user; +RESET SESSION AUTHORIZATION; +SELECT 'TEST PASSED' AS result;