Skip to content

Commit dea4643

Browse files
author
Agent
committed
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).
1 parent bc1b81c commit dea4643

7 files changed

Lines changed: 354 additions & 6 deletions

‎.changeset/fix-force-skip-multi-queue.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@pgflow/core': patch
3+
---
4+
5+
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.

‎pkgs/core/schemas/0076_function_assert_step_queue_available.sql‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,10 @@
99
-- - a queue name routed to by another concrete flow's steps, or defaulted
1010
-- to by another flow-mode flow (cross-flow reference);
1111
-- - an ambiguous case-insensitive match among listed PGMQ queues
12-
-- (external damage), even when this flow's definition owns the route.
12+
-- (external damage), even when this flow's definition owns the route;
13+
-- - an owned route whose PGMQ queue is not listed (external damage), so a
14+
-- queue that may contain outstanding task identities is never silently
15+
-- treated as available.
1316
--
1417
-- Allows one exact listed queue only when the existing definition of this
1518
-- exact flow owns that route (idempotent reuse). A name that is neither
@@ -69,11 +72,21 @@ begin
6972

7073
-- Owned by an existing definition of this exact flow: reuse idempotently.
7174
-- A verified definition owns every derived route, so its one exact listed
72-
-- queue is allowed here.
75+
-- queue is allowed here. An owned route without its listed queue is
76+
-- external damage: reuse would report the route available while polling
77+
-- fails, so reject it instead.
7378
if exists (
7479
select 1 from pgflow.steps as s
7580
where s.flow_slug = p_flow_slug and s.queue_name = p_queue_name
7681
) then
82+
if v_listed is null then
83+
raise exception
84+
'queue "%" owned by flow "%" is not listed in PGMQ',
85+
p_queue_name, p_flow_slug
86+
using detail = 'An owned route whose queue is missing may still hold outstanding task identities.',
87+
hint = 'Recreate the queue, or drop and recompile the flow definition.';
88+
end if;
89+
7790
return false;
7891
end if;
7992

‎pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ as $$
1111
DECLARE
1212
v_flow_slug text;
1313
v_total_skipped int := 0;
14+
v_archived_queues int;
1415
BEGIN
1516
-- Get flow_slug for this run
1617
SELECT r.flow_slug INTO v_flow_slug
@@ -122,10 +123,16 @@ BEGIN
122123
WHERE r.run_id = _cascade_force_skip_steps.run_id
123124
AND skipped_count.count > 0
124125
)
125-
SELECT skipped_count.count
126-
INTO v_total_skipped
126+
-- Consume every archived_messages row (COUNT(*)) in the same statement:
127+
-- SELECT INTO stops after its first row, and a direct LEFT JOIN of the
128+
-- CTE would leave every later queue's pgmq.archive group unevaluated, so
129+
-- its messages would recur indefinitely. The counted column lands in
130+
-- v_archived_queues the same way the other terminal cleanup functions
131+
-- force their archive CTE to run.
132+
SELECT skipped_count.count, archived_count.count
133+
INTO v_total_skipped, v_archived_queues
127134
FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count
128-
LEFT JOIN archived_messages ON true;
135+
LEFT JOIN (SELECT COUNT(*) AS count FROM archived_messages) archived_count ON true;
129136

130137
RETURN v_total_skipped;
131138
END;

‎pkgs/core/supabase/migrations/20260918213051_pgflow_fix_force_skip_multi_queue.sql‎

Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
-- Modify "_assert_step_queue_available" function
2+
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 $$
3+
declare
4+
v_owner_flow_slug text;
5+
v_listed text[];
6+
begin
7+
-- A name derived or referenced by another concrete flow is rejected
8+
select s.flow_slug into v_owner_flow_slug
9+
from pgflow.steps as s
10+
where s.queue_name = p_queue_name
11+
and lower(s.flow_slug) <> lower(p_flow_slug)
12+
limit 1;
13+
14+
if v_owner_flow_slug is null then
15+
select f.flow_slug into v_owner_flow_slug
16+
from pgflow.flows as f
17+
where f.queue_mode = 'flow'
18+
and lower(f.flow_slug) = p_queue_name
19+
and lower(f.flow_slug) <> lower(p_flow_slug)
20+
limit 1;
21+
end if;
22+
23+
if v_owner_flow_slug is not null then
24+
raise exception
25+
'cannot create flow "%": queue "%" is already used by another flow ("%")',
26+
p_flow_slug, p_queue_name, v_owner_flow_slug
27+
using detail = 'Generated per-step queue names must belong to exactly one concrete flow.',
28+
hint = 'Use a different concrete flow slug, or drop the conflicting definition.';
29+
end if;
30+
31+
-- Ambiguous normalized matches among listed queues are external damage
32+
select array_agg(listed.queue_name order by listed.queue_name)
33+
into v_listed
34+
from pgmq.list_queues() as listed
35+
where lower(listed.queue_name) = p_queue_name;
36+
37+
if v_listed is not null and cardinality(v_listed) > 1 then
38+
raise exception
39+
'queue "%" matches multiple listed PGMQ queues (%)',
40+
p_queue_name, v_listed
41+
using detail = 'An ambiguous case-insensitive match is external damage.',
42+
hint = 'Resolve the duplicate queue spellings manually, then retry.';
43+
end if;
44+
45+
-- Owned by an existing definition of this exact flow: reuse idempotently.
46+
-- A verified definition owns every derived route, so its one exact listed
47+
-- queue is allowed here. An owned route without its listed queue is
48+
-- external damage: reuse would report the route available while polling
49+
-- fails, so reject it instead.
50+
if exists (
51+
select 1 from pgflow.steps as s
52+
where s.flow_slug = p_flow_slug and s.queue_name = p_queue_name
53+
) then
54+
if v_listed is null then
55+
raise exception
56+
'queue "%" owned by flow "%" is not listed in PGMQ',
57+
p_queue_name, p_flow_slug
58+
using detail = 'An owned route whose queue is missing may still hold outstanding task identities.',
59+
hint = 'Recreate the queue, or drop and recompile the flow definition.';
60+
end if;
61+
62+
return false;
63+
end if;
64+
65+
if v_listed is not null then
66+
raise exception
67+
'cannot create flow "%": queue "%" is already listed in PGMQ and not owned by this flow',
68+
p_flow_slug, p_queue_name
69+
using detail = 'A missing definition must not adopt an already listed queue.',
70+
hint = 'Drop the conflicting queue or use a different concrete flow slug.';
71+
end if;
72+
73+
return true;
74+
end;
75+
$$;
76+
-- Modify "_cascade_force_skip_steps" function
77+
CREATE OR REPLACE FUNCTION "pgflow"."_cascade_force_skip_steps" ("run_id" uuid, "step_slug" text, "skip_reason" text) RETURNS integer LANGUAGE plpgsql AS $$
78+
DECLARE
79+
v_flow_slug text;
80+
v_total_skipped int := 0;
81+
v_archived_queues int;
82+
BEGIN
83+
-- Get flow_slug for this run
84+
SELECT r.flow_slug INTO v_flow_slug
85+
FROM pgflow.runs r
86+
WHERE r.run_id = _cascade_force_skip_steps.run_id;
87+
88+
IF v_flow_slug IS NULL THEN
89+
RAISE EXCEPTION 'Run not found: %', _cascade_force_skip_steps.run_id;
90+
END IF;
91+
92+
-- ==========================================
93+
-- SKIP STEPS IN TOPOLOGICAL ORDER
94+
-- ==========================================
95+
-- Use recursive CTE to find all downstream dependents,
96+
-- then skip them in topological order (by step_index)
97+
WITH RECURSIVE
98+
-- ---------- Find all downstream steps ----------
99+
downstream_steps AS (
100+
-- Base case: the trigger step
101+
SELECT
102+
s.flow_slug,
103+
s.step_slug,
104+
s.step_index,
105+
_cascade_force_skip_steps.skip_reason AS reason -- Original reason for trigger step
106+
FROM pgflow.steps s
107+
WHERE s.flow_slug = v_flow_slug
108+
AND s.step_slug = _cascade_force_skip_steps.step_slug
109+
110+
UNION ALL
111+
112+
-- Recursive case: steps that depend on already-found steps
113+
SELECT
114+
s.flow_slug,
115+
s.step_slug,
116+
s.step_index,
117+
'dependency_skipped'::text AS reason -- Downstream steps get this reason
118+
FROM pgflow.steps s
119+
JOIN pgflow.deps d ON d.flow_slug = s.flow_slug AND d.step_slug = s.step_slug
120+
JOIN downstream_steps ds ON ds.flow_slug = d.flow_slug AND ds.step_slug = d.dep_slug
121+
),
122+
-- ---------- Deduplicate and order by step_index ----------
123+
steps_to_skip AS (
124+
SELECT DISTINCT ON (ds.step_slug)
125+
ds.flow_slug,
126+
ds.step_slug,
127+
ds.step_index,
128+
ds.reason
129+
FROM downstream_steps ds
130+
ORDER BY ds.step_slug, ds.step_index -- Keep first occurrence (trigger step has original reason)
131+
),
132+
-- ---------- Skip the steps ----------
133+
skipped AS (
134+
UPDATE pgflow.step_states ss
135+
SET status = 'skipped',
136+
skip_reason = sts.reason,
137+
skipped_at = now(),
138+
remaining_tasks = NULL -- Clear remaining_tasks for skipped steps
139+
FROM steps_to_skip sts
140+
WHERE ss.run_id = _cascade_force_skip_steps.run_id
141+
AND ss.step_slug = sts.step_slug
142+
AND ss.status IN ('created', 'started') -- Only skip non-terminal steps
143+
RETURNING
144+
ss.*,
145+
-- Broadcast step:skipped event
146+
realtime.send(
147+
jsonb_build_object(
148+
'event_type', 'step:skipped',
149+
'run_id', ss.run_id,
150+
'flow_slug', ss.flow_slug,
151+
'step_slug', ss.step_slug,
152+
'status', 'skipped',
153+
'skip_reason', ss.skip_reason,
154+
'skipped_at', ss.skipped_at
155+
),
156+
concat('step:', ss.step_slug, ':skipped'),
157+
concat('pgflow:run:', ss.run_id),
158+
false
159+
) as _broadcast_result
160+
),
161+
-- ---------- Terminalize active tasks of newly skipped steps ----------
162+
skipped_tasks AS (
163+
UPDATE pgflow.step_tasks AS task
164+
SET status = 'skipped'
165+
WHERE task.run_id = _cascade_force_skip_steps.run_id
166+
AND task.step_slug IN (
167+
SELECT skipped_step.step_slug
168+
FROM skipped AS skipped_step
169+
)
170+
AND task.status IN ('queued', 'started')
171+
RETURNING task.message_id, task.queue_name
172+
),
173+
-- ---------- Archive queued/started task messages for skipped steps ----------
174+
-- Batched per stored queue route (#650)
175+
archived_messages AS (
176+
SELECT pgmq.archive(
177+
task.queue_name,
178+
ARRAY_AGG(task.message_id)
179+
) as result
180+
FROM skipped_tasks AS task
181+
WHERE task.message_id IS NOT NULL
182+
GROUP BY task.queue_name
183+
HAVING COUNT(task.message_id) > 0
184+
),
185+
-- ---------- Update run counters ----------
186+
run_updates AS (
187+
UPDATE pgflow.runs r
188+
SET remaining_steps = r.remaining_steps - skipped_count.count
189+
FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count
190+
WHERE r.run_id = _cascade_force_skip_steps.run_id
191+
AND skipped_count.count > 0
192+
)
193+
-- Consume every archived_messages row (COUNT(*)) in the same statement:
194+
-- SELECT INTO stops after its first row, and a direct LEFT JOIN of the
195+
-- CTE would leave every later queue's pgmq.archive group unevaluated, so
196+
-- its messages would recur indefinitely. The counted column lands in
197+
-- v_archived_queues the same way the other terminal cleanup functions
198+
-- force their archive CTE to run.
199+
SELECT skipped_count.count, archived_count.count
200+
INTO v_total_skipped, v_archived_queues
201+
FROM (SELECT COUNT(*) AS count FROM skipped) skipped_count
202+
LEFT JOIN (SELECT COUNT(*) AS count FROM archived_messages) archived_count ON true;
203+
204+
RETURN v_total_skipped;
205+
END;
206+
$$;

‎pkgs/core/supabase/migrations/atlas.sum‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
h1:KsVAXOPvkHDCj18n4kgWtwuS/HMiSC1dVMnNor++gpA=
1+
h1:mC3ikq4kkuBk1OJJeDAbv7aTAymhZljD1BYta6qu/f8=
22
20250429164909_pgflow_initial.sql h1:I3n/tQIg5Q5nLg7RDoU3BzqHvFVjmumQxVNbXTPG15s=
33
20250517072017_pgflow_fix_poll_for_tasks_to_use_separate_statement_for_polling.sql h1:wTuXuwMxVniCr3ONCpodpVWJcHktoQZIbqMZ3sUHKMY=
44
20250609105135_pgflow_add_start_tasks_and_started_status.sql h1:ggGanW4Wyt8Kv6TWjnZ00/qVb3sm+/eFVDjGfT8qyPg=
@@ -23,3 +23,4 @@ h1:KsVAXOPvkHDCj18n4kgWtwuS/HMiSC1dVMnNor++gpA=
2323
20260904095427_pgflow_task_lifecycle_hardening.sql h1:27b0BfBcQxeu5XSqVtQYvDCRzTsvLqzS/5hx14/2VyM=
2424
20260907082520_pgflow_remove_legacy_flow_compilation.sql h1:LNFDz+ZZlWb19FmWNPK57eiD+FXySMbStVij8MTSvDw=
2525
20260915074120_pgflow_private_step_queues.sql h1:+vsfsOyDaM8WISO/jxp4UBlTzPuQhg6RwBCiOAk5YAE=
26+
20260918213051_pgflow_fix_force_skip_multi_queue.sql h1:ibzV9U4G2t9K9T85/2HY996C/p1VZtDt0q9mcU60LpQ=

‎pkgs/core/supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
\set ON_ERROR_STOP on
2+
\set QUIET on
3+
4+
-- Force-skip across multiple private step queues (#651 review regression):
5+
-- a completed ancestor whose two children hold queued tasks in two separate
6+
-- private step queues must archive every queue's messages. The final
7+
-- SELECT INTO must not stop after the first archived queue row, or the
8+
-- second queue's message recurs indefinitely.
9+
begin;
10+
select plan(5);
11+
12+
select pgflow_tests.reset_db();
13+
14+
select pgflow.ensure_flow_compiled(
15+
'fskipmulti',
16+
'{
17+
"steps": [
18+
{"slug": "gate", "stepType": "single", "dependencies": [], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}},
19+
{"slug": "left", "stepType": "single", "dependencies": ["gate"], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}},
20+
{"slug": "right", "stepType": "single", "dependencies": ["gate"], "whenUnmet": "skip", "whenExhausted": "fail", "requiredInputPattern": {"defined": false}, "forbiddenInputPattern": {"defined": false}}
21+
]
22+
}'::jsonb,
23+
'step'
24+
);
25+
26+
select run_id as gate_run_id from pgflow.start_flow('fskipmulti', '{}') \gset
27+
28+
-- Claim the root task from its private queue and complete it
29+
select pgflow_tests.ensure_worker('fskipmulti__gate');
30+
select array_agg(msg_id) as gate_ids
31+
from pgmq.read_with_poll('fskipmulti__gate', 30, 1, 1, 50) \gset
32+
select pgflow.start_tasks(
33+
'fskipmulti',
34+
:'gate_ids'::bigint[],
35+
'11111111-1111-1111-1111-111111111111'::uuid,
36+
'fskipmulti__gate',
37+
'gate'
38+
);
39+
select pgflow.complete_task(:'gate_run_id'::uuid, 'gate', 0, '{}'::jsonb);
40+
41+
select is(
42+
(select count(*) from pgflow.step_tasks
43+
where run_id = :'gate_run_id'::uuid and status = 'queued'),
44+
2::bigint,
45+
'Setup: both children dispatched a queued task'
46+
);
47+
select is(
48+
(select (select count(*) from pgmq.q_fskipmulti__left)
49+
+ (select count(*) from pgmq.q_fskipmulti__right)),
50+
2::bigint,
51+
'Setup: one active message in each private child queue'
52+
);
53+
54+
-- Force-skip the completed ancestor; the cascade skips both children
55+
select pgflow._cascade_force_skip_steps(:'gate_run_id'::uuid, 'gate', 'condition_unmet');
56+
57+
select is(
58+
(select count(*) from pgmq.q_fskipmulti__left),
59+
0::bigint,
60+
'left child message left its private queue'
61+
);
62+
select is(
63+
(select count(*) from pgmq.q_fskipmulti__right),
64+
0::bigint,
65+
'right child message left its private queue'
66+
);
67+
select is(
68+
(select (select count(*) from pgmq.a_fskipmulti__left)
69+
+ (select count(*) from pgmq.a_fskipmulti__right)),
70+
2::bigint,
71+
'both child messages archived in their private queues'
72+
);
73+
74+
select * from finish();
75+
rollback;

0 commit comments

Comments
 (0)