diff --git a/.changeset/use-effective-step-timeout.md b/.changeset/use-effective-step-timeout.md new file mode 100644 index 000000000..4bd2bd42a --- /dev/null +++ b/.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. diff --git a/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql b/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql index 0b8f6646c..76b2a1b23 100644 --- a/pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql +++ b/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 diff --git a/pkgs/core/supabase/migrations/20260902012951_pgflow_effective_step_timeout.sql b/pkgs/core/supabase/migrations/20260902012951_pgflow_effective_step_timeout.sql new file mode 100644 index 000000000..5a48f475a --- /dev/null +++ b/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; +$$; diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index e85c1bcc1..428ae3147 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/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= diff --git a/pkgs/core/supabase/tests/requeue_stalled_tasks/effective_step_timeout.test.sql b/pkgs/core/supabase/tests/requeue_stalled_tasks/effective_step_timeout.test.sql new file mode 100644 index 000000000..987f2b2f7 --- /dev/null +++ b/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;