diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 2029e0e6..d7d0f67f 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -22,9 +22,10 @@ pg_durable is a PostgreSQL extension that brings durable, fault-tolerant functio 12. [Visualizing Functions](#visualizing-functions) 13. [Monitoring](#monitoring) 14. [User Isolation & Privileges](#user-isolation--privileges) -15. [Troubleshooting](#troubleshooting) -16. [Quick Reference Card](#quick-reference-card) -17. [Appendix: Test Data Setup](#appendix-test-data-setup) +15. [Configuration](#configuration) +16. [Troubleshooting](#troubleshooting) +17. [Quick Reference Card](#quick-reference-card) +18. [Appendix: Test Data Setup](#appendix-test-data-setup) --- @@ -54,6 +55,8 @@ pg_durable enables you to define and execute **durable SQL functions** entirely | **Eternal Loops** | Create forever-running jobs with `@>` operator or `df.loop()` | | **Signals** | Wait for external events with `df.wait_for_signal()` | | **Variable Substitution** | Pass results between steps using `$name` | +| **Automatic Retries** | SQL and HTTP nodes are retried on transient failure (configurable) | +| **Loop Resilience** | Failed loop iterations are absorbed; the loop continues | | **Labels** | Tag functions with friendly names | | **Visualization** | Preview function structure with `df.explain()` | | **Monitoring** | Query function status, history, and metrics | @@ -955,6 +958,28 @@ SELECT df.start( `df.break(value)` exits the loop and returns the value as the loop's final result. +### Loop Resilience + +Loops are designed as **long-running supervisors**. When a loop body fails (after exhausting its retry attempts), the loop absorbs the error and continues to the next iteration instead of failing the entire durable function. This is critical for heartbeat loops, polling jobs, and periodic sync tasks that encounter transient failures. + +**Key behaviors:** + +- A failed iteration is logged but does not propagate — the loop continues. +- Iteration metrics (successes, failures, consecutive failures) are tracked across iterations. +- If **10 consecutive iterations** fail, the loop terminates with an error. This protects against permanently broken queries (e.g., dropped table, revoked permissions). +- The while-condition is always evaluated even after a failed body. If the condition itself fails, the loop terminates. + +```sql +-- This loop will continue even if process_item() occasionally fails +SELECT df.start( + df.loop( + 'SELECT process_item()' ~> df.sleep(5), + 'SELECT count(*) > 0 FROM task_queue' + ), + 'resilient-processor' +); +``` + ### Stopping a Loop Externally ```sql @@ -1530,6 +1555,45 @@ GRANT SELECT, INSERT, UPDATE, DELETE ON df.vars TO username; ``` **Note:** These grants will become more restrictive as the security model evolves. + +## Configuration + +pg_durable exposes the following PostgreSQL GUC (Grand Unified Configuration) settings: + +| Setting | Type | Default | Reload? | Description | +|---------|------|---------|---------|-------------| +| `pg_durable.worker_role` | string | `azuresu` | Restart | PostgreSQL role used by the background worker | +| `pg_durable.database` | string | `postgres` | Restart | Database the background worker connects to | +| `pg_durable.max_retries` | integer | `3` | Reload | Maximum retry attempts for SQL and HTTP activities | + +### Retry Configuration + +The `pg_durable.max_retries` setting controls how many times a failed SQL or HTTP activity is retried before the error propagates. The default is 3 (one initial attempt plus two retries, using exponential backoff starting at 100ms). + +```sql +-- Check current setting +SHOW pg_durable.max_retries; + +-- Disable retries (fail immediately) +SET pg_durable.max_retries = 1; + +-- Increase retries for flaky environments +ALTER SYSTEM SET pg_durable.max_retries = 5; +SELECT pg_reload_conf(); -- apply without restart +``` + +The retry value is captured at `df.start()` time, so changing the GUC only affects newly started durable functions — already-running functions keep the retry count they were started with. + +**What is retried:** + +| Node type | Retried? | Notes | +|-----------|----------|-------| +| SQL (`df.sql()`) | Yes | Connection drops, transient DB errors | +| HTTP (`df.http()`) | Yes | Network timeouts, server errors | +| Sleep, Signal, Loop, If, Join, Race, Break | No | Control flow — not I/O activities | + +**Backoff strategy:** Exponential backoff with 100ms base, 2x multiplier, capped at 30s. This is not user-configurable. + ## Troubleshooting ### Extension Exists But Workflows Don't Start @@ -1747,6 +1811,10 @@ SELECT df.result('id'); -- Cancel SELECT df.cancel('id', 'reason'); + +-- Configuration +SHOW pg_durable.max_retries; -- check retry setting (default: 3) +SET pg_durable.max_retries = 1; -- disable retries for this session ``` --- diff --git a/docs/api-reference.md b/docs/api-reference.md index 9996d19a..0f34fa9b 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -34,6 +34,8 @@ Creates a SQL execution node. df.sql('SELECT * FROM users WHERE id = 1') ``` +**Retries:** SQL nodes are automatically retried on transient failure, controlled by `pg_durable.max_retries` (default 3). + --- ### df.seq(a, b) / `~>` operator @@ -138,6 +140,8 @@ df.if('SELECT count(*) > 0 FROM q', 'SELECT ''yes''', 'SELECT ''no''') Repeats body (forever or while condition is true). +Failed iterations are absorbed — the loop continues to the next iteration. If 10 consecutive iterations fail, the loop terminates. + | Parameter | Type | Auto-wrap | Description | |-----------|------|-----------|-------------| | `body` | TEXT | ✅ Auto-wrap | Node to repeat | @@ -220,6 +224,8 @@ df.wait_for_signal('approval', 3600) -- 1 hour timeout Makes an HTTP request. +**Retries:** HTTP nodes are automatically retried on failure, controlled by `pg_durable.max_retries` (default 3). + | Parameter | Type | Auto-wrap | Description | |-----------|------|-----------|-------------| | `url` | TEXT | ❌ Literal | Request URL (supports `$var` substitution) | diff --git a/docs/retries.md b/docs/retries.md new file mode 100644 index 00000000..b1f1d39e --- /dev/null +++ b/docs/retries.md @@ -0,0 +1,408 @@ +# Proposal: Retry Support for pg_durable + +**Status:** Proposal + +## Problem + +Today, any node failure (SQL error, terminated connection, transient network issue) immediately fails the entire durable function. There is no retry mechanism. This was confirmed experimentally: + +- **Loop:** A single iteration failure (e.g., backend terminated during `pg_sleep`) causes the entire loop to fail permanently. The `await?` operator in `execute_loop_node` propagates the error immediately. +- **Sequence:** A failed step fails the entire sequence. Subsequent steps are never executed. +- **HTTP:** A transient 5xx or network timeout fails the function. + +This is problematic for real-world use cases: heartbeat loops, polling jobs, webhook deliveries, and AI pipelines all encounter transient failures that should not kill the entire function. + +## Duroxide Retry Support + +Duroxide already has a fully implemented `schedule_activity_with_retry()` API. pg_durable simply doesn't use it yet — all 8 `schedule_activity` call sites in `execute_function_graph.rs` use the non-retry variant. + +### Duroxide `RetryPolicy` API + +```rust +use duroxide::{RetryPolicy, BackoffStrategy}; + +// Simple: 3 attempts with default exponential backoff (100ms base, 2x multiplier, 30s max) +let result = ctx.schedule_activity_with_retry("Task", input, RetryPolicy::new(3)).await?; + +// Custom: 5 attempts, fixed 1s backoff, 30s per-attempt timeout +let policy = RetryPolicy::new(5) + .with_timeout(Duration::from_secs(30)) + .with_backoff(BackoffStrategy::Fixed { delay: Duration::from_secs(1) }); +let result = ctx.schedule_activity_with_retry("Task", input, policy).await?; +``` + +**`RetryPolicy` fields:** + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `max_attempts` | `u32` | 3 | Total attempts including initial (must be ≥ 1) | +| `backoff` | `BackoffStrategy` | Exponential(100ms, 2.0, 30s) | Delay between retry attempts | +| `timeout` | `Option` | None | Per-attempt timeout; timeouts exit immediately (no retry) | + +**`BackoffStrategy` variants:** + +| Variant | Formula | Description | +|---------|---------|-------------| +| `None` | 0 | No delay between retries | +| `Fixed { delay }` | constant | Same delay every retry | +| `Linear { base, max }` | `base × attempt` | Linearly increasing, capped at max | +| `Exponential { base, multiplier, max }` | `base × multiplier^(attempt-1)` | Exponential growth, capped at max | + +**Key semantics:** +- Activity *errors* trigger retries (with backoff delay between attempts) +- *Timeouts* exit immediately — they are NOT retried +- Fully deterministic: each retry creates a new `ActivityScheduled` event in the history, replays correctly +- No worker/provider changes needed — retries are orchestration-level control flow + +**References:** +- [duroxide RetryPolicy API](https://github.com/microsoft/duroxide/blob/main/src/lib.rs#L1312-L1398) +- [schedule_activity_with_retry implementation](https://github.com/microsoft/duroxide/blob/main/src/lib.rs#L2805-L2855) +- [Activity Retry Policy proposal (IMPLEMENTED)](https://github.com/microsoft/duroxide/blob/main/docs/proposals-impl/activity-retry-policy.md) +- [Orchestration Guide — Retry with Backoff](https://github.com/microsoft/duroxide/blob/main/docs/ORCHESTRATION-GUIDE.md#L1356-L1425) +- [Retry tests](https://github.com/microsoft/duroxide/blob/main/tests/schedule_with_retry_tests.rs) + +--- + +## Design + +Two separate mechanisms: + +1. **Node retries** — retry individual activity calls (SQL, HTTP) on transient failure +2. **Loop resilience** — a failed iteration doesn't kill the loop; the loop absorbs the error and continues + +These are complementary. A SQL node inside a loop body might retry 3 times (node retry), and if it still fails, the loop absorbs that failure and moves to the next iteration (loop resilience). + +--- + +## 1. Node Retries + +### What changes + +Replace `ctx.schedule_activity(NAME, input)` with `ctx.schedule_activity_with_retry(NAME, input, policy)` for SQL and HTTP nodes in `src/orchestrations/execute_function_graph.rs`. + +### Which nodes get retries + +| Node type | Retryable? | Rationale | +|-----------|------------|-----------| +| SQL | Yes | Connection drops, transient DB errors | +| HTTP | Yes | Network timeouts, 5xx errors | +| SLEEP | No | Deterministic timer, no I/O | +| SIGNAL | No | Deterministic wait, no I/O | +| LOOP, IF, THEN, JOIN, RACE, BREAK | No | Control flow — not activities | +| `update-node-status` | Yes | Status writes should be best-effort resilient | +| `update-instance-status` | Yes | Same | +| `load-function-graph` | No | Already has its own polling/retry for transaction visibility | + +### Default retry policy + +``` +max_attempts: 3 +backoff: Exponential { base: 100ms, multiplier: 2.0, max: 30s } +timeout: None +``` + +This is duroxide's `RetryPolicy::default()`. Suitable for most transient failures (connection drops, brief outages). Three attempts with 100ms → 200ms → 400ms delays. + +### Code change (SQL node) + +```rust +// Before: +let result = ctx + .schedule_activity(activities::execute_sql::NAME, input.to_string()) + .await?; + +// After: +let result = ctx + .schedule_activity_with_retry( + activities::execute_sql::NAME, + input.to_string(), + retry_policy.clone(), + ) + .await?; +``` + +Where `retry_policy` is constructed from the configured setting (see Configuration section). + +### Code change (HTTP node) + +Same pattern — replace `schedule_activity` with `schedule_activity_with_retry` at the HTTP call site. + +### Code change (status updates) + +Status updates (`update-node-status`, `update-instance-status`) already use `let _ =` to ignore errors. Adding retry makes them more reliably delivered without changing error semantics: + +```rust +let _ = ctx + .schedule_activity_with_retry( + activities::update_node_status::NAME, + input.to_string(), + RetryPolicy::new(3).with_backoff(BackoffStrategy::Fixed { + delay: Duration::from_secs(1), + }), + ) + .await; +``` + +--- + +## 2. Loop Resilience + +### Problem + +A loop's purpose is long-running, repeated execution: heartbeats, polling, periodic sync. A single iteration failure should not be fatal. Today, `execute_loop_node` does: + +```rust +let body_result = Box::pin(execute_function_node_with_vars(...)).await?; +// ^ error kills the loop +``` + +### Design: loops absorb iteration failures + +When a loop body fails (after exhausting its own node retries), the loop catches the error, logs it, and continues to the next iteration via `continue_as_new`. The failed iteration is recorded but does not propagate. + +```rust +// Proposed change in execute_loop_node: +let body_result = Box::pin(execute_function_node_with_vars( + ctx, graph, body_id, results, exec_ctx, +)).await; + +match body_result { + Ok(result) => { + if is_break_signal(&result) { + return Ok(extract_break_value(&result)); + } + // Success — check condition and continue + } + Err(err) => { + ctx.trace_warn(format!("Loop iteration failed: {err}")); + // Record failure but continue to next iteration + } +} +``` + +The loop always continues (subject to its while-condition) regardless of whether the body succeeded or failed. Note: if the while-condition itself fails, that *does* fail the loop — the condition must be evaluable for the loop to make a continue/exit decision. + +### This is not configurable + +Loop resilience is always on. A loop that should fail on error can use `df.seq` outside the loop or use `df.if` inside the body to handle errors explicitly. The mental model: **a loop is a long-running supervisor** — it keeps going. + +### Loop iteration metrics + +Since failed iterations are absorbed, visibility is critical. The orchestration should track and log iteration outcomes. Proposed approach: + +**Trace logging per iteration:** +``` +Loop iteration 1: completed (result: ...) +Loop iteration 2: failed (error: SQL execution failed: connection terminated) +Loop iteration 3: completed (result: ...) +``` + +**Rolling window in orchestration state (passed via `continue_as_new`):** + +Track iteration outcomes in the `FunctionInput` state passed across `continue_as_new` boundaries: + +```rust +struct LoopMetrics { + total_iterations: u64, + total_successes: u64, + total_failures: u64, + // Last N iteration outcomes (ring buffer, carried across continue_as_new) + recent_outcomes: Vec, // last 5 +} + +struct IterationOutcome { + iteration: u64, + succeeded: bool, + error: Option, // truncated to ~200 chars +} +``` + +This state is: +- Logged as a trace at each iteration (visible in `~/.pgrx/17.log` / worker logs) +- Passed through `continue_as_new` so it survives history truncation +- Available for future monitoring queries (e.g., `df.loop_status(instance_id)`) + +**Example trace output:** +``` +Loop iteration 47: failed (error: connection terminated) +Loop stats: 47 iterations, 45 successes, 2 failures +Recent: [ok, ok, FAIL, ok, ok] +``` + +### Consecutive failure limit + +To prevent a loop from spinning forever on a permanently broken query, add a **consecutive failure limit**. If N consecutive iterations fail (default: 10), the loop fails. This protects against: +- Permanently dropped table +- Revoked permissions +- Invalid SQL that will never succeed + +This is a safety mechanism, not user-configurable in v1. The value 10 is chosen to tolerate bursts of transient failures while catching permanent ones. + +--- + +## 3. Configuration + +### Option A: GUC (global setting) — **Recommended for v1** + +Add a single GUC that controls the default retry count for all activity nodes: + +``` +pg_durable.max_retries = 3 (default) +``` + +**Pros:** +- Simple to implement — one place to read the setting +- Consistent behavior across all functions +- DBA can tune for the environment (set to 1 for fail-fast testing, 5 for flaky networks) +- No DSL changes needed + +**Implementation:** + +```rust +pub static MAX_RETRIES: GucSetting = GucSetting::::new(3); + +// In _PG_init(): +GucRegistry::define_int_guc( + c"pg_durable.max_retries", + c"Maximum retry attempts for SQL and HTTP activities (default 3)", + c"", + &MAX_RETRIES, + 1, // min + 100, // max + GucContext::Sighup, // reloadable without restart + GucFlags::default(), +); +``` + +The orchestration reads this value when constructing the retry policy. Since `GucContext::Sighup`, changes take effect on `pg_ctl reload` without restart. + +The backoff strategy is hardcoded to `Exponential { base: 100ms, multiplier: 2.0, max: 30s }` — suitable for most workloads and not worth exposing as a GUC in v1. + +### Option B: Per-function configuration (future) + +Allow users to set retry policy per `df.start()` call: + +```sql +SELECT df.start( + df.sql('SELECT process_batch()'), + 'batch-job', + retries => 5 +); +``` + +Or per node: + +```sql +df.sql('SELECT flaky_query()', retries => 10) +``` + +This requires: +- Adding a `retries` parameter to DSL functions +- Storing it in `df.nodes` +- Reading it in the orchestration per-node + +**Recommendation:** Defer to a future version. The GUC provides adequate control for v1, and per-node configuration adds complexity to the DSL, node schema, and orchestration. + +### Option C: Per-function via `df.start()` parameter (reasonable v1.1) + +A middle ground between A and B — set retry policy per durable function invocation: + +```sql +SELECT df.start( + df.sql('SELECT 1') ~> df.sql('SELECT 2'), + 'my-function', + max_retries => 5 +); +``` + +This is passed through `FunctionInput.vars` (or a dedicated field) and read by the orchestration. All nodes in that function share the same retry policy. This avoids both the rigidity of a global GUC and the complexity of per-node configuration. + +--- + +## Implementation Plan + +### Phase 1: Node retries via GUC + +1. Add `pg_durable.max_retries` GUC +2. In `execute_sql_node` and `execute_http_node`, replace `schedule_activity` with `schedule_activity_with_retry` using the GUC value +3. Add retry to `update_node_status` and `update_instance_status` calls (hardcoded 3 attempts, fixed 1s backoff) +4. Add E2E test: start function, kill backend mid-execution, verify function completes after retry + +### Phase 2: Loop resilience + +1. Change `execute_loop_node` to catch body errors instead of propagating +2. Add `LoopMetrics` struct, carry through `continue_as_new` +3. Add consecutive failure limit (10) +4. Add trace logging for iteration outcomes +5. Add E2E test: loop with intermittent failures, verify loop continues and metrics are logged + +### Phase 3: Per-function retries (future) + +1. Add `max_retries` parameter to `df.start()` +2. Carry through `FunctionInput` +3. Orchestration reads per-function setting, falls back to GUC + +--- + +## E2E Test Sketches + +### Test: Node retry on backend termination + +```sql +-- Start a function with a long-running SQL +SELECT df.start( + df.sql('SELECT pg_sleep(120)'), + 'retry-test' +) AS instance_id; + +-- Find and kill the backend +SELECT pg_terminate_backend(pid) +FROM pg_stat_activity +WHERE query = 'SELECT pg_sleep(120)' AND pid != pg_backend_pid(); + +-- With retries enabled, the function should retry the SQL node +-- and eventually complete (the retry will start a new pg_sleep) +-- For testing, use a shorter sleep or a query that succeeds on retry +``` + +### Test: Loop resilience + +```sql +-- Set up a table that causes failure on specific iterations +CREATE TABLE loop_test (iteration int, ts timestamptz DEFAULT now()); + +-- Create a function referenced by the SQL: +-- Fails on every 3rd call by dividing by zero +CREATE OR REPLACE FUNCTION flaky_insert(iter int) RETURNS void AS $$ +BEGIN + IF iter % 3 = 0 THEN + RAISE EXCEPTION 'simulated failure on iteration %', iter; + END IF; + INSERT INTO loop_test VALUES (iter); +END; +$$ LANGUAGE plpgsql; + +-- Loop that calls the flaky function +SELECT df.start( + df.loop( + df.sql('SELECT flaky_insert(nextval(''loop_counter''))'), + 'SELECT currval(''loop_counter'') < 10' + ), + 'loop-resilience-test' +); + +-- After completion: loop_test should have ~7 rows (iterations 1,2,4,5,7,8,10) +-- Iterations 3, 6, 9 failed but the loop continued +``` + +--- + +## Open Questions + +1. **Should retries apply to the while-condition of a loop?** Proposed: no — if the condition fails, the loop fails. The condition should be a simple, reliable query. A flaky condition makes the loop's behavior unpredictable. + +2. **Should `update_node_status` retries block the orchestration?** Today, status updates use `let _ =` (fire-and-forget for errors). With retry, a status update could block for seconds during backoff. Proposed: keep fire-and-forget semantics but use `schedule_activity_with_retry` — duroxide will retry in the background without blocking the orchestration's critical path. (Need to verify this is how duroxide handles it — TBD.) + +3. **Should we expose retry metrics via SQL?** E.g., `df.node_retries(instance_id)` showing how many retries each node needed. This is naturally available in duroxide's history (multiple `ActivityScheduled` events for the same logical activity). Proposed: defer to Phase 3. + +4. **Per-attempt timeout:** The GUC controls `max_attempts` only. Should we also expose a per-attempt timeout GUC (`pg_durable.activity_timeout`)? For SQL nodes, PostgreSQL's `statement_timeout` already provides this. For HTTP nodes, `timeout_seconds` is already in the `df.http()` DSL. Proposed: no additional GUC needed in v1. diff --git a/src/dsl.rs b/src/dsl.rs index 4e3d8289..258a6b8a 100644 --- a/src/dsl.rs +++ b/src/dsl.rs @@ -647,6 +647,8 @@ pub fn start( instance_id: instance_id.clone(), label: label.map(|s| s.to_string()), vars, + max_retries: crate::MAX_RETRIES.get() as u32, + loop_metrics: None, }; let input_json = serde_json::to_string(&input).unwrap_or(instance_id.clone()); diff --git a/src/lib.rs b/src/lib.rs index c63b0766..eb0edf19 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -17,6 +17,8 @@ pub static WORKER_ROLE: GucSetting> = pub static DATABASE: GucSetting> = GucSetting::>::new(Some(c"postgres")); +pub static MAX_RETRIES: GucSetting = GucSetting::::new(3); + // Module declarations pub mod activities; pub mod client; @@ -63,6 +65,17 @@ pub extern "C-unwind" fn _PG_init() { GucFlags::default(), ); + GucRegistry::define_int_guc( + c"pg_durable.max_retries", + c"Maximum retry attempts for SQL and HTTP activities (default 3)", + c"Controls how many times a failed SQL or HTTP activity is retried before the error propagates. Set to 1 to disable retries.", + &MAX_RETRIES, + 1, // min + 100, // max + GucContext::Sighup, + GucFlags::default(), + ); + worker::register_background_worker(); } diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index c8b4c701..7ffd1dc3 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -4,16 +4,22 @@ //! - No I/O except through activities //! - No random numbers, current time, or other non-deterministic sources //! - Same input must always produce the same scheduling decisions +//! +//! Note: `RetryPolicy` configuration is read from `FunctionInput.max_retries`, +//! which is captured at `df.start()` time. This is deterministic because the +//! value is fixed for the function's lifetime and stored in the orchestration +//! input. Duroxide's `schedule_activity_with_retry` replays from history for +//! already-completed activities, so the policy only affects new scheduling. use std::collections::HashMap; use std::time::Duration; -use duroxide::OrchestrationContext; +use duroxide::{BackoffStrategy, OrchestrationContext, RetryPolicy}; use crate::activities; use crate::types::{ evaluate_condition, substitute_all, substitute_all_raw, FunctionGraph, FunctionInput, - FunctionNode, SystemVars, + FunctionNode, LoopMetrics, SystemVars, }; /// Orchestration name for ExecuteFunctionGraph @@ -27,8 +33,24 @@ pub const SUBTREE_NAME: &str = "pg_durable::orchestration::execute-subtree"; struct ExecutionContext { vars: HashMap, label: Option, + /// Retry policy for SQL and HTTP activities, derived from `FunctionInput.max_retries`. + activity_retry_policy: RetryPolicy, + /// Loop metrics carried across `continue_as_new` boundaries. + loop_metrics: Option, +} + +/// Fixed retry policy for status updates (update-node-status, update-instance-status). +/// These are best-effort and use a simple fixed backoff. +fn status_update_retry_policy() -> RetryPolicy { + RetryPolicy::new(3).with_backoff(BackoffStrategy::Fixed { + delay: Duration::from_secs(1), + }) } +/// Maximum consecutive loop iteration failures before the loop is terminated. +/// Protects against permanently broken queries spinning forever. +const MAX_CONSECUTIVE_LOOP_FAILURES: u64 = 10; + /// Execute a complete function graph pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result { let input: FunctionInput = serde_json::from_str(&input_json) @@ -51,6 +73,8 @@ pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result Result = HashMap::new(); - // Create execution context with vars + // Create execution context with vars and retry policy let exec_ctx = ExecutionContext { vars: input.vars.clone(), label: input.label.clone(), + activity_retry_policy: RetryPolicy::new(input.max_retries), + loop_metrics: input.loop_metrics.clone(), }; let function_result = @@ -87,9 +113,10 @@ pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result Result { + metrics.record_success(); + ctx.trace_info(format!( + "Loop iteration {}: completed", + metrics.total_iterations + )); + result + } + Err(err) => { + metrics.record_failure(&err); + ctx.trace_warn(format!( + "Loop iteration {}: failed (error: {})", + metrics.total_iterations, err + )); + ctx.trace_info(format!("Loop stats: {}", metrics.summary())); + + // Check consecutive failure limit + if metrics.consecutive_failures >= MAX_CONSECUTIVE_LOOP_FAILURES { + return Err(format!( + "Loop terminated: {} consecutive failures exceeded limit of {}. Last error: {}", + metrics.consecutive_failures, MAX_CONSECUTIVE_LOOP_FAILURES, err + )); + } + + // Continue to next iteration despite failure + let new_input = FunctionInput { + instance_id: graph.instance_id.clone(), + label: exec_ctx.label.clone(), + vars: exec_ctx.vars.clone(), + max_retries: exec_ctx.activity_retry_policy.max_attempts, + loop_metrics: Some(metrics), + }; + ctx.trace_info("Continuing loop after failed iteration"); + return ctx + .continue_as_new( + serde_json::to_string(&new_input).unwrap_or(graph.instance_id.clone()), + ) + .await + .map(|_| "null".to_string()) + .map_err(|e| format!("continue_as_new failed: {e:?}")); + } + }; + + ctx.trace_info(format!("Loop stats: {}", metrics.summary())); // Check for break signal from body if is_break_signal(&body_result) { @@ -447,11 +541,13 @@ async fn execute_loop_node( } ctx.trace_info("Continuing as new for next loop iteration"); - // Preserve vars in continue_as_new input + // Preserve vars and metrics in continue_as_new input let new_input = FunctionInput { instance_id: graph.instance_id.clone(), label: exec_ctx.label.clone(), vars: exec_ctx.vars.clone(), + max_retries: exec_ctx.activity_retry_policy.max_attempts, + loop_metrics: Some(metrics), }; // duroxide 0.1.1: continue_as_new returns an awaitable future - return it directly @@ -762,7 +858,11 @@ async fn execute_http_node( ctx.trace_info(format!("Executing HTTP {method} {url}")); let result = ctx - .schedule_activity(activities::execute_http::NAME, final_config) + .schedule_activity_with_retry( + activities::execute_http::NAME, + final_config, + exec_ctx.activity_retry_policy.clone(), + ) .await?; // Store result if named diff --git a/src/types.rs b/src/types.rs index 6ba4ffe8..9cea14af 100644 --- a/src/types.rs +++ b/src/types.rs @@ -369,6 +369,108 @@ pub struct FunctionInput { pub label: Option, #[serde(default)] pub vars: std::collections::HashMap, + /// Maximum retry attempts for SQL and HTTP activities. + /// Captured from `pg_durable.max_retries` GUC at `df.start()` time. + #[serde(default = "default_max_retries")] + pub max_retries: u32, + /// Loop iteration metrics carried across `continue_as_new` boundaries. + #[serde(skip_serializing_if = "Option::is_none")] + pub loop_metrics: Option, +} + +fn default_max_retries() -> u32 { + 3 +} + +/// Tracks loop iteration outcomes across `continue_as_new` boundaries. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LoopMetrics { + pub total_iterations: u64, + pub total_successes: u64, + pub total_failures: u64, + pub consecutive_failures: u64, + /// Ring buffer of the last 5 iteration outcomes (carried across continue_as_new). + pub recent_outcomes: Vec, +} + +/// Outcome of a single loop iteration. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct IterationOutcome { + pub iteration: u64, + pub succeeded: bool, + /// Truncated error message (max ~200 chars) if the iteration failed. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +impl Default for LoopMetrics { + fn default() -> Self { + Self::new() + } +} + +impl LoopMetrics { + pub fn new() -> Self { + Self { + total_iterations: 0, + total_successes: 0, + total_failures: 0, + consecutive_failures: 0, + recent_outcomes: Vec::new(), + } + } + + pub fn record_success(&mut self) { + self.total_iterations += 1; + self.total_successes += 1; + self.consecutive_failures = 0; + self.push_outcome(IterationOutcome { + iteration: self.total_iterations, + succeeded: true, + error: None, + }); + } + + pub fn record_failure(&mut self, error: &str) { + self.total_iterations += 1; + self.total_failures += 1; + self.consecutive_failures += 1; + let truncated_error = if error.len() > 200 { + format!("{}...", &error[..200]) + } else { + error.to_string() + }; + self.push_outcome(IterationOutcome { + iteration: self.total_iterations, + succeeded: false, + error: Some(truncated_error), + }); + } + + fn push_outcome(&mut self, outcome: IterationOutcome) { + self.recent_outcomes.push(outcome); + // Keep only the last 5 + if self.recent_outcomes.len() > 5 { + self.recent_outcomes.remove(0); + } + } + + /// Summary string for trace logging. + pub fn summary(&self) -> String { + let recent: Vec<&str> = self + .recent_outcomes + .iter() + .map(|o| if o.succeeded { "ok" } else { "FAIL" }) + .collect(); + format!( + "{} iterations, {} successes, {} failures (consecutive: {}). Recent: [{}]", + self.total_iterations, + self.total_successes, + self.total_failures, + self.consecutive_failures, + recent.join(", ") + ) + } } /// Configuration for HTTP requests diff --git a/tests/e2e/sql/36_retry_node.sql b/tests/e2e/sql/36_retry_node.sql new file mode 100644 index 00000000..3bf129ff --- /dev/null +++ b/tests/e2e/sql/36_retry_node.sql @@ -0,0 +1,135 @@ +-- Test: Node retry on transient SQL failure +-- Tests that SQL activities are retried on failure up to max_retries attempts. +-- Uses a function that fails on the first N calls and succeeds on the Nth. +-- Expected: The durable function completes successfully after retries. +-- +-- Key design note: Each activity retry runs in a separate transaction. If the +-- function raises an exception, the transaction is rolled back, so any INSERTs +-- in that transaction are lost. We use SEQUENCES to count attempts because +-- nextval() is never rolled back, even if the calling transaction aborts. + +-- ============================================================================ +-- Test 1: SQL node succeeds after transient failures (retried by duroxide) +-- ============================================================================ + +DROP SEQUENCE IF EXISTS test_retry_seq; +CREATE SEQUENCE test_retry_seq START 1; +DROP TABLE IF EXISTS test_retry_log; +CREATE TABLE test_retry_log (id SERIAL, attempt INT, ts TIMESTAMP DEFAULT now()); + +-- This function tracks attempts via a sequence (survives rollback). +-- It fails the first 2 calls and succeeds on the 3rd. +CREATE OR REPLACE FUNCTION test_retry_flaky() RETURNS TEXT AS $$ +DECLARE + v_attempt INT; +BEGIN + v_attempt := nextval('test_retry_seq'); + + IF v_attempt < 3 THEN + RAISE EXCEPTION 'Transient failure on attempt %', v_attempt; + END IF; + + INSERT INTO test_retry_log (attempt) VALUES (v_attempt); + RETURN format('success on attempt %s', v_attempt); +END; +$$ LANGUAGE plpgsql; + +-- pg_durable.max_retries defaults to 3, so this should succeed: +-- Attempt 1: fail, Attempt 2: fail, Attempt 3: succeed +CREATE TEMP TABLE _test1_state AS +SELECT df.start( + df.sql('SELECT test_retry_flaky()'), + 'test-retry-succeed' +) AS instance_id; + +DO $$ +DECLARE + v_instance_id TEXT; + v_status TEXT; + v_logged INT; +BEGIN + SELECT instance_id INTO v_instance_id FROM _test1_state; + RAISE NOTICE 'Test 1 - SQL retry: instance %', v_instance_id; + + SELECT df.wait_for_completion(v_instance_id, 30) INTO v_status; + + SELECT COUNT(*) INTO v_logged FROM test_retry_log; + + IF v_status != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED [retry-succeed]: expected completed, got %. Rows logged: %', v_status, v_logged; + END IF; + + -- The successful attempt (3rd) should have inserted a row + IF v_logged < 1 THEN + RAISE EXCEPTION 'TEST FAILED [retry-succeed]: expected at least 1 logged row, got %', v_logged; + END IF; + + RAISE NOTICE 'PASSED: SQL node retried and succeeded (% rows logged)', + v_logged; +END $$; + +DROP TABLE _test1_state; + +-- ============================================================================ +-- Test 2: SQL node fails permanently when retries are exhausted +-- ============================================================================ + +DROP SEQUENCE IF EXISTS test_retry_fail_seq; +CREATE SEQUENCE test_retry_fail_seq START 1; + +-- This function always fails. Sequence tracks how many times it was called. +CREATE OR REPLACE FUNCTION test_retry_always_fail() RETURNS TEXT AS $$ +DECLARE + v_attempt INT; +BEGIN + v_attempt := nextval('test_retry_fail_seq'); + RAISE EXCEPTION 'permanent failure on attempt %', v_attempt; +END; +$$ LANGUAGE plpgsql; + +-- With default max_retries=3, this function should fail after 3 attempts +CREATE TEMP TABLE _test2_state AS +SELECT df.start( + df.sql('SELECT test_retry_always_fail()'), + 'test-retry-exhaust' +) AS instance_id; + +DO $$ +DECLARE + v_instance_id TEXT; + v_status TEXT; + v_attempts INT; +BEGIN + SELECT instance_id INTO v_instance_id FROM _test2_state; + RAISE NOTICE 'Test 2 - Retry exhausted: instance %', v_instance_id; + + SELECT df.wait_for_completion(v_instance_id, 30) INTO v_status; + + -- Sequence value tells us how many times the function was called + SELECT last_value INTO v_attempts FROM test_retry_fail_seq; + + IF v_status != 'failed' THEN + RAISE EXCEPTION 'TEST FAILED [retry-exhaust]: expected failed, got %. Attempts: %', v_status, v_attempts; + END IF; + + -- Should have exactly 3 attempts (max_retries=3) + IF v_attempts != 3 THEN + RAISE EXCEPTION 'TEST FAILED [retry-exhaust]: expected 3 attempts, got %', v_attempts; + END IF; + + RAISE NOTICE 'PASSED: SQL node failed after exhausting % retry attempts', v_attempts; +END $$; + +DROP TABLE _test2_state; + +-- ============================================================================ +-- Cleanup +-- ============================================================================ + +DROP FUNCTION IF EXISTS test_retry_flaky(); +DROP FUNCTION IF EXISTS test_retry_always_fail(); +DROP TABLE IF EXISTS test_retry_log; +DROP SEQUENCE IF EXISTS test_retry_seq; +DROP SEQUENCE IF EXISTS test_retry_fail_seq; + +SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/37_loop_resilience.sql b/tests/e2e/sql/37_loop_resilience.sql new file mode 100644 index 00000000..25691894 --- /dev/null +++ b/tests/e2e/sql/37_loop_resilience.sql @@ -0,0 +1,144 @@ +-- Test: Loop resilience - failed iterations don't kill the loop +-- Tests that a loop continues executing after individual iteration failures. +-- Expected: Loop absorbs iteration failures and continues to next iteration. + +-- ============================================================================ +-- Test 1: Loop continues after intermittent failures +-- ============================================================================ + +DROP TABLE IF EXISTS test_loop_resilience; +CREATE TABLE test_loop_resilience (id SERIAL, iteration INT, ts TIMESTAMP DEFAULT now()); +DROP SEQUENCE IF EXISTS loop_resilience_seq; +CREATE SEQUENCE loop_resilience_seq START 1; + +-- This function fails on every 3rd call (iterations 3, 6, 9) but succeeds otherwise. +-- Uses a shared sequence so retries of the same iteration count separately. +CREATE OR REPLACE FUNCTION test_loop_flaky_insert() RETURNS TEXT AS $$ +DECLARE + v_iter INT; +BEGIN + v_iter := nextval('loop_resilience_seq'); + + IF v_iter % 3 = 0 THEN + RAISE EXCEPTION 'simulated failure on call %', v_iter; + END IF; + + INSERT INTO test_loop_resilience (iteration) VALUES (v_iter); + RETURN format('inserted iteration %s', v_iter); +END; +$$ LANGUAGE plpgsql; + +-- Loop with a flaky body. The body SQL calls test_loop_flaky_insert which +-- fails on every 3rd call. With loop resilience, the loop should continue +-- past failures and run until the condition becomes false. +-- +-- Note: Because retries also consume sequence values, the actual sequence values +-- in the table will vary. We verify the loop completed and inserted some rows. +CREATE TEMP TABLE _test1_state AS +SELECT df.start( + df.loop( + df.sql('SELECT test_loop_flaky_insert()'), + 'SELECT currval(''loop_resilience_seq'') < 10' + ), + 'test-loop-resilience' +) AS instance_id; + +DO $$ +DECLARE + v_instance_id TEXT; + v_status TEXT; + v_cnt INT; + v_seq_val INT; +BEGIN + SELECT instance_id INTO v_instance_id FROM _test1_state; + RAISE NOTICE 'Test 1 - Loop resilience: instance %', v_instance_id; + + -- Wait longer since the loop has multiple iterations with retries and backoff + SELECT df.wait_for_completion(v_instance_id, 60) INTO v_status; + + SELECT COUNT(*) INTO v_cnt FROM test_loop_resilience; + SELECT last_value INTO v_seq_val FROM loop_resilience_seq; + + IF v_status != 'completed' THEN + RAISE EXCEPTION 'TEST FAILED [loop-resilience]: expected completed, got %. Rows: %, seq: %', + v_status, v_cnt, v_seq_val; + END IF; + + -- We should have some rows (iterations that succeeded) + IF v_cnt < 1 THEN + RAISE EXCEPTION 'TEST FAILED [loop-resilience]: expected at least 1 row, got %', v_cnt; + END IF; + + RAISE NOTICE 'PASSED: Loop continued past failures. % successful inserts, sequence reached %', + v_cnt, v_seq_val; +END $$; + +DROP TABLE _test1_state; + +-- ============================================================================ +-- Test 2: Loop terminates after too many consecutive failures +-- ============================================================================ + +DROP TABLE IF EXISTS test_loop_consecutive_fail; +CREATE TABLE test_loop_consecutive_fail (id SERIAL, attempt INT, ts TIMESTAMP DEFAULT now()); + +-- This function always fails. With retry count of 1 (no retries) and +-- consecutive failure limit of 10, the loop should fail after 10 iterations. +CREATE OR REPLACE FUNCTION test_loop_always_fail() RETURNS TEXT AS $$ +BEGIN + INSERT INTO test_loop_consecutive_fail (attempt) VALUES ( + (SELECT COALESCE(MAX(attempt), 0) + 1 FROM test_loop_consecutive_fail) + ); + RAISE EXCEPTION 'always fails'; +END; +$$ LANGUAGE plpgsql; + +-- Set max_retries to 1 (no retries) so each iteration fails quickly +SET pg_durable.max_retries = 1; + +CREATE TEMP TABLE _test2_state AS +SELECT df.start( + df.loop( + df.sql('SELECT test_loop_always_fail()') + ), + 'test-consecutive-fail' +) AS instance_id; + +-- Reset max_retries to default +SET pg_durable.max_retries = 3; + +DO $$ +DECLARE + v_instance_id TEXT; + v_status TEXT; + v_attempts INT; +BEGIN + SELECT instance_id INTO v_instance_id FROM _test2_state; + RAISE NOTICE 'Test 2 - Consecutive failure limit: instance %', v_instance_id; + + -- This should fail once the consecutive failure limit (10) is hit + SELECT df.wait_for_completion(v_instance_id, 60) INTO v_status; + + SELECT COUNT(*) INTO v_attempts FROM test_loop_consecutive_fail; + + IF v_status != 'failed' THEN + RAISE EXCEPTION 'TEST FAILED [consecutive-fail]: expected failed, got %. Attempts: %', + v_status, v_attempts; + END IF; + + RAISE NOTICE 'PASSED: Loop terminated after % consecutive failures', v_attempts; +END $$; + +DROP TABLE _test2_state; + +-- ============================================================================ +-- Cleanup +-- ============================================================================ + +DROP FUNCTION IF EXISTS test_loop_flaky_insert(); +DROP FUNCTION IF EXISTS test_loop_always_fail(); +DROP TABLE IF EXISTS test_loop_resilience; +DROP TABLE IF EXISTS test_loop_consecutive_fail; +DROP SEQUENCE IF EXISTS loop_resilience_seq; + +SELECT 'TEST PASSED' AS result;