fix(edge-worker): await worker startup in integration tests - #680
Merged
Conversation
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.
|
Contributor
Author
This stack of pull requests is managed by Graphite. Learn more about stacking. |
|
View your CI Pipeline Execution ↗ for commit cae4772
💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗ ☁️ Nx Cloud last updated this comment at |
|
View your CI Pipeline Execution ↗ for commit cae4772
💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗ ☁️ Nx Cloud last updated this comment at |
Contributor
Author
Contributor
🚀 Production Deployment: Website✅ Successfully deployed to production! 🔗 Production URL: https://pgflow.dev 📝 Details:
Deployed at: 2026-09-13T19:59:50+02:00 |
jumski
added a commit
that referenced
this pull request
Sep 13, 2026
## 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. - Store canonical `queue_name` on 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. - **Breaking claim API:** SQL `pgflow.start_tasks(flow_slug, msg_ids, worker_id, queue_name)` and TypeScript `startTasks()` 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. - **No hot-path queue listing:** PGMQ already normalizes message-operation names. Remove `_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. - Preserve exact-flow ownership checks, ambiguous-match rejection before destructive work, transactional deletion, single-read claiming, and unmatched/wrong-flow message preservation. Keep accepted `set_vt_batch()` and direct archive pruning. - Regenerate the unreleased `20260913093141_pgflow_persist_queue_identity` migration: 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**. - Full core pgTAP: **300 files / 1,464 tests pass**, including populated 0.16.0 upgrade and conflict rollback fixtures. - Edge-worker unit: **254 pass**; edge-worker E2E and portable-runtime E2E: **12 each pass**; client E2E: **47 pass**. - Migration/generated-type checks, website lint/build/link checks, changeset validation and post-commit affected prepush: **pass** (32/32 prepush tasks). - Current-head GitHub CI **passes edge-worker integration**, core pgTAP, build/test, edge-worker E2E, client/CLI E2E, smoke, type infrastructure and changesets. The local integration timeout did not recur in CI. - **Remaining CI blocker:** portable-runtime E2E is **11/12**. `portable runtimes - max_concurrency works in supabase` fails with `Timeout 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
30 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Make the integration-test harness await worker startup.
startWorker()inpkgs/edge-worker/tests/integration/_helpers.tsfiredworker.startOnlyOnce()fire-and-forget, so the startup transaction (ensure_flow_compiled:create_flow/add_step, or destructive local recompile viadelete_flow_and_data) could race the test body and commit across test boundaries, interleaving with the next test'sreset_db. A stray commit then left a listed PGMQ queue with no owning flow row, and a latercreate_flowrejected it (queue "X" is already in use by another owner). Startup errors now fail at the call site instead of surfacing via a laterworker.stop().Test-only change:
_helpers.ts, 31 call sites across 9 integration test files, andperformanceMapFlow'sstartWorkers. 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.tsfocused runs pass twice consecutively (no orphan-queue leak across runs);queueIdentity.test.tslikewisepnpm nx test:integration edge-worker: 63 passed, 0 failednx affected --target=prepush: 32/32 successfulPart of #650