From cb52bb5036fadca2007f04ad0ea2f700d9a1bb0f Mon Sep 17 00:00:00 2001 From: Agent Date: Tue, 1 Sep 2026 21:14:59 +0000 Subject: [PATCH] fix(core): use effective step timeout for stalled tasks Stalled recovery used only the flow timeout, so step overrides could leave tasks started long after their PGMQ messages became visible. Use the same step-over-flow timeout selection as start_tasks while preserving the 30-second recovery buffer, requeue limits, permanent-stall behavior, and the #645 parent-state guards. The PGMQ-only two-second visibility margin remains unchanged. --- .changeset/use-effective-step-timeout.md | 5 + .../0062_function_requeue_stalled_tasks.sql | 14 +- ...02012951_pgflow_effective_step_timeout.sql | 95 +++++++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- .../effective_step_timeout.test.sql | 130 ++++++++++++++++++ 5 files changed, 241 insertions(+), 6 deletions(-) create mode 100644 .changeset/use-effective-step-timeout.md create mode 100644 pkgs/core/supabase/migrations/20260902012951_pgflow_effective_step_timeout.sql create mode 100644 pkgs/core/supabase/tests/requeue_stalled_tasks/effective_step_timeout.test.sql 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;