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/20260902005317_pgflow_failed_run_terminalization.sql b/pkgs/core/supabase/migrations/20260902005317_pgflow_failed_run_terminalization.sql new file mode 100644 index 000000000..c2f053cfd --- /dev/null +++ b/pkgs/core/supabase/migrations/20260902005317_pgflow_failed_run_terminalization.sql @@ -0,0 +1,1258 @@ +-- 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..e85c1bcc1 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:T6owyW5rB37cPjny53IxN6P6+KLzA2CUTT4z01T9Y+U= +h1:vahrstrzyG/2HD7neJ3HHsQNmR8Mj4LBcp5Ik/HGebc= 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= +20260902005317_pgflow_failed_run_terminalization.sql h1:agqIyQVyIS95DY8PI9QWg0KyJ3kGVMGrhQ+02FKZ8Xo= 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/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/). +