fix(core): use effective step timeout for stalled tasks

.changeset/use-effective-step-timeout.md

@@ -0,0 +1,5 @@
+---
+"@pgflow/core": patch
+---
+
+Requeue stalled tasks using the effective step timeout instead of waiting for the flow timeout.

pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql

@@ -1,4 +1,6 @@
--- Requeue stalled tasks that have been in 'started' status longer than their timeout + 30s buffer
+-- Requeue stalled tasks that have been in 'started' status longer than their effective
+-- timeout (step override with flow fallback) + 30s buffer. This matches the effective
+-- timeout used by start_tasks() for PGMQ visibility, without its +2s visibility margin.
 -- This handles tasks that got stuck when workers crashed without completing them
 create or replace function pgflow.requeue_stalled_tasks()
 returns int
@@ -10,7 +12,7 @@ declare
   result_count int := 0;
   max_requeues constant int := 3;
 begin
-  -- Find and requeue stalled tasks (where started_at > timeout + 30s buffer)
+  -- Find and requeue stalled tasks (where started_at > effective timeout + 30s buffer)
   -- Tasks with requeued_count >= max_requeues will have their message archived
   -- but status left as 'started' for easy identification via requeued_count column
   -- Eligibility requires the parent run AND parent step to still be 'started':
@@ -22,17 +24,19 @@ begin
       st.task_index,
       st.message_id,
       r.flow_slug,
-      st.requeued_count,
-      f.opt_timeout
+      st.requeued_count
     from pgflow.step_tasks st
     join pgflow.runs r on r.run_id = st.run_id
     join pgflow.step_states ss on ss.run_id = st.run_id and ss.step_slug = st.step_slug
     join pgflow.flows f on f.flow_slug = r.flow_slug
+    join pgflow.steps s on s.flow_slug = r.flow_slug and s.step_slug = st.step_slug
     where st.status = 'started'
       and r.status = 'started'
       and ss.status = 'started'
       and st.permanently_stalled_at is null
-      and st.started_at < now() - (f.opt_timeout * interval '1 second') - interval '30 seconds'
+      and st.started_at < now()
+        - (coalesce(s.opt_timeout, f.opt_timeout) * interval '1 second')
+        - interval '30 seconds'
     for update of st skip locked
   ),
   -- Separate tasks that can be requeued from those that exceeded max requeues

pkgs/core/supabase/migrations/20260902012951_pgflow_effective_step_timeout.sql

@@ -0,0 +1,95 @@
+-- Modify "requeue_stalled_tasks" function
+CREATE OR REPLACE FUNCTION "pgflow"."requeue_stalled_tasks" () RETURNS integer LANGUAGE plpgsql SECURITY DEFINER SET "search_path" = '' AS $$
+declare
+  result_count int := 0;
+  max_requeues constant int := 3;
+begin
+  -- Find and requeue stalled tasks (where started_at > effective timeout + 30s buffer)
+  -- Tasks with requeued_count >= max_requeues will have their message archived
+  -- but status left as 'started' for easy identification via requeued_count column
+  -- Eligibility requires the parent run AND parent step to still be 'started':
+  -- stale rows on failed runs or terminal steps must not be revived (#645).
+  with stalled_tasks as (
+    select
+      st.run_id,
+      st.step_slug,
+      st.task_index,
+      st.message_id,
+      r.flow_slug,
+      st.requeued_count
+    from pgflow.step_tasks st
+    join pgflow.runs r on r.run_id = st.run_id
+    join pgflow.step_states ss on ss.run_id = st.run_id and ss.step_slug = st.step_slug
+    join pgflow.flows f on f.flow_slug = r.flow_slug
+    join pgflow.steps s on s.flow_slug = r.flow_slug and s.step_slug = st.step_slug
+    where st.status = 'started'
+      and r.status = 'started'
+      and ss.status = 'started'
+      and st.permanently_stalled_at is null
+      and st.started_at < now()
+        - (coalesce(s.opt_timeout, f.opt_timeout) * interval '1 second')
+        - interval '30 seconds'
+    for update of st skip locked
+  ),
+  -- Separate tasks that can be requeued from those that exceeded max requeues
+  to_requeue as (
+    select * from stalled_tasks where requeued_count < max_requeues
+  ),
+  to_archive as (
+    select * from stalled_tasks where requeued_count >= max_requeues
+  ),
+  -- Update tasks that will be requeued
+  requeued as (
+    update pgflow.step_tasks st
+    set
+      status = 'queued',
+      started_at = null,
+      last_worker_id = null,
+      requeued_count = st.requeued_count + 1,
+      last_requeued_at = now()
+    from to_requeue tr
+    where st.run_id = tr.run_id
+      and st.step_slug = tr.step_slug
+      and st.task_index = tr.task_index
+    returning tr.flow_slug as queue_name, tr.message_id
+  ),
+  -- Make requeued messages visible immediately (batched per queue)
+  visibility_reset as (
+    select pgflow.set_vt_batch(
+      r.queue_name,
+      array_agg(r.message_id),
+      array_agg(0)  -- all offsets are 0 (immediate visibility)
+    )
+    from requeued r
+    where r.message_id is not null
+    group by r.queue_name
+  ),
+  -- Mark tasks as permanently stalled before archiving
+  mark_permanently_stalled as (
+    update pgflow.step_tasks st
+    set permanently_stalled_at = now()
+    from to_archive ta
+    where st.run_id = ta.run_id
+      and st.step_slug = ta.step_slug
+      and st.task_index = ta.task_index
+    returning st.run_id
+  ),
+  -- Archive messages for tasks that exceeded max requeues (batched per queue)
+  archived as (
+    select pgmq.archive(ta.flow_slug, array_agg(ta.message_id))
+    from to_archive ta
+    where ta.message_id is not null
+    group by ta.flow_slug
+  ),
+  -- Force execution of visibility_reset CTE
+  _vr as (select count(*) from visibility_reset),
+  -- Force execution of mark_permanently_stalled CTE
+  _mps as (select count(*) from mark_permanently_stalled),
+  -- Force execution of archived CTE
+  _ar as (select count(*) from archived)
+  select count(*) into result_count
+  from requeued, _vr, _mps, _ar;
+
+  return result_count;
+end;
+$$;

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:vahrstrzyG/2HD7neJ3HHsQNmR8Mj4LBcp5Ik/HGebc=
+h1:7L7TQAgkuEri2ucryxVs4mSra1W+DJ4Ezl4jehNrfDk=
 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=
@@ -22,3 +22,4 @@ h1:vahrstrzyG/2HD7neJ3HHsQNmR8Mj4LBcp5Ik/HGebc=
 20260607175525_pgflow_worker_start_mode.sql h1:PFAfoGaHe5stKF7YAFg6AqBxmRisqDvV60vVpnnVdBE=
 20260827180017_pgflow_terminalize_skipped_tasks.sql h1:Aq4zYSUp707UiDxi08FQeXlRw2lkIonL8O6yi1LfU/g=
 20260902005317_pgflow_failed_run_terminalization.sql h1:agqIyQVyIS95DY8PI9QWg0KyJ3kGVMGrhQ+02FKZ8Xo=
+20260902012951_pgflow_effective_step_timeout.sql h1:+7sOuiyHB3IG0OxdjfZj1/Ush4fy5P2/tp8DASG85lE=

pkgs/core/supabase/tests/requeue_stalled_tasks/effective_step_timeout.test.sql

@@ -0,0 +1,130 @@
+-- Test: requeue_stalled_tasks uses the effective step timeout
+-- Effective timeout: coalesce(step.opt_timeout, flow.opt_timeout)
+-- Stalled when started_at is strictly older than effective timeout + 30s buffer
+begin;
+select plan(10);
+
+select pgflow_tests.reset_db();
+
+-- ==========================================
+-- Case 1: step timeout 5 overrides flow timeout 60
+-- ==========================================
+select pgflow.create_flow('step_short', null, null, 60);
+select pgflow.add_step('step_short', 'step_a', timeout => 5);
+
+select pgflow.start_flow('step_short', '"x"'::jsonb);
+select pgflow_tests.ensure_worker('step_short');
+select pgflow_tests.read_and_start('step_short', 30, 1);
+
+-- Exactly 35s old: boundary of 5s + 30s buffer, not strictly older
+update pgflow.step_tasks
+set queued_at = now() - interval '39 seconds',
+    started_at = now() - interval '35 seconds'
+where flow_slug = 'step_short';
+
+select is(
+  pgflow.requeue_stalled_tasks(),
+  0,
+  'step 5 / flow 60: exactly 35s old is not requeued (strict boundary)'
+);
+
+select is(
+  (select status from pgflow.step_tasks where flow_slug = 'step_short'),
+  'started',
+  'step 5 / flow 60: boundary task stays started'
+);
+
+-- 36s old: strictly past effective timeout 5s + 30s buffer
+update pgflow.step_tasks
+set queued_at = now() - interval '40 seconds',
+    started_at = now() - interval '36 seconds'
+where flow_slug = 'step_short';
+
+select is(
+  pgflow.requeue_stalled_tasks(),
+  1,
+  'step 5 / flow 60: 36s old requeued per effective step timeout'
+);
+
+select is(
+  (select status from pgflow.step_tasks where flow_slug = 'step_short'),
+  'queued',
+  'step 5 / flow 60: 36s old task becomes queued'
+);
+
+-- ==========================================
+-- Case 2: step timeout 60 overrides flow timeout 5
+-- ==========================================
+select pgflow.create_flow('step_long', null, null, 5);
+select pgflow.add_step('step_long', 'step_a', timeout => 60);
+
+select pgflow.start_flow('step_long', '"x"'::jsonb);
+select pgflow_tests.ensure_worker('step_long');
+select pgflow_tests.read_and_start('step_long', 30, 1);
+
+-- 36s old: past flow timeout 5s + 30s buffer, but effective timeout is 60s
+update pgflow.step_tasks
+set queued_at = now() - interval '40 seconds',
+    started_at = now() - interval '36 seconds'
+where flow_slug = 'step_long';
+
+select is(
+  pgflow.requeue_stalled_tasks(),
+  0,
+  'step 60 / flow 5: 36s old not requeued despite flow timeout 5s'
+);
+
+select is(
+  (select status from pgflow.step_tasks where flow_slug = 'step_long'),
+  'started',
+  'step 60 / flow 5: 36s old task stays started'
+);
+
+-- 91s old: strictly past effective timeout 60s + 30s buffer
+update pgflow.step_tasks
+set queued_at = now() - interval '95 seconds',
+    started_at = now() - interval '91 seconds'
+where flow_slug = 'step_long';
+
+select is(
+  pgflow.requeue_stalled_tasks(),
+  1,
+  'step 60 / flow 5: 91s old requeued per effective step timeout'
+);
+
+select is(
+  (select status from pgflow.step_tasks where flow_slug = 'step_long'),
+  'queued',
+  'step 60 / flow 5: 91s old task becomes queued'
+);
+
+-- ==========================================
+-- Case 3: null step timeout inherits flow timeout 5
+-- ==========================================
+select pgflow.create_flow('step_null', null, null, 5);
+select pgflow.add_step('step_null', 'step_a');
+
+select pgflow.start_flow('step_null', '"x"'::jsonb);
+select pgflow_tests.ensure_worker('step_null');
+select pgflow_tests.read_and_start('step_null', 30, 1);
+
+-- 36s old: effective timeout is inherited flow timeout 5s + 30s buffer
+update pgflow.step_tasks
+set queued_at = now() - interval '40 seconds',
+    started_at = now() - interval '36 seconds'
+where flow_slug = 'step_null';
+
+select is(
+  pgflow.requeue_stalled_tasks(),
+  1,
+  'null step timeout inherits flow 5: 36s old requeued'
+);
+
+select is(
+  (select status from pgflow.step_tasks where flow_slug = 'step_null'),
+  'queued',
+  'null step timeout inherits flow 5: 36s old task becomes queued'
+);
+
+select finish();
+rollback;