From 5668a962c8c9718ad14c0632e4aaeb53c0e43fba Mon Sep 17 00:00:00 2001 From: Agent Date: Fri, 18 Sep 2026 21:34:24 +0000 Subject: [PATCH] fix(core): consume all archived queues on force-skip; reject missing pgmq queue in assert_step_queue_available Two #651 review findings. _cascade_force_skip_steps ended its CTE chain with 'LEFT JOIN archived_messages ON true' feeding a SELECT INTO. SELECT INTO stops after its first row, so the executor shut down the join after the first per-queue pgmq.archive group and every later queue's group never ran: force-skipping a completed ancestor whose children hold queued tasks in separate private step queues left the later queues' messages in place, where they recur indefinitely. The final statement now counts every archived_messages row (same full-consumption pattern as complete_task/fail_task), with the count landing in v_archived_queues; putting the aggregate in an unreferenced join column instead let the planner skip the CTE entirely, so the counted column must stay in the target list. _assert_step_queue_available returned false (route available) for a route owned by this flow's definition even when v_listed was NULL, i.e. the PGMQ queue did not exist. A dropped queue then let ensure_flow_compiled report 'verified' while worker polling would fail, and a silently treated-as-available queue may still hold outstanding task identities. An owned route with no listed queue is now external damage and raises: 'queue "..." owned by flow "..." is not listed in PGMQ'. pgTAP: archives_task_messages_from_all_private_queues (two private queues, both drained and archived), owned_route_requires_listed_queue (dropped queue fails verified startup). Migration 20260918213051_pgflow_fix_force_skip_multi_queue (CREATE OR REPLACE only). --- .changeset/fix-force-skip-multi-queue.md | 5 + ...6_function_assert_step_queue_available.sql | 19 +- ...100_function__cascade_force_skip_steps.sql | 13 +- ...3051_pgflow_fix_force_skip_multi_queue.sql | 206 ++++++++++++++++++ pkgs/core/supabase/migrations/atlas.sum | 3 +- ..._messages_from_all_private_queues.test.sql | 75 +++++++ ...owned_route_requires_listed_queue.test.sql | 41 ++++ 7 files changed, 356 insertions(+), 6 deletions(-) create mode 100644 .changeset/fix-force-skip-multi-queue.md create mode 100644 pkgs/core/supabase/migrations/20260918213051_pgflow_fix_force_skip_multi_queue.sql create mode 100644 pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql create mode 100644 pkgs/core/supabase/tests/queue_mode/owned_route_requires_listed_queue.test.sql diff --git a/.changeset/fix-force-skip-multi-queue.md b/.changeset/fix-force-skip-multi-queue.md new file mode 100644 index 000000000..4a7e94bfe --- /dev/null +++ b/.changeset/fix-force-skip-multi-queue.md @@ -0,0 +1,5 @@ +--- +'@pgflow/core': patch +--- + +Fix force-skip consuming only the first queue's archived messages when active tasks span multiple private step queues, and stop `assert_step_queue_available()` from reporting an owned route as available when its PGMQ queue is missing. diff --git a/pkgs/core/schemas/0076_function_assert_step_queue_available.sql b/pkgs/core/schemas/0076_function_assert_step_queue_available.sql index 792a4cfe0..16769443e 100644 --- a/pkgs/core/schemas/0076_function_assert_step_queue_available.sql +++ b/pkgs/core/schemas/0076_function_assert_step_queue_available.sql @@ -9,7 +9,12 @@ -- - a queue name routed to by another concrete flow's steps, or defaulted -- to by another flow-mode flow (cross-flow reference); -- - an ambiguous case-insensitive match among listed PGMQ queues --- (external damage), even when this flow's definition owns the route. +-- (external damage), even when this flow's definition owns the route; +-- - an owned route whose PGMQ queue is not listed: pgflow never drops an +-- owned queue itself, so something outside pgflow did (a manual +-- pgmq.drop_queue, a customized prune_data_older_than, or a restore that +-- skipped the pgmq tables) — reject instead of verifying a startup whose +-- polling cannot work. -- -- Allows one exact listed queue only when the existing definition of this -- exact flow owns that route (idempotent reuse). A name that is neither @@ -69,11 +74,21 @@ begin -- Owned by an existing definition of this exact flow: reuse idempotently. -- A verified definition owns every derived route, so its one exact listed - -- queue is allowed here. + -- queue is allowed here. An owned route without its listed queue is + -- external damage: reuse would report the route available while polling + -- fails, so reject it instead. if exists ( select 1 from pgflow.steps as s where s.flow_slug = p_flow_slug and s.queue_name = p_queue_name ) then + if v_listed is null then + raise exception + 'queue "%" owned by flow "%" is not listed in PGMQ', + p_queue_name, p_flow_slug + using detail = 'pgflow never drops an owned queue itself: a manual pgmq.drop_queue, a customized prune_data_older_than, or a restore that skipped the pgmq tables did. Historical task rows still reference message ids from the dropped queue.', + hint = 'Check why the queue disappeared; if the loss is intended, drop the flow definition and recompile it fresh.'; + end if; + return false; end if; diff --git a/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql b/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql index 8762c42ae..fc2850264 100644 --- a/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql +++ b/pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql @@ -11,6 +11,7 @@ as $$ DECLARE v_flow_slug text; v_total_skipped int := 0; + v_archived_queues int; BEGIN -- Get flow_slug for this run SELECT r.flow_slug INTO v_flow_slug @@ -122,10 +123,16 @@ BEGIN WHERE r.run_id = _cascade_force_skip_steps.run_id AND skipped_count.count > 0 ) - SELECT skipped_count.count - INTO v_total_skipped + -- Consume every archived_messages row (COUNT(*)) in the same statement: + -- SELECT INTO stops after its first row, and a direct LEFT JOIN of the + -- CTE would leave every later queue's pgmq.archive group unevaluated, so + -- its messages would recur indefinitely. The counted column lands in + -- v_archived_queues the same way the other terminal cleanup functions + -- force their archive CTE to run. + SELECT skipped_count.count, archived_count.count + INTO v_total_skipped, v_archived_queues FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count - LEFT JOIN archived_messages ON true; + LEFT JOIN (SELECT COUNT(*) AS count FROM archived_messages) archived_count ON true; RETURN v_total_skipped; END; diff --git a/pkgs/core/supabase/migrations/20260918213051_pgflow_fix_force_skip_multi_queue.sql b/pkgs/core/supabase/migrations/20260918213051_pgflow_fix_force_skip_multi_queue.sql new file mode 100644 index 000000000..7134f28dd --- /dev/null +++ b/pkgs/core/supabase/migrations/20260918213051_pgflow_fix_force_skip_multi_queue.sql @@ -0,0 +1,206 @@ +-- Modify "_assert_step_queue_available" function +CREATE OR REPLACE FUNCTION "pgflow"."_assert_step_queue_available" ("p_flow_slug" text, "p_queue_name" text) RETURNS boolean LANGUAGE plpgsql SET "search_path" = '' AS $$ +declare + v_owner_flow_slug text; + v_listed text[]; +begin + -- A name derived or referenced by another concrete flow is rejected + select s.flow_slug into v_owner_flow_slug + from pgflow.steps as s + where s.queue_name = p_queue_name + and lower(s.flow_slug) <> lower(p_flow_slug) + limit 1; + + if v_owner_flow_slug is null then + select f.flow_slug into v_owner_flow_slug + from pgflow.flows as f + where f.queue_mode = 'flow' + and lower(f.flow_slug) = p_queue_name + and lower(f.flow_slug) <> lower(p_flow_slug) + limit 1; + end if; + + if v_owner_flow_slug is not null then + raise exception + 'cannot create flow "%": queue "%" is already used by another flow ("%")', + p_flow_slug, p_queue_name, v_owner_flow_slug + using detail = 'Generated per-step queue names must belong to exactly one concrete flow.', + hint = 'Use a different concrete flow slug, or drop the conflicting definition.'; + end if; + + -- Ambiguous normalized matches among listed queues are external damage + select array_agg(listed.queue_name order by listed.queue_name) + into v_listed + from pgmq.list_queues() as listed + where lower(listed.queue_name) = p_queue_name; + + if v_listed is not null and cardinality(v_listed) > 1 then + raise exception + 'queue "%" matches multiple listed PGMQ queues (%)', + p_queue_name, v_listed + using detail = 'An ambiguous case-insensitive match is external damage.', + hint = 'Resolve the duplicate queue spellings manually, then retry.'; + end if; + + -- Owned by an existing definition of this exact flow: reuse idempotently. + -- A verified definition owns every derived route, so its one exact listed + -- queue is allowed here. An owned route without its listed queue is + -- external damage: reuse would report the route available while polling + -- fails, so reject it instead. + if exists ( + select 1 from pgflow.steps as s + where s.flow_slug = p_flow_slug and s.queue_name = p_queue_name + ) then + if v_listed is null then + raise exception + 'queue "%" owned by flow "%" is not listed in PGMQ', + p_queue_name, p_flow_slug + using detail = 'An owned route whose queue is missing may still hold outstanding task identities.', + hint = 'Recreate the queue, or drop and recompile the flow definition.'; + end if; + + return false; + end if; + + if v_listed is not null then + raise exception + 'cannot create flow "%": queue "%" is already listed in PGMQ and not owned by this flow', + p_flow_slug, p_queue_name + using detail = 'A missing definition must not adopt an already listed queue.', + hint = 'Drop the conflicting queue or use a different concrete flow slug.'; + end if; + + return true; +end; +$$; +-- Modify "_cascade_force_skip_steps" function +CREATE OR REPLACE FUNCTION "pgflow"."_cascade_force_skip_steps" ("run_id" uuid, "step_slug" text, "skip_reason" text) RETURNS integer LANGUAGE plpgsql AS $$ +DECLARE + v_flow_slug text; + v_total_skipped int := 0; + v_archived_queues int; +BEGIN + -- Get flow_slug for this run + SELECT r.flow_slug INTO v_flow_slug + FROM pgflow.runs r + WHERE r.run_id = _cascade_force_skip_steps.run_id; + + IF v_flow_slug IS NULL THEN + RAISE EXCEPTION 'Run not found: %', _cascade_force_skip_steps.run_id; + END IF; + + -- ========================================== + -- SKIP STEPS IN TOPOLOGICAL ORDER + -- ========================================== + -- Use recursive CTE to find all downstream dependents, + -- then skip them in topological order (by step_index) + WITH RECURSIVE + -- ---------- Find all downstream steps ---------- + downstream_steps AS ( + -- Base case: the trigger step + SELECT + s.flow_slug, + s.step_slug, + s.step_index, + _cascade_force_skip_steps.skip_reason AS reason -- Original reason for trigger step + FROM pgflow.steps s + WHERE s.flow_slug = v_flow_slug + AND s.step_slug = _cascade_force_skip_steps.step_slug + + UNION ALL + + -- Recursive case: steps that depend on already-found steps + SELECT + s.flow_slug, + s.step_slug, + s.step_index, + 'dependency_skipped'::text AS reason -- Downstream steps get this reason + FROM pgflow.steps s + JOIN pgflow.deps d ON d.flow_slug = s.flow_slug AND d.step_slug = s.step_slug + JOIN downstream_steps ds ON ds.flow_slug = d.flow_slug AND ds.step_slug = d.dep_slug + ), + -- ---------- Deduplicate and order by step_index ---------- + steps_to_skip AS ( + SELECT DISTINCT ON (ds.step_slug) + ds.flow_slug, + ds.step_slug, + ds.step_index, + ds.reason + FROM downstream_steps ds + ORDER BY ds.step_slug, ds.step_index -- Keep first occurrence (trigger step has original reason) + ), + -- ---------- Skip the steps ---------- + skipped AS ( + UPDATE pgflow.step_states ss + SET status = 'skipped', + skip_reason = sts.reason, + skipped_at = now(), + remaining_tasks = NULL -- Clear remaining_tasks for skipped steps + FROM steps_to_skip sts + WHERE ss.run_id = _cascade_force_skip_steps.run_id + AND ss.step_slug = sts.step_slug + AND ss.status IN ('created', 'started') -- Only skip non-terminal steps + RETURNING + ss.*, + -- Broadcast step:skipped event + realtime.send( + jsonb_build_object( + 'event_type', 'step:skipped', + 'run_id', ss.run_id, + 'flow_slug', ss.flow_slug, + 'step_slug', ss.step_slug, + 'status', 'skipped', + 'skip_reason', ss.skip_reason, + 'skipped_at', ss.skipped_at + ), + concat('step:', ss.step_slug, ':skipped'), + concat('pgflow:run:', ss.run_id), + 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, task.queue_name + ), + -- ---------- Archive queued/started task messages for skipped steps ---------- + -- Batched per stored queue route (#650) + archived_messages AS ( + SELECT pgmq.archive( + task.queue_name, + ARRAY_AGG(task.message_id) + ) as result + FROM skipped_tasks AS task + WHERE task.message_id IS NOT NULL + GROUP BY task.queue_name + HAVING COUNT(task.message_id) > 0 + ), + -- ---------- Update run counters ---------- + run_updates AS ( + UPDATE pgflow.runs r + SET remaining_steps = r.remaining_steps - skipped_count.count + FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count + WHERE r.run_id = _cascade_force_skip_steps.run_id + AND skipped_count.count > 0 + ) + -- Consume every archived_messages row (COUNT(*)) in the same statement: + -- SELECT INTO stops after its first row, and a direct LEFT JOIN of the + -- CTE would leave every later queue's pgmq.archive group unevaluated, so + -- its messages would recur indefinitely. The counted column lands in + -- v_archived_queues the same way the other terminal cleanup functions + -- force their archive CTE to run. + SELECT skipped_count.count, archived_count.count + INTO v_total_skipped, v_archived_queues + FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count + LEFT JOIN (SELECT COUNT(*) AS count FROM archived_messages) archived_count ON true; + + RETURN v_total_skipped; +END; +$$; diff --git a/pkgs/core/supabase/migrations/atlas.sum b/pkgs/core/supabase/migrations/atlas.sum index c9ffbd276..d5c85d5ca 100644 --- a/pkgs/core/supabase/migrations/atlas.sum +++ b/pkgs/core/supabase/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:KsVAXOPvkHDCj18n4kgWtwuS/HMiSC1dVMnNor++gpA= +h1:mC3ikq4kkuBk1OJJeDAbv7aTAymhZljD1BYta6qu/f8= 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= @@ -23,3 +23,4 @@ h1:KsVAXOPvkHDCj18n4kgWtwuS/HMiSC1dVMnNor++gpA= 20260904095427_pgflow_task_lifecycle_hardening.sql h1:27b0BfBcQxeu5XSqVtQYvDCRzTsvLqzS/5hx14/2VyM= 20260907082520_pgflow_remove_legacy_flow_compilation.sql h1:LNFDz+ZZlWb19FmWNPK57eiD+FXySMbStVij8MTSvDw= 20260915074120_pgflow_private_step_queues.sql h1:+vsfsOyDaM8WISO/jxp4UBlTzPuQhg6RwBCiOAk5YAE= +20260918213051_pgflow_fix_force_skip_multi_queue.sql h1:ibzV9U4G2t9K9T85/2HY996C/p1VZtDt0q9mcU60LpQ= diff --git a/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql b/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql new file mode 100644 index 000000000..60af1f2d7 --- /dev/null +++ b/pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql @@ -0,0 +1,75 @@ +\set ON_ERROR_STOP on +\set QUIET on + +-- Force-skip across multiple private step queues (#651 review regression): +-- a completed ancestor whose two children hold queued tasks in two separate +-- private step queues must archive every queue's messages. The final +-- SELECT INTO must not stop after the first archived queue row, or the +-- second queue's message recurs indefinitely. +begin; +select plan(5); + +select pgflow_tests.reset_db(); + +select pgflow.ensure_flow_compiled( + 'fskipmulti', + '{ + "steps": [ + {"slug": "gate", "stepType": "single", "dependencies": [], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}}, + {"slug": "left", "stepType": "single", "dependencies": ["gate"], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}}, + {"slug": "right", "stepType": "single", "dependencies": ["gate"], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}} + ] + }'::jsonb, + 'step' +); + +select run_id as gate_run_id from pgflow.start_flow('fskipmulti', '{}') \gset + +-- Claim the root task from its private queue and complete it +select pgflow_tests.ensure_worker('fskipmulti__gate'); +select array_agg(msg_id) as gate_ids +from pgmq.read_with_poll('fskipmulti__gate', 30, 1, 1, 50) \gset +select pgflow.start_tasks( + 'fskipmulti', + :'gate_ids'::bigint[], + '11111111-1111-1111-1111-111111111111'::uuid, + 'fskipmulti__gate', + 'gate' +); +select pgflow.complete_task(:'gate_run_id'::uuid, 'gate', 0, '{}'::jsonb); + +select is( + (select count(*) from pgflow.step_tasks + where run_id = :'gate_run_id'::uuid and status = 'queued'), + 2::bigint, + 'Setup: both children dispatched a queued task' +); +select is( + (select (select count(*) from pgmq.q_fskipmulti__left) + + (select count(*) from pgmq.q_fskipmulti__right)), + 2::bigint, + 'Setup: one active message in each private child queue' +); + +-- Force-skip the completed ancestor; the cascade skips both children +select pgflow._cascade_force_skip_steps(:'gate_run_id'::uuid, 'gate', 'condition_unmet'); + +select is( + (select count(*) from pgmq.q_fskipmulti__left), + 0::bigint, + 'left child message left its private queue' +); +select is( + (select count(*) from pgmq.q_fskipmulti__right), + 0::bigint, + 'right child message left its private queue' +); +select is( + (select (select count(*) from pgmq.a_fskipmulti__left) + + (select count(*) from pgmq.a_fskipmulti__right)), + 2::bigint, + 'both child messages archived in their private queues' +); + +select * from finish(); +rollback; diff --git a/pkgs/core/supabase/tests/queue_mode/owned_route_requires_listed_queue.test.sql b/pkgs/core/supabase/tests/queue_mode/owned_route_requires_listed_queue.test.sql new file mode 100644 index 000000000..eaf5c352f --- /dev/null +++ b/pkgs/core/supabase/tests/queue_mode/owned_route_requires_listed_queue.test.sql @@ -0,0 +1,41 @@ +\set ON_ERROR_STOP on +\set QUIET on + +-- Owned-route preflight (#651 review regression): a route owned by an +-- existing definition of this exact flow is reusable only when its PGMQ +-- queue is actually listed. A dropped queue must fail verification instead +-- of reporting verified while polling would fail. +begin; +select plan(2); + +select pgflow_tests.reset_db(); + +select pgflow.ensure_flow_compiled( + 'ownedqueue', + '{"steps": [{"slug": "only", "stepType": "single", "dependencies": [], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}}]}'::jsonb, + 'step' +); + +select is( + (select count(*) from pgmq.list_queues() where queue_name = 'ownedqueue__only'), + 1::bigint, + 'Setup: the owned route has a listed PGMQ queue' +); + +-- External damage: the queue is dropped while the definition still owns the route +select pgmq.drop_queue('ownedqueue__only'); + +select throws_ok( + $$ + select pgflow.ensure_flow_compiled( + 'ownedqueue', + '{"steps": [{"slug": "only", "stepType": "single", "dependencies": [], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}}]}'::jsonb, + 'step' + ) + $$, + 'queue "ownedqueue__only" owned by flow "ownedqueue" is not listed in PGMQ', + 'a verified startup rejects an owned route whose PGMQ queue is missing' +); + +select * from finish(); +rollback;