feat: persist queue state in database (#650) by jumski · Pull Request #677 · pgflow-dev/pgflow

jumski · 2026-09-10T05:30:36Z

Summary

Persist queue state in the database for issue #650: queue messages and archives become owned, auditable pgflow state with a coordinated schema + worker upgrade.

Queue-aware claiming: pgflow.claim_tasks takes ordered parent/state/task/queue locks, classifies the complete batch under those locks, and builds results exclusively from the guarded UPDATE .. RETURNING. Envelope rules reject malformed and foreign identities (exact durable pair wins over malformed components; a valid address identifying different work is fatal).
Migration 20260910104929_pgflow_persist_queue.sql: transactional, with a full old-schema preflight that mirrors _inspect_generated_queue's physical contract (columns, types, PKs, usable indexes, sequence pg_depend association, extension membership, envelope contradictions) under bounded 5-second locks. atlas.sum refreshed.
Audit: core/queries/PRE_MIGRATION_CHECK_650.sql — copyable, read-only, 0.16.0-compatible; reports physical malformations, envelope contradictions, and bounded samples.
Worker protocol: fatal ClaimTasksResult discriminated union, queue_name on StepTaskRecord, string message IDs end to end, and a startup fence — 0.16.0 workers fail fast at the removed ensure_flow_compiled(text,jsonb) signature before registration (QueueProtocolMismatchError).
Upgrade fixture (test:upgrade:queue): proves 0.16.0 rejection pre-registration, atomic rejection of malformed objects, bounded concurrency behavior, and state survival across migration.
Pruning/deletion: pruning validates every route via _inspect_generated_queue before queue access; delete_flow_and_data verifies table/sequence/metadata absence after drop_queue.
Docs: queue-identity upgrade guide, data model, naming rules, context msg_id strings, deletion order, claim_tasks current-surface descriptions.

Three reviewed cycles: review-initial (15 findings), review-correction-1 (7), review-correction-2 APPROVED (independent reviewer).

Checks

test:pgtap core — 300 files / 1522 tests (both upgrade fixtures)
test:upgrade:queue core — PASS (missing-column, malformed-objects, queue_create scenarios)
test:integration edge-worker — 64/65 (performanceMapFlow timeout pre-existing, reproduced identically on unmodified baseline; logs in run evidence)
e2e client — 47/47; test:unit edge-worker — 258; test:types core/dsl/client/edge-worker — green
verify-migrations, verify-gen-types, verify-exports, test:node, smoke:bun — green
build website + lint website — green; nx affected --target=prepush --base=origin/main — 32/32

Closes #650

Checkpoint commit of the issue #650 implementation, made mid-delivery to preserve the working tree; Task 10 documentation is still pending. Before: the scheduler kept run/step/task state in memory only, so a worker restart lost in-flight work. Now: queue state lives in Postgres; the worker claims work through queue-backed claim/startup paths and resumes from the persisted snapshot after a restart. Key pieces: - schemas 0030/0050/0060/0070/0100/0120 and migration 20260909210645_pgflow_persist_queue.sql (queue tables, identity, ownership, validate_flow_shape, claim functions) - pgTAP: queue_identity suite, delete_flow_and_data queue_ownership, maintenance queue_snapshot_pruning - edge-worker: queueClaim/queueStartup integration tests, StepTaskPoller classification unit tests, core errors module - PRE_MIGRATION_CHECK_650.sql audit script plus upgrade_queue_fixture/ and run-queue-upgrade-fixture runner (Nx target test:upgrade:queue) to validate the in-place upgrade path - run-queue-upgrade-fixture calls pg_dump without -X (fixup applied during delivery) Validation state at checkpoint: core/dsl/client/edge-worker tests and typechecks green, full pgTAP suite green, queueClaim/queueStartup integration green; test:upgrade:queue on late fixture iterations.

changeset-bot · 2026-09-10T05:30:39Z

⚠️ No Changeset found

Latest commit: afda120

Merging this PR will not cause a version bump for any packages. If these changes should not result in a new version, you're good to go. If these changes should result in a version bump, you need to add a changeset.

This PR includes no changesets

When changesets are added to this PR, you'll see the packages that this PR includes changesets for and the associated semver types

Click here to learn what changesets are, and how to add one.

Click here if you're a maintainer who wants to add a changeset to this PR

jumski · 2026-09-10T05:30:49Z

feat: persist queue state in database (#650) #677 👈 (View in Graphite)
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

nx-cloud · 2026-09-10T05:31:33Z

View your CI Pipeline Execution ↗ for commit afda120

Command	Status	Duration	Result
`nx run-many -t build --projects=dsl,core,cli`	✅ Succeeded	6s	View ↗

💡 Verify your cache is correct by running tasks in a sandbox. Read docs ↗

☁️ Nx Cloud last updated this comment at 2026-09-10 17:51:33 UTC

Complete the coordinated schema and worker change for issue #650 on top of checkpoint 6142ab1: queue-identity claiming, transactional migration with full old-schema preflight, copyable audit, upgrade fixture, released-worker compatibility fence, and documentation. Why: queue state lived only in PGMQ, so messages could not be audited, pruned, or recovered with the flows they belong to, and 0.16.0 workers could silently interoperate with an incompatible schema. Implemented in three reviewed cycles (review-initial BLOCKED 15 findings, review-correction-1 BLOCKED 7, review-correction-2 APPROVED on openai/gpt-5.6-sol:xhigh): - claim_tasks rebuilt: ordered parent/state/task/queue locks, complete classification under locks, results only from guarded UPDATE..RETURNING; envelope rules for malformed and foreign identities (queue_identity/claim_envelope pgTAP). - Migration regenerated as 20260910104929_pgflow_persist_queue.sql: preflight mirrors the full _inspect_generated_queue physical contract (columns, types, PKs, usable indexes, sequence pg_depend association, extension membership, envelope contradictions) with bounded 5s locks; atlas.sum refreshed. - PRE_MIGRATION_CHECK_650.sql audit gains physical and envelope-contradiction reporting with bounded samples; to_regprocedure for the absent pruning helper. - Pruning helper validates every route via _inspect_generated_queue under the pgmq.meta fence before any queue access. - delete_flow_and_data verifies queue tables, sequence, and metadata absence after drop_queue. - Worker: fatal ClaimTasksResult discriminated union, queue_name on StepTaskRecord, string message IDs end to end, QueueProtocolMismatch fence on startup (3-arg ensure_flow_compiled). - Upgrade fixture proves 0.16.0 workers fail at the removed signature pre-registration and state survives migration; startup probe verification no longer masked by CONNECTION_ENDED suppression. - Docs: update-pgflow queue-identity upgrade section, data model, naming rules, context msg_id strings, deletion order, claim_tasks current-surface descriptions. Checks (logs retained in run evidence): test:pgtap core 300 files/1522 tests; test:upgrade:queue PASS; test:integration edge-worker 64/65 (performanceMapFlow timeout pre-existing on unmodified baseline, reproduced and documented); e2e client 47/47; test:unit edge-worker 258; verify-migrations, verify-gen-types, verify-exports, test:node, smoke:bun, website build+lint all green. Closes #650

github-actions · 2026-09-10T17:59:21Z

🔍 Preview Deployment: Website

✅ Deployment successful!

🔗 Preview URL: https://pr-677.pgflow.pages.dev

📝 Details:

Branch: 650-persist-queue
Commit: 5b19c4c3adb356f897ebd178b7a8799ba72bcbe8
View Logs

_Last updated: _

jumski

i reviewed only sql changes and have huge amount of issues with them

pkgs/core/queries/PRE_MIGRATION_CHECK_650.sql: this query is huge - is it safe to run something like this on user production database???

pkgs/core/schemas/0070_functions_generated_queues.sql: this whole file is very strange. wtf. why we validate pgmq internals/implementation details like sequences?

pkgs/core/schemas/0120_function_claim_tasks.sql: this function is monstrous

jumski · 2026-09-11T17:46:24Z

  step_type text not null default 'single',
  step_index int not null default 0,
  deps_count int not null default 0 check (deps_count >= 0),
+  queue_name text not null,


i hope migration backfills with nullable and ooonly adds not null aftger the backfill

jumski · 2026-09-11T17:46:24Z

+
+-- Case-insensitive namespace uniqueness: concrete spelling is preserved, but
+-- case-only aliases would collide on generated queue names, so they are
+-- rejected atomically by declarative indexes (#650).


need this documented in docs

jumski · 2026-09-11T17:46:25Z

+create trigger keep_task_queue_name
+after update on pgflow.step_tasks
+referencing old table as old_tasks new table as new_tasks
+for each statement execute function pgflow._keep_task_queue_name();


im not really sure if we should introduce this trigger. is there any pgflow's code that could attempt mutation? because if there isnt, it protects from user doing those changes, right? and there are lot of other columns that should be immutable by design too - flow_slug, step_slug, step_index and many others. so we only protect one column here. and the step_tasks updates are kinda hot path right? we update them multiple times per lifecycle of a step_task. please reconsider if we should maybe remove this trigger?

jumski · 2026-09-11T17:46:25Z

-  with stalled_tasks as (
+  --
+  -- Lock order (#650): eligible parent runs are locked first (ordered by
+  -- run_id), then eligible step states (ordered by (run_id, step_slug)), then


should we order by run_id, step_slug, or maybe (run_id, step_index) ??

jumski · 2026-09-11T17:46:25Z

+  -- Lock order (#650): eligible parent runs are locked first (ordered by
+  -- run_id), then eligible step states (ordered by (run_id, step_slug)), then
+  -- task rows (ordered by (run_id, step_slug, task_index)) - as three
+  -- sequential lock sets, not one joined FOR UPDATE, so parent rows are


step_index would guarantee parent locked before children, which cannot be said on ordering by step_slug

jumski · 2026-09-11T17:46:25Z

-GROUP BY r.flow_slug
-HAVING COUNT(st.message_id) > 0;
+-- For failed tasks: archive the message grouped by the task's queue snapshot
+FOR v_archive_batch IN


jumski · 2026-09-11T17:46:25Z

+perform 1 from pgflow.flows f
+where f.flow_slug = start_flow.flow_slug
+for key share;


jumski · 2026-09-11T17:46:25Z

+  execute format(
+    'select coalesce(jsonb_agg(jsonb_build_object(''msg_id'', q.msg_id, ''message'', q.message)), ''[]''::jsonb)
+     from pgmq.%I q where q.msg_id = any($1)',
+    v_qtable
+  ) into v_bodies using v_ids;


why we are using pgmq internals again? this is smelly bad code

jumski · 2026-09-11T17:46:25Z

+    'select coalesce(jsonb_agg(jsonb_build_object(''msg_id'', q.msg_id, ''message'', q.message)), ''[]''::jsonb)
+     from pgmq.%I q where q.msg_id = any($1)',


this compromises pgmq performance guarantees

jumski · 2026-09-11T17:46:25Z

+    if v_classification.task_run_id is not null then
+      -- ==========================================
+      -- EXACT DURABLE PAIR: the pair wins over the envelope. Malformed or
+      -- absent components never contradict it; only a VALID address that
+      -- positively identifies different work is fatal.
+      -- ==========================================
+      if v_body_flow is not null and v_body_flow is distinct from claim_tasks.flow_slug then
+        v_errors := v_errors || jsonb_build_object(
+          'queue_name', queue_name, 'message_id', v_classification.msg_id::text, 'reason', 'wrong_route');
+        v_fatal := true;
+      elsif v_run_valid and v_body_run::uuid is distinct from v_classification.task_run_id then
+        v_errors := v_errors || jsonb_build_object(
+          'queue_name', queue_name, 'message_id', v_classification.msg_id::text, 'reason', 'unsupported_work');
+        v_fatal := true;
+      elsif v_body_step is not null and v_body_step is distinct from v_classification.task_step then
+        v_errors := v_errors || jsonb_build_object(
+          'queue_name', queue_name, 'message_id', v_classification.msg_id::text, 'reason', 'unsupported_work');
+        v_fatal := true;
+      elsif v_index_valid and (v_body_index)::int is distinct from v_classification.task_index then
+        v_errors := v_errors || jsonb_build_object(
+          'queue_name', queue_name, 'message_id', v_classification.msg_id::text, 'reason', 'unsupported_work');
+        v_fatal := true;
+      elsif v_classification.task_status in ('completed', 'failed', 'skipped', 'cancelled') then
+        -- Terminal task: idempotent archive (archive ignores already-archived)
+        v_terminal_ids := array_append(v_terminal_ids, v_classification.msg_id);
+      elsif v_classification.permanently_stalled_at is not null then
+        -- Permanent stall: preserve status/history, archive idempotently
+        v_terminal_ids := array_append(v_terminal_ids, v_classification.msg_id);
+      elsif v_classification.task_status = 'started'
+            and v_classification.run_status = 'started'
+            and v_classification.step_status = 'started' then
+        v_defer_ids := array_append(v_defer_ids, v_classification.msg_id);
+      elsif v_classification.task_status = 'queued'
+            and v_classification.run_status = 'started'
+            and v_classification.step_status = 'started' then
+        v_claim_ids := array_append(v_claim_ids, v_classification.msg_id);
+      else
+        -- Active task with incompatible parent state
+        v_errors := v_errors || jsonb_build_object(
+          'queue_name', queue_name, 'message_id', v_classification.msg_id::text, 'reason', 'unsupported_work');
+        v_fatal := true;
+      end if;
+    else


why this is needed?

jumski deployed to preview September 10, 2026 12:43 — with GitHub Actions Active

jumski force-pushed the 650-persist-queue branch from 8c36a21 to afda120 Compare September 10, 2026 17:49

jumski deployed to preview September 10, 2026 17:57 — with GitHub Actions Active

jumski commented Sep 11, 2026

View reviewed changes

jumski mentioned this pull request Sep 11, 2026

Enforce immutability of persisted pgflow identities #678

Open

		'select coalesce(jsonb_agg(jsonb_build_object(''msg_id'', q.msg_id, ''message'', q.message)), ''[]''::jsonb)
		from pgmq.%I q where q.msg_id = any($1)',

Uh oh!

Conversation

jumski commented Sep 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Checks

Uh oh!

changeset-bot Bot commented Sep 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

⚠️ No Changeset found

Uh oh!

jumski commented Sep 10, 2026

Uh oh!

nx-cloud Bot commented Sep 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Sep 10, 2026

🔍 Preview Deployment: Website

Uh oh!

jumski left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jumski commented Sep 10, 2026 •

edited

Loading

changeset-bot Bot commented Sep 10, 2026 •

edited

Loading

nx-cloud Bot commented Sep 10, 2026 •

edited

Loading