fix(worker): coordinate pending starts with invocation handshakes by jumski · Pull Request #711 · pgflow-dev/pgflow

jumski · 2026-11-02T09:06:54Z

Summary

Prevent duplicate HTTP worker starts while initialization remains pending. Debounce stays a retry throttle, not a readiness signal.

ensure_workers / authenticated manual or local request
  reserve_worker_invocation(function_name)
    lock registration; skip healthy or unexpired pending starts
    reserve invocation_id with a 60-second database-clock deadline
  dispatch { invocation_id }
    reconcile duplicate deliveries / acknowledge an existing healthy worker
    initialize and compile
    accept_worker_invocation
      atomically consume the current invocation and register the ready worker
    on failure: dismiss only the matching pending invocation

Include bounded startup recovery, the generated migration and types, protocol regressions, and SQL-first rollout documentation. A separate formatting-only commit fixes five array-spacing violations required by the pre-push hook; Atlas confirms no schema drift. Process workers remain outside HTTP scheduling. Lifetime ownership and stale-worker fencing remain in #697.

Fixes #696.

Evidence

Before: cron could dispatch another start after the 6-second debounce while the first worker still initialized. A request receipt did not prove readiness.
After: shared SQL reservations prevent another pending start; real-SQL tests cover independent runtimes, duplicates, expiry, superseded requests, lost acknowledgements, and manual/local compatibility.
Before: review regressions exposed a replacement with an aborted signal after watchdog recovery and a delayed reconciliation that stopped a healthy worker.
After: both regressions pass and prove actual task completion after recovery.
Passed: migration verification and migration-backed generated-type checks; full pgTAP (321 files, 1,702 assertions, including upgrade fixtures); integration (81 tests); Edge Function E2E (13 tests); portable-runtime E2E (14 tests across Supabase, Node, and Bun).
Passed: worker unit tests (279 tests plus 17 steps), typecheck/type tests/lint, Node/Bun package smoke checks, and website build/internal links. Standards and Spec reviews, correction rechecks, and generated-migration reviews report zero remaining blockers.
Push gate passed: all 28 Nx tasks completed successfully, including core SQL lint, core/client Vitest type checks, and all affected package pre-push targets. Changeset validation passed. Known baseline issue outside this gate: four type errors remain in unchanged demo sources; its focused production build passes.

Merge Danger

Door: two-way

The database migration adds nullable invocation fields and startup functions without deleting data or changing existing worker registrations. Worker rollback remains compatible; database rollback needs manual SQL because no automatic down migration exists.

Apply the SQL migration before the worker update. Mixed-version rollout remains compatible, but old workers do not enforce the new handshake. Lost or hung startup recovery can take up to 60 seconds.

Blast Radius: startup

Changes affect cron, authenticated manual, and local HTTP worker startup. Authentication remains enforced. Hosted runtimes cannot be forcibly evicted; lifetime ownership and task fencing are intentionally outside this PR.

changeset-bot · 2026-11-02T09:07:00Z

🦋 Changeset detected

Latest commit: 61c75a5

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/edge-worker	Patch
pgflow	Patch
@pgflow/client	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-11-02T09:07:08Z

fix(worker): coordinate pending starts with invocation handshakes #711 👈 (View in Graphite)
main

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

nx-cloud · 2026-11-02T09:07:49Z

View your CI Pipeline Execution ↗ for commit 61c75a5

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

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

☁️ Nx Cloud last updated this comment at 2026-11-02 12:41:49 UTC

jumski · 2026-11-02T10:27:17Z

+  invocation_id uuid,
+  invocation_expires_at timestamptz,
+  invocation_status text check (invocation_status in ('pending', 'ready', 'failed')),
+  invocation_worker_id uuid,


do we allow it to be null? shouldn't it be an fkey?

null is intentional: it is set only when a handshake reaches ready. Left as a soft pointer on purpose — pgflow never deletes worker rows (only deprecates), so an FK would add migration churn without behavior change. Can revisit if workers ever gain cleanup.

jumski · 2026-11-02T10:27:17Z

+  select wf.* into registration
+  from pgflow.worker_functions as wf
+  where wf.function_name = reserve_worker_invocation.function_name
+  for update;


shouldn't we order here so the locking is deterministic?

This is a point lookup on the primary key, so at most one row locks here. The multi-function lock order is pinned in ensure_workers, whose reservations CTE locks in order by wf.function_name.

jumski · 2026-11-02T10:27:17Z

+  if not found or not registration.enabled or registration.start_mode <> 'http' then
+    return null;
+  end if;


why not put the filters like enabled and start_mode into the select wf.* into registrations above? i think this whole function could be language sql/declarative if we just filter appropriate rows? do we need to throw any errors? null (empty result set) can be declaratively returned, right?

Done — now language sql with a locked eligible CTE and one update ... returning. No error paths existed to preserve: null is the only non-reserve result. One real trap surfaced while converting: not (invocation_status = 'pending' and ...) is NULL for a fresh registration, which the plpgsql IF treated as false but WHERE semantics filter out — predicates are now NULL-safe (is distinct from). The full ensure_workers pgTAP suite covers it.

jumski · 2026-11-02T10:27:17Z

+-- Validate every HTTP delivery, including deliveries to surviving runtimes.
+-- A receipt never clears pending state. An existing healthy worker may finish
+-- the handshake without registering a new worker or starting another loop.
+create or replace function pgflow.reconcile_worker_invocation(


is it possible to make this a declarative, language sql function by leveraging cte's etc

Kept plpgsql here and in accept_worker_invocation. Both are multi-exit state machines (pending/ready/rejected plus a conditional update and an idempotent early return); the CTE rewrite of reserve was clean, but these read clearer imperatively. reserve_worker_invocation — the one you also flagged — is now declarative.

jumski · 2026-11-02T10:27:17Z

  with
-    -- Detect environment
-    env as (
+    env as materialized (


why we need to materialize?

PG12+ inlines plain CTEs. materialized pins is_local() and the Vault reads to one evaluation per statement (consistent environment detection, no repeated decryption); the old code relied on pre-PG12 always-materialized CTEs.

jumski · 2026-11-02T10:27:17Z

-    debounce_passed as (
-      select wf.function_name, wf.debounce
-      from http_worker_functions as wf
-      where wf.last_invoked_at is null
-        or wf.last_invoked_at < now() - wf.debounce
-    ),
-
-    -- Find functions that have at least one alive worker
-    functions_with_alive_workers as (
-      select distinct w.function_name
-      from pgflow.workers as w
-      inner join debounce_passed as dp on w.function_name = dp.function_name
-      where w.stopped_at is null
-        and w.deprecated_at is null
-        and w.last_heartbeat_at > now() - dp.debounce
-    ),
-
-    -- Determine which functions should be invoked
-    -- Local mode: all enabled functions (bypass debounce AND alive workers check)
-    -- Production mode: only functions that pass debounce AND have no alive workers
-    functions_to_invoke as (
-      select wf.function_name
-      from http_worker_functions as wf
-      where pgflow.is_local() = true  -- Local: all enabled HTTP functions
-        or (
-          -- Production: debounce + no alive workers
-          wf.function_name in (select dp.function_name from debounce_passed as dp)
-          and wf.function_name not in (select faw.function_name from functions_with_alive_workers as faw)
-        )
-    ),
-
-    -- Make HTTP requests and capture request_ids
-    http_requests as (
-      select
-        fti.function_name,
-        net.http_post(
-          url => c.base_url || '/' || fti.function_name,
-          headers => case
-            when e.is_local then '{}'::jsonb
-            else jsonb_build_object(
-              'Content-Type', 'application/json',
-              'Authorization', 'Bearer ' || c.service_role_key
-            )
-          end,
-          body => '{}'::jsonb
-        ) as request_id
-      from functions_to_invoke as fti
      cross join credentials as c
-      cross join env as e
-      where c.base_url is not null
-        and (e.is_local or c.service_role_key is not null)
-    ),
-
-    -- Update last_invoked_at for invoked functions
-    updated as (
-      update pgflow.worker_functions as wf
-      set last_invoked_at = clock_timestamp()
-      from http_requests as hr
-      where wf.function_name = hr.function_name
-      returning wf.function_name


why all this is removed?

The debounce/alive-worker CTE chain moved into reserve_worker_invocation, which manual startup now shares — the comments moved with the logic. Restored concise per-CTE comments in ensure_workers (env/credentials, one reservation per function, async delivery).

jumski · 2026-11-02T10:27:18Z

-  msg_ids BIGINT[],
-  vt_offsets INTEGER[]
+  msg_ids BIGINT [],
+  vt_offsets INTEGER []


why those changes? is it core:fix-sql? we dont want to introduce like a huge diff/migration due to styling/lint fixes... im pretty sure it is somewhat misconfigured or version-mismatched or wrongly invoked sqruff that causes those...

You were right — version mismatch. Local sqruff 0.39.0 (mise pipx) flags bigint[] in function signatures; CI's setup-sqruff action installs latest (0.44.0), which instead flags the spaced form, so the style commit would have failed CI lint. Dropped 53dcb9a, reset the branch to the compact form, pre-push green under 0.44.0. Upgrading the local sqruff install is the durable fix.

jumski · 2026-11-02T10:27:18Z

+    if (invocationId) {
+      const accept = () => withStartupTimeout(this.sql, (sql) => sql<WorkerRow[]>`
+        SELECT * FROM pgflow.accept_worker_invocation(
+          ${edgeFunctionName}, ${invocationId}::uuid, ${workerId}::uuid,
+          ${queueName}, ${pgflowVersion}
+        )
+      `, invocationId);
+      // A connection can fail after COMMIT. Repeat the idempotent operation
+      // once to recover the same worker, never to authorize another runtime.
+      const [worker] = await accept().catch(() => accept());
+      if (!worker) throw new WorkerInvocationRejectedError();
+      return worker;
+    }
+


is it good to have a separate query? shouldn't this be baked into start_worker or somethign that we already ccall?

This branch replaces the plain INSERT inside onWorkerStarted — the start call we already make — so it is one roundtrip, not an extra query: consume + register atomically. Baking it into ensure_flow_compiled would tie handshake acceptance to flow compilation and change that function's contract.

jumski · 2026-11-02T10:27:18Z

          routes.map(({ stepSlug, queueName }) => ({ stepSlug, queueName })) as unknown as Json
        );
-    const [result] = await this.sql<{ result: EnsureFlowCompiledResult }[]>`
+    const [result] = await withStartupTimeout(this.sql, (sql) => sql<{ result: EnsureFlowCompiledResult }[]>`


can we reliably say that when timeout was triggered, the ensure_flow_compiled have surely rollback?

Yes. ensure_flow_compiled runs as a single statement inside sql.begin: statement_timeout aborts it server-side, postgres.js rolls the transaction back, and if the connection dies the server discards the uncommitted transaction — idle_in_transaction_session_timeout=60s bounds the half-open case. No autonomous transactions (no dblink anywhere in schemas), so nothing can escape the rollback.

jumski · 2026-11-02T10:27:18Z

@@ -9,7 +9,7 @@ import { Aside, CardGrid, LinkCard } from "@astrojs/starlight/components";



File-level: should we have a dedicated page explaining this reconcillation and shit? or some worker torubleshooting?

Went with a section: added Troubleshooting worker startup to worker-management (startup_state, invocation_error, both 409 reasons, duplicate-live-workers guidance) instead of a new page — the handshake explanation already lives in that page's Startup handshake section.

Debounce measured HTTP dispatch time rather than readiness, so slow initialization let cron enqueue duplicate live workers. Reserve one database-clock invocation per function, then atomically accept its current unexpired token with ready-worker registration instead of extending debounce. Use a 60-second pending deadline, idempotent acknowledgement, safe dismissal, and bounded runtime/SQL startup recovery. Authenticated manual and local requests share reservations; process workers stay outside HTTP scheduling. Include the generated migration, types, regression coverage, and SQL-first mixed-version rollout guidance. Lifetime ownership and task fencing remain in #697. Review feedback: reserve_worker_invocation is now a declarative language sql function. Its eligibility predicates are NULL-safe because a fresh registration has no invocation state; the previous `not (status = 'pending' and ...)` form filtered fresh rows out under WHERE semantics while the plpgsql IF treated the NULL as false. reconcile_worker_invocation and accept_worker_invocation stay plpgsql: they are multi-exit state machines whose early returns read clearer imperatively. invocation_worker_id stays a soft pointer without a foreign key: pgflow never deletes worker rows, so the constraint would add migration churn without behavior. A startup troubleshooting section joins the worker management docs. The four baseline array-spacing lint failures were a local sqruff 0.39.0 defect in function signatures; CI installs a current sqruff that passes the original compact form, so no reformat ships and the earlier style commit was dropped. Validation: full pgTAP (1702 assertions), the real-SQL worker-invocation integration file (11 tests), Edge Function E2E (13 tests), portable-runtime E2E (14 tests), migration verification, generated-type checks, and the website build all pass, with the full 81-test integration suite from the pre-review delivery. Fixes #696

github-actions · 2026-11-02T12:51:03Z

🔍 Preview Deployment: Website

✅ Deployment successful!

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

📝 Details:

Branch: 696-worker-startup
Commit: b1c9109597f2798d88bcae6fc8787f0bce388073
View Logs

_Last updated: _

jumski commented Nov 2, 2026

View reviewed changes

jumski force-pushed the 696-worker-startup branch from 53dcb9a to a5a20db Compare November 2, 2026 10:35

jumski deployed to preview November 2, 2026 10:46 — with GitHub Actions Active

jumski force-pushed the 696-worker-startup branch from a5a20db to 63bf753 Compare November 2, 2026 11:08

jumski deployed to preview November 2, 2026 11:18 — with GitHub Actions Active

jumski force-pushed the 696-worker-startup branch from 63bf753 to 61c75a5 Compare November 2, 2026 12:40

jumski deployed to preview November 2, 2026 12:49 — with GitHub Actions Active

This was referenced Nov 2, 2026

Design safe worker-update SQL helpers and linked-project CLI commands #712

Open

fix(core): terminalize permanently stalled tasks through whenExhausted #713

Open

		@@ -9,7 +9,7 @@ import { Aside, CardGrid, LinkCard } from "@astrojs/starlight/components";

Conversation

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

Uh oh!

Summary

Evidence

Merge Danger

Uh oh!

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

Uh oh!

🦋 Changeset detected

Uh oh!

jumski commented Nov 2, 2026

Uh oh!

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

Uh oh!

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!

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!

github-actions Bot commented Nov 2, 2026

🔍 Preview Deployment: Website

Uh oh!

This branch was successfully deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jumski commented Nov 2, 2026 •

edited

Loading

changeset-bot Bot commented Nov 2, 2026 •

edited

Loading

nx-cloud Bot commented Nov 2, 2026 •

edited

Loading