Add a node failure policy: retries, and what happens when retrying doesn't help

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

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"

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

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