feat(core): persist physical queue identity for flow tasks - #679
Conversation
🦋 Changeset detectedLatest commit: 01147fe 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 01147fe
💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗ ☁️ Nx Cloud last updated this comment at |
startWorker() fired worker.startOnlyOnce() fire-and-forget, so the startup transaction (ensure_flow_compiled: create_flow/add_step, or destructive local recompile via delete_flow_and_data) raced the test body and could commit across test boundaries, interleaving with the next test's reset_db. A stray commit then left a listed PGMQ queue with no owning flow row, and the next create_flow correctly rejected it: cannot create flow "X": queue "X" is already in use by another owner. Fix: make startWorker() async and await startup; await all 31 call sites and performanceMapFlow's startWorkers. Startup errors now fail at the call site instead of surfacing via a later worker.stop(). Product code, SQL, and migrations are untouched; the ownership guard stays as approved in #650. mapFlow.test.ts focused runs pass twice consecutively; full integration suite 63 passed, 0 failed. Refs #650. CI: edge-worker-integration on PR #679.
🔍 Preview Deployment: Website✅ Deployment successful! 🔗 Preview URL: https://pr-679.pgflow.pages.dev 📝 Details:
_Last updated: _ |
This stack of pull requests is managed by Graphite. Learn more about stacking. |
jumski
left a comment
jumski
left a comment
There was a problem hiding this comment.
the main issue i have is this strange backward compatibility that calls pgmq.list_queues() in a hot path. we need to find a better solution, even if it would involve forcing users to redeplooy their flows or drain queues
pkgs/core/schemas/0035_function_effective_queue_name.sql: this function is called in hot path and it internally calls pgmq.list_queues - i dont think this is optimal nor acceptable, we need to find a better wayy. i dont really understand the issue here why this is even needed?
pkgs/core/scripts/run-upgrade-fixture: why this is needed? for testing the migration?
| expectTypeOf(client.startTasks).toBeFunction(); | ||
| expectTypeOf(client.startTasks).parameters.toMatchTypeOf< | ||
| [string, number[], string] | ||
| [string, string[], string, string?] |
There was a problem hiding this comment.
why queuue_name argument is optional?
| archived_messages AS ( | ||
| SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(task.message_id)) as result | ||
| SELECT pgmq.archive( | ||
| pgflow._effective_queue_name(task.queue_name), |
There was a problem hiding this comment.
this is hot path and this function does list_queues and shit, we need to find a better way - like asked above, please explain why its even needed?
| returns pgflow.flows | ||
| language sql | ||
| set search_path to '' | ||
| language plpgsql |
There was a problem hiding this comment.
why the switch to plpgsql?
| -- | ||
| -- The flow's queues are dropped through their persisted definition routes | ||
| -- (steps.queue_name), plus the default queue for an empty flow (#650). Queues | ||
| -- created by older releases keep their original listed spelling; the drop |
There was a problem hiding this comment.
i think we should migrate old messages to new qaueues maybe?
| workerId: string | ||
| msgIds: string[], | ||
| workerId: string, | ||
| queueName?: string |
There was a problem hiding this comment.
why we made this optional?
| -- Queues created by older pgflow releases keep their original mixed-case | ||
| -- spelling in pgmq (#650). Every message operation resolves the stored | ||
| -- canonical name through pgmq.list_queues() instead of creating a second | ||
| -- metadata entry: dispatch, claim, visibility, completion, retry, recovery. |
There was a problem hiding this comment.
would it be possible to migrate old queue names to new format or reject migration and tell user how to fix his data? i would really like to avoid having this backward compatibility support. if it is not possible, maybe a column indicating that given step uses old style queue, so we dont need to invoke the list_queues? and later prevent users from re-deploying the old style? ideally we get rid of it immediately tho
| * Resolves the physical spelling of a canonical (lowercase) queue name | ||
| * through pgmq.list_queues() (#650). Queues created by older pgflow | ||
| * releases keep their original mixed-case spelling. |
There was a problem hiding this comment.
i dont like this 'older flows keep' thing
## Summary Make the integration-test harness await worker startup. `startWorker()` in `pkgs/edge-worker/tests/integration/_helpers.ts` fired `worker.startOnlyOnce()` fire-and-forget, so the startup transaction (`ensure_flow_compiled`: `create_flow`/`add_step`, or destructive local recompile via `delete_flow_and_data`) could race the test body and commit across test boundaries, interleaving with the next test's `reset_db`. A stray commit then left a listed PGMQ queue with no owning flow row, and a later `create_flow` rejected it (`queue "X" is already in use by another owner`). Startup errors now fail at the call site instead of surfacing via a later `worker.stop()`. Test-only change: `_helpers.ts`, 31 call sites across 9 integration test files, and `performanceMapFlow`'s `startWorkers`. No product code, SQL, or migrations. Base of the #679 stack (queue identity feature); that PR's CI originally failed on this race. ## Checks - `mapFlow.test.ts` focused runs pass twice consecutively (no orphan-queue leak across runs); `queueIdentity.test.ts` likewise - Full `pnpm nx test:integration edge-worker`: 63 passed, 0 failed - `nx affected --target=prepush`: 32/32 successful Part of #650
Persist canonical queue routes on steps and immutable-by-runtime task snapshots so queue-scoped message IDs cannot cross task identity. Route dispatch, claims, lifecycle cleanup and optional pruning through stored routes while preserving concurrency, terminal-state and visibility safeguards. Require the canonical queue_name argument in SQL and TypeScript claims and remove the released three-argument overload. PGMQ already normalizes message-operation names, so remove hot-path spelling resolution, session caches and worker lookup plumbing. Keep public queue listing for provisioning collisions and fresh destructive deletion only. Backfill populated 0.16.0 databases before enforcing constraints, with bounded lock waits and atomic rollback on conflicts. Document the breaking maintenance upgrade: pause producers, save worker states, drain active work, stop workers and other writers, migrate, deploy matching packages and restore prior states. Queued messages remain intact; no queue recreation or mixed-version rolling upgrade. Add identity, isolation, no-listing, required-signature and populated-upgrade regressions; retain the new integration test with awaited startup from the parent harness fix. Closes #650.
🚀 Production Deployment: Website✅ Successfully deployed to production! 🔗 Production URL: https://pgflow.dev 📝 Details:
Deployed at: 2026-09-13T20:08:29+02:00 |
1 participant
Summary
Persist physical queue identity independently of flow identity while keeping today's one-flow/one-queue behavior (#650). Replacement for rejected #677. Stacked on the test-harness startup fix #680; each PR contains one commit.
queue_nameon steps and task snapshots; enforce partial(queue_name, message_id)uniqueness while retaining nullable message IDs and the task primary key. Lifecycle operations use stored routes, grouped by queue, with existing concurrency, terminal-state and visibility safeguards.pgflow.start_tasks(flow_slug, msg_ids, worker_id, queue_name)and TypeScriptstartTasks()require the canonical queue argument (lower(flow_slug)today). The released three-argument SQL overload is removed; no default, fallback or compatibility shim remains. JavaScript message IDs are exact decimal strings._effective_queue_name, its session cache, and worker spelling-resolution plumbing. Pruning also uses canonical routes directly. Public listing remains for provisioning collisions and fresh original-spelling resolution before destructive queue deletion. Existing mixed-case queues remain usable without moving messages or recreating queues.set_vt_batch()and direct archive pruning.20260913093141_pgflow_persist_queue_identitymigration: nullable adds, populated backfill, constraints, then new functions and removal of the old claim overload. Bounded lock waits and atomic rollback safeguards remain. Update the maintenance guide, pruning instructions, lifecycle diagram and 0.17.0 news article.Maintenance upgrade
No mixed-version rolling upgrade. Pause producers; record worker enablement and running process states before changing them; drain active handlers under the old schema; stop workers and re-invocation; quiesce maintenance/recovery/definition writers; apply migrations to the explicit target through Supabase's runner; replace any installed pruning helper; deploy matching packages and callers; restore the saved states and resume producers.
Draining active work does not mean emptying queues. Queued tasks and messages survive the upgrade. A failed migration rolls back; after success, old three-argument workers cannot simply restart.
Checks
Three GLM implementation / independent Sol review rounds completed. Final code and documentation review: APPROVED.
portable runtimes - max_concurrency works in supabasefails withTimeout after 60000ms waiting for test_seq to increment by 200. One targeted failed-job rerun reproduced the same failure. Local portable-runtime suites passed twice; that does not establish the cause of the CI failure. No timeout or assertion was weakened. Three authorized correction/review rounds are exhausted.Closes #650