fix(core): guarantee task visibility extension in start_tasks() by jumski · Pull Request #667 · pgflow-dev/pgflow

jumski · 2026-09-04T10:15:59Z

Summary

start_tasks() invoked pgflow.set_vt_batch() inside an unreferenced SELECT CTE, so PostgreSQL could skip the call and leave a claimed task with only the short initial PGMQ read visibility. A message then became visible while its step_tasks row stayed started, and workers repeatedly read it as a task that never ran.

The final statement now consumes a guard over the renamed visibility_reset CTE, making the visibility update structurally required. When set_vt_batch() updates fewer messages than were claimed, the statement fails with start_tasks(): visibility updated N of M claimed messages, rolling back the task transitions and attempt increments and returning nothing.
Effective delay stays coalesce(step.opt_timeout, flow.opt_timeout) + 2, matching the fix(core): use effective step timeout for stalled tasks #664 stalled-recovery threshold. A repeatedly visible message whose task is already started claims nothing and consumes no attempt — a benign duplicate, per Add private per-step queues with typed step workers #651/Add explicit shared queues with multi-flow dispatch #652 direction.
Consolidates the three unreleased Core migrations on main (terminalize_skipped_tasks Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638/Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649, failed_run_terminalization Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645/fix(core): terminalize unfinished tasks when runs fail #663, effective_step_timeout requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621/fix(core): use effective step timeout for stalled tasks #664) plus the development temp migration into one Atlas-generated task_lifecycle_hardening migration against the 0.15.0 baseline (Release 0.15.1 — task lifecycle hardening #666). Both migration-only data repairs are preserved in order: active tasks under skipped steps become skipped before remaining active tasks under failed runs become cancelled.
Adds a 0.15.0 upgrade fixture (supabase/upgrade_fixture/ + scripts/run-upgrade-fixture): starts from the 0.15.0 baseline, applies the consolidated migration, and asserts both historical repairs plus final runtime visibility behavior. Wired into the core test:pgtap nx target so CI executes it.
New pgTAP coverage for all seven Make start_tasks() apply task visibility before handlers run #656 test bullets: effective-timeout visibility (shorter / longer / null fallback), rollback on total and partial visibility failure, benign duplicate batches, and mixed-batch atomicity. Visibility-delay checks use clock_timestamp() to match set_vt_batch(), and repeated duplicates recheck that the attempt count stays unchanged.
Removes the unsupported sqruff 0.39.0 --force option so the required pnpm nx fix-sql core migration step runs normally.
One @pgflow/core patch changeset.

Checks

pnpm nx verify-migrations core — pass (incl. schemas-synced)
pnpm nx gen-types core + verify-gen-types core --skip-nx-cache — pass
pnpm nx fix-sql core — pass (37 files, nothing to fix)
pnpm nx test:pgtap core --skip-nx-cache — pass (288 files, 1352 tests; focused start_tasks suites 26/26; upgrade fixture: PASS)
pnpm nx affected --target=prepush --base=origin/main --head=HEAD — 28/28 successful
atlas migrate validate — pass; atlas.sum re-hashed after the manual repair edits

Closes #656

start_tasks() invoked pgflow.set_vt_batch() inside an unreferenced SELECT CTE, so PostgreSQL could skip the call and leave a claimed task with only the short initial PGMQ read visibility (#656). The final statement now consumes a guard over the renamed visibility_reset CTE: execution is structurally required, and when set_vt_batch() updates fewer messages than were claimed the statement fails ("start_tasks(): visibility updated N of M claimed messages"), rolling back the task transitions and attempt increments and returning nothing. Effective delay stays coalesce(step.opt_timeout, flow.opt_timeout) + 2, matching the #664 stalled-recovery threshold. A repeatedly visible message whose task is already started claims nothing and consumes no attempt (benign duplicate). Consolidates the three unreleased Core migrations on main (terminalize_skipped_tasks #638/#649, failed_run_terminalization #645/#663, effective_step_timeout #621/#664) plus the dev temp migration into one Atlas-generated task_lifecycle_hardening migration against the 0.15.0 baseline. Both migration-only data repairs are preserved in order: active tasks under skipped steps become skipped before remaining active tasks under failed runs become cancelled. A 0.15.0 upgrade fixture (supabase/upgrade_fixture + scripts/run-upgrade-fixture) applies the consolidated migration and asserts both repairs plus runtime visibility, wired into the core test:pgtap target. Adds pgTAP coverage for all seven issue #656 test bullets: effective-timeout visibility (shorter/longer/null fallback), rollback on total and partial visibility failure, benign duplicate batches, and mixed-batch atomicity.

changeset-bot · 2026-09-04T10:16:07Z

🦋 Changeset detected

Latest commit: 6ab5d5b

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 5 packages

Name	Type
@pgflow/core	Patch
pgflow	Patch
@pgflow/client	Patch
@pgflow/edge-worker	Patch
@pgflow/dsl	Patch

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

jumski · 2026-09-04T10:16:13Z

fix(core): guarantee task visibility extension in start_tasks() #667 👈 (View in Graphite)
main

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

nx-cloud · 2026-09-04T10:17:01Z

View your CI Pipeline Execution ↗ for commit 6ab5d5b

Command	Status	Duration	Result
`nx run edge-worker:test:integration`	✅ Succeeded	5m 58s	View ↗
`nx run client:e2e`	✅ Succeeded	3m 23s	View ↗
`nx run edge-worker:e2e:portable-runtimes`	✅ Succeeded	1m 1s	View ↗
`nx affected -t verify-exports --base=origin/mai...`	✅ Succeeded	6s	View ↗
`nx run core:pgtap`	✅ Succeeded	2m	View ↗
`nx affected -t build --configuration=production...`	✅ Succeeded	3s	View ↗
`nx affected -t lint typecheck test --parallel -...`	✅ Succeeded	2m 17s	View ↗
`nx run cli:e2e`	✅ Succeeded	6s	View ↗
`Additional runs (2)`	✅ Succeeded	...	View ↗

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

☁️ Nx Cloud last updated this comment at 2026-09-04 12:48:01 UTC

Use clock_timestamp() in visibility-delay pgTAP assertions because set_vt_batch() uses wall-clock time while now() stays fixed at transaction start. Recheck attempts_count after a repeatedly visible started message so benign duplicates remain pinned end to end. Remove the unsupported sqruff --force option from the core fix-sql target. sqruff 0.39.0 rejects that flag, which blocked the schema-first migration workflow required for issue #656.

jumski

See inline comments.

jumski · 2026-09-04T19:48:41Z

+  -- means a visibility extension did not run (#656). SQL functions cannot
+  -- RAISE, so the mismatch branch casts a descriptive message to int4:
+  -- the cast error fails the whole statement, rolling back the task
+  -- transition and attempt increment, and returns nothing.


is it really the best approach? can't we use "or raise"?

jumski · 2026-09-04T19:48:41Z

+          'start_tasks(): visibility updated %s of %s claimed messages',
+          updated.updated_count,
+          claimed.claimed_count
+        )::int4


ok i get it now, the error message will surface as it will be casted to int4: 'cannot cast to int4' i think it is acceptable

at what circumstance it can happen?

jumski · 2026-09-04T19:48:41Z

  left join deps_outputs dep_out on
    dep_out.run_id = st.run_id and
    dep_out.step_slug = st.step_slug
+  cross join _vr


are those additional cross joins optimal or impacting performance much?

jumski · 2026-09-04T19:48:41Z

+psql_in < "$consolidated"
+
+psql_in < supabase/upgrade_fixture/assertions.sql
+echo "upgrade fixture: PASS"


why is this whole file even needed?

jumski · 2026-09-04T19:48:41Z

+       (select ids from vispartial_msgs),
+       '11111111-1111-1111-1111-111111111111'::uuid
+     ) $$,
+  'invalid input syntax for type integer: "start_tasks(): visibility updated 1 of 2 claimed messages"',


im thinking - is it optimal to fail whole batch if one of the messages is missing? just curious, convince me it is

jumski · 2026-09-04T19:48:41Z

+  end if;
+end $$;
+
+select 'PASS: 0.15.0 upgrade fixture (repairs + runtime)' as result;


why this is needed?

jumski · 2026-09-04T19:48:41Z

@@ -0,0 +1,88 @@
+-- Upgrade fixture seed: stale data both consolidated-migration repairs target.


why this is needed?

jumski · 2026-09-04T20:09:57Z

Merge activity

Sep 4, 8:09 PM UTC: A user started a stack merge that includes this pull request via Graphite.
Sep 4, 8:10 PM UTC: @jumski merged this pull request with Graphite.

github-actions · 2026-09-04T20:11:42Z

🚀 Production Deployment: Website

✅ Successfully deployed to production!

🔗 Production URL: https://pgflow.dev

📝 Details:

Commit: 078aa1e43caae4b4930a2d1d29a55773814f7d42
View Logs

Deployed at: 2026-09-04T22:10:06+02:00

jumski deployed to preview September 4, 2026 10:40 — with GitHub Actions Active

jumski deployed to preview September 4, 2026 12:48 — with GitHub Actions Active

jumski commented Sep 4, 2026

View reviewed changes

jumski merged commit 078aa1e into main Sep 4, 2026
16 checks passed

jumski mentioned this pull request Sep 4, 2026

Release 0.15.1 — task lifecycle hardening #666

Closed

12 tasks

		@@ -0,0 +1,88 @@
		-- Upgrade fixture seed: stale data both consolidated-migration repairs target.

Uh oh!

Conversation

jumski commented Sep 4, 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 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🦋 Changeset detected

Uh oh!

jumski commented Sep 4, 2026

Uh oh!

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

Uh oh!

Uh oh!

jumski left a comment

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

jumski Sep 4, 2026

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Merge activity

Uh oh!

Uh oh!

github-actions Bot commented Sep 4, 2026

🚀 Production Deployment: Website

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jumski commented Sep 4, 2026 •

edited

Loading

changeset-bot Bot commented Sep 4, 2026 •

edited

Loading

nx-cloud Bot commented Sep 4, 2026 •

edited

Loading

jumski commented Sep 4, 2026 •

edited

Loading