Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs

.changeset/terminalize-skipped-tasks.md

@@ -0,0 +1,9 @@
+---
+"@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.
+
+Tasks are now terminalized before their queue messages are archived, preserving the task-before-queue lock order, and `start_tasks` only returns rows it actually claimed, so workers no longer execute tasks a concurrent skip already marked `skipped`.
+
+Fixes #638

NOMENCLATURE_GUIDE.md

@@ -142,13 +142,15 @@ 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
 
 - `queued` - Task queued in PGMQ
 - `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
 

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')

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 (

pkgs/core/schemas/0100_function_fail_task.sql

@@ -18,6 +18,7 @@ DECLARE
   v_flow_slug_for_deps text;
   v_prev_step_status text;
   v_flow_slug text;
+  v_skipped_message_ids bigint[];
 begin
 
 -- If run is already failed, no retries allowed
@@ -67,7 +68,7 @@ IF v_prev_step_status IS NOT NULL AND v_prev_step_status != 'started' THEN
     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
@@ -214,16 +215,27 @@ 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;
+  -- 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(

pkgs/core/schemas/0120_function_start_tasks.sql

@@ -8,7 +8,7 @@ volatile
 set search_path to ''
 language sql
 as $$
-  with tasks as (
+  with task_candidates as (
     select
       task.flow_slug,
       task.run_id,
@@ -29,17 +29,27 @@ as $$
           and ss.status = 'started'
       )
   ),
-  start_tasks_update as (
+  -- Claim rows with a guarded update and return only what was actually
+  -- claimed. A concurrent skip can win the row lock between the candidate
+  -- select and this update; the status = 'queued' recheck then claims nothing,
+  -- so no stale candidate row must escape to the worker (#638).
+  tasks as (
     update pgflow.step_tasks
     set
       attempts_count = attempts_count + 1,
       status = 'started',
       started_at = now(),
       last_worker_id = worker_id
-    from tasks
-    where step_tasks.message_id = tasks.message_id
-      and step_tasks.flow_slug = tasks.flow_slug
+    from task_candidates as candidate
+    where step_tasks.message_id = candidate.message_id
+      and step_tasks.flow_slug = candidate.flow_slug
       and step_tasks.status = 'queued'
+    returning
+      step_tasks.flow_slug,
+      step_tasks.run_id,
+      step_tasks.step_slug,
+      step_tasks.task_index,
+      step_tasks.message_id
   ),
   runs as (
     select