From daa17533e3026847328a123c0430df474858ade1 Mon Sep 17 00:00:00 2001 From: Agent Date: Mon, 2 Nov 2026 21:23:56 +0100 Subject: [PATCH] fix(core): terminalize permanently stalled tasks through whenExhausted Recovery archived permanently stalled task messages but left their tasks and runs active. Treat the permanent-stall marker as exhaustion in fail_task() without inflating attempts_count, and use a separate escalation cron transaction to apply the existing fail, skip, or skip-cascade policy to marked started and queued tasks. Preserve terminal-state guards, diagnostic markers, stored queue routing, and installed recovery schedules and activation choices. Restore saved cron booleans exactly in the upgrade guidance rather than enabling every paused job; reject malformed saved state and require the snapshot to survive the SQL session. Preserve recorded enabled values in the deployment guide too. Combine the changes from dd41e5dfe and d4831bd79 onto the independent 689-terminalize-stalled branch based on main. Recalculate atlas.sum for this branch without the unrelated worker-startup migration from #696. The original commits remain recoverable at backup/689-misplaced-d4831bd79. The original implementation commit records pgTAP, worker, E2E, migration/type, and website checks; the documentation correction records rolled-back restoration checks and website lint/build. This relocation preserves their code changes but does not treat those earlier results as fresh runtime validation on the new base. Repository hooks run normally for this commit and submission. Fixes #689 --- .changeset/permanently-stalled-escalation.md | 5 + ...ction_setup_requeue_stalled_tasks_cron.sql | 41 +- ..._function_escalate_permanently_stalled.sql | 80 +++ pkgs/core/schemas/0100_function_fail_task.sql | 24 +- pkgs/core/scripts/run-upgrade-fixture | 59 ++ pkgs/core/src/database-types.ts | 1 + ...32_pgflow_escalate_permanently_stalled.sql | 550 ++++++++++++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- ...rmanently_stalled_marker_exhausts.test.sql | 91 +++ .../requeue_stalled_tasks/cron_setup.test.sql | 25 +- .../escalate_permanently_stalled.test.sql | 317 ++++++++++ .../assertions_escalation_cron.sql | 59 ++ .../concepts/failures-and-termination.mdx | 14 +- .../deploy/supabase/update-deployed-flows.mdx | 16 +- .../deploy/troubleshooting-stalled-tasks.mdx | 86 ++- 15 files changed, 1339 insertions(+), 32 deletions(-) create mode 100644 .changeset/permanently-stalled-escalation.md create mode 100644 pkgs/core/schemas/0064_function_escalate_permanently_stalled.sql create mode 100644 pkgs/core/supabase/migrations/20261102150832_pgflow_escalate_permanently_stalled.sql create mode 100644 pkgs/core/supabase/tests/fail_task_when_exhausted/permanently_stalled_marker_exhausts.test.sql create mode 100644 pkgs/core/supabase/tests/requeue_stalled_tasks/escalate_permanently_stalled.test.sql create mode 100644 pkgs/core/supabase/upgrade_fixture/assertions_escalation_cron.sql diff --git a/.changeset/permanently-stalled-escalation.md b/.changeset/permanently-stalled-escalation.md new file mode 100644 index 000000000..3bd79619f --- /dev/null +++ b/.changeset/permanently-stalled-escalation.md @@ -0,0 +1,5 @@ +--- +"@pgflow/core": minor +--- + +Terminalize permanently stalled tasks through the step's `whenExhausted` policy. A new `pgflow.escalate_permanently_stalled()` function, scheduled as its own `pgflow_escalate_permanently_stalled` cron job, hands tasks marked with `permanently_stalled_at` to `fail_task()`, which now treats the marker as exhaustion regardless of remaining attempts: the run fails, the step skips or cascades per policy, dependents start, and siblings are cancelled or archived. `attempts_count` stays untouched as committed task claims, and ordinary failure callbacks on a marked task exhaust it instead of requeueing against its archived message. Includes the SQL migration (which preserves an absent or paused recovery cron job and copies the installed recovery interval), generated types, and upgrade guidance with a stamped-backlog inventory query. diff --git a/pkgs/core/schemas/0063_function_setup_requeue_stalled_tasks_cron.sql b/pkgs/core/schemas/0063_function_setup_requeue_stalled_tasks_cron.sql index 10df5292e..be5dba18e 100644 --- a/pkgs/core/schemas/0063_function_setup_requeue_stalled_tasks_cron.sql +++ b/pkgs/core/schemas/0063_function_setup_requeue_stalled_tasks_cron.sql @@ -9,31 +9,56 @@ set search_path = pgflow, cron, pg_temp as $$ declare job_id bigint; + escalate_job_id bigint; begin - -- Remove existing job if any + -- Remove existing jobs if any begin perform cron.unschedule('pgflow_requeue_stalled_tasks'); exception when others then null; end; - -- Schedule the new job + begin + perform cron.unschedule('pgflow_escalate_permanently_stalled'); + exception + when others then null; + end; + + -- Schedule the recovery job job_id := cron.schedule( job_name => 'pgflow_requeue_stalled_tasks', schedule => setup_requeue_stalled_tasks_cron.cron_interval, command => 'select pgflow.requeue_stalled_tasks()' ); - return format('Scheduled pgflow_requeue_stalled_tasks (every %s, job_id=%s)', - setup_requeue_stalled_tasks_cron.cron_interval, job_id); + -- Schedule escalation as its own job so it runs in a separate transaction + -- after recovery commits: recovery locks step_tasks first, while + -- fail_task() (driven by escalation) locks the run and step first (#689). + -- Holding one job's locks across both calls would invert that order + -- against live workers. + escalate_job_id := cron.schedule( + job_name => 'pgflow_escalate_permanently_stalled', + schedule => setup_requeue_stalled_tasks_cron.cron_interval, + command => 'select pgflow.escalate_permanently_stalled()' + ); + + return format( + 'Scheduled pgflow_requeue_stalled_tasks (every %s, job_id=%s) and pgflow_escalate_permanently_stalled (job_id=%s)', + setup_requeue_stalled_tasks_cron.cron_interval, + job_id, + escalate_job_id + ); end; $$; comment on function pgflow.setup_requeue_stalled_tasks_cron(text) is -'Sets up cron job to automatically requeue stalled tasks. -Schedules pgflow_requeue_stalled_tasks at the specified cron_interval (default: 15 seconds). -Replaces existing job if it exists (idempotent). -Returns a confirmation message with job ID.'; +'Sets up cron jobs to automatically recover and terminalize stalled tasks. +Schedules pgflow_requeue_stalled_tasks and pgflow_escalate_permanently_stalled +at the specified cron_interval (default: 15 seconds). +Escalation runs as a separate job so recovery commits before fail_task() +locks are taken (#689). +Replaces existing jobs if they exist (idempotent). +Returns a confirmation message with job IDs.'; -- Automatically set up the cron job when migration runs select pgflow.setup_requeue_stalled_tasks_cron(); diff --git a/pkgs/core/schemas/0064_function_escalate_permanently_stalled.sql b/pkgs/core/schemas/0064_function_escalate_permanently_stalled.sql new file mode 100644 index 000000000..c1d1fd51d --- /dev/null +++ b/pkgs/core/schemas/0064_function_escalate_permanently_stalled.sql @@ -0,0 +1,80 @@ +-- Terminalize permanently stalled tasks through the step's when_exhausted +-- policy (#689). requeue_stalled_tasks() archives the message and stamps +-- permanently_stalled_at but leaves the task nonterminal; this function hands +-- each marked task to fail_task(), which treats the marker as exhaustion and +-- decides whether the run fails, the step skips, dependents start, and +-- siblings are cancelled or archived. +-- +-- Lock order matches fail_task(): lock the run and step first, then the task, +-- so recovery (which locks step_tasks first) and live workers cannot deadlock. +-- The candidate loop is a deliberate exception to the declarative-first SQL +-- style: each candidate needs its own lock-and-recheck sequence before +-- fail_task() runs, which a set operation cannot interleave. +-- Runs as its own cron job, in a separate transaction from recovery, so +-- recovery's locks are released before escalation begins. Safe to repeat: +-- candidates are rechecked under lock, and rows already terminalized by +-- fail_task() lose their nonterminal status and stop matching. +create or replace function pgflow.escalate_permanently_stalled() +returns int +language plpgsql +security definer +set search_path = '' +as $$ +declare + v_candidate record; + v_run_status text; + v_step_status text; + v_task_status text; + v_stalled_at timestamptz; + v_escalated int := 0; +begin + -- Pick up every marked nonterminal task, including marked queued leftovers + -- requeued by older callbacks and rows stamped by earlier deployments + for v_candidate in + select + st.run_id, + st.step_slug, + st.task_index + from pgflow.step_tasks st + where st.status in ('started', 'queued') + and st.permanently_stalled_at is not null + order by st.run_id, st.step_slug, st.task_index + loop + -- Lock and recheck the run and step: if either left 'started', that + -- transition owns the task's terminal state and escalation must not + -- fight it + select r.status, step_state.status into v_run_status, v_step_status + from pgflow.runs r + join pgflow.step_states step_state on step_state.run_id = r.run_id + where r.run_id = v_candidate.run_id + and step_state.step_slug = v_candidate.step_slug + for update of r, step_state; + + if v_run_status = 'started' and v_step_status = 'started' then + -- Lock and recheck the task: recovery or a live callback may have + -- transitioned it while this pass waited for the run and step locks + select st.status, st.permanently_stalled_at + into v_task_status, v_stalled_at + from pgflow.step_tasks st + where st.run_id = v_candidate.run_id + and st.step_slug = v_candidate.step_slug + and st.task_index = v_candidate.task_index + for update; + + if v_task_status in ('started', 'queued') and v_stalled_at is not null then + -- fail_task() treats the marker as exhaustion and applies + -- when_exhausted; attempts_count stays untouched (#689) + perform pgflow.fail_task( + run_id => v_candidate.run_id, + step_slug => v_candidate.step_slug, + task_index => v_candidate.task_index, + error_message => 'permanently stalled: worker did not report after 3 requeues' + ); + v_escalated := v_escalated + 1; + end if; + end if; + end loop; + + return v_escalated; +end; +$$; diff --git a/pkgs/core/schemas/0100_function_fail_task.sql b/pkgs/core/schemas/0100_function_fail_task.sql index 72e73da2e..e72daaeba 100644 --- a/pkgs/core/schemas/0100_function_fail_task.sql +++ b/pkgs/core/schemas/0100_function_fail_task.sql @@ -87,23 +87,37 @@ WITH flow_info AS ( fail_or_retry_task as ( UPDATE pgflow.step_tasks as task SET + -- Exhaustion: attempts_count reached max_attempts, or stalled-task + -- recovery gave up on the task (permanently_stalled_at set). A marked + -- task exhausts regardless of its remaining attempts: its message was + -- archived, so a retry could never be delivered. attempts_count stays + -- untouched; it records committed claims (#689). status = CASE - WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued' + WHEN task.permanently_stalled_at IS NULL + AND 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 + WHEN task.permanently_stalled_at IS NULL + AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL + ELSE now() END, started_at = CASE - WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL + WHEN task.permanently_stalled_at IS NULL + AND 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' + AND ( + task.status = 'started' + -- Marked queued leftovers from older callbacks: terminalize them too. + -- Ordinary unmarked queued tasks stay unchanged (a recovery requeue + -- owns their next attempt). + OR (task.status = 'queued' AND task.permanently_stalled_at IS NOT NULL) + ) RETURNING * ), -- Determine if task exhausted retries and get when_exhausted mode diff --git a/pkgs/core/scripts/run-upgrade-fixture b/pkgs/core/scripts/run-upgrade-fixture index 84c188c7f..7658eb94c 100755 --- a/pkgs/core/scripts/run-upgrade-fixture +++ b/pkgs/core/scripts/run-upgrade-fixture @@ -218,3 +218,62 @@ SQL done echo "upgrade fixture: PASS (0.16.0 slug conflict rollbacks)" + +# ===================================================================== +# Fixture 5: escalation cron scheduling across upgrade states (#689) +# ===================================================================== +# The escalation migration schedules pgflow_escalate_permanently_stalled +# only where scheduled recovery already exists, copies the installed +# recovery interval, and mirrors the recovery job's paused state. Proves: +# 1. active recovery at a custom interval gains an active escalation job +# at that interval, while the recovery job itself is untouched, +# 2. paused (active = false) recovery gains a paused escalation job, +# 3. an intentionally absent recovery job gains no escalation job. +ESCALATION_BASELINE='20261102104928_pgflow_worker_startup_handshake.sql' +escalation_migration=$(ls supabase/migrations/*_pgflow_escalate_permanently_stalled.sql) + +for cron_case in active paused absent; do + start_container "pgflow-upgrade-fixture-escalation-${cron_case}" + + echo "upgrade fixture escalation ${cron_case}: applying supabase baseline schema" + psql_in < atlas/supabase-baseline-schema.sql + + echo "upgrade fixture escalation ${cron_case}: applying pgflow migrations up to ($ESCALATION_BASELINE)" + apply_migrations_up_to "$ESCALATION_BASELINE" + + case "$cron_case" in + active) + psql_in <<'SQL' +SELECT cron.alter_job( + job_id => job.jobid, + schedule => '45 seconds' +) +FROM cron.job job +WHERE job.jobname = 'pgflow_requeue_stalled_tasks'; +SQL + ;; + paused) + psql_in <<'SQL' +SELECT cron.alter_job( + job_id => job.jobid, + schedule => '45 seconds', + active => false +) +FROM cron.job job +WHERE job.jobname = 'pgflow_requeue_stalled_tasks'; +SQL + ;; + absent) + psql_in <<'SQL' +SELECT cron.unschedule('pgflow_requeue_stalled_tasks'); +SQL + ;; + esac + + echo "upgrade fixture escalation ${cron_case}: applying migration $(basename "$escalation_migration")" + psql_in < "$escalation_migration" + + sed "s/__CASE__/${cron_case}/g" supabase/upgrade_fixture/assertions_escalation_cron.sql | psql_in +done + +echo "upgrade fixture: PASS (escalation cron scheduling states)" diff --git a/pkgs/core/src/database-types.ts b/pkgs/core/src/database-types.ts index 737d0c72b..99d1a2d04 100644 --- a/pkgs/core/src/database-types.ts +++ b/pkgs/core/src/database-types.ts @@ -582,6 +582,7 @@ export type Database = { request_id: number }[] } + escalate_permanently_stalled: { Args: never; Returns: number } fail_task: { Args: { error_message: string diff --git a/pkgs/core/supabase/migrations/20261102150832_pgflow_escalate_permanently_stalled.sql b/pkgs/core/supabase/migrations/20261102150832_pgflow_escalate_permanently_stalled.sql new file mode 100644 index 000000000..0c25cd556 --- /dev/null +++ b/pkgs/core/supabase/migrations/20261102150832_pgflow_escalate_permanently_stalled.sql @@ -0,0 +1,550 @@ +-- 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_archived_queues int; +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, through the task's stored queue + -- snapshot (#650) + 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; + +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 + -- Exhaustion: attempts_count reached max_attempts, or stalled-task + -- recovery gave up on the task (permanently_stalled_at set). A marked + -- task exhausts regardless of its remaining attempts: its message was + -- archived, so a retry could never be delivered. attempts_count stays + -- untouched; it records committed claims (#689). + status = CASE + WHEN task.permanently_stalled_at IS NULL + AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued' + ELSE 'failed' + END, + failed_at = CASE + WHEN task.permanently_stalled_at IS NULL + AND task.attempts_count < (SELECT opt_max_attempts FROM config) THEN NULL + ELSE now() + END, + started_at = CASE + WHEN task.permanently_stalled_at IS NULL + AND 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' + -- Marked queued leftovers from older callbacks: terminalize them too. + -- Ordinary unmarked queued tasks stay unchanged (a recovery requeue + -- owns their next attempt). + OR (task.status = 'queued' AND task.permanently_stalled_at IS NOT NULL) + ) + 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, then + -- archive their messages batched per stored queue route (#650); the + -- archive reads the terminalized rows through the CTE. + 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, task.queue_name + ), + archived_messages AS ( + SELECT pgmq.archive( + st.queue_name, + ARRAY_AGG(st.message_id) + ) + FROM skipped_tasks st + WHERE st.message_id IS NOT NULL + GROUP BY st.queue_name + ) + SELECT COUNT(*)::int INTO v_archived_queues + FROM archived_messages; + + -- 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 batched per stored queue route (#650). Lock-order invariant: +-- always lock/update step_tasks before PGMQ queue rows; the archive reads the +-- terminalized rows through the CTE. 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, task.queue_name + ), + archived_messages AS ( + SELECT pgmq.archive( + ct.queue_name, + ARRAY_AGG(ct.message_id) + ) + FROM cancelled_tasks ct + WHERE ct.message_id IS NOT NULL + GROUP BY ct.queue_name + ) + SELECT COUNT(*)::int INTO v_archived_queues + FROM archived_messages; +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 + st.queue_name, + 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.queue_name, + qt.message_id, + qt.calculated_delay + ) + FROM queued_tasks qt + WHERE EXISTS (SELECT 1 FROM queued_tasks) +); + +-- For failed tasks: archive the message, through the task's stored queue +-- snapshot (#650) +PERFORM pgmq.archive(st.queue_name, 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.status = 'failed' + AND st.message_id IS NOT NULL +GROUP BY st.queue_name +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; +$$; +-- Create "escalate_permanently_stalled" function +CREATE FUNCTION "pgflow"."escalate_permanently_stalled" () RETURNS integer LANGUAGE plpgsql SECURITY DEFINER SET "search_path" = '' AS $$ +declare + v_candidate record; + v_run_status text; + v_step_status text; + v_task_status text; + v_stalled_at timestamptz; + v_escalated int := 0; +begin + -- Pick up every marked nonterminal task, including marked queued leftovers + -- requeued by older callbacks and rows stamped by earlier deployments + for v_candidate in + select + st.run_id, + st.step_slug, + st.task_index + from pgflow.step_tasks st + where st.status in ('started', 'queued') + and st.permanently_stalled_at is not null + order by st.run_id, st.step_slug, st.task_index + loop + -- Lock and recheck the run and step: if either left 'started', that + -- transition owns the task's terminal state and escalation must not + -- fight it + select r.status, step_state.status into v_run_status, v_step_status + from pgflow.runs r + join pgflow.step_states step_state on step_state.run_id = r.run_id + where r.run_id = v_candidate.run_id + and step_state.step_slug = v_candidate.step_slug + for update of r, step_state; + + if v_run_status = 'started' and v_step_status = 'started' then + -- Lock and recheck the task: recovery or a live callback may have + -- transitioned it while this pass waited for the run and step locks + select st.status, st.permanently_stalled_at + into v_task_status, v_stalled_at + from pgflow.step_tasks st + where st.run_id = v_candidate.run_id + and st.step_slug = v_candidate.step_slug + and st.task_index = v_candidate.task_index + for update; + + if v_task_status in ('started', 'queued') and v_stalled_at is not null then + -- fail_task() treats the marker as exhaustion and applies + -- when_exhausted; attempts_count stays untouched (#689) + perform pgflow.fail_task( + run_id => v_candidate.run_id, + step_slug => v_candidate.step_slug, + task_index => v_candidate.task_index, + error_message => 'permanently stalled: worker did not report after 3 requeues' + ); + v_escalated := v_escalated + 1; + end if; + end if; + end loop; + + return v_escalated; +end; +$$; +-- Modify "setup_requeue_stalled_tasks_cron" function +CREATE OR REPLACE FUNCTION "pgflow"."setup_requeue_stalled_tasks_cron" ("cron_interval" text DEFAULT '15 seconds') RETURNS text LANGUAGE plpgsql SECURITY DEFINER SET "search_path" = pgflow, cron, pg_temp AS $$ +declare + job_id bigint; + escalate_job_id bigint; +begin + -- Remove existing jobs if any + begin + perform cron.unschedule('pgflow_requeue_stalled_tasks'); + exception + when others then null; + end; + + begin + perform cron.unschedule('pgflow_escalate_permanently_stalled'); + exception + when others then null; + end; + + -- Schedule the recovery job + job_id := cron.schedule( + job_name => 'pgflow_requeue_stalled_tasks', + schedule => setup_requeue_stalled_tasks_cron.cron_interval, + command => 'select pgflow.requeue_stalled_tasks()' + ); + + -- Schedule escalation as its own job so it runs in a separate transaction + -- after recovery commits: recovery locks step_tasks first, while + -- fail_task() (driven by escalation) locks the run and step first (#689). + -- Holding one job's locks across both calls would invert that order + -- against live workers. + escalate_job_id := cron.schedule( + job_name => 'pgflow_escalate_permanently_stalled', + schedule => setup_requeue_stalled_tasks_cron.cron_interval, + command => 'select pgflow.escalate_permanently_stalled()' + ); + + return format( + 'Scheduled pgflow_requeue_stalled_tasks (every %s, job_id=%s) and pgflow_escalate_permanently_stalled (job_id=%s)', + setup_requeue_stalled_tasks_cron.cron_interval, + job_id, + escalate_job_id + ); +end; +$$; +-- Set comment to function: "setup_requeue_stalled_tasks_cron" +COMMENT ON FUNCTION "pgflow"."setup_requeue_stalled_tasks_cron" IS 'Sets up cron jobs to automatically recover and terminalize stalled tasks. +Schedules pgflow_requeue_stalled_tasks and pgflow_escalate_permanently_stalled +at the specified cron_interval (default: 15 seconds). +Escalation runs as a separate job so recovery commits before fail_task() +locks are taken (#689). +Replaces existing jobs if they exist (idempotent). +Returns a confirmation message with job IDs.'; +-- enabled, an operator's custom recovery interval is preserved by copying +-- the installed schedule instead of resetting it, and a paused recovery +-- job stays paused for escalation too. +SELECT cron.schedule( + job_name => 'pgflow_escalate_permanently_stalled', + schedule => ( + SELECT cron.job.schedule + FROM cron.job + WHERE cron.job.jobname = 'pgflow_requeue_stalled_tasks' + ), + command => 'select pgflow.escalate_permanently_stalled()' +) +WHERE EXISTS ( + SELECT 1 FROM cron.job + WHERE cron.job.jobname = 'pgflow_requeue_stalled_tasks' +); + +-- cron.schedule() always creates an active job; mirror the recovery job's +-- paused state onto escalation so an upgrade cannot resume paused recovery +-- through the back door. +SELECT cron.alter_job( + job_id => escalation.jobid, + active => recovery.active +) +FROM cron.job escalation +JOIN cron.job recovery ON recovery.jobname = 'pgflow_requeue_stalled_tasks' +WHERE escalation.jobname = 'pgflow_escalate_permanently_stalled'; diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index 771681f6d..7180018db 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:sOlj3XB6+Tss5pInYvuwe/yxDhipAVGK/ZF49e+gLAo= +h1:/C0cW94KLrjD4rV0Wywgvq251dge+WilvBbGoIgjuF0= 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= @@ -26,3 +26,4 @@ h1:sOlj3XB6+Tss5pInYvuwe/yxDhipAVGK/ZF49e+gLAo= 20260919152659_pgflow_fix_force_skip_multi_queue.sql h1:DC7uQRwjpOm3A4NtDYb3SjbSSqQix4MlQ4+BnhGQ/Q8= 20260920093533_pgflow_telemetry.sql h1:dp3gNRSUpTNoPH9w4n9P7CBhXFoJspQ+hP9MfQ1u3Fg= 20261001194538_pgflow_telemetry_endpoint_fix.sql h1:/ROJiUnHzOt6xsw9f877wY6KlrVOFxpjGBuFuweX6Is= +20261102150832_pgflow_escalate_permanently_stalled.sql h1:6OqZdfyNTMj26Dj5bn8sUvL1Xhw7DbH4KRVJfXYyImk= diff --git a/pkgs/core/supabase/tests/fail_task_when_exhausted/permanently_stalled_marker_exhausts.test.sql b/pkgs/core/supabase/tests/fail_task_when_exhausted/permanently_stalled_marker_exhausts.test.sql new file mode 100644 index 000000000..1d5977e42 --- /dev/null +++ b/pkgs/core/supabase/tests/fail_task_when_exhausted/permanently_stalled_marker_exhausts.test.sql @@ -0,0 +1,91 @@ +-- Test: a non-null permanently_stalled_at makes a nonterminal task exhausted +-- inside fail_task(), regardless of its effective max_attempts (#689 ADR: +-- permanently stalled tasks exhaust recovery without rewriting claim counts). +-- Ordinary failure callbacks must exhaust a marked task rather than requeue it +-- against its archived message. +begin; +select plan(6); + +select pgflow_tests.reset_db(); + +-- Marked started task with attempts left: fail_task must exhaust, not retry +select pgflow.create_flow('fail_flow', max_attempts => 3); +select pgflow.add_step('fail_flow', 'step_a'); +select pgflow.start_flow('fail_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('fail_flow'); +select pgflow_tests.read_and_start('fail_flow', 30, 1); + +update pgflow.step_tasks +set permanently_stalled_at = now(), attempts_count = 1 +where flow_slug = 'fail_flow'; + +select pgflow.fail_task( + (select run_id from pgflow.runs where flow_slug = 'fail_flow'), + 'step_a', + 0, + 'handler error' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'fail_flow'), + 'failed', + 'marked task exhausts even with attempts_count below max_attempts' +); + +select is( + (select attempts_count from pgflow.step_tasks where flow_slug = 'fail_flow'), + 1, + 'attempts_count is preserved as committed task claims' +); + +select is( + (select status from pgflow.runs where flow_slug = 'fail_flow'), + 'failed', + 'marked task failure applies when_exhausted=fail to the run' +); + +-- Marked queued leftover (requeued by an older callback against an archived +-- message): fail_task must terminalize it, not leave it queued +select pgflow.create_flow('leftover_flow', max_attempts => 3); +select pgflow.add_step('leftover_flow', 'step_a'); +select pgflow.start_flow('leftover_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('leftover_flow'); +select pgflow_tests.read_and_start('leftover_flow', 30, 1); + +update pgflow.step_tasks +set + permanently_stalled_at = now(), + status = 'queued', + started_at = null, + attempts_count = 1 +where flow_slug = 'leftover_flow'; + +select pgflow.fail_task( + (select run_id from pgflow.runs where flow_slug = 'leftover_flow'), + 'step_a', + 0, + 'late handler error' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'leftover_flow'), + 'failed', + 'marked queued leftover terminalizes instead of staying queued' +); + +select is( + (select status from pgflow.runs where flow_slug = 'leftover_flow'), + 'failed', + 'marked queued leftover fails its run' +); + +select is( + (select error_message from pgflow.step_tasks where flow_slug = 'leftover_flow'), + 'late handler error', + 'callback error message is preserved' +); + +select finish(); +rollback; diff --git a/pkgs/core/supabase/tests/requeue_stalled_tasks/cron_setup.test.sql b/pkgs/core/supabase/tests/requeue_stalled_tasks/cron_setup.test.sql index c469ae01e..90e5f4f09 100644 --- a/pkgs/core/supabase/tests/requeue_stalled_tasks/cron_setup.test.sql +++ b/pkgs/core/supabase/tests/requeue_stalled_tasks/cron_setup.test.sql @@ -1,6 +1,6 @@ -- Test: setup_requeue_stalled_tasks_cron() function begin; -select plan(6); +select plan(9); select pgflow_tests.reset_db(); @@ -50,5 +50,28 @@ select ok( 'Cron job should call requeue_stalled_tasks' ); +-- Test 7: Escalation job is created alongside the recovery job +select ok( + exists( + select 1 from cron.job + where jobname = 'pgflow_escalate_permanently_stalled' + ), + 'Cron job pgflow_escalate_permanently_stalled should be created' +); + +-- Test 8: Escalation job shares the same schedule +select is( + (select schedule from cron.job where jobname = 'pgflow_escalate_permanently_stalled'), + '30 seconds', + 'Escalation job should follow the configured interval' +); + +-- Test 9: Escalation job command calls escalate_permanently_stalled +select is( + (select command from cron.job where jobname = 'pgflow_escalate_permanently_stalled'), + 'select pgflow.escalate_permanently_stalled()', + 'Escalation job should call escalate_permanently_stalled in its own transaction' +); + select finish(); rollback; diff --git a/pkgs/core/supabase/tests/requeue_stalled_tasks/escalate_permanently_stalled.test.sql b/pkgs/core/supabase/tests/requeue_stalled_tasks/escalate_permanently_stalled.test.sql new file mode 100644 index 000000000..dcf6ee0cf --- /dev/null +++ b/pkgs/core/supabase/tests/requeue_stalled_tasks/escalate_permanently_stalled.test.sql @@ -0,0 +1,317 @@ +-- Test: escalate_permanently_stalled() terminalizes stamped tasks through +-- the step's when_exhausted policy (#689). requeue_stalled_tasks() archives +-- and stamps them; escalation is a separate call that hands each task to +-- fail_task() so the existing exhaustion policy owns every transition. +begin; +select plan(30); + +select pgflow_tests.reset_db(); + +-------------------------------------------------------------------------------- +-- Failure mode: single-step flow terminalizes to failed run +-------------------------------------------------------------------------------- +select pgflow.create_flow('fail_flow', timeout => 5); +select pgflow.add_step('fail_flow', 'step_a'); +select pgflow.start_flow('fail_flow', '{"input": "test"}'::jsonb); + +select pgflow_tests.ensure_worker('fail_flow'); +select pgflow_tests.read_and_start('fail_flow', 30, 1); + +-- Stall the fourth claim past the requeue cap +update pgflow.step_tasks +set + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds', + requeued_count = 3 +where flow_slug = 'fail_flow'; + +select is( + pgflow.requeue_stalled_tasks(), + 0, + 'requeue past the cap returns 0 (archive and stamp, no requeue)' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'fail_flow'), + 'started', + 'task remains started after stamping; escalation is a separate step' +); + +select is( + pgflow.escalate_permanently_stalled(), + 1, + 'escalate terminalizes the permanently stalled task' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'fail_flow'), + 'failed', + 'escalated task becomes failed' +); + +select is( + (select status from pgflow.step_states where flow_slug = 'fail_flow'), + 'failed', + 'step becomes failed through when_exhausted default' +); + +select is( + (select status from pgflow.runs where flow_slug = 'fail_flow'), + 'failed', + 'run becomes failed' +); + +select ok( + (select permanently_stalled_at from pgflow.step_tasks where flow_slug = 'fail_flow') is not null, + 'permanently_stalled_at is retained for investigation' +); + +select is( + (select error_message from pgflow.step_tasks where flow_slug = 'fail_flow'), + 'permanently stalled: worker did not report after 3 requeues', + 'escalation records the permanently stalled error message' +); + +select is( + pgflow.escalate_permanently_stalled(), + 0, + 'repeated escalate call is a no-op' +); + +-------------------------------------------------------------------------------- +-- Skip mode: skipped step starts its dependent +-------------------------------------------------------------------------------- +select pgflow.create_flow('skip_flow', timeout => 5); +select pgflow.add_step('skip_flow', 'process', when_exhausted => 'skip'); +select pgflow.add_step('skip_flow', 'finalize', array['process']); +select pgflow.start_flow('skip_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('skip_flow'); +select pgflow_tests.read_and_start('skip_flow', 30, 1); + +update pgflow.step_tasks +set + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds', + requeued_count = 3 +where flow_slug = 'skip_flow'; + +select is( + (select pgflow.requeue_stalled_tasks() + pgflow.escalate_permanently_stalled()), + 1, + 'skip flow escalates one stalled task' +); + +select is( + (select status from pgflow.step_states where flow_slug = 'skip_flow' and step_slug = 'process'), + 'skipped', + 'skip mode skips the stalled step instead of failing the run' +); + +select is( + (select status from pgflow.step_states where flow_slug = 'skip_flow' and step_slug = 'finalize'), + 'started', + 'dependent step starts after the skipped step' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'skip_flow' and step_slug = 'finalize'), + 'queued', + 'dependent step has a queued task' +); + +select is( + (select status from pgflow.runs where flow_slug = 'skip_flow'), + 'started', + 'run continues after the skipped step' +); + +-------------------------------------------------------------------------------- +-- Skip-cascade mode: dependents are skipped and the run completes +-------------------------------------------------------------------------------- +select pgflow.create_flow('cascade_flow', timeout => 5); +select pgflow.add_step('cascade_flow', 'process', when_exhausted => 'skip-cascade'); +select pgflow.add_step('cascade_flow', 'dependent', array['process']); +select pgflow.start_flow('cascade_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('cascade_flow'); +select pgflow_tests.read_and_start('cascade_flow', 30, 1); + +update pgflow.step_tasks +set + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds', + requeued_count = 3 +where flow_slug = 'cascade_flow'; + +select is( + (select pgflow.requeue_stalled_tasks() + pgflow.escalate_permanently_stalled()), + 1, + 'cascade flow escalates one stalled task' +); + +select is( + (select status from pgflow.step_states where flow_slug = 'cascade_flow' and step_slug = 'dependent'), + 'skipped', + 'skip-cascade skips the dependent step' +); + +select is( + (select status from pgflow.runs where flow_slug = 'cascade_flow'), + 'completed', + 'run completes after the cascade skip' +); + +-------------------------------------------------------------------------------- +-- Map step: stalled task fails, active siblings are skipped +-------------------------------------------------------------------------------- +select pgflow.create_flow('map_flow', timeout => 5); +select pgflow.add_step('map_flow', 'items', step_type => 'map', when_exhausted => 'skip'); +select pgflow.start_flow('map_flow', '[10, 20, 30]'::jsonb); + +select pgflow_tests.ensure_worker('map_flow'); +select pgflow_tests.read_and_start('map_flow', 30, 1); + +update pgflow.step_tasks +set + queued_at = now() - interval '40 seconds', + started_at = now() - interval '36 seconds', + requeued_count = 3 +where flow_slug = 'map_flow' and task_index = 0; + +select is( + (select pgflow.requeue_stalled_tasks() + pgflow.escalate_permanently_stalled()), + 1, + 'map flow escalates one stalled task' +); + +select results_eq( + $$ select status from pgflow.step_tasks + where flow_slug = 'map_flow' order by task_index $$, + $$ values ('failed'), ('skipped'), ('skipped') $$, + 'stalled map task fails while queued siblings are skipped' +); + +select is( + (select status from pgflow.step_states where flow_slug = 'map_flow'), + 'skipped', + 'map step is skipped' +); + +select is( + (select status from pgflow.runs where flow_slug = 'map_flow'), + 'completed', + 'run completes after the map step skip' +); + +-------------------------------------------------------------------------------- +-- Previously stamped rows: rows stamped by earlier deployments escalate too +-------------------------------------------------------------------------------- +select pgflow.create_flow('stale_flow', timeout => 5); +select pgflow.add_step('stale_flow', 'step_a'); +select pgflow.start_flow('stale_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('stale_flow'); +select pgflow_tests.read_and_start('stale_flow', 30, 1); + +-- Simulate a row stamped by an earlier deployment without requeued_count = 3 +update pgflow.step_tasks +set permanently_stalled_at = now() - interval '1 hour' +where flow_slug = 'stale_flow'; + +select is( + pgflow.escalate_permanently_stalled(), + 1, + 'previously stamped row escalates without another requeue sweep' +); + +select is( + (select status from pgflow.runs where flow_slug = 'stale_flow'), + 'failed', + 'previously stamped row fails its run' +); + +select is( + pgflow.escalate_permanently_stalled(), + 0, + 'escalate is safe to repeat after catching up old rows' +); + +-------------------------------------------------------------------------------- +-- Marked queued leftovers: requeued by older callbacks, escalated here +-------------------------------------------------------------------------------- +select pgflow.create_flow('leftover_flow', timeout => 5); +select pgflow.add_step('leftover_flow', 'step_a'); +select pgflow.start_flow('leftover_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('leftover_flow'); +select pgflow_tests.read_and_start('leftover_flow', 30, 1); + +-- Simulate an older callback that requeued the marked task against its +-- archived message +update pgflow.step_tasks +set + permanently_stalled_at = now() - interval '1 hour', + status = 'queued', + started_at = null +where flow_slug = 'leftover_flow'; + +select is( + pgflow.escalate_permanently_stalled(), + 1, + 'marked queued leftover escalates' +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'leftover_flow'), + 'failed', + 'marked queued leftover terminalizes instead of staying queued' +); + +select is( + (select status from pgflow.runs where flow_slug = 'leftover_flow'), + 'failed', + 'marked queued leftover fails its run' +); + +-------------------------------------------------------------------------------- +-- Completion that wins before escalation remains completed +-------------------------------------------------------------------------------- +select pgflow.create_flow('late_flow', timeout => 5); +select pgflow.add_step('late_flow', 'step_a'); +select pgflow.start_flow('late_flow', '{}'::jsonb); + +select pgflow_tests.ensure_worker('late_flow'); +select pgflow_tests.read_and_start('late_flow', 30, 1); + +update pgflow.step_tasks +set permanently_stalled_at = now() +where flow_slug = 'late_flow'; + +select pgflow.complete_task( + (select run_id from pgflow.runs where flow_slug = 'late_flow'), + 'step_a', + 0, + '{"result": "late but valid"}'::jsonb +); + +select is( + (select status from pgflow.step_tasks where flow_slug = 'late_flow'), + 'completed', + 'completion that wins before escalation remains completed' +); + +select is( + (select status from pgflow.runs where flow_slug = 'late_flow'), + 'completed', + 'winning completion completes the run' +); + +select is( + pgflow.escalate_permanently_stalled(), + 0, + 'escalation skips the completed task' +); + +select finish(); +rollback; diff --git a/pkgs/core/supabase/upgrade_fixture/assertions_escalation_cron.sql b/pkgs/core/supabase/upgrade_fixture/assertions_escalation_cron.sql new file mode 100644 index 000000000..5e71bb064 --- /dev/null +++ b/pkgs/core/supabase/upgrade_fixture/assertions_escalation_cron.sql @@ -0,0 +1,59 @@ +-- Assertions for the escalation cron upgrade fixture (#689). +-- __CASE__ expands to active | paused | absent. + +do $$ +declare + v_recovery record; + v_escalation record; +begin + select jobid, schedule, active into v_recovery + from cron.job + where cron.job.jobname = 'pgflow_requeue_stalled_tasks'; + + select jobid, schedule, active into v_escalation + from cron.job + where cron.job.jobname = 'pgflow_escalate_permanently_stalled'; + + -- The recovery job itself is never rescheduled or toggled by the upgrade. + if v_recovery.jobid is null then + if '__CASE__' = 'absent' then + if v_escalation.jobid is not null then + raise exception 'absent recovery: escalation job was scheduled anyway'; + end if; + raise notice 'PASS: absent recovery schedules no escalation job'; + return; + end if; + raise exception '__CASE__: recovery job disappeared during upgrade'; + end if; + + if v_recovery.schedule <> '45 seconds' then + raise exception '__CASE__: recovery schedule changed to %', v_recovery.schedule; + end if; + + if v_escalation.jobid is null then + raise exception '__CASE__: escalation job missing'; + end if; + + if v_escalation.schedule <> v_recovery.schedule then + raise exception '__CASE__: escalation schedule % does not copy recovery %', + v_escalation.schedule, v_recovery.schedule; + end if; + + if '__CASE__' = 'active' then + if not v_recovery.active then + raise exception 'active: recovery job was paused by the upgrade'; + end if; + if not v_escalation.active then + raise exception 'active: escalation job is not active'; + end if; + else + if v_recovery.active then + raise exception 'paused: recovery job was activated by the upgrade'; + end if; + if v_escalation.active then + raise exception 'paused: escalation job is active despite paused recovery'; + end if; + end if; + + raise notice 'PASS: __CASE__ recovery state preserved and mirrored onto escalation'; +end $$; diff --git a/pkgs/website/src/content/docs/concepts/failures-and-termination.mdx b/pkgs/website/src/content/docs/concepts/failures-and-termination.mdx index 251e21ccb..3f88450fe 100644 --- a/pkgs/website/src/content/docs/concepts/failures-and-termination.mdx +++ b/pkgs/website/src/content/docs/concepts/failures-and-termination.mdx @@ -36,7 +36,7 @@ A failed task does not always produce a failed step. After the task exhausts its | `failed` | The task or step encountered a real failure. | | `skipped` | Orchestration decided that the work should not run or continue. | | `cancelled` | A failed run invalidated unfinished task work. | -| `permanently_stalled_at` | Recovery stopped after repeated worker-loss requeues. The task status remains `started`. | +| `permanently_stalled_at` | Recovery stopped after repeated worker-loss requeues. The timestamp stays on the terminalized task for investigation. |