|
| 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 = '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.', |
| 59 | + hint = 'Check why the queue disappeared; if the loss is intended, drop the flow definition and recompile it fresh.'; |
| 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 | +$$; |
0 commit comments