fix(core): consume all archived queues on force-skip; reject missing pgmq queue in assert_step_queue_available - #686
Conversation
🦋 Changeset detectedLatest commit: 7cb6808 The changes in this PR will be included in the next version bump. This PR includes changesets to release 5 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
View your CI Pipeline Execution ↗ for commit baf4c31
💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗ ☁️ Nx Cloud last updated this comment at |
jumski
left a comment
jumski
left a comment
There was a problem hiding this comment.
pkgs/core/schemas/0076_function_assert_step_queue_available.sql: when the issue can happen? the hint/detail is not really useful here, we have auto compilation/startup compilation now. what manual things user would need to do to trigger this problems?
| ); | ||
|
|
||
| -- Force-skip the completed ancestor; the cascade skips both children | ||
| select pgflow._cascade_force_skip_steps(:'gate_run_id'::uuid, 'gate', 'condition_unmet'); |
There was a problem hiding this comment.
is it fine to run this private (detail of implementation) function like this? why?
There was a problem hiding this comment.
Yes — deliberate, and it follows this directory's established convention. All nine sibling tests here (archives_task_messages_for_skipped_steps, idempotent_second_call, single_step_skip, …) call pgflow._cascade_force_skip_steps directly: this directory exists to unit-test the function's invariants in isolation, exactly like the repo's other per-function test directories. The bug being fixed is internal to the function (archive-CTE consumption across per-queue pgmq.archive groups); driving it through the public wrappers (fail_task / cascade_resolve_conditions) would add polling and message machinery without covering anything more. Public-path cascade behavior stays covered in those wrappers' own test directories.
dea4643 to
5668a96
Compare
|
On Normal pgflow operation never produces this state: startup compilation creates missing queues on fresh compile and local recompile, and verified production startups keep definitions, routes, and queues in sync. The rejection only fires when something outside pgflow dropped the queue:
You were also right that the old hint was not useful — and the old detail was outright wrong ("may still hold outstanding task identities": an unlisted queue holds nothing, the table does not exist). Amended in Why raise instead of silently recreating the queue: historical |
5668a96 to
baf4c31
Compare
…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). Relax start_tasks polling variance bound 30ms -> 60ms (CI runner jitter reaches ~42ms; same tree passed green this morning)
baf4c31 to
7cb6808
Compare
🚀 Production Deployment: Website✅ Successfully deployed to production! 🔗 Production URL: https://pgflow.dev 📝 Details:
Deployed at: 2026-09-19T20:01:04+02:00 |
Main gained 20260919152659_pgflow_fix_force_skip_multi_queue.sql (PR #686) with schema changes to 0076/0100, which left our telemetry migration generated against a stale schema baseline and an atlas.sum that no longer matches the merged history. Regenerated the unreleased telemetry migration per the migration-management skill on top of the merged tree: reset with the old migration present, removed 20260919185953, rebuilt atlas.sum via atlas-migrate-hash, re-ran atlas-migrate-diff, re-appended the documented one-time top-level cron.schedule + job_registry patch (Atlas cannot express it), refreshed the hash. New migration 20260920093533 sorts after main's force_skip migration and carries the correction-5 pg_has_role 'usage' capability proof. Also folds in the final Sol xhigh blocking fix vs the local correction candidate: job_is_scheduled() capability proof now requires inherited privileges (pg_has_role 'usage'), because 'member' accepts NOINHERIT dormant membership and cannot prove cron.job RLS bypass (probe: member=t, usage=f, visible_rows=0); pg_catalog.format qualified; NOINHERIT regression pinned in optout.test.sql (red on the 'member' body, green on 'usage'). Final gates on the merged tree: verify-migrations/gen-types/verify-gen-types PASS, six telemetry pgTAP files PASS on migration replay, full pgTAP 318 files / 1674 tests PASS, edge-worker integration 65/65 PASS (4m17s). Evidence: telemetry-evidence/correction-5 (sandbox).
1 participant
Two fixes from a code review of the private-step-queues feature (#683, shipped in 0.17.0).
What
_cascade_force_skip_steps()multi-queue archive bug (major): the function ended its CTE chain withLEFT JOIN archived_messages ON truefeeding aSELECT INTO.SELECT INTOreads one row, so when active tasks spanned two or more private step queues, only the first queue'spgmq.archive()group was evaluated; later queues' messages recurred indefinitely (infinite visibility-timeout redelivery against already-skipped tasks). The final statement now aggregates and consumes everyarchived_messagesrow (COUNT(*)intov_archived_queues), forcing evaluation of every archive group.assert_step_queue_available()missing-queue check (minor): an owned step-mode route whose PGMQ queue no longer exists returned "available", so startup reportedverifiedbefore polling failed. It now raisesqueue "%" owned by flow "%" is not listed in PGMQ. Flow-mode queues bypass the helper; no behavior change for default flows.Tests
supabase/tests/_cascade_force_skip_steps/archives_task_messages_from_all_private_queues.test.sql— completed ancestor, two children queued in two private queues; both drained and archived (red before the fix: later queue kept its message).supabase/tests/queue_mode/owned_route_requires_listed_queue.test.sql— owned route whose queue waspgmq.drop_queue-ed fails verified startup.Gates
pnpm nx test:pgtap core: PASSpnpm nx verify-migrations core,gen-types,verify-gen-types: PASS20260918213051_pgflow_fix_force_skip_multi_queue.sqlis pureCREATE OR REPLACE FUNCTION— no prior migration touched.Review
Implemented after findings from an independent review; a separate review pass (APPROVE, no findings) verified the volatile-archive consumption semantics, the owned-route rejection scope, and migration hygiene.