From 165cfa244debdd71d6d04fc767ec5c9288d8ed04 Mon Sep 17 00:00:00 2001 From: Agent Date: Tue, 1 Sep 2026 14:46:29 +0000 Subject: [PATCH 1/4] fix(core): terminalize unfinished tasks when runs fail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Failed runs archive their queue messages but currently leave queued and started task rows active, allowing stale recovery and callback races to misrepresent or revive terminal work. Mark only unfinished tasks as cancelled before queue archival, preserve completed and genuinely failed outcomes, guard stalled recovery by parent state, and repair historical failed runs. Worker abort behavior and the #621 timeout correction remain separate work. Review round 1 changes: - Add late_callbacks_post_lock_race.test.sql: deterministic two-session dblink test that exercises the post-lock guards the single-session tests cannot reach. A ctrl session holds FOR UPDATE on the run row; late complete_task (single-to-map type-violating completion) and fail_task callbacks block after their pre-lock checks; ctrl then fails the run and commits. Asserts the cancelled task row, runs.failed_at, remaining_steps, event counts, and queue state stay untouched after both callbacks resume. Mutation-checked: removing the post-lock guard fails 8 of 15 asserts. dblink mechanics learned the hard way: pg_stat_activity snapshots are cached per transaction (poll through a dedicated autocommit probe connection), reset_db() does not clear pgflow.workers (cleanup deletes the test worker), and a queue read in this transaction would block the cleanup's drop_queue (read the queue through the probe connection). - cancelled_task_late_callbacks_are_idempotent.test.sql: run_before now includes run_id. Previously the inner run_id resolved to the outer query, making both comparison predicates tautologies instead of binding the idempotency assertions to the target run. - Trim trailing whitespace in 0062_function_requeue_stalled_tasks.sql (source of the migration lint failure) and regenerate the temp migration (repair UPDATE unchanged, rehashed). - edge-worker-integration CI failures on the first run were flake: same job green on main and green on the --failed rerun of this PR's run (33521700558, attempt 2), so no code change. Review round 2 changes: - late_callbacks_post_lock_race.test.sql: the late fail_task() callback now targets single_step, the task the winning boom failure cancels. Targeting boom (the culprit) only exercised the terminal-step guard: boom's step state is 'failed' after the winning failure, so that callback never needed the post-lock failed-run guard. single_step's step state stays 'started' under a failed run, so the post-lock guard is the only thing protecting the cancelled row, which is the regression this test exists to catch. - The wait proof now uses pg_blocking_pids() instead of a co-existence check: each callback backend's blocking chain (recursive, because PostgreSQL queues row-lock waiters FIFO — the second callback's direct blocker is the first callback, not ctrl) must terminate at the ctrl backend before the winning failure commits. Sleep-only/timing-only evidence is gone. - After both callbacks resume: single_step history (status, output, error_message, attempts_count, completed_at, failed_at) asserted unchanged after BOTH late callbacks, plus direct assertions that boom stays genuinely failed with the winning error_message. - Bounded lock_timeout (30s) on the late callback sessions so a stuck setup errors loudly instead of hanging; ctrl keeps its 5s bound; session/worker cleanup unchanged. - Mutation check re-run against the new test: removing only the fail_task post-lock failed-run guard fails exactly 2 of 18 asserts — runs.failed_at wiped to NULL by the guardless run UPDATE (the cancelled task matches no rows in fail_or_retry_task, so maybe_fail_step is empty and run_update takes the ELSE branches) and a duplicate run:failed event (run_update still returns 'failed', so v_run_failed re-fires the event). Guard restored, test green again 10/10 consecutive runs. --- .changeset/cancel-failed-run-tasks.md | 5 + NOMENCLATURE_GUIDE.md | 1 + pkgs/core/README.md | 7 +- pkgs/core/schemas/0060_tables_runtime.sql | 2 +- .../0062_function_requeue_stalled_tasks.sql | 9 +- ...00_function_cascade_resolve_conditions.sql | 105 +- .../schemas/0100_function_complete_task.sql | 99 +- pkgs/core/schemas/0100_function_fail_task.sql | 71 +- ...pgflow_temp_failed_run_terminalization.sql | 1259 +++++++++++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- .../no_mutations_on_failed_run.test.sql | 12 +- ...met_fail_archives_active_messages.test.sql | 109 +- .../archive_sibling_map_tasks.test.sql | 168 ++- ...ask_late_callbacks_are_idempotent.test.sql | 130 ++ .../late_callbacks_post_lock_race.test.sql | 324 +++++ .../ignores_terminal_parent_state.test.sql | 136 ++ .../cancels_unfinished_tasks.test.sql | 164 +++ .../src/content/docs/concepts/data-model.mdx | 4 +- 18 files changed, 2403 insertions(+), 205 deletions(-) create mode 100644 .changeset/cancel-failed-run-tasks.md create mode 100644 pkgs/core/supabase/migrations/20260901183450_pgflow_temp_failed_run_terminalization.sql create mode 100644 pkgs/core/supabase/tests/fail_task_when_exhausted/cancelled_task_late_callbacks_are_idempotent.test.sql create mode 100644 pkgs/core/supabase/tests/fail_task_when_exhausted/late_callbacks_post_lock_race.test.sql create mode 100644 pkgs/core/supabase/tests/requeue_stalled_tasks/ignores_terminal_parent_state.test.sql create mode 100644 pkgs/core/supabase/tests/type_violations/cancels_unfinished_tasks.test.sql diff --git a/.changeset/cancel-failed-run-tasks.md b/.changeset/cancel-failed-run-tasks.md new file mode 100644 index 000000000..cac7539e7 --- /dev/null +++ b/.changeset/cancel-failed-run-tasks.md @@ -0,0 +1,5 @@ +--- +"@pgflow/core": patch +--- + +Mark unfinished tasks as cancelled when their run fails, prevent late callbacks and stalled recovery from reviving them, and repair active tasks on historical failed runs. diff --git a/NOMENCLATURE_GUIDE.md b/NOMENCLATURE_GUIDE.md index b75668a74..e891f0ac2 100644 --- a/NOMENCLATURE_GUIDE.md +++ b/NOMENCLATURE_GUIDE.md @@ -151,6 +151,7 @@ Slugs are unique text identifiers with specific rules: - `completed` - Task completed successfully - `failed` - Task failed (may be retried or permanent) - `skipped` - Task was cancelled because its parent step was skipped +- `cancelled` - Task was cancelled because its run failed before the task finished; `runs.failed_at` is the cancellation time ## Configuration Terms diff --git a/pkgs/core/README.md b/pkgs/core/README.md index 9c894cc35..2b48363d1 100644 --- a/pkgs/core/README.md +++ b/pkgs/core/README.md @@ -348,9 +348,12 @@ The system handles failures by: - Marking the step as 'failed' - Marking the run as 'failed' - Archiving the message in PGMQ - - **Archiving all queued messages for the failed run** (preventing orphaned messages) + - **Marking every remaining queued or started task in the run as 'cancelled'** and archiving their messages (the failed-run invariant: a failed run has no task rows left in `queued` or `started`) 4. Additional failure handling: - - **No retries on already-failed runs** - tasks are immediately marked as failed + - **No retries on already-failed runs** - late `fail_task()` callbacks keep cancelled tasks cancelled + - **Cancellation wins over late callbacks** - late `complete_task()` or `fail_task()` cannot revive or rewrite a cancelled task; a committed completed task stays completed + - **`cancelled` differs from `failed` and `skipped`** - `failed` counts real handler failures, `skipped` is step-skip policy, `cancelled` is orchestration state invalidated by the run failure; `runs.failed_at` is the cancellation time + - **Database cancellation does not terminate handlers** - JavaScript already executing in a worker is not forcibly stopped and external side effects are not undone - **Graceful type constraint violations** - handled without exceptions when single steps feed map steps - **Stores invalid output on type violations** - captures the output that caused the violation for debugging - **Performance-optimized message archiving** using indexed queries diff --git a/pkgs/core/schemas/0060_tables_runtime.sql b/pkgs/core/schemas/0060_tables_runtime.sql index 0cc4f3fa8..c487d687f 100644 --- a/pkgs/core/schemas/0060_tables_runtime.sql +++ b/pkgs/core/schemas/0060_tables_runtime.sql @@ -104,7 +104,7 @@ create table pgflow.step_tasks ( foreign key (run_id, step_slug) references pgflow.step_states(run_id, step_slug), constraint valid_status check ( - status in ('queued', 'started', 'completed', 'failed', 'skipped') + status in ('queued', 'started', 'completed', 'failed', 'skipped', 'cancelled') ), constraint output_valid_only_for_completed check ( output is null or status in ('completed', 'failed') diff --git a/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql b/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql index 60924873c..0b8f6646c 100644 --- a/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql +++ b/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql @@ -13,6 +13,8 @@ begin -- Find and requeue stalled tasks (where started_at > timeout + 30s buffer) -- Tasks with requeued_count >= max_requeues will have their message archived -- but status left as 'started' for easy identification via requeued_count column + -- Eligibility requires the parent run AND parent step to still be 'started': + -- stale rows on failed runs or terminal steps must not be revived (#645). with stalled_tasks as ( select st.run_id, @@ -24,8 +26,11 @@ begin f.opt_timeout from pgflow.step_tasks st join pgflow.runs r on r.run_id = st.run_id + join pgflow.step_states ss on ss.run_id = st.run_id and ss.step_slug = st.step_slug join pgflow.flows f on f.flow_slug = r.flow_slug where st.status = 'started' + and r.status = 'started' + and ss.status = 'started' and st.permanently_stalled_at is null and st.started_at < now() - (f.opt_timeout * interval '1 second') - interval '30 seconds' for update of st skip locked @@ -84,9 +89,9 @@ begin _vr as (select count(*) from visibility_reset), -- Force execution of mark_permanently_stalled CTE _mps as (select count(*) from mark_permanently_stalled), - -- Force execution of archived CTE + -- Force execution of archived CTE _ar as (select count(*) from archived) - select count(*) into result_count + select count(*) into result_count from requeued, _vr, _mps, _ar; return result_count; diff --git a/pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql b/pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql index cf410028b..8fe445c8e 100644 --- a/pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql +++ b/pkgs/core/schemas/0100_function_cascade_resolve_conditions.sql @@ -17,6 +17,9 @@ DECLARE v_iteration_count int := 0; v_max_iterations int := 50; v_processed_count int; + v_run_transitioned boolean; + v_flow_slug text; + v_cancelled_message_ids bigint[]; BEGIN -- ========================================== -- GUARD: Early return if run is already terminal @@ -108,54 +111,72 @@ BEGIN -- Note: Cannot use "v_first_fail IS NOT NULL" because records with NULL fields -- evaluate to NULL in IS NOT NULL checks. Use FOUND instead. IF FOUND THEN - UPDATE pgflow.step_states - SET status = 'failed', - failed_at = now(), - error_message = 'Condition not met' - WHERE pgflow.step_states.run_id = cascade_resolve_conditions.run_id - AND pgflow.step_states.step_slug = v_first_fail.step_slug; - + -- Fail the run only if it is still started. The conditional UPDATE takes + -- the run row lock and rechecks status atomically, so replayed or + -- concurrent calls cannot duplicate the terminal transition or its events. UPDATE pgflow.runs SET status = 'failed', failed_at = now() - WHERE pgflow.runs.run_id = cascade_resolve_conditions.run_id; + WHERE pgflow.runs.run_id = cascade_resolve_conditions.run_id + AND pgflow.runs.status = 'started' + RETURNING true INTO v_run_transitioned; + + IF v_run_transitioned THEN + UPDATE pgflow.step_states + SET status = 'failed', + failed_at = now(), + error_message = 'Condition not met' + WHERE pgflow.step_states.run_id = cascade_resolve_conditions.run_id + AND pgflow.step_states.step_slug = v_first_fail.step_slug; + + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:failed', + 'run_id', cascade_resolve_conditions.run_id, + 'step_slug', v_first_fail.step_slug, + 'status', 'failed', + 'error_message', 'Condition not met', + 'failed_at', now() + ), + concat('step:', v_first_fail.step_slug, ':failed'), + concat('pgflow:run:', cascade_resolve_conditions.run_id), + false + ); - PERFORM realtime.send( - jsonb_build_object( - 'event_type', 'step:failed', - 'run_id', cascade_resolve_conditions.run_id, - 'step_slug', v_first_fail.step_slug, - 'status', 'failed', - 'error_message', 'Condition not met', - 'failed_at', now() - ), - concat('step:', v_first_fail.step_slug, ':failed'), - concat('pgflow:run:', cascade_resolve_conditions.run_id), - false - ); + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'run:failed', + 'run_id', cascade_resolve_conditions.run_id, + 'flow_slug', v_first_fail.flow_slug, + 'status', 'failed', + 'error_message', 'Condition not met', + 'failed_at', now() + ), + 'run:failed', + concat('pgflow:run:', cascade_resolve_conditions.run_id), + false + ); - PERFORM realtime.send( - jsonb_build_object( - 'event_type', 'run:failed', - 'run_id', cascade_resolve_conditions.run_id, - 'flow_slug', v_first_fail.flow_slug, - 'status', 'failed', - 'error_message', 'Condition not met', - 'failed_at', now() - ), - 'run:failed', - concat('pgflow:run:', cascade_resolve_conditions.run_id), - false - ); + -- Terminalize every unfinished task across all branches as cancelled, + -- capturing their message ids for archival below. Lock-order invariant: + -- always lock/update step_tasks before PGMQ queue rows. + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = cascade_resolve_conditions.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ) + SELECT ARRAY_AGG(ct.message_id) INTO v_cancelled_message_ids + FROM cancelled_tasks ct + WHERE ct.message_id IS NOT NULL; - PERFORM pgmq.archive(r.flow_slug, ARRAY_AGG(st.message_id)) - FROM pgflow.step_tasks st - JOIN pgflow.runs r ON st.run_id = r.run_id - WHERE st.run_id = cascade_resolve_conditions.run_id - AND st.status IN ('queued', 'started') - AND st.message_id IS NOT NULL - GROUP BY r.flow_slug - HAVING COUNT(st.message_id) > 0; + -- Archive the cancelled task messages captured above (only after their + -- task rows are terminalized) + IF v_cancelled_message_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_first_fail.flow_slug, v_cancelled_message_ids); + END IF; + END IF; RETURN false; END IF; diff --git a/pkgs/core/schemas/0100_function_complete_task.sql b/pkgs/core/schemas/0100_function_complete_task.sql index 884bd54e1..6b593bebd 100644 --- a/pkgs/core/schemas/0100_function_complete_task.sql +++ b/pkgs/core/schemas/0100_function_complete_task.sql @@ -14,6 +14,7 @@ declare v_dependent_map_slug text; v_run_record pgflow.runs%ROWTYPE; v_step_record pgflow.step_states%ROWTYPE; + v_violation_archived_ids bigint[]; begin -- ========================================== @@ -40,6 +41,27 @@ WHERE pgflow.step_states.run_id = complete_task.run_id AND pgflow.step_states.step_slug = complete_task.step_slug FOR UPDATE; +-- ========================================== +-- GUARD: Run failed while this callback waited for the lock +-- ========================================== +-- The failed-run guard above ran before the failure committed. Recheck under +-- lock so cancellation wins: archived message stays archived, task row keeps +-- its terminal status, and no events or counters are emitted. +IF v_run_record.status = 'failed' THEN + -- Archive the task message if present (no-op when already archived) + PERFORM pgflow._archive_task_message( + complete_task.run_id, + complete_task.step_slug, + complete_task.task_index + ); + -- Return the current task row without any mutations + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + RETURN; +END IF; + -- ========================================== -- GUARD: Late callback - step not started -- ========================================== @@ -84,6 +106,20 @@ LIMIT 1; -- Handle type violation if detected IF v_dependent_map_slug IS NOT NULL THEN + -- Mark current task as failed FIRST and store the output that caused the + -- violation, so the task row is terminal before any queue row is touched. + UPDATE pgflow.step_tasks + SET status = 'failed', + failed_at = now(), + output = complete_task.output, -- Store the output that caused the violation + error_message = '[TYPE_VIOLATION] Produced ' || + CASE WHEN complete_task.output IS NULL THEN 'null' + ELSE jsonb_typeof(complete_task.output) END || + ' instead of array' + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + -- Mark run as failed immediately UPDATE pgflow.runs SET status = 'failed', @@ -105,30 +141,6 @@ IF v_dependent_map_slug IS NOT NULL THEN false ); - -- Archive all active messages (both queued and started) to prevent orphaned messages - PERFORM pgmq.archive( - v_run_record.flow_slug, - array_agg(st.message_id) - ) - FROM pgflow.step_tasks st - WHERE st.run_id = complete_task.run_id - AND st.status IN ('queued', 'started') - AND st.message_id IS NOT NULL - HAVING count(*) > 0; -- Only call archive if there are messages to archive - - -- Mark current task as failed and store the output - UPDATE pgflow.step_tasks - SET status = 'failed', - failed_at = now(), - output = complete_task.output, -- Store the output that caused the violation - error_message = '[TYPE_VIOLATION] Produced ' || - CASE WHEN complete_task.output IS NULL THEN 'null' - ELSE jsonb_typeof(complete_task.output) END || - ' instead of array' - WHERE pgflow.step_tasks.run_id = complete_task.run_id - AND pgflow.step_tasks.step_slug = complete_task.step_slug - AND pgflow.step_tasks.task_index = complete_task.task_index; - -- Mark step state as failed UPDATE pgflow.step_states SET status = 'failed', @@ -159,16 +171,37 @@ IF v_dependent_map_slug IS NOT NULL THEN false ); - -- Archive the current task's message (it was started, now failed) - PERFORM pgmq.archive( - v_run_record.flow_slug, - st.message_id -- Single message, use scalar form + -- Terminalize every other unfinished task as cancelled, capturing their + -- message ids for archival below. Lock-order invariant: always lock/update + -- step_tasks before PGMQ queue rows. The culprit task is already terminal + -- (failed above), so it is excluded from the cancellation set. + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = complete_task.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ), + culprit_task AS ( + -- Terminal culprit row: safe to read for its message id after terminalization + SELECT st.message_id + FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index + AND st.message_id IS NOT NULL ) - FROM pgflow.step_tasks st - WHERE st.run_id = complete_task.run_id - AND st.step_slug = complete_task.step_slug - AND st.task_index = complete_task.task_index - AND st.message_id IS NOT NULL; + SELECT ARRAY_AGG(ids.message_id) INTO v_violation_archived_ids + FROM ( + SELECT message_id FROM culprit_task + UNION ALL + SELECT message_id FROM cancelled_tasks WHERE message_id IS NOT NULL + ) ids; + + -- Archive the culprit and cancelled task messages (only after their task rows are terminalized) + IF v_violation_archived_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_run_record.flow_slug, v_violation_archived_ids); + END IF; -- Return the failed task row (API contract: always return task row) RETURN QUERY diff --git a/pkgs/core/schemas/0100_function_fail_task.sql b/pkgs/core/schemas/0100_function_fail_task.sql index bf3d7c00f..6da065e4f 100644 --- a/pkgs/core/schemas/0100_function_fail_task.sql +++ b/pkgs/core/schemas/0100_function_fail_task.sql @@ -17,31 +17,18 @@ DECLARE v_task_exhausted boolean; v_flow_slug_for_deps text; v_prev_step_status text; + v_run_status text; v_flow_slug text; v_skipped_message_ids bigint[]; + v_cancelled_message_ids bigint[]; begin --- If run is already failed, no retries allowed +-- If run is already failed, no retries allowed. +-- Cancellation wins: tasks terminalized by the run failure (failed culprit or +-- cancelled siblings) keep their terminal status. This late callback only +-- archives any still-active message and returns the current row unchanged. IF EXISTS (SELECT 1 FROM pgflow.runs WHERE pgflow.runs.run_id = fail_task.run_id AND pgflow.runs.status = 'failed') THEN - UPDATE pgflow.step_tasks - SET status = 'failed', - failed_at = now(), - error_message = fail_task.error_message - WHERE pgflow.step_tasks.run_id = fail_task.run_id - AND pgflow.step_tasks.step_slug = fail_task.step_slug - AND pgflow.step_tasks.task_index = fail_task.task_index - AND pgflow.step_tasks.status = 'started'; - - -- Archive the task's message - PERFORM pgmq.archive(r.flow_slug, ARRAY_AGG(st.message_id)) - FROM pgflow.step_tasks st - JOIN pgflow.runs r ON st.run_id = r.run_id - WHERE st.run_id = fail_task.run_id - AND st.step_slug = fail_task.step_slug - AND st.task_index = fail_task.task_index - AND st.message_id IS NOT NULL - GROUP BY r.flow_slug - HAVING COUNT(st.message_id) > 0; + PERFORM pgflow._archive_task_message(fail_task.run_id, fail_task.step_slug, fail_task.task_index); RETURN QUERY SELECT * FROM pgflow.step_tasks WHERE pgflow.step_tasks.run_id = fail_task.run_id @@ -50,15 +37,27 @@ IF EXISTS (SELECT 1 FROM pgflow.runs WHERE pgflow.runs.run_id = fail_task.run_id RETURN; END IF; --- Late callback guard: lock run + step rows and use current step status +-- Late callback guard: lock run + step rows and use current statuses -- under lock so concurrent fail_task calls cannot read stale status. -SELECT ss.status, r.flow_slug INTO v_prev_step_status, v_flow_slug +SELECT ss.status, r.status, r.flow_slug INTO v_prev_step_status, v_run_status, v_flow_slug FROM pgflow.runs r JOIN pgflow.step_states ss ON ss.run_id = r.run_id WHERE ss.run_id = fail_task.run_id AND ss.step_slug = fail_task.step_slug FOR UPDATE OF r, ss; +-- Recheck under lock: the run may have failed while this callback waited +-- for the lock (the EXISTS guard above ran before the failure committed). +IF v_run_status = 'failed' THEN + PERFORM pgflow._archive_task_message(fail_task.run_id, fail_task.step_slug, fail_task.task_index); + + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = fail_task.run_id + AND pgflow.step_tasks.step_slug = fail_task.step_slug + AND pgflow.step_tasks.task_index = fail_task.task_index; + RETURN; +END IF; + IF v_prev_step_status IS NOT NULL AND v_prev_step_status != 'started' THEN -- Archive the task message if present PERFORM pgmq.archive(v_flow_slug, ARRAY_AGG(st.message_id)) @@ -327,16 +326,26 @@ IF v_run_failed THEN END; END IF; --- Archive all active messages (both queued and started) when run fails +-- Terminalize unfinished tasks as cancelled when the run fails, then archive +-- their messages. Lock-order invariant: always lock/update step_tasks before +-- PGMQ queue rows. The culprit task is already terminal (failed or requeued by +-- fail_or_retry_task), so only unfinished queued/started siblings are cancelled. IF v_run_failed THEN - PERFORM pgmq.archive(r.flow_slug, ARRAY_AGG(st.message_id)) - FROM pgflow.step_tasks st - JOIN pgflow.runs r ON st.run_id = r.run_id - WHERE st.run_id = fail_task.run_id - AND st.status IN ('queued', 'started') - AND st.message_id IS NOT NULL - GROUP BY r.flow_slug - HAVING COUNT(st.message_id) > 0; + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = fail_task.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ) + SELECT ARRAY_AGG(ct.message_id) INTO v_cancelled_message_ids + FROM cancelled_tasks ct + WHERE ct.message_id IS NOT NULL; + + -- Archive the cancelled task messages captured above (only after their task rows are terminalized) + IF v_cancelled_message_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_flow_slug, v_cancelled_message_ids); + END IF; END IF; -- For queued tasks: delay the message for retry with exponential backoff diff --git a/pkgs/core/supabase/migrations/20260901183450_pgflow_temp_failed_run_terminalization.sql b/pkgs/core/supabase/migrations/20260901183450_pgflow_temp_failed_run_terminalization.sql new file mode 100644 index 000000000..d6206eb96 --- /dev/null +++ b/pkgs/core/supabase/migrations/20260901183450_pgflow_temp_failed_run_terminalization.sql @@ -0,0 +1,1259 @@ +-- Modify "step_tasks" table +ALTER TABLE "pgflow"."step_tasks" DROP CONSTRAINT "valid_status", ADD CONSTRAINT "valid_status" CHECK (status = ANY (ARRAY['queued'::text, 'started'::text, 'completed'::text, 'failed'::text, 'skipped'::text, 'cancelled'::text])); +-- Modify "cascade_resolve_conditions" function +CREATE OR REPLACE FUNCTION "pgflow"."cascade_resolve_conditions" ("run_id" uuid) RETURNS boolean LANGUAGE plpgsql SET "search_path" = '' AS $$ +DECLARE + v_run_input jsonb; + v_run_status text; + v_first_fail record; + v_iteration_count int := 0; + v_max_iterations int := 50; + v_processed_count int; + v_run_transitioned boolean; + v_flow_slug text; + v_cancelled_message_ids bigint[]; +BEGIN + -- ========================================== + -- GUARD: Early return if run is already terminal + -- ========================================== + SELECT r.status, r.input INTO v_run_status, v_run_input + FROM pgflow.runs r + WHERE r.run_id = cascade_resolve_conditions.run_id; + + IF v_run_status IN ('failed', 'completed') THEN + RETURN v_run_status != 'failed'; + END IF; + + -- ========================================== + -- ITERATE UNTIL CONVERGENCE + -- ========================================== + -- After skipping steps, dependents may become ready and need evaluation. + -- Loop until no more steps are processed. + LOOP + v_iteration_count := v_iteration_count + 1; + IF v_iteration_count > v_max_iterations THEN + RAISE EXCEPTION 'cascade_resolve_conditions exceeded safety limit of % iterations', v_max_iterations; + END IF; + + v_processed_count := 0; + + -- ========================================== + -- PHASE 1a: CHECK FOR FAIL CONDITIONS + -- ========================================== + -- Find first step (by topological order) with unmet condition and 'fail' mode. + -- Condition is unmet when: + -- (required_input_pattern is set AND input does NOT contain it) OR + -- (forbidden_input_pattern is set AND input DOES contain it) + WITH steps_with_conditions AS ( + SELECT + step_state.flow_slug, + step_state.step_slug, + step.required_input_pattern, + step.forbidden_input_pattern, + step.when_unmet, + step.deps_count, + step.step_index + FROM pgflow.step_states AS step_state + JOIN pgflow.steps AS step + ON step.flow_slug = step_state.flow_slug + AND step.step_slug = step_state.step_slug + WHERE step_state.run_id = cascade_resolve_conditions.run_id + AND step_state.status = 'created' + AND step_state.remaining_deps = 0 + AND (step.required_input_pattern IS NOT NULL OR step.forbidden_input_pattern IS NOT NULL) + ), + step_deps_output AS ( + SELECT + swc.step_slug, + jsonb_object_agg(dep_state.step_slug, dep_state.output) AS deps_output + FROM steps_with_conditions swc + JOIN pgflow.deps dep ON dep.flow_slug = swc.flow_slug AND dep.step_slug = swc.step_slug + JOIN pgflow.step_states dep_state + ON dep_state.run_id = cascade_resolve_conditions.run_id + AND dep_state.step_slug = dep.dep_slug + AND dep_state.status = 'completed' -- Only completed deps (not skipped) + WHERE swc.deps_count > 0 + GROUP BY swc.step_slug + ), + condition_evaluations AS ( + SELECT + swc.*, + -- condition_met = (if IS NULL OR input @> if) AND (ifNot IS NULL OR NOT(input @> ifNot)) + (swc.required_input_pattern IS NULL OR + CASE WHEN swc.deps_count = 0 THEN v_run_input ELSE COALESCE(sdo.deps_output, '{}'::jsonb) END @> swc.required_input_pattern) + AND + (swc.forbidden_input_pattern IS NULL OR + NOT (CASE WHEN swc.deps_count = 0 THEN v_run_input ELSE COALESCE(sdo.deps_output, '{}'::jsonb) END @> swc.forbidden_input_pattern)) + AS condition_met + FROM steps_with_conditions swc + LEFT JOIN step_deps_output sdo ON sdo.step_slug = swc.step_slug + ) + SELECT + flow_slug, + step_slug, + required_input_pattern, + forbidden_input_pattern + INTO v_first_fail + FROM condition_evaluations + WHERE NOT condition_met AND when_unmet = 'fail' + ORDER BY step_index + LIMIT 1; + + -- Handle fail mode: fail step and run, return false + -- Note: Cannot use "v_first_fail IS NOT NULL" because records with NULL fields + -- evaluate to NULL in IS NOT NULL checks. Use FOUND instead. + IF FOUND THEN + -- Fail the run only if it is still started. The conditional UPDATE takes + -- the run row lock and rechecks status atomically, so replayed or + -- concurrent calls cannot duplicate the terminal transition or its events. + UPDATE pgflow.runs + SET status = 'failed', + failed_at = now() + WHERE pgflow.runs.run_id = cascade_resolve_conditions.run_id + AND pgflow.runs.status = 'started' + RETURNING true INTO v_run_transitioned; + + IF v_run_transitioned THEN + UPDATE pgflow.step_states + SET status = 'failed', + failed_at = now(), + error_message = 'Condition not met' + WHERE pgflow.step_states.run_id = cascade_resolve_conditions.run_id + AND pgflow.step_states.step_slug = v_first_fail.step_slug; + + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:failed', + 'run_id', cascade_resolve_conditions.run_id, + 'step_slug', v_first_fail.step_slug, + 'status', 'failed', + 'error_message', 'Condition not met', + 'failed_at', now() + ), + concat('step:', v_first_fail.step_slug, ':failed'), + concat('pgflow:run:', cascade_resolve_conditions.run_id), + false + ); + + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'run:failed', + 'run_id', cascade_resolve_conditions.run_id, + 'flow_slug', v_first_fail.flow_slug, + 'status', 'failed', + 'error_message', 'Condition not met', + 'failed_at', now() + ), + 'run:failed', + concat('pgflow:run:', cascade_resolve_conditions.run_id), + false + ); + + -- Terminalize every unfinished task across all branches as cancelled, + -- capturing their message ids for archival below. Lock-order invariant: + -- always lock/update step_tasks before PGMQ queue rows. + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = cascade_resolve_conditions.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ) + SELECT ARRAY_AGG(ct.message_id) INTO v_cancelled_message_ids + FROM cancelled_tasks ct + WHERE ct.message_id IS NOT NULL; + + -- Archive the cancelled task messages captured above (only after their + -- task rows are terminalized) + IF v_cancelled_message_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_first_fail.flow_slug, v_cancelled_message_ids); + END IF; + END IF; + + RETURN false; + END IF; + + -- ========================================== + -- PHASE 1b: HANDLE SKIP CONDITIONS (with propagation) + -- ========================================== + -- Skip steps with unmet conditions and whenUnmet='skip'. + -- Also decrement remaining_deps on dependents and set initial_tasks=0 for map dependents. + WITH steps_with_conditions AS ( + SELECT + step_state.flow_slug, + step_state.step_slug, + step.required_input_pattern, + step.forbidden_input_pattern, + step.when_unmet, + step.deps_count, + step.step_index + FROM pgflow.step_states AS step_state + JOIN pgflow.steps AS step + ON step.flow_slug = step_state.flow_slug + AND step.step_slug = step_state.step_slug + WHERE step_state.run_id = cascade_resolve_conditions.run_id + AND step_state.status = 'created' + AND step_state.remaining_deps = 0 + AND (step.required_input_pattern IS NOT NULL OR step.forbidden_input_pattern IS NOT NULL) + ), + step_deps_output AS ( + SELECT + swc.step_slug, + jsonb_object_agg(dep_state.step_slug, dep_state.output) AS deps_output + FROM steps_with_conditions swc + JOIN pgflow.deps dep ON dep.flow_slug = swc.flow_slug AND dep.step_slug = swc.step_slug + JOIN pgflow.step_states dep_state + ON dep_state.run_id = cascade_resolve_conditions.run_id + AND dep_state.step_slug = dep.dep_slug + AND dep_state.status = 'completed' -- Only completed deps (not skipped) + WHERE swc.deps_count > 0 + GROUP BY swc.step_slug + ), + condition_evaluations AS ( + SELECT + swc.*, + -- condition_met = (if IS NULL OR input @> if) AND (ifNot IS NULL OR NOT(input @> ifNot)) + (swc.required_input_pattern IS NULL OR + CASE WHEN swc.deps_count = 0 THEN v_run_input ELSE COALESCE(sdo.deps_output, '{}'::jsonb) END @> swc.required_input_pattern) + AND + (swc.forbidden_input_pattern IS NULL OR + NOT (CASE WHEN swc.deps_count = 0 THEN v_run_input ELSE COALESCE(sdo.deps_output, '{}'::jsonb) END @> swc.forbidden_input_pattern)) + AS condition_met + FROM steps_with_conditions swc + LEFT JOIN step_deps_output sdo ON sdo.step_slug = swc.step_slug + ), + unmet_skip_steps AS ( + SELECT * FROM condition_evaluations + WHERE NOT condition_met AND when_unmet = 'skip' + ), + skipped_steps AS ( + UPDATE pgflow.step_states ss + SET status = 'skipped', + skip_reason = 'condition_unmet', + skipped_at = now() + FROM unmet_skip_steps uss + WHERE ss.run_id = cascade_resolve_conditions.run_id + AND ss.step_slug = uss.step_slug + AND ss.status = 'created' + RETURNING + ss.*, + realtime.send( + jsonb_build_object( + 'event_type', 'step:skipped', + 'run_id', ss.run_id, + 'flow_slug', ss.flow_slug, + 'step_slug', ss.step_slug, + 'status', 'skipped', + 'skip_reason', 'condition_unmet', + 'skipped_at', ss.skipped_at + ), + concat('step:', ss.step_slug, ':skipped'), + concat('pgflow:run:', ss.run_id), + false + ) AS _broadcast_result + ), + -- NEW: Update dependent steps (decrement remaining_deps by count of skipped parents, set initial_tasks=0 for maps) + skipped_parent_counts AS ( + -- Count how many skipped parents each child has + SELECT + dep.step_slug AS child_step_slug, + dep.flow_slug AS child_flow_slug, + COUNT(*) AS skipped_parent_count + FROM skipped_steps parent + JOIN pgflow.deps dep ON dep.flow_slug = parent.flow_slug AND dep.dep_slug = parent.step_slug + GROUP BY dep.step_slug, dep.flow_slug + ), + dependent_updates AS ( + UPDATE pgflow.step_states child_state + SET remaining_deps = child_state.remaining_deps - spc.skipped_parent_count, + -- If child is a map step and this skipped step is its only dependency, + -- set initial_tasks = 0 (skipped dep = empty array) + initial_tasks = CASE + WHEN child_step.step_type = 'map' AND child_step.deps_count = 1 THEN 0 + ELSE child_state.initial_tasks + END + FROM skipped_parent_counts spc + JOIN pgflow.steps child_step ON child_step.flow_slug = spc.child_flow_slug AND child_step.step_slug = spc.child_step_slug + WHERE child_state.run_id = cascade_resolve_conditions.run_id + AND child_state.step_slug = spc.child_step_slug + ), + run_update AS ( + UPDATE pgflow.runs r + SET remaining_steps = r.remaining_steps - (SELECT COUNT(*) FROM skipped_steps) + WHERE r.run_id = cascade_resolve_conditions.run_id + AND (SELECT COUNT(*) FROM skipped_steps) > 0 + ) + SELECT COUNT(*)::int INTO v_processed_count FROM skipped_steps; + + -- ========================================== + -- PHASE 1c: HANDLE SKIP-CASCADE CONDITIONS + -- ========================================== + -- Call _cascade_force_skip_steps for each step with unmet condition and whenUnmet='skip-cascade'. + -- Process in topological order; _cascade_force_skip_steps is idempotent. + PERFORM pgflow._cascade_force_skip_steps(cascade_resolve_conditions.run_id, ready_step.step_slug, 'condition_unmet') + FROM pgflow.step_states AS ready_step + JOIN pgflow.steps AS step + ON step.flow_slug = ready_step.flow_slug + AND step.step_slug = ready_step.step_slug + LEFT JOIN LATERAL ( + SELECT jsonb_object_agg(dep_state.step_slug, dep_state.output) AS deps_output + FROM pgflow.deps dep + JOIN pgflow.step_states dep_state + ON dep_state.run_id = cascade_resolve_conditions.run_id + AND dep_state.step_slug = dep.dep_slug + AND dep_state.status = 'completed' -- Only completed deps (not skipped) + WHERE dep.flow_slug = ready_step.flow_slug + AND dep.step_slug = ready_step.step_slug + ) AS agg_deps ON step.deps_count > 0 + WHERE ready_step.run_id = cascade_resolve_conditions.run_id + AND ready_step.status = 'created' + AND ready_step.remaining_deps = 0 + AND (step.required_input_pattern IS NOT NULL OR step.forbidden_input_pattern IS NOT NULL) + AND step.when_unmet = 'skip-cascade' + -- Condition is NOT met when: (if fails) OR (ifNot fails) + AND NOT ( + (step.required_input_pattern IS NULL OR + CASE WHEN step.deps_count = 0 THEN v_run_input ELSE COALESCE(agg_deps.deps_output, '{}'::jsonb) END @> step.required_input_pattern) + AND + (step.forbidden_input_pattern IS NULL OR + NOT (CASE WHEN step.deps_count = 0 THEN v_run_input ELSE COALESCE(agg_deps.deps_output, '{}'::jsonb) END @> step.forbidden_input_pattern)) + ) + ORDER BY step.step_index; + + -- Check if run was failed during cascade (e.g., if _cascade_force_skip_steps triggers fail) + SELECT r.status INTO v_run_status + FROM pgflow.runs r + WHERE r.run_id = cascade_resolve_conditions.run_id; + + IF v_run_status IN ('failed', 'completed') THEN + RETURN v_run_status != 'failed'; + END IF; + + -- Exit loop if no steps were processed in this iteration + EXIT WHEN v_processed_count = 0; + END LOOP; + + RETURN true; +END; +$$; +-- Modify "complete_task" function +CREATE OR REPLACE FUNCTION "pgflow"."complete_task" ("run_id" uuid, "step_slug" text, "task_index" integer, "output" jsonb) RETURNS SETOF "pgflow"."step_tasks" LANGUAGE plpgsql SET "search_path" = '' AS $$ +declare + v_step_state pgflow.step_states%ROWTYPE; + v_dependent_map_slug text; + v_run_record pgflow.runs%ROWTYPE; + v_step_record pgflow.step_states%ROWTYPE; + v_violation_archived_ids bigint[]; +begin + +-- ========================================== +-- GUARD: No mutations on failed runs +-- ========================================== +IF EXISTS (SELECT 1 FROM pgflow.runs WHERE pgflow.runs.run_id = complete_task.run_id AND pgflow.runs.status = 'failed') THEN + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + RETURN; +END IF; + +-- ========================================== +-- LOCK ACQUISITION AND TYPE VALIDATION +-- ========================================== +-- Acquire locks first to prevent race conditions +SELECT * INTO v_run_record FROM pgflow.runs +WHERE pgflow.runs.run_id = complete_task.run_id +FOR UPDATE; + +SELECT * INTO v_step_record FROM pgflow.step_states +WHERE pgflow.step_states.run_id = complete_task.run_id + AND pgflow.step_states.step_slug = complete_task.step_slug +FOR UPDATE; + +-- ========================================== +-- GUARD: Run failed while this callback waited for the lock +-- ========================================== +-- The failed-run guard above ran before the failure committed. Recheck under +-- lock so cancellation wins: archived message stays archived, task row keeps +-- its terminal status, and no events or counters are emitted. +IF v_run_record.status = 'failed' THEN + -- Archive the task message if present (no-op when already archived) + PERFORM pgflow._archive_task_message( + complete_task.run_id, + complete_task.step_slug, + complete_task.task_index + ); + -- Return the current task row without any mutations + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + RETURN; +END IF; + +-- ========================================== +-- GUARD: Late callback - step not started +-- ========================================== +-- If the step is not in 'started' state, this is a late callback. +-- Do not mutate step_states or runs, archive message, return task row. +IF v_step_record.status != 'started' THEN + -- Archive the task message if present (prevents stuck work) + PERFORM pgmq.archive( + v_run_record.flow_slug, + st.message_id + ) + FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index + AND st.message_id IS NOT NULL; + -- Return the current task row without any mutations + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + RETURN; +END IF; + +-- Check for type violations AFTER acquiring locks +SELECT child_step.step_slug INTO v_dependent_map_slug +FROM pgflow.deps dependency +JOIN pgflow.steps child_step ON child_step.flow_slug = dependency.flow_slug + AND child_step.step_slug = dependency.step_slug +JOIN pgflow.steps parent_step ON parent_step.flow_slug = dependency.flow_slug + AND parent_step.step_slug = dependency.dep_slug +JOIN pgflow.step_states child_state ON child_state.flow_slug = child_step.flow_slug + AND child_state.step_slug = child_step.step_slug +WHERE dependency.dep_slug = complete_task.step_slug -- parent is the completing step + AND dependency.flow_slug = v_run_record.flow_slug + AND parent_step.step_type = 'single' -- Only validate single steps + AND child_step.step_type = 'map' + AND child_state.run_id = complete_task.run_id + AND child_state.initial_tasks IS NULL + AND (complete_task.output IS NULL OR jsonb_typeof(complete_task.output) != 'array') +LIMIT 1; + +-- Handle type violation if detected +IF v_dependent_map_slug IS NOT NULL THEN + -- Mark current task as failed FIRST and store the output that caused the + -- violation, so the task row is terminal before any queue row is touched. + UPDATE pgflow.step_tasks + SET status = 'failed', + failed_at = now(), + output = complete_task.output, -- Store the output that caused the violation + error_message = '[TYPE_VIOLATION] Produced ' || + CASE WHEN complete_task.output IS NULL THEN 'null' + ELSE jsonb_typeof(complete_task.output) END || + ' instead of array' + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + + -- Mark run as failed immediately + UPDATE pgflow.runs + SET status = 'failed', + failed_at = now() + WHERE pgflow.runs.run_id = complete_task.run_id; + + -- Broadcast run:failed event + -- Uses PERFORM pattern to ensure execution (proven reliable pattern in this function) + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'run:failed', + 'run_id', complete_task.run_id, + 'flow_slug', v_run_record.flow_slug, + 'status', 'failed', + 'failed_at', now() + ), + 'run:failed', + concat('pgflow:run:', complete_task.run_id), + false + ); + + -- Mark step state as failed + UPDATE pgflow.step_states + SET status = 'failed', + failed_at = now(), + error_message = '[TYPE_VIOLATION] Map step ' || v_dependent_map_slug || + ' expects array input but dependency ' || complete_task.step_slug || + ' produced ' || CASE WHEN complete_task.output IS NULL THEN 'null' + ELSE jsonb_typeof(complete_task.output) END + WHERE pgflow.step_states.run_id = complete_task.run_id + AND pgflow.step_states.step_slug = complete_task.step_slug; + + -- Broadcast step:failed event + -- Uses PERFORM pattern to ensure execution (proven reliable pattern in this function) + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:failed', + 'run_id', complete_task.run_id, + 'step_slug', complete_task.step_slug, + 'status', 'failed', + 'error_message', '[TYPE_VIOLATION] Map step ' || v_dependent_map_slug || + ' expects array input but dependency ' || complete_task.step_slug || + ' produced ' || CASE WHEN complete_task.output IS NULL THEN 'null' + ELSE jsonb_typeof(complete_task.output) END, + 'failed_at', now() + ), + concat('step:', complete_task.step_slug, ':failed'), + concat('pgflow:run:', complete_task.run_id), + false + ); + + -- Terminalize every other unfinished task as cancelled, capturing their + -- message ids for archival below. Lock-order invariant: always lock/update + -- step_tasks before PGMQ queue rows. The culprit task is already terminal + -- (failed above), so it is excluded from the cancellation set. + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = complete_task.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ), + culprit_task AS ( + -- Terminal culprit row: safe to read for its message id after terminalization + SELECT st.message_id + FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index + AND st.message_id IS NOT NULL + ) + SELECT ARRAY_AGG(ids.message_id) INTO v_violation_archived_ids + FROM ( + SELECT message_id FROM culprit_task + UNION ALL + SELECT message_id FROM cancelled_tasks WHERE message_id IS NOT NULL + ) ids; + + -- Archive the culprit and cancelled task messages (only after their task rows are terminalized) + IF v_violation_archived_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_run_record.flow_slug, v_violation_archived_ids); + END IF; + + -- Return the failed task row (API contract: always return task row) + RETURN QUERY + SELECT * FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index; + RETURN; +END IF; + +-- ========================================== +-- MAIN CTE CHAIN: Update task and propagate changes +-- ========================================== +WITH +-- ---------- Task completion ---------- +-- Update the task record with completion status and output +task AS ( + UPDATE pgflow.step_tasks + SET + status = 'completed', + completed_at = now(), + output = complete_task.output + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index + AND pgflow.step_tasks.status = 'started' + RETURNING * +), +-- ---------- Get step type for output handling ---------- +step_def AS ( + SELECT step.step_type + FROM pgflow.steps step + JOIN pgflow.runs run ON run.flow_slug = step.flow_slug + WHERE run.run_id = complete_task.run_id + AND step.step_slug = complete_task.step_slug +), +-- ---------- Step state update ---------- +-- Decrement remaining_tasks and potentially mark step as completed +-- Also store output atomically with status transition to completed +step_state AS ( + UPDATE pgflow.step_states + SET + status = CASE + WHEN pgflow.step_states.remaining_tasks = 1 THEN 'completed' -- Will be 0 after decrement + ELSE 'started' + END, + completed_at = CASE + WHEN pgflow.step_states.remaining_tasks = 1 THEN now() -- Will be 0 after decrement + ELSE NULL + END, + remaining_tasks = pgflow.step_states.remaining_tasks - 1, + -- Store output atomically with completion (only when remaining_tasks = 1, meaning step completes) + output = CASE + -- Single step: store task output directly when completing + WHEN (SELECT step_type FROM step_def) = 'single' AND pgflow.step_states.remaining_tasks = 1 THEN + complete_task.output + -- Map step: aggregate on completion (ordered by task_index) + WHEN (SELECT step_type FROM step_def) = 'map' AND pgflow.step_states.remaining_tasks = 1 THEN + (SELECT COALESCE(jsonb_agg(all_outputs.output ORDER BY all_outputs.task_index), '[]'::jsonb) + FROM ( + -- All previously completed tasks + SELECT st.output, st.task_index + FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.status = 'completed' + UNION ALL + -- Current task being completed (not yet visible as completed in snapshot) + SELECT complete_task.output, complete_task.task_index + ) all_outputs) + ELSE pgflow.step_states.output + END + FROM task + WHERE pgflow.step_states.run_id = complete_task.run_id + AND pgflow.step_states.step_slug = complete_task.step_slug + RETURNING pgflow.step_states.* +), +-- ---------- Dependency resolution ---------- +-- Find all child steps that depend on the completed parent step (only if parent completed) +child_steps AS ( + SELECT deps.step_slug AS child_step_slug + FROM pgflow.deps deps + JOIN step_state parent_state ON parent_state.status = 'completed' AND deps.flow_slug = parent_state.flow_slug + WHERE deps.dep_slug = complete_task.step_slug -- dep_slug is the parent, step_slug is the child + ORDER BY deps.step_slug -- Ensure consistent ordering +), +-- ---------- Lock child steps ---------- +-- Acquire locks on all child steps before updating them +child_steps_lock AS ( + SELECT * FROM pgflow.step_states + WHERE pgflow.step_states.run_id = complete_task.run_id + AND pgflow.step_states.step_slug IN (SELECT child_step_slug FROM child_steps) + FOR UPDATE +), +-- ---------- Update child steps ---------- +-- Decrement remaining_deps and resolve NULL initial_tasks for map steps +child_steps_update AS ( + UPDATE pgflow.step_states child_state + SET remaining_deps = child_state.remaining_deps - 1, + -- Resolve NULL initial_tasks for child map steps + -- This is where child maps learn their array size from the parent + -- This CTE only runs when the parent step is complete (see child_steps JOIN) + initial_tasks = CASE + WHEN child_step.step_type = 'map' AND child_state.initial_tasks IS NULL THEN + CASE + WHEN parent_step.step_type = 'map' THEN + -- Map->map: Count all completed tasks from parent map + -- We add 1 because the current task is being completed in this transaction + -- but isn't yet visible as 'completed' in the step_tasks table + -- TODO: Refactor to use future column step_states.total_tasks + -- Would eliminate the COUNT query and just use parent_state.total_tasks + (SELECT COUNT(*)::int + 1 + FROM pgflow.step_tasks parent_tasks + WHERE parent_tasks.run_id = complete_task.run_id + AND parent_tasks.step_slug = complete_task.step_slug + AND parent_tasks.status = 'completed' + AND parent_tasks.task_index != complete_task.task_index) + ELSE + -- Single->map: Use output array length (single steps complete immediately) + CASE + WHEN complete_task.output IS NOT NULL + AND jsonb_typeof(complete_task.output) = 'array' THEN + jsonb_array_length(complete_task.output) + ELSE NULL -- Keep NULL if not an array + END + END + ELSE child_state.initial_tasks -- Keep existing value (including NULL) + END + FROM child_steps children + JOIN pgflow.steps child_step ON child_step.flow_slug = (SELECT r.flow_slug FROM pgflow.runs r WHERE r.run_id = complete_task.run_id) + AND child_step.step_slug = children.child_step_slug + JOIN pgflow.steps parent_step ON parent_step.flow_slug = (SELECT r.flow_slug FROM pgflow.runs r WHERE r.run_id = complete_task.run_id) + AND parent_step.step_slug = complete_task.step_slug + WHERE child_state.run_id = complete_task.run_id + AND child_state.step_slug = children.child_step_slug +) +-- ---------- Update run remaining_steps ---------- +-- Decrement the run's remaining_steps counter if step completed +UPDATE pgflow.runs +SET remaining_steps = pgflow.runs.remaining_steps - 1 +FROM step_state +WHERE pgflow.runs.run_id = complete_task.run_id + AND step_state.status = 'completed'; + +-- ========================================== +-- POST-COMPLETION ACTIONS +-- ========================================== + +-- ---------- Get updated state for broadcasting ---------- +SELECT * INTO v_step_state FROM pgflow.step_states +WHERE pgflow.step_states.run_id = complete_task.run_id AND pgflow.step_states.step_slug = complete_task.step_slug; + +-- ---------- Handle step completion ---------- +IF v_step_state.status = 'completed' THEN + -- Broadcast step:completed event FIRST (before cascade) + -- This ensures parent broadcasts before its dependent children + -- Use stored output from step_states (set atomically during status transition) + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:completed', + 'run_id', complete_task.run_id, + 'step_slug', complete_task.step_slug, + 'status', 'completed', + 'output', v_step_state.output, -- Use stored output instead of re-aggregating + 'completed_at', v_step_state.completed_at + ), + concat('step:', complete_task.step_slug, ':completed'), + concat('pgflow:run:', complete_task.run_id), + false + ); + + -- THEN evaluate conditions on newly-ready dependent steps + -- This must happen before cascade_complete_taskless_steps so that + -- skipped steps can set initial_tasks=0 for their map dependents + IF NOT pgflow.cascade_resolve_conditions(complete_task.run_id) THEN + -- Run was failed due to a condition with when_unmet='fail' + -- Archive the current task's message before returning + PERFORM pgmq.archive( + (SELECT r.flow_slug FROM pgflow.runs r WHERE r.run_id = complete_task.run_id), + (SELECT st.message_id FROM pgflow.step_tasks st + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index) + ); + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = complete_task.run_id + AND pgflow.step_tasks.step_slug = complete_task.step_slug + AND pgflow.step_tasks.task_index = complete_task.task_index; + RETURN; + END IF; + + -- THEN cascade complete any taskless steps that are now ready + -- This ensures dependent children broadcast AFTER their parent + PERFORM pgflow.cascade_complete_taskless_steps(complete_task.run_id); +END IF; + +-- ---------- Archive completed task message ---------- +-- Move message from active queue to archive table +PERFORM ( + WITH completed_tasks AS ( + SELECT r.flow_slug, st.message_id + FROM pgflow.step_tasks st + JOIN pgflow.runs r ON st.run_id = r.run_id + WHERE st.run_id = complete_task.run_id + AND st.step_slug = complete_task.step_slug + AND st.task_index = complete_task.task_index + AND st.status = 'completed' + ) + SELECT pgmq.archive(ct.flow_slug, ct.message_id) + FROM completed_tasks ct + WHERE EXISTS (SELECT 1 FROM completed_tasks) +); + +-- ---------- Trigger next steps ---------- +-- Start any steps that are now ready (deps satisfied) +PERFORM pgflow.start_ready_steps(complete_task.run_id); + +-- Check if the entire run is complete +PERFORM pgflow.maybe_complete_run(complete_task.run_id); + +-- ---------- Return completed task ---------- +RETURN QUERY SELECT * +FROM pgflow.step_tasks AS step_task +WHERE step_task.run_id = complete_task.run_id + AND step_task.step_slug = complete_task.step_slug + AND step_task.task_index = complete_task.task_index; + +end; +$$; +-- Modify "requeue_stalled_tasks" function +CREATE OR REPLACE FUNCTION "pgflow"."requeue_stalled_tasks" () RETURNS integer LANGUAGE plpgsql SECURITY DEFINER SET "search_path" = '' AS $$ +declare + result_count int := 0; + max_requeues constant int := 3; +begin + -- Find and requeue stalled tasks (where started_at > timeout + 30s buffer) + -- Tasks with requeued_count >= max_requeues will have their message archived + -- but status left as 'started' for easy identification via requeued_count column + -- Eligibility requires the parent run AND parent step to still be 'started': + -- stale rows on failed runs or terminal steps must not be revived (#645). + with stalled_tasks as ( + select + st.run_id, + st.step_slug, + st.task_index, + st.message_id, + r.flow_slug, + st.requeued_count, + f.opt_timeout + from pgflow.step_tasks st + join pgflow.runs r on r.run_id = st.run_id + join pgflow.step_states ss on ss.run_id = st.run_id and ss.step_slug = st.step_slug + join pgflow.flows f on f.flow_slug = r.flow_slug + where st.status = 'started' + and r.status = 'started' + and ss.status = 'started' + and st.permanently_stalled_at is null + and st.started_at < now() - (f.opt_timeout * interval '1 second') - interval '30 seconds' + for update of st skip locked + ), + -- Separate tasks that can be requeued from those that exceeded max requeues + to_requeue as ( + select * from stalled_tasks where requeued_count < max_requeues + ), + to_archive as ( + select * from stalled_tasks where requeued_count >= max_requeues + ), + -- Update tasks that will be requeued + requeued as ( + update pgflow.step_tasks st + set + status = 'queued', + started_at = null, + last_worker_id = null, + requeued_count = st.requeued_count + 1, + last_requeued_at = now() + from to_requeue tr + where st.run_id = tr.run_id + and st.step_slug = tr.step_slug + and st.task_index = tr.task_index + returning tr.flow_slug as queue_name, tr.message_id + ), + -- Make requeued messages visible immediately (batched per queue) + visibility_reset as ( + select pgflow.set_vt_batch( + r.queue_name, + array_agg(r.message_id), + array_agg(0) -- all offsets are 0 (immediate visibility) + ) + from requeued r + where r.message_id is not null + group by r.queue_name + ), + -- Mark tasks as permanently stalled before archiving + mark_permanently_stalled as ( + update pgflow.step_tasks st + set permanently_stalled_at = now() + from to_archive ta + where st.run_id = ta.run_id + and st.step_slug = ta.step_slug + and st.task_index = ta.task_index + returning st.run_id + ), + -- Archive messages for tasks that exceeded max requeues (batched per queue) + archived as ( + select pgmq.archive(ta.flow_slug, array_agg(ta.message_id)) + from to_archive ta + where ta.message_id is not null + group by ta.flow_slug + ), + -- Force execution of visibility_reset CTE + _vr as (select count(*) from visibility_reset), + -- Force execution of mark_permanently_stalled CTE + _mps as (select count(*) from mark_permanently_stalled), + -- Force execution of archived CTE + _ar as (select count(*) from archived) + select count(*) into result_count + from requeued, _vr, _mps, _ar; + + return result_count; +end; +$$; +-- Modify "fail_task" function +CREATE OR REPLACE FUNCTION "pgflow"."fail_task" ("run_id" uuid, "step_slug" text, "task_index" integer, "error_message" text) RETURNS SETOF "pgflow"."step_tasks" LANGUAGE plpgsql SET "search_path" = '' AS $$ +DECLARE + v_run_failed boolean; + v_step_failed boolean; + v_step_skipped boolean; + v_when_exhausted text; + v_task_exhausted boolean; + v_flow_slug_for_deps text; + v_prev_step_status text; + v_run_status text; + v_flow_slug text; + v_skipped_message_ids bigint[]; + v_cancelled_message_ids bigint[]; +begin + +-- If run is already failed, no retries allowed. +-- Cancellation wins: tasks terminalized by the run failure (failed culprit or +-- cancelled siblings) keep their terminal status. This late callback only +-- archives any still-active message and returns the current row unchanged. +IF EXISTS (SELECT 1 FROM pgflow.runs WHERE pgflow.runs.run_id = fail_task.run_id AND pgflow.runs.status = 'failed') THEN + PERFORM pgflow._archive_task_message(fail_task.run_id, fail_task.step_slug, fail_task.task_index); + + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = fail_task.run_id + AND pgflow.step_tasks.step_slug = fail_task.step_slug + AND pgflow.step_tasks.task_index = fail_task.task_index; + RETURN; +END IF; + +-- Late callback guard: lock run + step rows and use current statuses +-- under lock so concurrent fail_task calls cannot read stale status. +SELECT ss.status, r.status, r.flow_slug INTO v_prev_step_status, v_run_status, v_flow_slug +FROM pgflow.runs r +JOIN pgflow.step_states ss ON ss.run_id = r.run_id +WHERE ss.run_id = fail_task.run_id + AND ss.step_slug = fail_task.step_slug +FOR UPDATE OF r, ss; + +-- Recheck under lock: the run may have failed while this callback waited +-- for the lock (the EXISTS guard above ran before the failure committed). +IF v_run_status = 'failed' THEN + PERFORM pgflow._archive_task_message(fail_task.run_id, fail_task.step_slug, fail_task.task_index); + + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = fail_task.run_id + AND pgflow.step_tasks.step_slug = fail_task.step_slug + AND pgflow.step_tasks.task_index = fail_task.task_index; + RETURN; +END IF; + +IF v_prev_step_status IS NOT NULL AND v_prev_step_status != 'started' THEN + -- Archive the task message if present + PERFORM pgmq.archive(v_flow_slug, ARRAY_AGG(st.message_id)) + FROM pgflow.step_tasks st + WHERE st.run_id = fail_task.run_id + AND st.step_slug = fail_task.step_slug + AND st.task_index = fail_task.task_index + AND st.message_id IS NOT NULL + HAVING COUNT(st.message_id) > 0; + + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = fail_task.run_id + AND pgflow.step_tasks.step_slug = fail_task.step_slug + AND pgflow.step_tasks.task_index = fail_task.task_index; + RETURN; +END IF; + +WITH flow_info AS ( + SELECT r.flow_slug + FROM pgflow.runs r + WHERE r.run_id = fail_task.run_id +), + config AS ( + SELECT + COALESCE(s.opt_max_attempts, f.opt_max_attempts) AS opt_max_attempts, + COALESCE(s.opt_base_delay, f.opt_base_delay) AS opt_base_delay, + s.when_exhausted + FROM pgflow.steps s + JOIN pgflow.flows f ON f.flow_slug = s.flow_slug + JOIN flow_info fi ON fi.flow_slug = s.flow_slug + WHERE s.flow_slug = fi.flow_slug AND s.step_slug = fail_task.step_slug +), +fail_or_retry_task as ( + UPDATE pgflow.step_tasks as task + SET + status = CASE + WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued' + ELSE 'failed' + END, + failed_at = CASE + WHEN task.attempts_count >= (SELECT opt_max_attempts FROM config) THEN now() + ELSE NULL + END, + started_at = CASE + WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL + ELSE task.started_at + END, + error_message = fail_task.error_message + WHERE task.run_id = fail_task.run_id + AND task.step_slug = fail_task.step_slug + AND task.task_index = fail_task.task_index + AND task.status = 'started' + RETURNING * +), + -- Determine if task exhausted retries and get when_exhausted mode + task_status AS ( + SELECT + (select status from fail_or_retry_task) AS new_task_status, + (select when_exhausted from config) AS when_exhausted_mode, + -- Task is exhausted when it's failed (no more retries) + ((select status from fail_or_retry_task) = 'failed') AS is_exhausted +), +maybe_fail_step AS ( + UPDATE pgflow.step_states + SET + -- Status logic: + -- - If task not exhausted (retrying): keep current status + -- - If exhausted AND when_exhausted='fail': set to 'failed' + -- - If exhausted AND when_exhausted IN ('skip', 'skip-cascade'): set to 'skipped' + status = CASE + WHEN NOT (select is_exhausted from task_status) THEN pgflow.step_states.status + WHEN (select when_exhausted_mode from task_status) = 'fail' THEN 'failed' + ELSE 'skipped' -- skip or skip-cascade + END, + failed_at = CASE + WHEN (select is_exhausted from task_status) AND (select when_exhausted_mode from task_status) = 'fail' THEN now() + ELSE NULL + END, + error_message = CASE + WHEN (select is_exhausted from task_status) THEN fail_task.error_message + ELSE NULL + END, + skip_reason = CASE + WHEN (select is_exhausted from task_status) AND (select when_exhausted_mode from task_status) IN ('skip', 'skip-cascade') THEN 'handler_failed' + ELSE pgflow.step_states.skip_reason + END, + skipped_at = CASE + WHEN (select is_exhausted from task_status) AND (select when_exhausted_mode from task_status) IN ('skip', 'skip-cascade') THEN now() + ELSE pgflow.step_states.skipped_at + END, + -- Clear remaining_tasks when skipping (required by remaining_tasks_state_consistency constraint) + remaining_tasks = CASE + WHEN (select is_exhausted from task_status) AND (select when_exhausted_mode from task_status) IN ('skip', 'skip-cascade') THEN NULL + ELSE pgflow.step_states.remaining_tasks + END + FROM fail_or_retry_task + WHERE pgflow.step_states.run_id = fail_task.run_id + AND pgflow.step_states.step_slug = fail_task.step_slug + RETURNING pgflow.step_states.* +), +run_update AS ( + -- Update run status: only fail when when_exhausted='fail' and step was failed + UPDATE pgflow.runs + SET status = CASE + WHEN (select status from maybe_fail_step) = 'failed' THEN 'failed' + ELSE status + END, + failed_at = CASE + WHEN (select status from maybe_fail_step) = 'failed' THEN now() + ELSE NULL + END, + -- Decrement remaining_steps only on FIRST transition to skipped + -- (not when step was already skipped and a second task fails) + -- Uses PL/pgSQL variable captured before CTE chain + remaining_steps = CASE + WHEN (select status from maybe_fail_step) = 'skipped' + AND v_prev_step_status != 'skipped' + THEN pgflow.runs.remaining_steps - 1 + ELSE pgflow.runs.remaining_steps + END + WHERE pgflow.runs.run_id = fail_task.run_id + RETURNING pgflow.runs.status +) +SELECT + COALESCE((SELECT status = 'failed' FROM run_update), false), + COALESCE((SELECT status = 'failed' FROM maybe_fail_step), false), + COALESCE((SELECT status = 'skipped' FROM maybe_fail_step), false), + COALESCE((SELECT is_exhausted FROM task_status), false) +INTO v_run_failed, v_step_failed, v_step_skipped, v_task_exhausted; + + -- Capture when_exhausted mode for later skip handling + SELECT s.when_exhausted INTO v_when_exhausted + FROM pgflow.steps s +JOIN pgflow.runs r ON r.flow_slug = s.flow_slug + WHERE r.run_id = fail_task.run_id + AND s.step_slug = fail_task.step_slug; + +-- Send broadcast event for step failure if the step was failed +IF v_task_exhausted AND v_step_failed THEN + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:failed', + 'run_id', fail_task.run_id, + 'step_slug', fail_task.step_slug, + 'status', 'failed', + 'error_message', fail_task.error_message, + 'failed_at', now() + ), + concat('step:', fail_task.step_slug, ':failed'), + concat('pgflow:run:', fail_task.run_id), + false + ); +END IF; + +-- Handle step skipping (when_exhausted = 'skip' or 'skip-cascade') + IF v_task_exhausted AND v_step_skipped THEN + -- Lock-order invariant: always lock/update step_tasks before PGMQ queue rows. + -- requeue_stalled_tasks() uses the same order; archiving queue rows first + -- deadlocks the two transactions against each other. + -- Terminalize all still-active sibling task rows for the skipped step, + -- capturing their message ids for archival below. + WITH skipped_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'skipped' + WHERE task.run_id = fail_task.run_id + AND task.step_slug = fail_task.step_slug + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ) + SELECT ARRAY_AGG(st.message_id) INTO v_skipped_message_ids + FROM skipped_tasks st + WHERE st.message_id IS NOT NULL; + + -- Archive the sibling task messages captured above (only after their task rows are terminalized) + IF v_skipped_message_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_flow_slug, v_skipped_message_ids); + END IF; + + -- Send broadcast event for step skipped + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'step:skipped', + 'run_id', fail_task.run_id, + 'step_slug', fail_task.step_slug, + 'status', 'skipped', + 'skip_reason', 'handler_failed', + 'error_message', fail_task.error_message, + 'skipped_at', now() + ), + concat('step:', fail_task.step_slug, ':skipped'), + concat('pgflow:run:', fail_task.run_id), + false + ); + + -- For skip-cascade: cascade skip to all downstream dependents + IF v_when_exhausted = 'skip-cascade' THEN + PERFORM pgflow._cascade_force_skip_steps(fail_task.run_id, fail_task.step_slug, 'handler_failed'); + ELSE + -- For plain 'skip': decrement remaining_deps on dependent steps + -- (This mirrors the pattern in cascade_resolve_conditions.sql for when_unmet='skip') + SELECT flow_slug INTO v_flow_slug_for_deps + FROM pgflow.runs + WHERE pgflow.runs.run_id = fail_task.run_id; + + UPDATE pgflow.step_states AS child_state + SET remaining_deps = child_state.remaining_deps - 1, + -- If child is a map step and this skipped step is its only dependency, + -- set initial_tasks = 0 (skipped dep = empty array) + initial_tasks = CASE + WHEN child_step.step_type = 'map' AND child_step.deps_count = 1 THEN 0 + ELSE child_state.initial_tasks + END + FROM pgflow.deps AS dep + JOIN pgflow.steps AS child_step ON child_step.flow_slug = dep.flow_slug AND child_step.step_slug = dep.step_slug + WHERE child_state.run_id = fail_task.run_id + AND dep.flow_slug = v_flow_slug_for_deps + AND dep.dep_slug = fail_task.step_slug + AND child_state.step_slug = dep.step_slug; + + -- Evaluate conditions on newly-ready dependent steps + -- This must happen before cascade_complete_taskless_steps so that + -- skipped steps can set initial_tasks=0 for their map dependents + IF NOT pgflow.cascade_resolve_conditions(fail_task.run_id) THEN + -- Run was failed due to a condition with when_unmet='fail' + -- Archive the failed task's message before returning + PERFORM pgflow._archive_task_message(fail_task.run_id, fail_task.step_slug, fail_task.task_index); + -- Return the task row (API contract) + RETURN QUERY SELECT * FROM pgflow.step_tasks + WHERE pgflow.step_tasks.run_id = fail_task.run_id + AND pgflow.step_tasks.step_slug = fail_task.step_slug + AND pgflow.step_tasks.task_index = fail_task.task_index; + RETURN; + END IF; + + -- Auto-complete taskless steps (e.g., map steps with initial_tasks=0 from skipped dep) + PERFORM pgflow.cascade_complete_taskless_steps(fail_task.run_id); + + -- Start steps that became ready after condition resolution and taskless completion + PERFORM pgflow.start_ready_steps(fail_task.run_id); + END IF; + + -- Try to complete the run (remaining_steps may now be 0) + PERFORM pgflow.maybe_complete_run(fail_task.run_id); +END IF; + +-- Send broadcast event for run failure if the run was failed +IF v_run_failed THEN + DECLARE + v_flow_slug text; + BEGIN + SELECT flow_slug INTO v_flow_slug FROM pgflow.runs WHERE pgflow.runs.run_id = fail_task.run_id; + + PERFORM realtime.send( + jsonb_build_object( + 'event_type', 'run:failed', + 'run_id', fail_task.run_id, + 'flow_slug', v_flow_slug, + 'status', 'failed', + 'error_message', fail_task.error_message, + 'failed_at', now() + ), + 'run:failed', + concat('pgflow:run:', fail_task.run_id), + false + ); + END; +END IF; + +-- Terminalize unfinished tasks as cancelled when the run fails, then archive +-- their messages. Lock-order invariant: always lock/update step_tasks before +-- PGMQ queue rows. The culprit task is already terminal (failed or requeued by +-- fail_or_retry_task), so only unfinished queued/started siblings are cancelled. +IF v_run_failed THEN + WITH cancelled_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'cancelled' + WHERE task.run_id = fail_task.run_id + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ) + SELECT ARRAY_AGG(ct.message_id) INTO v_cancelled_message_ids + FROM cancelled_tasks ct + WHERE ct.message_id IS NOT NULL; + + -- Archive the cancelled task messages captured above (only after their task rows are terminalized) + IF v_cancelled_message_ids IS NOT NULL THEN + PERFORM pgmq.archive(v_flow_slug, v_cancelled_message_ids); + END IF; +END IF; + +-- For queued tasks: delay the message for retry with exponential backoff +PERFORM ( + WITH retry_config AS ( + SELECT + COALESCE(s.opt_base_delay, f.opt_base_delay) AS base_delay + FROM pgflow.steps s + JOIN pgflow.flows f ON f.flow_slug = s.flow_slug + JOIN pgflow.runs r ON r.flow_slug = f.flow_slug + WHERE r.run_id = fail_task.run_id + AND s.step_slug = fail_task.step_slug + ), + queued_tasks AS ( + SELECT + r.flow_slug, + st.message_id, + pgflow.calculate_retry_delay((SELECT base_delay FROM retry_config), st.attempts_count) AS calculated_delay + FROM pgflow.step_tasks st + JOIN pgflow.runs r ON st.run_id = r.run_id + WHERE st.run_id = fail_task.run_id + AND st.step_slug = fail_task.step_slug + AND st.task_index = fail_task.task_index + AND st.status = 'queued' + ) + SELECT pgmq.set_vt(qt.flow_slug, qt.message_id, qt.calculated_delay) + FROM queued_tasks qt + WHERE EXISTS (SELECT 1 FROM queued_tasks) +); + +-- For failed tasks: archive the message +PERFORM pgmq.archive(r.flow_slug, ARRAY_AGG(st.message_id)) +FROM pgflow.step_tasks st +JOIN pgflow.runs r ON st.run_id = r.run_id +WHERE st.run_id = fail_task.run_id + AND st.step_slug = fail_task.step_slug + AND st.task_index = fail_task.task_index + AND st.status = 'failed' + AND st.message_id IS NOT NULL +GROUP BY r.flow_slug +HAVING COUNT(st.message_id) > 0; + +return query select * +from pgflow.step_tasks st +where st.run_id = fail_task.run_id + and st.step_slug = fail_task.step_slug + and st.task_index = fail_task.task_index; + +end; +$$; + +-- DATA REPAIR: Terminalize unfinished tasks on historical failed runs +-- ========================================== +-- Historical failure paths archived messages but left sibling task rows +-- queued/started. The migration's constraint change above permits 'cancelled', +-- so repair active rows attached to failed runs. Completed and genuinely +-- failed tasks keep their outcomes; all history fields are preserved. +-- runs.failed_at remains the cancellation time; no new columns are added. + +UPDATE pgflow.step_tasks AS task +SET status = 'cancelled' +FROM pgflow.runs AS run +WHERE run.run_id = task.run_id + AND run.status = 'failed' + AND task.status IN ('queued', 'started'); diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index a99720676..45ea27349 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:T6owyW5rB37cPjny53IxN6P6+KLzA2CUTT4z01T9Y+U= +h1:cdRdUjC74IbEKXRAHDb7liq+1AAMIWD3mR5ezWhlJoI= 20250429164909_pgflow_initial.sql h1:I3n/tQIg5Q5nLg7RDoU3BzqHvFVjmumQxVNbXTPG15s= 20250517072017_pgflow_fix_poll_for_tasks_to_use_separate_statement_for_polling.sql h1:wTuXuwMxVniCr3ONCpodpVWJcHktoQZIbqMZ3sUHKMY= 20250609105135_pgflow_add_start_tasks_and_started_status.sql h1:ggGanW4Wyt8Kv6TWjnZ00/qVb3sm+/eFVDjGfT8qyPg= @@ -21,3 +21,4 @@ h1:T6owyW5rB37cPjny53IxN6P6+KLzA2CUTT4z01T9Y+U= 20260214181656_pgflow_step_conditions.sql h1:rHQnXCeZ/QGxPlChdTMxumtsTtYHr1ej183Dd+auw34= 20260607175525_pgflow_worker_start_mode.sql h1:PFAfoGaHe5stKF7YAFg6AqBxmRisqDvV60vVpnnVdBE= 20260827180017_pgflow_terminalize_skipped_tasks.sql h1:Aq4zYSUp707UiDxi08FQeXlRw2lkIonL8O6yi1LfU/g= +20260901183450_pgflow_temp_failed_run_terminalization.sql h1:yaXA7EtqXxlTjH+w60OpFyBnynExtUJzDjrI1Hv0jis= diff --git a/pkgs/core/supabase/tests/complete_task/no_mutations_on_failed_run.test.sql b/pkgs/core/supabase/tests/complete_task/no_mutations_on_failed_run.test.sql index f380602d8..19456ca5a 100644 --- a/pkgs/core/supabase/tests/complete_task/no_mutations_on_failed_run.test.sql +++ b/pkgs/core/supabase/tests/complete_task/no_mutations_on_failed_run.test.sql @@ -34,11 +34,11 @@ select pgflow.fail_task(:'run_id', 'step2', 0, 'Simulated failure'); -- Now try to complete step1 (race condition - worker doesn't know run failed) select pgflow.complete_task(:'run_id', 'step1', 0, '{"result": "test"}'::jsonb); --- Verify task was NOT marked as completed +-- Verify task was NOT marked as completed: it was terminalized as cancelled select is( status, - 'started', - 'Task should remain in started status when run is failed' + 'cancelled', + 'Task should be cancelled when the run fails before it completes' ) from pgflow.step_tasks where run_id = :'run_id' and step_slug = 'step1' and task_index = 0; @@ -73,13 +73,13 @@ select is( 'Run should be in failed status' ) from pgflow.runs where run_id = :'run_id'; --- Verify step1 remains started (not completed) +-- Verify step1 history is preserved (attempts kept from start_tasks) select is( attempts_count, 1, - 'Step1 attempts_count should be 1 after start_tasks' + 'Step1 attempts_count should stay 1 after cancellation and late callback' ) from pgflow.step_tasks where run_id = :'run_id' and step_slug = 'step1' and task_index = 0; select finish(); -rollback; \ No newline at end of file +rollback; diff --git a/pkgs/core/supabase/tests/condition_evaluation/dependent_unmet_fail_archives_active_messages.test.sql b/pkgs/core/supabase/tests/condition_evaluation/dependent_unmet_fail_archives_active_messages.test.sql index 034672cd1..f6c7ffc20 100644 --- a/pkgs/core/supabase/tests/condition_evaluation/dependent_unmet_fail_archives_active_messages.test.sql +++ b/pkgs/core/supabase/tests/condition_evaluation/dependent_unmet_fail_archives_active_messages.test.sql @@ -1,5 +1,5 @@ begin; -select plan(6); +select plan(11); select pgflow_tests.reset_db(); @@ -35,20 +35,47 @@ select ok( 'should have active messages before failure' ); -with started as ( - select * from pgflow_tests.read_and_start('dependent_fail_archive', qty => 10) -), -target as ( - select run_id, step_slug, task_index - from started - where step_slug = 'first' - limit 1 -) -select pgflow.complete_task( - (select run_id from target), - (select step_slug from target), - (select task_index from target), - '{"ok": false}'::jsonb +-- Install a guard on the queue table: archiving (DELETE) is only allowed once the +-- owning step_tasks row left queued/started. If cascade_resolve_conditions archives +-- messages before terminalizing their tasks, this trigger raises. +create or replace function pg_temp.assert_task_terminalized_before_archive() +returns trigger language plpgsql as $$ +declare + v_flow_slug text := substr(tg_table_name, 3); -- strip 'q_' prefix + v_status text; +begin + select st.status into v_status + from pgflow.step_tasks st + join pgflow.runs r on r.run_id = st.run_id + where r.flow_slug = v_flow_slug + and st.message_id = old.msg_id; + + if v_status in ('queued', 'started') then + raise exception 'message % archived before its task was terminalized', old.msg_id; + end if; + + return old; +end; +$$; + +create trigger assert_terminalized_before_archive +before delete on pgmq.q_dependent_fail_archive +for each row execute function pg_temp.assert_task_terminalized_before_archive(); + +-- Start both root tasks ('second' is the independent branch that stays unfinished) +select * from pgflow_tests.read_and_start('dependent_fail_archive', qty => 10); + +-- Complete 'first' with an output that leaves the checker condition unmet +select lives_ok( + $$ + select pgflow.complete_task( + (select run_id from run_ids), + 'first', + 0, + '{"ok": false}'::jsonb + ) + $$, + 'condition failure should archive active messages only after terminalizing their tasks' ); select is( @@ -72,6 +99,39 @@ select is( 'checker should fail due to unmet condition' ); +select is( + ( + select status + from pgflow.step_tasks + where run_id = (select run_id from run_ids) + and step_slug = 'first' + ), + 'completed', + 'completed trigger task should stay completed' +); + +select is( + ( + select status + from pgflow.step_tasks + where run_id = (select run_id from run_ids) + and step_slug = 'second' + ), + 'cancelled', + 'independent unfinished task should become cancelled' +); + +select is( + ( + select count(*)::int + from pgflow.step_tasks + where run_id = (select run_id from run_ids) + and status in ('queued', 'started') + ), + 0, + 'failed run should have zero task rows with status queued or started' +); + select is( ( select count(*) @@ -92,15 +152,20 @@ select is( 'previously active messages should be in archive' ); +-- Replay the failure route: events must not duplicate, terminal state must hold select is( - ( - select error_message - from pgflow.step_states - where run_id = (select run_id from run_ids) - and step_slug = 'checker' + pgflow.cascade_resolve_conditions((select run_id from run_ids)), + false, + 'replayed cascade_resolve_conditions should return false without duplicating transitions' +); + +select is( + pgflow_tests.count_realtime_events( + 'run:failed', + (select run_id from run_ids) ), - 'Condition not met', - 'checker failure should use stable condition error message' + 1::int, + 'replayed failure route should not duplicate run:failed events' ); drop table if exists run_ids; diff --git a/pkgs/core/supabase/tests/fail_task/archive_sibling_map_tasks.test.sql b/pkgs/core/supabase/tests/fail_task/archive_sibling_map_tasks.test.sql index 660c0465c..6876c82f2 100644 --- a/pkgs/core/supabase/tests/fail_task/archive_sibling_map_tasks.test.sql +++ b/pkgs/core/supabase/tests/fail_task/archive_sibling_map_tasks.test.sql @@ -1,10 +1,11 @@ begin; -select plan(8); +select plan(10); select pgflow_tests.reset_db(); --- Test: fail_task should archive all sibling task messages for map steps +-- Test: fail_task should cancel unfinished sibling tasks and archive their messages +-- Reproduction of #645: task 0 fails, task 1 started -> cancelled, task 2 queued -> cancelled --- Create flow with a map step +-- Create flow with a map step that fails the run on exhaustion select pgflow.create_flow('test_map_fail'); select pgflow.add_step( flow_slug => 'test_map_fail', @@ -13,49 +14,72 @@ select pgflow.add_step( max_attempts => 1 ); --- Start flow with 5 array elements -select run_id as test_run_id from pgflow.start_flow('test_map_fail', '["a", "b", "c", "d", "e"]'::jsonb) \gset +-- Start flow with 3 array elements +select run_id as test_run_id from pgflow.start_flow('test_map_fail', '["a", "b", "c"]'::jsonb) \gset --- Verify all 5 messages are in queue +-- Verify all 3 messages are in queue select is( (select count(*) from pgmq.q_test_map_fail), - 5::bigint, - 'Should have 5 messages in queue for 5 map tasks' -); - --- Verify all 5 tasks are created -select is( - (select count(*)::integer from pgflow.step_tasks - where run_id = :'test_run_id'::uuid - and step_slug = 'map_step' - and status = 'queued'), - 5, - 'Should have 5 queued tasks' + 3::bigint, + 'Should have 3 messages in queue for 3 map tasks' ); -- Ensure worker exists for polling -select pgflow_tests.ensure_worker('test_map_fail'); - --- Start task 0 (simulating Edge Worker behavior) --- Note: read_and_start will start one of the tasks (we'll use it for testing) -WITH task AS ( - SELECT * FROM pgflow_tests.read_and_start('test_map_fail', 1, 1) - LIMIT 1 -) -SELECT step_slug FROM task; - --- Get the actual task_index of the started task for later reference -select task_index as started_task_index from pgflow.step_tasks -where run_id = :'test_run_id'::uuid - and step_slug = 'map_step' - and status = 'started' \gset - --- Fail the started task -select pgflow.fail_task( - :'test_run_id'::uuid, - 'map_step', - :'started_task_index'::integer, - 'Task failed!' +select pgflow_tests.ensure_worker('test_map_fail') as test_worker_id \gset + +-- Start task 0 (will be the failing task) +select message_id as msg_0 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 0 \gset +select pgflow.start_tasks('test_map_fail', array[:'msg_0'::bigint], :'test_worker_id'::uuid); + +-- Start task 1 (unfinished started sibling) +select message_id as msg_1 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1 \gset +select pgflow.start_tasks('test_map_fail', array[:'msg_1'::bigint], :'test_worker_id'::uuid); + +-- Task 2 stays queued + +-- Install a guard on the queue table: archiving (DELETE) is only allowed once the +-- owning step_tasks row left queued/started. If fail_task archives sibling messages +-- before terminalizing the sibling tasks, this trigger raises. +create or replace function pg_temp.assert_task_terminalized_before_archive() +returns trigger language plpgsql as $$ +declare + v_flow_slug text := substr(tg_table_name, 3); -- strip 'q_' prefix + v_status text; +begin + select st.status into v_status + from pgflow.step_tasks st + join pgflow.runs r on r.run_id = st.run_id + where r.flow_slug = v_flow_slug + and st.message_id = old.msg_id; + + if v_status in ('queued', 'started') then + raise exception 'message % archived before its task was terminalized', old.msg_id; + end if; + + return old; +end; +$$; + +create trigger assert_terminalized_before_archive +before delete on pgmq.q_test_map_fail +for each row execute function pg_temp.assert_task_terminalized_before_archive(); + +-- psql cannot interpolate :'test_run_id' inside dollar quotes, so pass it via temp table +select :'test_run_id'::uuid as run_id into temporary test_run_ids; + +-- Fail task 0 (max_attempts=1 -> immediate exhaustion -> run fails) +select lives_ok( + $$ + select pgflow.fail_task( + (select run_id from test_run_ids), + 'map_step', + 0, + 'Task failed!' + ) + $$, + 'fail_task should archive sibling messages only after terminalizing their tasks' ); -- Test: Run should be marked as failed @@ -65,49 +89,65 @@ select is( 'Run should be marked as failed after task failure' ); --- Test: Failed task should have status 'failed' +-- Test: task 0 failed, task 1 (started) cancelled, task 2 (queued) cancelled +select results_eq( + format($$ + select task_index, status + from pgflow.step_tasks + where run_id = '%s'::uuid and step_slug = 'map_step' + order by task_index + $$, :'test_run_id'), + $$ values (0, 'failed'), (1, 'cancelled'), (2, 'cancelled') $$, + 'Task statuses should be (0, failed), (1, cancelled), (2, cancelled)' +); + +-- CRITICAL TEST: No active task rows remain on the failed run select is( - (select status from pgflow.step_tasks + (select count(*)::int from pgflow.step_tasks where run_id = :'test_run_id'::uuid - and step_slug = 'map_step' - and task_index = :'started_task_index'::integer), + and status in ('queued', 'started')), + 0, + 'Failed run should have zero task rows with status queued or started' +); + +-- Test: Step state should be marked as failed +select is( + (select status from pgflow.step_states + where run_id = :'test_run_id'::uuid + and step_slug = 'map_step'), 'failed', - 'Started task should be marked as failed' + 'Map step should be marked as failed' ); --- CRITICAL TEST: All sibling task messages should be archived (removed from queue) +-- CRITICAL TEST: All messages should be archived (removed from queue) select is( (select count(*) from pgmq.q_test_map_fail), 0::bigint, - 'All 5 messages should be archived (removed from queue) when one map task fails' + 'All 3 messages should be archived (removed from queue) when one map task fails' ); -- Test: Verify messages were actually archived, not deleted select is( (select count(*) from pgmq.a_test_map_fail), - 5::bigint, - 'All 5 messages should be in archive table' + 3::bigint, + 'All 3 messages should be in archive table' ); --- Test: All sibling tasks should remain in 'queued' status +-- Test: Cancellation preserves history fields (worker identity, attempts, timestamps) select is( - (select count(*)::integer from pgflow.step_tasks - where run_id = :'test_run_id'::uuid - and step_slug = 'map_step' - and task_index != :'started_task_index'::integer - and status = 'queued'), - 4, - 'Sibling tasks should remain in queued status' + (select last_worker_id = :'test_worker_id'::uuid + from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1), + true, + 'Cancelled started task should preserve last_worker_id' ); --- Test: Step state should be marked as failed select is( - (select status from pgflow.step_states - where run_id = :'test_run_id'::uuid - and step_slug = 'map_step'), - 'failed', - 'Map step should be marked as failed' + (select attempts_count from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1), + 1, + 'Cancelled started task should preserve attempts_count' ); select * from finish(); -rollback; \ No newline at end of file +rollback; diff --git a/pkgs/core/supabase/tests/fail_task_when_exhausted/cancelled_task_late_callbacks_are_idempotent.test.sql b/pkgs/core/supabase/tests/fail_task_when_exhausted/cancelled_task_late_callbacks_are_idempotent.test.sql new file mode 100644 index 000000000..e80bed616 --- /dev/null +++ b/pkgs/core/supabase/tests/fail_task_when_exhausted/cancelled_task_late_callbacks_are_idempotent.test.sql @@ -0,0 +1,130 @@ +-- Test: late complete_task() and fail_task() callbacks after cancellation are idempotent +-- Cancellation wins: repeated late callbacks must not revive cancelled rows, rewrite +-- history, change parent state or counters, or duplicate events. +begin; +select plan(10); +select pgflow_tests.reset_db(); + +-- Map step with one allowed attempt: exhausting task 0 fails the run +select pgflow.create_flow('late_callback_test', max_attempts => 1); +select pgflow.add_step( + flow_slug => 'late_callback_test', + step_slug => 'map_step', + step_type => 'map' +); + +select run_id as test_run_id from pgflow.start_flow('late_callback_test', '["a", "b"]'::jsonb) \gset + +-- Start both tasks +select pgflow_tests.ensure_worker('late_callback_test') as test_worker_id \gset + +select message_id as msg_0 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 0 \gset +select pgflow.start_tasks('late_callback_test', array[:'msg_0'::bigint], :'test_worker_id'::uuid); + +select message_id as msg_1 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1 \gset +select pgflow.start_tasks('late_callback_test', array[:'msg_1'::bigint], :'test_worker_id'::uuid); + +-- Fail task 0: run fails, task 1 becomes cancelled +select pgflow.fail_task(:'test_run_id'::uuid, 'map_step', 0, 'Task 0 failed'); + +select is( + (select status from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1), + 'cancelled', + 'Sibling task should be cancelled when the run fails' +); + +-- Snapshot state before the late callbacks +create temporary table task_before as +select status, attempts_count, error_message, output, started_at, completed_at, + failed_at, last_worker_id +from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1; + +create temporary table run_before as +select run_id, status, failed_at, remaining_steps from pgflow.runs +where run_id = :'test_run_id'::uuid; + +-- Late callbacks, each repeated +select pgflow.complete_task(:'test_run_id'::uuid, 'map_step', 1, '{"late": true}'::jsonb); +select pgflow.complete_task(:'test_run_id'::uuid, 'map_step', 1, '{"late": true}'::jsonb); +select pgflow.fail_task(:'test_run_id'::uuid, 'map_step', 1, 'Late failure'); +select pgflow.fail_task(:'test_run_id'::uuid, 'map_step', 1, 'Late failure'); + +-- Cancelled row is unchanged +select results_eq( + $$ + select status, attempts_count, error_message, output, started_at, + completed_at, failed_at, last_worker_id + from pgflow.step_tasks + where run_id = (select run_id from run_before limit 1) + and step_slug = 'map_step' and task_index = 1 + $$, + $$ select status, attempts_count, error_message, output, started_at, + completed_at, failed_at, last_worker_id from task_before $$, + 'Repeated late callbacks should leave the cancelled row and its history unchanged' +); + +-- Late complete_task must not write an output +select is( + (select output from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1), + null, + 'Late complete_task should not store output on cancelled task' +); + +-- Late fail_task must not write an error +select is( + (select error_message from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 1), + null, + 'Late fail_task should not write error_message on cancelled task' +); + +-- Parent state unchanged +select results_eq( + $$ select status, failed_at, remaining_steps from pgflow.runs + where run_id = (select run_id from run_before limit 1) $$, + $$ select status, failed_at, remaining_steps from run_before $$, + 'Late callbacks should not change run status, failed_at, or counters' +); + +select is( + (select status from pgflow.step_states + where run_id = :'test_run_id'::uuid and step_slug = 'map_step'), + 'failed', + 'Step state should stay failed after late callbacks' +); + +-- Culprit task stays failed +select is( + (select status from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'map_step' and task_index = 0), + 'failed', + 'Genuinely failed culprit task should stay failed' +); + +-- Events are not duplicated +select is( + pgflow_tests.count_realtime_events('run:failed', :'test_run_id'::uuid), + 1::int, + 'Late callbacks should not duplicate run:failed events' +); + +select is( + pgflow_tests.count_realtime_events('step:failed', :'test_run_id'::uuid, 'map_step'), + 1::int, + 'Late callbacks should not duplicate step:failed events' +); + +-- No active queue rows reappear +select is( + (select count(*) from pgmq.q_late_callback_test), + 0::bigint, + 'Late callbacks should leave the queue empty' +); + +select * from finish(); +rollback; diff --git a/pkgs/core/supabase/tests/fail_task_when_exhausted/late_callbacks_post_lock_race.test.sql b/pkgs/core/supabase/tests/fail_task_when_exhausted/late_callbacks_post_lock_race.test.sql new file mode 100644 index 000000000..1676b9c21 --- /dev/null +++ b/pkgs/core/supabase/tests/fail_task_when_exhausted/late_callbacks_post_lock_race.test.sql @@ -0,0 +1,324 @@ +-- Test: late complete_task()/fail_task() callbacks blocked on the run row lock +-- while the run fails underneath them (the post-lock race). +-- +-- The single-session tests in this directory commit the run failure before the +-- callback starts, so the callbacks always exit at the pre-lock EXISTS guard. +-- This test uses two dblink sessions to force the interleaving that guard +-- cannot cover: +-- +-- 1. ctrl session locks the run row (SELECT ... FOR UPDATE) in an open txn +-- 2. late complete_task() and late fail_task() — both aimed at single_step, +-- the task the winning boom failure will CANCEL — pass their pre-lock +-- checks (run still 'started') and block on the run row lock. +-- pg_blocking_pids() proves the ctrl backend is in each callback's +-- blocking chain (second waiter queues behind the first; see the wait +-- loop below). single_step's step state stays 'started' when the run +-- fails, so the terminal-step guard cannot stop either callback: only +-- the post-lock failed-run guard protects the cancelled task here. +-- 3. ctrl fails the run (fail_task on the culprit boom) and commits +-- 4. both late callbacks resume, see status='failed' under the lock, and +-- must return without any mutation +-- +-- The late completion is a single-to-map type violation (non-array output for +-- a step with a dependent map step): without the post-lock guard it would +-- rewrite the cancelled task to failed, move runs.failed_at, and duplicate +-- failure events. The late fail_task on the cancelled sibling would fall +-- through to the run UPDATE and wipe runs.failed_at while re-emitting +-- run:failed. +begin; +select plan(18); + +create extension if not exists dblink; + +-- Self-heal: terminate ctrl/late sessions leaked by a previously crashed run +-- of this test. They hold locks that would make the ctrl setup below hang. +select count(pg_terminate_backend(pid)) as terminated_stale_sessions +from pg_stat_activity +where application_name in ('race_ctrl', 'race_late_complete', 'race_late_fail') + and pid <> pg_backend_pid(); + +-- Connection string for the ctrl/late dblink sessions (same DB as this test). +-- The setup must be committed by the ctrl session: rows created in this +-- transaction are invisible to dblink sessions. +select format( + 'hostaddr=%s port=%s dbname=%s user=postgres password=postgres application_name=', + coalesce(host(inet_server_addr()), '127.0.0.1'), + inet_server_port(), + current_database() +) as conn_base \gset + +select dblink_connect('ctrl', :'conn_base' || 'race_ctrl'); +-- Fail fast (loud test error) if leaked locks from a crashed run would hang +-- setup/cleanup instead of blocking forever. +select dblink_exec('ctrl', 'set lock_timeout = 5000'); + +-- Committed setup: single_step feeds map_step (map, never started, so its +-- initial_tasks IS NULL and a non-array completion would be a TYPE_VIOLATION). +-- boom is an independent root whose exhaustion (max_attempts=1) fails the run +-- and cancels the started single_step task. +select dblink_exec('ctrl', $$do $do$ begin perform pgflow_tests.reset_db(); end $do$;$$); +select dblink_exec('ctrl', $$do $do$ begin perform pgflow.create_flow('race_flow', max_attempts => 1); end $do$;$$); +select dblink_exec('ctrl', $$do $do$ begin perform pgflow.add_step('race_flow', 'single_step'); end $do$;$$); +select dblink_exec('ctrl', $$do $do$ begin perform pgflow.add_step('race_flow', 'map_step', ARRAY['single_step'], step_type => 'map'); end $do$;$$); +select dblink_exec('ctrl', $$do $do$ begin perform pgflow.add_step('race_flow', 'boom'); end $do$;$$); + +select run_id from dblink('ctrl', $$select run_id from pgflow.start_flow('race_flow', '{}'::jsonb)$$) as t(run_id uuid) \gset + +select dblink_exec('ctrl', $$do $do$ begin perform pgflow_tests.ensure_worker('race_flow'); end $do$;$$); + +select msg_single from dblink('ctrl', $$select message_id from pgflow.step_tasks where step_slug = 'single_step' and task_index = 0$$) as t(msg_single bigint) \gset +select msg_boom from dblink('ctrl', $$select message_id from pgflow.step_tasks where step_slug = 'boom' and task_index = 0$$) as t(msg_boom bigint) \gset + +select dblink_exec( + 'ctrl', + format( + $$do $do$ begin perform pgflow.start_tasks('race_flow', ARRAY[%s, %s]::bigint[], '11111111-1111-1111-1111-111111111111'::uuid); end $do$;$$, + :'msg_single', :'msg_boom' + ) +); + +-- RACE SETUP: ctrl holds the run row lock; both late callbacks block after +-- their pre-lock checks, at the FOR UPDATE inside the functions. +select dblink_exec('ctrl', 'begin'); +select dblink_exec('ctrl', format($$do $do$ begin perform 1 from pgflow.runs where run_id = %L for update; end $do$;$$, :'run_id')); + +-- Both late callbacks target single_step: the task the winning boom failure +-- will cancel (its step state stays 'started', so the terminal-step guard +-- cannot cover them). Late fail_task must NOT target the culprit boom — +-- boom's step state is terminal after the failure, so that callback would +-- never exercise the post-lock failed-run guard. +select dblink_connect('late_complete', :'conn_base' || 'race_late_complete'); +-- Bound the lock wait: a stuck setup errors loudly instead of hanging. +select dblink_exec('late_complete', 'set lock_timeout = 30000'); +select dblink_send_query( + 'late_complete', + format( + $$select status, output is null as output_is_null, error_message is null as error_is_null + from pgflow.complete_task(%L, 'single_step', 0, '{"not": "an array"}'::jsonb) as cb$$, + :'run_id' + ) +); + +select dblink_connect('late_fail', :'conn_base' || 'race_late_fail'); +select dblink_exec('late_fail', 'set lock_timeout = 30000'); +select dblink_send_query( + 'late_fail', + format( + $$select status, error_message, failed_at is null as failed_at_is_null + from pgflow.fail_task(%L, 'single_step', 0, 'late failure report') as ft$$, + :'run_id' + ) +); + +-- Probe connection for polling pg_stat_activity. Each dblink() call on it is a +-- single autocommit statement with a FRESH activity snapshot; this test +-- transaction's own pg_stat_activity view is cached from its first use +-- (the terminate above) and would never show the late sessions. +select dblink_connect('probe', :'conn_base' || 'race_probe'); + +-- Deterministic: wait until BOTH late callbacks are blocked by the ctrl +-- backend specifically. pg_blocking_pids() proves ctrl is in each callback's +-- blocking chain — not just that some lock wait exists. The chain matters: +-- PostgreSQL queues row-lock waiters FIFO, so the second callback blocks on +-- the first callback's tuple lock (its direct blocker), not on ctrl itself; +-- both still resume only after ctrl commits. +do $do$ +declare + blocked bigint; + deadline timestamptz := clock_timestamp() + interval '10 seconds'; +begin + perform pg_sleep(0.2); -- let the late callbacks reach their lock waits + loop + select blocked_count into blocked + from dblink('probe', $q$ + with recursive blockers as ( + select late.pid as late_pid, b.pid as blocker_pid + from pg_stat_activity late + cross join lateral unnest(pg_blocking_pids(late.pid)) as b(pid) + where late.application_name in ('race_late_complete', 'race_late_fail') + and late.wait_event_type = 'Lock' + union + select bl.late_pid, nb.pid + from blockers bl + join pg_stat_activity blocker on blocker.pid = bl.blocker_pid + cross join lateral unnest(pg_blocking_pids(blocker.pid)) as nb(pid) + ) + select count(distinct late_pid) as blocked_count + from blockers + where blocker_pid in ( + select pid from pg_stat_activity where application_name = 'race_ctrl' + ) + $q$) as t(blocked_count bigint); + exit when blocked = 2; + if clock_timestamp() > deadline then + raise exception 'late callbacks never blocked on the ctrl backend (blocked by ctrl: %/2)', blocked; + end if; + perform pg_sleep(0.05); + end loop; +end +$do$; + +-- Fail the run and commit: the lock releases and both late callbacks resume +-- with a committed failed run. +select dblink_exec('ctrl', format($$do $do$ begin perform pgflow.fail_task(%L, 'boom', 0, 'boom failed for the race test'); end $do$;$$, :'run_id')); +select dblink_exec('ctrl', 'commit'); + +-- Wait until both async late-callback queries have finished. +do $do$ +declare + deadline timestamptz := clock_timestamp() + interval '10 seconds'; +begin + loop + exit when dblink_is_busy('late_complete') = 0 and dblink_is_busy('late_fail') = 0; + if clock_timestamp() > deadline then + raise exception 'late callbacks did not finish after the run failure commit'; + end if; + perform pg_sleep(0.05); + end loop; +end +$do$; + +-- Late complete_task returns the cancelled row untouched: no output stored, +-- no TYPE_VIOLATION error written. (One get_result call: the result set is +-- consumed by the first fetch, so assert on all columns at once.) +select is( + (status, output_is_null, error_is_null), + ('cancelled'::text, true, true), + 'Late complete_task should return the cancelled task row unchanged: no output, no TYPE_VIOLATION error' +) from dblink_get_result('late_complete') as r(status text, output_is_null boolean, error_is_null boolean); + +-- Late fail_task also returns the cancelled sibling row untouched: no late +-- error message, no failed_at timestamp. +select is( + (status, error_message, failed_at_is_null), + ('cancelled'::text, null::text, true), + 'Late fail_task should return the cancelled sibling task row unchanged: no error, no failed_at' +) from dblink_get_result('late_fail') as r(status text, error_message text, failed_at_is_null boolean); + +select dblink_disconnect('late_complete'); +select dblink_disconnect('late_fail'); + +-- Cancelled task history is preserved after BOTH late callbacks. +select is( + (select status from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + 'cancelled', + 'single_step task should stay cancelled after the race' +); + +select is( + (select output from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + null, + 'single_step task output should stay null after the race' +); + +select is( + (select error_message from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + null, + 'single_step task error_message should stay null after the late fail_task' +); + +select is( + (select attempts_count from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + 1, + 'single_step attempts_count should stay 1 after the race' +); + +select is( + (select completed_at from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + null, + 'single_step completed_at should stay null after the race' +); + +select is( + (select failed_at from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'single_step' and task_index = 0), + null, + 'single_step failed_at should stay null after the race' +); + +-- The culprit stays genuinely failed with the winning failure's history. +select is( + (select status from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'boom' and task_index = 0), + 'failed', + 'boom task should stay genuinely failed after the race' +); + +select is( + (select error_message from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'boom' and task_index = 0), + 'boom failed for the race test', + 'boom error_message should keep the winning failure report, not a late callback one' +); + +-- Run state: failed once, failed_at not moved, counters not decremented again. +select is( + (select status from pgflow.runs where run_id = :'run_id'::uuid), + 'failed', + 'Run should be failed after the race' +); + +select is( + (select failed_at from pgflow.runs where run_id = :'run_id'::uuid), + (select failed_at from pgflow.step_tasks + where run_id = :'run_id'::uuid and step_slug = 'boom' and task_index = 0), + 'runs.failed_at should stay the run-failure timestamp, not be moved or wiped by a late callback' +); + +select is( + (select remaining_steps from pgflow.runs where run_id = :'run_id'::uuid), + 3, + 'remaining_steps should stay unchanged by the late callbacks' +); + +select is( + (select status from pgflow.step_states + where run_id = :'run_id'::uuid and step_slug = 'single_step'), + 'started', + 'single_step state should stay started (run failure does not terminalize step states)' +); + +-- No duplicate failure events from the late callbacks. +select is( + pgflow_tests.count_realtime_events('run:failed', :'run_id'::uuid), + 1, + 'run:failed should be sent exactly once' +); + +select is( + pgflow_tests.count_realtime_events('step:failed', :'run_id'::uuid, 'single_step'), + 0, + 'late callbacks should not emit a step:failed event for the cancelled single_step' +); + +select is( + pgflow_tests.count_realtime_events('step:failed', :'run_id'::uuid, 'boom'), + 1, + 'step:failed for the culprit should be sent exactly once' +); + +-- Queue is empty: culprit archived by the failure path, cancelled sibling +-- archived by terminalization, late-callback archival is a no-op. +-- Read via the probe connection: a direct read here would hold an +-- AccessShareLock that blocks the cleanup's drop_queue on this queue. +select is( + (select c from dblink('probe', 'select count(*) from pgmq.q_race_flow') as t(c bigint)), + 0::bigint, + 'Queue should be empty after the race' +); + +-- Cleanup committed data created by the dblink sessions (this transaction's +-- own changes roll back with the test). reset_db() does not clear pgflow.workers, +-- so remove the test worker explicitly. +select dblink_exec('ctrl', $$do $do$ begin perform pgflow_tests.reset_db(); end $do$;$$); +select dblink_exec('ctrl', $$delete from pgflow.workers where queue_name = 'race_flow'$$); +select dblink_disconnect('ctrl'); +select dblink_disconnect('probe'); + +select * from finish(); +rollback; diff --git a/pkgs/core/supabase/tests/requeue_stalled_tasks/ignores_terminal_parent_state.test.sql b/pkgs/core/supabase/tests/requeue_stalled_tasks/ignores_terminal_parent_state.test.sql new file mode 100644 index 000000000..66dda2902 --- /dev/null +++ b/pkgs/core/supabase/tests/requeue_stalled_tasks/ignores_terminal_parent_state.test.sql @@ -0,0 +1,136 @@ +-- Test: requeue_stalled_tasks only recovers tasks on started runs under started steps +-- A stale started task on a failed run or under a terminal step is ignored. +begin; +select plan(10); + +select pgflow_tests.reset_db(); + +-- ========================================== +-- Scenario 1: terminal step, started run (skipped step leaves stale rows ignored) +-- ========================================== +select pgflow.create_flow('skip_stall_test', null, null, 5); +select pgflow.add_step( + flow_slug => 'skip_stall_test', + step_slug => 'map_a', + step_type => 'map', + max_attempts => 0, + when_exhausted => 'skip' +); +select pgflow.add_step('skip_stall_test', 'root_b'); + +-- 2-element array: map_a gets tasks 0 and 1; root_b gets 1 task +select run_id as skip_run_id from pgflow.start_flow('skip_stall_test', '["x", "y"]'::jsonb) \gset + +select pgflow_tests.ensure_worker('skip_stall_test'); + +-- Start map_a task 0 and root_b task +select * from pgflow_tests.read_and_start('skip_stall_test', 30, 10); + +-- Fail map_a task 0 with when_exhausted='skip': step skipped, run continues +select pgflow.fail_task(:'skip_run_id'::uuid, 'map_a', 0, 'skip me'); + +select is( + (select status from pgflow.step_states + where run_id = :'skip_run_id'::uuid and step_slug = 'map_a'), + 'skipped', + 'map_a step should be skipped (terminal) while run stays started' +); + +select is( + (select status from pgflow.runs where run_id = :'skip_run_id'::uuid), + 'started', + 'Run should stay started after skip' +); + +-- Backdate root_b (started run + started step): genuine recovery target +update pgflow.step_tasks +set queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds' +where run_id = :'skip_run_id'::uuid and step_slug = 'root_b'; + +-- Simulate a legacy stale row: map_a task 1 was terminalized as skipped, +-- rewrite it to started to reproduce pre-#645 data under a terminal step +update pgflow.step_tasks +set status = 'started', + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds' +where run_id = :'skip_run_id'::uuid and step_slug = 'map_a' and task_index = 1; + +select is( + pgflow.requeue_stalled_tasks(), + 1, + 'Only the started task under a started step should be requeued' +); + +select is( + (select status from pgflow.step_tasks + where run_id = :'skip_run_id'::uuid and step_slug = 'root_b'), + 'queued', + 'Genuine stalled task on started run and step should be requeued' +); + +select is( + (select status from pgflow.step_tasks + where run_id = :'skip_run_id'::uuid and step_slug = 'map_a' and task_index = 1), + 'started', + 'Stale started task under a terminal step should be ignored' +); + +-- ========================================== +-- Scenario 2: failed run (stale started rows ignored) +-- ========================================== +select pgflow.create_flow('failed_stall_test', max_attempts => 1, timeout => 5); +select pgflow.add_step('failed_stall_test', 'step_a'); +select pgflow.add_step('failed_stall_test', 'step_b'); + +select run_id as fail_run_id from pgflow.start_flow('failed_stall_test', '{}') \gset + +select pgflow_tests.ensure_worker('failed_stall_test'); +select * from pgflow_tests.read_and_start('failed_stall_test', 30, 10); + +-- Fail step_a: run fails and step_b task is terminalized as cancelled +select pgflow.fail_task(:'fail_run_id'::uuid, 'step_a', 0, 'boom'); + +select is( + (select status from pgflow.step_tasks + where run_id = :'fail_run_id'::uuid and step_slug = 'step_b'), + 'cancelled', + 'step_b task should be cancelled when the run fails' +); + +-- Simulate a legacy stale row: rewrite the cancelled row to started +update pgflow.step_tasks +set status = 'started', + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds' +where run_id = :'fail_run_id'::uuid and step_slug = 'step_b'; + +select is( + pgflow.requeue_stalled_tasks(), + 0, + 'Stale started task on a failed run should be ignored' +); + +select is( + (select status from pgflow.step_tasks + where run_id = :'fail_run_id'::uuid and step_slug = 'step_b'), + 'started', + 'Failed-run stale task should stay untouched (migration owns its repair)' +); + +-- The failed run's task must not gain requeue history +select is( + (select requeued_count from pgflow.step_tasks + where run_id = :'fail_run_id'::uuid and step_slug = 'step_b'), + 0, + 'Failed-run stale task should not gain requeue history' +); + +select is( + (select count(*)::int from pgmq.q_failed_stall_test), + 0, + 'Failed-run stale task should not get a new visible queue message' +); + +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/type_violations/cancels_unfinished_tasks.test.sql b/pkgs/core/supabase/tests/type_violations/cancels_unfinished_tasks.test.sql new file mode 100644 index 000000000..b9a6d8d2e --- /dev/null +++ b/pkgs/core/supabase/tests/type_violations/cancels_unfinished_tasks.test.sql @@ -0,0 +1,164 @@ +-- Test: type violations cancel unfinished tasks across independent branches +-- The directly invalid task stays failed with its output and TYPE_VIOLATION error; +-- completed work stays completed; queued and started unrelated tasks become cancelled. +begin; +select plan(9); +select pgflow_tests.reset_db(); + +select pgflow.create_flow('type_violation_cancel'); +select pgflow.add_step( + flow_slug => 'type_violation_cancel', + step_slug => 'producer', + step_type => 'single' +); +select pgflow.add_step( + flow_slug => 'type_violation_cancel', + step_slug => 'branch1', + deps_slugs => array['producer'], + step_type => 'single' +); +select pgflow.add_step( + flow_slug => 'type_violation_cancel', + step_slug => 'branch2', + deps_slugs => array['producer'], + step_type => 'single' +); +select pgflow.add_step( + flow_slug => 'type_violation_cancel', + step_slug => 'branch3', + deps_slugs => array['producer'], + step_type => 'single' +); +-- This map step expects arrays from branch1 +select pgflow.add_step( + flow_slug => 'type_violation_cancel', + step_slug => 'consumer_map', + deps_slugs => array['branch1'], + step_type => 'map' +); + +-- Start flow +select run_id as test_run_id from pgflow.start_flow('type_violation_cancel', '{}') \gset + +-- Start and complete producer to spawn the branches +select pgflow_tests.ensure_worker('type_violation_cancel'); +select * from pgflow_tests.read_and_start('type_violation_cancel', 1, 1) limit 1; +select pgflow.complete_task(:'test_run_id'::uuid, 'producer', 0, '{"data": "test"}'::jsonb); + +-- Install a guard on the queue table: archiving (DELETE) is only allowed once the +-- owning step_tasks row left queued/started. If the type-violation path archives +-- messages before terminalizing their tasks, this trigger raises. +create or replace function pg_temp.assert_task_terminalized_before_archive() +returns trigger language plpgsql as $$ +declare + v_flow_slug text := substr(tg_table_name, 3); -- strip 'q_' prefix + v_status text; +begin + select st.status into v_status + from pgflow.step_tasks st + join pgflow.runs r on r.run_id = st.run_id + where r.flow_slug = v_flow_slug + and st.message_id = old.msg_id; + + if v_status in ('queued', 'started') then + raise exception 'message % archived before its task was terminalized', old.msg_id; + end if; + + return old; +end; +$$; + +create trigger assert_terminalized_before_archive +before delete on pgmq.q_type_violation_cancel +for each row execute function pg_temp.assert_task_terminalized_before_archive(); + +-- psql cannot interpolate :'test_run_id' inside dollar quotes, so pass it via temp table +select :'test_run_id'::uuid as run_id into temporary test_run_ids; + +-- Start branch1 and branch2 (started siblings); branch3 stays queued +select message_id as msg_b1 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'branch1' \gset +select pgflow.start_tasks('type_violation_cancel', array[:'msg_b1'::bigint], '11111111-1111-1111-1111-111111111111'::uuid); + +select message_id as msg_b2 from pgflow.step_tasks +where run_id = :'test_run_id'::uuid and step_slug = 'branch2' \gset +select pgflow.start_tasks('type_violation_cancel', array[:'msg_b2'::bigint], '11111111-1111-1111-1111-111111111111'::uuid); + +-- Trigger type violation by completing branch1 with a non-array (consumer_map expects array) +select lives_ok( + $$ + select pgflow.complete_task( + (select run_id from test_run_ids), + 'branch1', + 0, + '{"not": "an array"}'::jsonb + ) + $$, + 'type violation should archive messages only after terminalizing their tasks' +); + +-- Invalid task failed with preserved output and TYPE_VIOLATION error +select is( + (select status from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'branch1'), + 'failed', + 'Directly invalid task should be failed' +); + +select is( + (select output from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'branch1'), + '{"not": "an array"}'::jsonb, + 'Invalid task should preserve the output that caused the violation' +); + +select matches( + (select error_message from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'branch1'), + '^\[TYPE_VIOLATION\].*', + 'Invalid task should keep its TYPE_VIOLATION error' +); + +-- Completed independent work stays completed +select is( + (select status from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and step_slug = 'producer'), + 'completed', + 'Previously completed task should stay completed' +); + +-- Unfinished tasks become cancelled +select results_eq( + format($$ + select step_slug, status + from pgflow.step_tasks + where run_id = '%s'::uuid and step_slug in ('branch2', 'branch3') + order by step_slug + $$, :'test_run_id'), + $$ values ('branch2', 'cancelled'), ('branch3', 'cancelled') $$, + 'Started and queued unrelated tasks should become cancelled' +); + +-- Failed-run invariant +select is( + (select count(*)::int from pgflow.step_tasks + where run_id = :'test_run_id'::uuid and status in ('queued', 'started')), + 0, + 'Failed run should have zero task rows with status queued or started' +); + +select is( + (select status from pgflow.runs where run_id = :'test_run_id'::uuid), + 'failed', + 'Run should be failed after the type violation' +); + +-- All messages archived +select is( + (select count(*) from pgmq.q_type_violation_cancel), + 0::bigint, + 'All messages should be archived after the type violation' +); + +select * from finish(); +rollback; diff --git a/pkgs/website/src/content/docs/concepts/data-model.mdx b/pkgs/website/src/content/docs/concepts/data-model.mdx index ba2717899..119cc3090 100644 --- a/pkgs/website/src/content/docs/concepts/data-model.mdx +++ b/pkgs/website/src/content/docs/concepts/data-model.mdx @@ -81,8 +81,10 @@ These tables track the execution state of flow instances: - Single steps create 1 task, map steps create N tasks - Each task has retry counter and attempts tracking - Contains `task_index` for map task array elements -- Tracks task status (`queued`, `started`, `completed`, `failed`, `skipped`) +- Tracks task status (`queued`, `started`, `completed`, `failed`, `skipped`, `cancelled`) - `skipped` marks the logical orchestration state: the parent step was skipped, so the task will never run; an already-running handler is not forcibly terminated +- `cancelled` marks unfinished tasks whose run failed: the task that caused the failure stays `failed` (with its error), completed work stays `completed`, and every remaining queued or started task becomes `cancelled`. `runs.failed_at` is the cancellation time; there is no separate `cancelled_at` column +- Database cancellation does not terminate JavaScript already executing in a worker and does not undo external side effects the handler already performed Created and modified during execution. Each run creates new records tracking progress from start to completion. From b4f88e1e8174dcd823c117a1c605942e01b25ea0 Mon Sep 17 00:00:00 2001 From: Agent Date: Wed, 2 Sep 2026 00:40:09 +0000 Subject: [PATCH 2/4] docs: explain failure and task termination Failure behavior spans handler retries, skipped steps, failed-run task cancellation, and stalled recovery, but the existing pages explain each path separately. Add a Concepts mental model with D2 state diagrams and cross-link it from build, operations, data-model, and reference pages. Extend the diagram theme for started, skipped, and cancelled task states. --- pkgs/website/astro.config.mjs | 5 + pkgs/website/src/assets/pgflow-theme.d2 | 15 +- .../content/docs/build/graceful-failure.mdx | 2 + .../src/content/docs/build/retrying-steps.mdx | 2 + .../src/content/docs/concepts/data-model.mdx | 2 + .../concepts/failures-and-termination.mdx | 253 ++++++++++++++++++ .../docs/concepts/how-pgflow-works.mdx | 5 + .../src/content/docs/concepts/index.mdx | 5 + pkgs/website/src/content/docs/d2-guide.mdx | 4 +- .../content/docs/deploy/monitor-execution.mdx | 9 +- .../deploy/troubleshooting-stalled-tasks.mdx | 2 + .../configuration/step-execution.mdx | 2 + 12 files changed, 300 insertions(+), 6 deletions(-) create mode 100644 pkgs/website/src/content/docs/concepts/failures-and-termination.mdx diff --git a/pkgs/website/astro.config.mjs b/pkgs/website/astro.config.mjs index 8698cf921..ca61b17f4 100644 --- a/pkgs/website/astro.config.mjs +++ b/pkgs/website/astro.config.mjs @@ -185,6 +185,7 @@ export default defineConfig({ 'get-started/flows/run-flow', 'concepts/how-pgflow-works', 'concepts/data-model', + 'concepts/failures-and-termination', 'concepts/understanding-flows', 'build/create-reusable-tasks', 'deploy/monitor-execution', @@ -381,6 +382,10 @@ export default defineConfig({ link: '/concepts/three-layer-architecture/', }, { label: 'Data model', link: '/concepts/data-model/' }, + { + label: 'Failures and termination', + link: '/concepts/failures-and-termination/', + }, { label: 'Startup Compilation', link: '/concepts/startup-compilation/', diff --git a/pkgs/website/src/assets/pgflow-theme.d2 b/pkgs/website/src/assets/pgflow-theme.d2 index 65fc375b8..ee850ccd9 100644 --- a/pkgs/website/src/assets/pgflow-theme.d2 +++ b/pkgs/website/src/assets/pgflow-theme.d2 @@ -92,11 +92,15 @@ classes: { style.stroke-dash: 3 } - # Task state classes (queued, completed, failed) + # Task state classes (queued, started, completed, failed, skipped, cancelled) task_queued: { style.fill: "#95a0a3" style.stroke: "#4a5759" } + task_started: { + style.fill: "#34578f" + style.stroke: "#5c8dd6" + } task_completed: { style.fill: "#247056" style.stroke: "#33cc7f" @@ -105,4 +109,13 @@ classes: { style.fill: "#a33636" style.stroke: "#e85c5c" } + task_skipped: { + style.fill: "#4a5759" + style.stroke: "#6b7a7d" + style.stroke-dash: 3 + } + task_cancelled: { + style.fill: "#a87c45" + style.stroke: "#d9a66e" + } } diff --git a/pkgs/website/src/content/docs/build/graceful-failure.mdx b/pkgs/website/src/content/docs/build/graceful-failure.mdx index 0ca1f5b06..82662f1be 100644 --- a/pkgs/website/src/content/docs/build/graceful-failure.mdx +++ b/pkgs/website/src/content/docs/build/graceful-failure.mdx @@ -9,6 +9,8 @@ import { Aside, CardGrid, LinkCard } from '@astrojs/starlight/components'; The `whenExhausted` option controls what happens when a step fails after exhausting all retry attempts. Instead of failing the entire run, you can skip the step and continue. +See [Failures and termination](/concepts/failures-and-termination/) for how each mode changes task, step, and run state. + ## Quick Example ```typescript diff --git a/pkgs/website/src/content/docs/build/retrying-steps.mdx b/pkgs/website/src/content/docs/build/retrying-steps.mdx index 98034129a..09b53cf1e 100644 --- a/pkgs/website/src/content/docs/build/retrying-steps.mdx +++ b/pkgs/website/src/content/docs/build/retrying-steps.mdx @@ -9,6 +9,8 @@ import { Aside, CardGrid, LinkCard } from '@astrojs/starlight/components'; Configure retry behavior based on step reliability characteristics. Set conservative flow-level defaults and override per-step as needed. +For the complete lifecycle after an error, see [Failures and termination](/concepts/failures-and-termination/). +