From 5ec80bf6436f00f3dc3a76c65dc0c48e41acbfae Mon Sep 17 00:00:00 2001 From: Agent Date: Fri, 21 Aug 2026 23:09:38 +0200 Subject: [PATCH] fix(core): terminalize skipped tasks --- .changeset/terminalize-skipped-tasks.md | 5 + NOMENCLATURE_GUIDE.md | 2 + pkgs/core/schemas/0060_tables_runtime.sql | 2 +- ...100_function__cascade_force_skip_steps.sql | 23 +- pkgs/core/schemas/0100_function_fail_task.sql | 7 + ...05023_pgflow_terminalize_skipped_tasks.sql | 512 ++++++++++++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- ...s_task_messages_for_skipped_steps.test.sql | 14 +- ...preexisting_skipped_step_messages.test.sql | 20 +- .../idempotent_second_call.test.sql | 16 +- ..._skip_does_not_mutate_step_or_run.test.sql | 12 +- ..._double_decrement_remaining_steps.test.sql | 10 +- .../skip_archives_sibling_messages.test.sql | 31 +- .../src/content/docs/concepts/data-model.mdx | 4 +- 14 files changed, 645 insertions(+), 16 deletions(-) create mode 100644 .changeset/terminalize-skipped-tasks.md create mode 100644 pkgs/core/supabase/migrations/20260821205023_pgflow_terminalize_skipped_tasks.sql diff --git a/.changeset/terminalize-skipped-tasks.md b/.changeset/terminalize-skipped-tasks.md new file mode 100644 index 000000000..2abd5ea47 --- /dev/null +++ b/.changeset/terminalize-skipped-tasks.md @@ -0,0 +1,5 @@ +--- +"@pgflow/core": patch +--- + +Terminalize queued and started task rows when their parent step is skipped: sibling tasks of a step skipped via `whenExhausted: 'skip'`/`'skip-cascade'` (and cascade-skipped steps) now end as `skipped` instead of staying `queued`/`started` forever, and a migration repairs existing rows. diff --git a/NOMENCLATURE_GUIDE.md b/NOMENCLATURE_GUIDE.md index 88c3ce90b..b75668a74 100644 --- a/NOMENCLATURE_GUIDE.md +++ b/NOMENCLATURE_GUIDE.md @@ -142,6 +142,7 @@ Slugs are unique text identifiers with specific rules: - `started` - Step is executing - `completed` - Step completed successfully - `failed` - Step failed permanently +- `skipped` - Step was skipped due to failed dependency, unmet condition, or exhausted retries ### Task Statuses @@ -149,6 +150,7 @@ Slugs are unique text identifiers with specific rules: - `started` - Task is executing - `completed` - Task completed successfully - `failed` - Task failed (may be retried or permanent) +- `skipped` - Task was cancelled because its parent step was skipped ## Configuration Terms diff --git a/pkgs/core/schemas/0060_tables_runtime.sql b/pkgs/core/schemas/0060_tables_runtime.sql index e3883b771..0cc4f3fa8 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') + status in ('queued', 'started', 'completed', 'failed', 'skipped') ), constraint output_valid_only_for_completed check ( output is null or status in ('completed', 'failed') diff --git a/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql b/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql index 7f41905da..5cc14c16c 100644 --- a/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql +++ b/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql @@ -90,15 +90,24 @@ BEGIN false ) as _broadcast_result ), + -- ---------- Terminalize active tasks of newly skipped steps ---------- + skipped_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'skipped' + WHERE task.run_id = _cascade_force_skip_steps.run_id + AND task.step_slug IN ( + SELECT skipped_step.step_slug + FROM skipped AS skipped_step + ) + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ), -- ---------- Archive queued/started task messages for skipped steps ---------- archived_messages AS ( - SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(st.message_id)) as result - FROM pgflow.step_tasks st - WHERE st.run_id = _cascade_force_skip_steps.run_id - AND st.step_slug IN (SELECT sk.step_slug FROM skipped sk) - AND st.status IN ('queued', 'started') - AND st.message_id IS NOT NULL - HAVING COUNT(st.message_id) > 0 + SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(task.message_id)) as result + FROM skipped_tasks AS task + WHERE task.message_id IS NOT NULL + HAVING COUNT(task.message_id) > 0 ), -- ---------- Update run counters ---------- run_updates AS ( diff --git a/pkgs/core/schemas/0100_function_fail_task.sql b/pkgs/core/schemas/0100_function_fail_task.sql index d1e0c7bb3..60e2fc782 100644 --- a/pkgs/core/schemas/0100_function_fail_task.sql +++ b/pkgs/core/schemas/0100_function_fail_task.sql @@ -225,6 +225,13 @@ END IF; GROUP BY r.flow_slug HAVING COUNT(st.message_id) > 0; + -- Terminalize all still-active sibling task rows for the skipped step + 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'); + -- Send broadcast event for step skipped PERFORM realtime.send( jsonb_build_object( diff --git a/pkgs/core/supabase/migrations/20260821205023_pgflow_terminalize_skipped_tasks.sql b/pkgs/core/supabase/migrations/20260821205023_pgflow_terminalize_skipped_tasks.sql new file mode 100644 index 000000000..bcd7cfb7e --- /dev/null +++ b/pkgs/core/supabase/migrations/20260821205023_pgflow_terminalize_skipped_tasks.sql @@ -0,0 +1,512 @@ +-- 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])); +-- Modify "_cascade_force_skip_steps" function +CREATE OR REPLACE FUNCTION "pgflow"."_cascade_force_skip_steps" ("run_id" uuid, "step_slug" text, "skip_reason" text) RETURNS integer LANGUAGE plpgsql AS $$ +DECLARE + v_flow_slug text; + v_total_skipped int := 0; +BEGIN + -- Get flow_slug for this run + SELECT r.flow_slug INTO v_flow_slug + FROM pgflow.runs r + WHERE r.run_id = _cascade_force_skip_steps.run_id; + + IF v_flow_slug IS NULL THEN + RAISE EXCEPTION 'Run not found: %', _cascade_force_skip_steps.run_id; + END IF; + + -- ========================================== + -- SKIP STEPS IN TOPOLOGICAL ORDER + -- ========================================== + -- Use recursive CTE to find all downstream dependents, + -- then skip them in topological order (by step_index) + WITH RECURSIVE + -- ---------- Find all downstream steps ---------- + downstream_steps AS ( + -- Base case: the trigger step + SELECT + s.flow_slug, + s.step_slug, + s.step_index, + _cascade_force_skip_steps.skip_reason AS reason -- Original reason for trigger step + FROM pgflow.steps s + WHERE s.flow_slug = v_flow_slug + AND s.step_slug = _cascade_force_skip_steps.step_slug + + UNION ALL + + -- Recursive case: steps that depend on already-found steps + SELECT + s.flow_slug, + s.step_slug, + s.step_index, + 'dependency_skipped'::text AS reason -- Downstream steps get this reason + FROM pgflow.steps s + JOIN pgflow.deps d ON d.flow_slug = s.flow_slug AND d.step_slug = s.step_slug + JOIN downstream_steps ds ON ds.flow_slug = d.flow_slug AND ds.step_slug = d.dep_slug + ), + -- ---------- Deduplicate and order by step_index ---------- + steps_to_skip AS ( + SELECT DISTINCT ON (ds.step_slug) + ds.flow_slug, + ds.step_slug, + ds.step_index, + ds.reason + FROM downstream_steps ds + ORDER BY ds.step_slug, ds.step_index -- Keep first occurrence (trigger step has original reason) + ), + -- ---------- Skip the steps ---------- + skipped AS ( + UPDATE pgflow.step_states ss + SET status = 'skipped', + skip_reason = sts.reason, + skipped_at = now(), + remaining_tasks = NULL -- Clear remaining_tasks for skipped steps + FROM steps_to_skip sts + WHERE ss.run_id = _cascade_force_skip_steps.run_id + AND ss.step_slug = sts.step_slug + AND ss.status IN ('created', 'started') -- Only skip non-terminal steps + RETURNING + ss.*, + -- Broadcast step:skipped event + 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', ss.skip_reason, + 'skipped_at', ss.skipped_at + ), + concat('step:', ss.step_slug, ':skipped'), + concat('pgflow:run:', ss.run_id), + false + ) as _broadcast_result + ), + -- ---------- Terminalize active tasks of newly skipped steps ---------- + skipped_tasks AS ( + UPDATE pgflow.step_tasks AS task + SET status = 'skipped' + WHERE task.run_id = _cascade_force_skip_steps.run_id + AND task.step_slug IN ( + SELECT skipped_step.step_slug + FROM skipped AS skipped_step + ) + AND task.status IN ('queued', 'started') + RETURNING task.message_id + ), + -- ---------- Archive queued/started task messages for skipped steps ---------- + archived_messages AS ( + SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(task.message_id)) as result + FROM skipped_tasks AS task + WHERE task.message_id IS NOT NULL + HAVING COUNT(task.message_id) > 0 + ), + -- ---------- Update run counters ---------- + run_updates AS ( + UPDATE pgflow.runs r + SET remaining_steps = r.remaining_steps - skipped_count.count + FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count + WHERE r.run_id = _cascade_force_skip_steps.run_id + AND skipped_count.count > 0 + ) + SELECT skipped_count.count + INTO v_total_skipped + FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count + LEFT JOIN archived_messages ON true; + + RETURN v_total_skipped; +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_flow_slug text; +begin + +-- If run is already failed, no retries allowed +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; + + 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 step status +-- 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 +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; + +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 + -- Archive all queued/started sibling task messages for this step + 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.status IN ('queued', 'started') + AND st.message_id IS NOT NULL + GROUP BY r.flow_slug + HAVING COUNT(st.message_id) > 0; + + -- Terminalize all still-active sibling task rows for the skipped step + 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'); + + -- 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; + +-- Archive all active messages (both queued and started) when run fails +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; +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 orphaned active tasks under skipped steps +-- ========================================== +-- Historical skip paths archived messages but left sibling task rows +-- queued/started. Repairs skipped steps whether the containing run is still +-- started or already completed; task rows on failed runs are out of scope (#645). +-- Does not re-archive messages: the historical skip paths already archived them. + +UPDATE pgflow.step_tasks AS task +SET status = 'skipped' +FROM pgflow.step_states AS step +WHERE step.run_id = task.run_id + AND step.step_slug = task.step_slug + AND step.status = 'skipped' + AND task.status IN ('queued', 'started'); diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index 314597418..2cecd193b 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:ugGS1dSXEdS8KOgQWhOdxWCovACZT4bTwWa9HVb7F70= +h1:46fDMXXm0wZFVz0yrDglH6WPG/sWhdXEziPUasjx4IU= 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= @@ -20,3 +20,4 @@ h1:ugGS1dSXEdS8KOgQWhOdxWCovACZT4bTwWa9HVb7F70= 20260124113408_pgflow_auth_secret_support.sql h1:i/s1JkBqRElN6FOYFQviJt685W08SuSo30aP25lNlLc= 20260214181656_pgflow_step_conditions.sql h1:rHQnXCeZ/QGxPlChdTMxumtsTtYHr1ej183Dd+auw34= 20260607175525_pgflow_worker_start_mode.sql h1:PFAfoGaHe5stKF7YAFg6AqBxmRisqDvV60vVpnnVdBE= +20260821205023_pgflow_terminalize_skipped_tasks.sql h1:vvgekFm/mBJPgYn9q6I/w5pDnyxQaQXa26xkKj9lpV8= diff --git a/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_for_skipped_steps.test.sql b/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_for_skipped_steps.test.sql index 438b20052..116333826 100644 --- a/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_for_skipped_steps.test.sql +++ b/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_for_skipped_steps.test.sql @@ -2,7 +2,7 @@ \set QUIET on begin; -select plan(5); +select plan(6); select pgflow_tests.reset_db(); @@ -61,5 +61,17 @@ select ok( 'Archive should contain all 3 map_a task messages' ); +-- Test: Active tasks of a directly cascaded skipped step become skipped +select results_eq( + $$ + select task_index, status + from pgflow.step_tasks + where flow_slug = 'cascade_skip_archive' and step_slug = 'map_a' + order by task_index + $$, + $$ values (0, 'skipped'), (1, 'skipped'), (2, 'skipped') $$, + 'Started and queued tasks of cascaded skipped step should become skipped' +); + select * from finish(); rollback; diff --git a/pkgs/core/supabase/tests/_cascade_force_skip_steps/does_not_archive_preexisting_skipped_step_messages.test.sql b/pkgs/core/supabase/tests/_cascade_force_skip_steps/does_not_archive_preexisting_skipped_step_messages.test.sql index cc69acb50..0dae6c2e1 100644 --- a/pkgs/core/supabase/tests/_cascade_force_skip_steps/does_not_archive_preexisting_skipped_step_messages.test.sql +++ b/pkgs/core/supabase/tests/_cascade_force_skip_steps/does_not_archive_preexisting_skipped_step_messages.test.sql @@ -2,7 +2,7 @@ \set QUIET on begin; -select plan(4); +select plan(6); select pgflow_tests.reset_db(); @@ -65,5 +65,23 @@ select is( 'Target step should be marked skipped' ); +select is( + (select count(*) from pgflow.step_tasks + where flow_slug = 'cascade_skip_preexisting' + and step_slug = 'target' + and status = 'skipped'), + 2::bigint, + 'Tasks of step newly skipped by this cascade call should become skipped' +); + +select is( + (select count(*) from pgflow.step_tasks + where flow_slug = 'cascade_skip_preexisting' + and step_slug = 'already_skipped' + and status = 'queued'), + 2::bigint, + 'Task rows under preexisting skipped step should remain untouched (queued)' +); + select * from finish(); rollback; diff --git a/pkgs/core/supabase/tests/_cascade_force_skip_steps/idempotent_second_call.test.sql b/pkgs/core/supabase/tests/_cascade_force_skip_steps/idempotent_second_call.test.sql index a509c00ba..ed87e27c4 100644 --- a/pkgs/core/supabase/tests/_cascade_force_skip_steps/idempotent_second_call.test.sql +++ b/pkgs/core/supabase/tests/_cascade_force_skip_steps/idempotent_second_call.test.sql @@ -2,7 +2,7 @@ \set QUIET on begin; -select plan(5); +select plan(7); select pgflow_tests.reset_db(); @@ -33,6 +33,13 @@ select is( 'First call should skip 2 steps (map_step + dependent_step)' ); +select is( + (select count(*) from pgflow.step_tasks + where flow_slug = 'idempotent_test' and step_slug = 'map_step' and status = 'skipped'), + 3::bigint, + 'First call should terminalize started and queued map_step tasks to skipped' +); + create temporary table after_first as select (select remaining_steps from pgflow.runs where run_id = (select run_id from test_run)) as remaining_steps, @@ -71,5 +78,12 @@ select is( 'Archive count should be unchanged after second call' ); +select is( + (select count(*) from pgflow.step_tasks + where flow_slug = 'idempotent_test' and step_slug = 'map_step' and status = 'skipped'), + 3::bigint, + 'Task statuses should be unchanged after second call' +); + select * from finish(); rollback; diff --git a/pkgs/core/supabase/tests/complete_task/late_complete_after_skip_does_not_mutate_step_or_run.test.sql b/pkgs/core/supabase/tests/complete_task/late_complete_after_skip_does_not_mutate_step_or_run.test.sql index ef22a6cf3..c562fa9fc 100644 --- a/pkgs/core/supabase/tests/complete_task/late_complete_after_skip_does_not_mutate_step_or_run.test.sql +++ b/pkgs/core/supabase/tests/complete_task/late_complete_after_skip_does_not_mutate_step_or_run.test.sql @@ -1,7 +1,7 @@ -- Test: Late complete after step is skipped should not mutate step or run state -- Verifies defense-in-depth: callbacks cannot change state after step is no longer started begin; -select plan(4); +select plan(5); select pgflow_tests.reset_db(); -- Setup: Create flow with map_a (skip on exhaust) and independent 'other' step @@ -86,5 +86,15 @@ select is( 'remaining_steps should not be decremented by late complete' ); +-- Verify sibling task row remains skipped (not revived or rewritten by late callback) +select is( + (select status from pgflow.step_tasks + where run_id = :'test_run_id'::uuid + and step_slug = 'map_a' + and task_index = 1), + 'skipped', + 'Sibling task should remain skipped after late complete' +); + select * from finish(); rollback; diff --git a/pkgs/core/supabase/tests/fail_task_when_exhausted/late_fail_after_skip_does_not_double_decrement_remaining_steps.test.sql b/pkgs/core/supabase/tests/fail_task_when_exhausted/late_fail_after_skip_does_not_double_decrement_remaining_steps.test.sql index e3b774a8c..a85ed4ba7 100644 --- a/pkgs/core/supabase/tests/fail_task_when_exhausted/late_fail_after_skip_does_not_double_decrement_remaining_steps.test.sql +++ b/pkgs/core/supabase/tests/fail_task_when_exhausted/late_fail_after_skip_does_not_double_decrement_remaining_steps.test.sql @@ -1,5 +1,5 @@ begin; -select plan(6); +select plan(7); select pgflow_tests.reset_db(); -- Setup: Create a flow with map step (max_attempts=>0, when_exhausted=>'skip') and 'other' step @@ -68,5 +68,13 @@ select is( 'remaining_steps unchanged after late fail (no double-decrement)' ); +-- Verify late-failed task row remains skipped (late callback must not rewrite it) +select is( + (select st.status from pgflow.step_tasks st, after_first_skip a + where st.run_id = a.run_id and st.step_slug = 'map_a' and st.task_index = 1), + 'skipped', + 'Late fail callback should not rewrite skipped sibling task row' +); + select finish(); rollback; diff --git a/pkgs/core/supabase/tests/fail_task_when_exhausted/skip_archives_sibling_messages.test.sql b/pkgs/core/supabase/tests/fail_task_when_exhausted/skip_archives_sibling_messages.test.sql index 19cf517a6..4a37d1796 100644 --- a/pkgs/core/supabase/tests/fail_task_when_exhausted/skip_archives_sibling_messages.test.sql +++ b/pkgs/core/supabase/tests/fail_task_when_exhausted/skip_archives_sibling_messages.test.sql @@ -1,7 +1,7 @@ -- Test: when_exhausted='skip' should archive all queued/started sibling task messages -- Verifies that when a map step transitions to skipped, sibling messages are archived begin; -select plan(6); +select plan(9); select pgflow_tests.reset_db(); -- Setup: Create flow with single root map step (max_attempts=0, when_exhausted='skip') @@ -85,5 +85,34 @@ select is( 'Step should be skipped when when_exhausted=skip' ); +-- Test: Failed task stays failed; started and queued siblings become skipped +select results_eq( + format($$ + select task_index, status + from pgflow.step_tasks + where run_id = '%s'::uuid and step_slug = 'map_a' + order by task_index + $$, :'test_run_id'), + $$ values (0, 'failed'), (1, 'skipped'), (2, 'skipped') $$, + 'Task statuses should be (0, failed), (1, skipped), (2, skipped) after skip' +); + +-- Test: Skipped step should have zero queued/started task rows +select is( + (select count(*)::int from pgflow.step_tasks + where run_id = :'test_run_id'::uuid + and step_slug = 'map_a' + and status in ('queued', 'started')), + 0, + 'Skipped step should have zero task rows with status queued or started' +); + +-- Test: Run should complete once its only step is skipped +select is( + (select status from pgflow.runs where run_id = :'test_run_id'::uuid), + 'completed', + 'Run should be completed after its only step was skipped' +); + select * from finish(); rollback; diff --git a/pkgs/website/src/content/docs/concepts/data-model.mdx b/pkgs/website/src/content/docs/concepts/data-model.mdx index 20bc0f785..ba2717899 100644 --- a/pkgs/website/src/content/docs/concepts/data-model.mdx +++ b/pkgs/website/src/content/docs/concepts/data-model.mdx @@ -72,7 +72,7 @@ These tables track the execution state of flow instances: - Maintains `remaining_steps` counter for completion detection **`step_states`** - State of individual steps within a run -- Tracks step status (`created`, `started`, `completed`, `failed`) +- Tracks step status (`created`, `started`, `completed`, `failed`, `skipped`) - For map steps, tracks `initial_tasks` and `remaining_tasks` counts - Stores step output when complete (for both single and map steps) - Coordinates step-level completion @@ -81,6 +81,8 @@ These tables track the execution state of flow instances: - Single steps create 1 task, map steps create N tasks - Each task has retry counter and attempts tracking - Contains `task_index` for map task array elements +- Tracks task status (`queued`, `started`, `completed`, `failed`, `skipped`) +- `skipped` marks the logical orchestration state: the parent step was skipped, so the task will never run; an already-running handler is not forcibly terminated Created and modified during execution. Each run creates new records tracking progress from start to completion.