Epic: staged private per-step queues and queue identity

[jumski]

Outcome

Deliver private per-step queues in stages while preserving one-flow/one-queue execution for plain flows. This release tightens slug validation and requires a coordinated package/database upgrade:

one concrete flow
  -> one private queue per step
  -> one independently configured worker pool per step

This removes cross-step starvation while preserving one typed DAG, automatic dependency release, and one run. Stable aliases, custom queue names, and multi-flow sharing remain deferred follow-ups that require separate evidence.

Why staged delivery

The terminal-state, claim-visibility, and stalled-recovery prerequisites shipped in 0.15.1. #647 removed the legacy compiler through #672 and shipped separately in 0.16.0 through #674.

The queue work starts from 0.16.0. #650 stops treating flow_slug as queue identity and establishes the queue-aware claim contract. #651 adds generated private per-step queues and exact step selection on that foundation. Neither stage carries a temporary legacy compiler or a second claim protocol.

A downstream validation then decides whether private per-step queues are useful enough to release and whether any deferred routing work has a concrete use case.

Current ownership model

Flow
  DAG, handlers, conditions, and execution semantics

StepQueuedFlow
  opt-in private queue-per-step deployment mode

Persisted step route
  generated physical queue for future tasks and ownership evidence

Task
  concrete flow slug and immutable queue snapshot

Plain flow worker
  one concrete flow and its generated default queue

Step worker
  one concrete flow and one generated private step queue

The current stages add no pgflow.queues registry. Generated private ownership derives from the concrete flow's default queue or complete persisted step route.

Compatibility defaults

plain Flow
  every step queue = lower(flow.slug)
  worker queue = the same generated default
  no worker selector

withStepQueues(flow)
  every step gets a generated private queue
  worker requires typed stepSlug

No queue option is added to .step(), .array(), or .map() in these stages.

Committed stages

Correctness prerequisites

• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638 — skipped paths terminalize their task rows, fixed by Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649 on 2026-08-28.
• Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645 — failed runs terminalize unfinished task rows and recovery ignores terminal runs and steps, fixed by fix(core): terminalize unfinished tasks when runs fail #663 on 2026-09-02.
• requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 — stalled recovery uses the effective step timeout, fixed by fix(core): use effective step timeout for stalled tasks #664 on 2026-09-02.
• Make start_tasks() apply task visibility before handlers run #656 — start_tasks() applies task visibility before handlers run, fixed by fix(core): guarantee task visibility extension in start_tasks() #667.

The complete correctness set shipped in 0.15.1 through #666.

Stage 0 — startup-only compilation (released)

• Remove legacy flow compilation before queue identity and routing #647 — Removed ControlPlane, migration-based flow compilation, and the obsolete FlowWorkerConfig.compilation option; implemented by refactor: make worker startup the only flow compilation path #672 and released in 0.16.0 through Version Packages #674.

Queue identity and route metadata extend only this startup compiler. The published 0.16.0 worker already calls the two-argument startup function; removal of the older three-argument function does not reject that worker.

Stage 1 — database queue identity

• Persist physical queue identity for flow tasks #650 — Persist canonical queue identity on steps and immutable queue snapshots on tasks.

This stage changes no routing and adds no queue registry. It updates every send, claim, visibility, archive, retry, skip, recovery, pruning, and deletion path to use (queue_name, message_id). It also owns the shared slug rules, upgrade audit, startup compatibility boundary, and message classification/fatal-result handling that #651 extends.

Stage 2 — private per-step queues

• Add private per-step queues with typed step workers #651 — Add withStepQueues(flow) and typed per-step workers.

Generated queues are lowercase, private to one concrete flow, collision-checked, and at most 47 characters. Plain flows keep the same worker API and routing, subject to the new shared slug restrictions.

Validation gate

• Validate one concrete StepQueuedFlow in a downstream application or focused spike before the stable Stage 2 release.

The validation must measure whether generated private queues remove meaningful application-owned queue and worker code while preserving domain correctness. It must not require aliases, named shared queues, or multi-flow workers.

Choose the validation application or focused application spike before this gate, not before #650. Older linked application plans that require aliases or shared routing do not add prerequisites to the committed queue stages. An internal two-step queue test provides early feedback but does not substitute for application-level validation.

Required rollout documentation

• Document the worker-function enable fence for production deployments #654 — Document the current one-function enable fence for in-place production replacement, fixed by docs: fence worker_functions row before deprecating workers in update guide #670.
• Add private per-step queues with typed step workers #651 — Document complete per-step replacement, new-version queue coverage, and old-version drain.

#654 is complete. Finish #651's rollout documentation before the first stable release that changes worker placement.

Independent integration parity

• Allow flow handlers to select the next retry delay #655 — Allow handlers to select the next retry delay.

#655 does not change queue architecture or block #650/#651 implementation. If the chosen application needs provider-selected Retry-After, implement the minimum delay override before claiming that application's retry parity, or report the gap explicitly. Keep attempt accounting unchanged; do not treat a retry-policy gap as evidence that private queue placement failed.

Deferred follow-ups

• Add stable flow aliases through DeployedFlow startup deployment #648 — stable aliases through DeployedFlow;
• Add explicit shared queues with multi-flow dispatch #652 — explicit shared queues and multi-flow dispatch;
• Design queue-less manual completion tasks (queue: false) #661 — queue-less manual completion tasks.

These issues are not completion requirements or prerequisites for this epic. Their APIs, storage models, and release order remain separate decisions. Start #648 or #652 only after Stage 2 validation records a concrete need; #652 also requires explicit evidence that grouping or sharing queues is valuable.

Cross-stage invariants

Startup compilation

The released #647 implementation makes worker startup the only supported flow compiler. Worker startup sends the complete ordered structural definition. Postgres resolves deployment metadata, validates shape, queue mode, and the ordered route map before mutation, and persists them under the flow lock. #650 establishes an explicit queue-capable startup contract against the 0.16.0 baseline; #651 extends it without a temporary intermediate protocol.

#650 and #651 extend only that startup path. Production mismatches fail without destructive recompilation. Existing automatic local same-slug recompilation remains unchanged. Handler code and TypeScript contracts that shape comparison cannot observe still require developer-managed concrete versioning.

Queue identity

PGMQ message identity is:

(queue_name, message_id)

Every queue operation uses the immutable task snapshot. flow_slug never substitutes for queue identity after #650.

Generated private ownership

Generated default and per-step queues are private pgflow resources. Ownership derives from the concrete flow's persisted default or complete step route rather than a separate registry.

For a missing concrete definition, compilation rejects an existing physical queue with a generated name instead of silently adopting it. An exact existing definition may verify and reuse its generated queues idempotently. Cross-flow and case-alias collisions fail before mutation.

Resolve exact PGMQ metadata spelling from pgmq.meta when a metadata-sensitive operation needs it. Do not persist a duplicate spelling solely for generated private queues.

Keep the flow lock and make case-insensitive uniqueness and ownership rejection atomic across compilation, provisioning, local recompilation, and deletion. Naming restrictions remove ambiguity between valid generated names, not conflicts with queues created outside pgflow. Never silently adopt a physical queue that appears during provisioning. Do not add a global lock solely to compensate for ambiguous slug concatenation.

Whole-flow deletion validates the complete private route, preserves the established runtime-row-before-queue lock order, and drops the private queues and archives transactionally. Do not archive individual messages immediately before dropping both tables. #650 also covers the upgrade of the optional manually installed pruning function.

If #652 is adopted, it owns the minimum metadata needed for explicit shared queues without changing task identity again.

Shared slug rules and upgrade audit

Use one validation contract for flow and step slugs in TypeScript and SQL:

• no leading or trailing _;
• no __ anywhere;
• single internal underscores and camelCase remain valid;
• existing character, leading-digit, length, and reserved-word restrictions remain;
• reject case-only duplicate flow slugs, and case-only duplicate step slugs within one flow, because queue names normalize to lowercase.

__ is reserved for generated queue names. Do not add internal-step flags, alternate validators, or bypasses for hypothetical ghost steps.

This is a breaking validation change. #650 provides a read-only pre-upgrade SQL report with exact incompatible flow/step names and reasons, including definitions without active runs, case-only conflicts, and queue/data hazards. Incompatible existing data blocks the migration without automatic renaming or deletion. The migration rechecks under its locks; an earlier successful audit cannot replace that check.

Generated per-step names

Use one fixed compatibility contract:

maximum length = 47
readable = lower(flow_slug || '__' || step_slug)
fallback = lower(flow_slug || '__' || zero_based_step_index)

Use the readable name when it fits, otherwise the index fallback. Reject the complete flow before mutation if both exceed 47.

TypeScript mirrors the algorithm for synchronous feedback. SQL remains authoritative and also calls pgmq.validate_queue_name().

Flow immutability and versions

A concrete flow_slug identifies one immutable production shape and queue mode. Shape or queue-mode changes require a new concrete slug.

Generated queues use concrete slugs. Existing tasks never move to another concrete version.

Worker coverage

One worker polls one queue. A private step worker supports one exact (flow_slug, step_slug) pair.

Compilation does not wait for every worker. Missing workers leave tasks queued durably. Startup logs identify only the worker's selected flow, step, and queue. #651 supplies a copyable post-deployment query that compares persisted routes with live worker rows; operators check coverage before switching callers to a new concrete version. Add no startup coordinator or global readiness protocol.

Message classification and unexpected work

#650 establishes this protocol once; #651 adds the exact step subscription. Inspect the complete batch before mutation and use persisted task identity before judging message appearance:

queued exact eligible task -> claim
started exact task -> benign duplicate visibility; consume no attempt
terminal exact task -> archive idempotently
clearly foreign message with no matching task -> archive, warn, continue
apparently genuine pgflow work with missing task or wrong route -> fatal unsupported work

A body that looks unusual must not cause archival of an existing task. If identity or origin is ambiguous, preserve the message and use the fatal path. If no fatal message exists, valid work and clearly foreign messages may be processed in the same transaction.

If any message is fatal, claim none and perform no partial task mutation or archival. Commit the complete batch visibility reset and persistent HTTP restart pause, then emit one fatal error and stop. Return an explicit outcome rather than raising a SQL error that rolls back the reset/pause. Ordinary database failures remain retryable. Warnings and fatal errors include queue/message identifiers, never message bodies.

Document that applications must not send directly to pgflow-owned queues. Preserve current timing: claim visibility uses the effective step timeout plus 2 seconds; stalled recovery uses the effective timeout plus 30 seconds. Defer a visible started task toward its existing recovery deadline, not a fresh full timeout on every read. Never classify it as corruption.

Production rollout

Use #654's enable fence for current in-place worker replacement. #651 extends it to the complete affected step-worker set and documents new concrete-version drain.

• In-place replacement disables, drains, deploys, and restores the affected functions.
• A new concrete version starts healthy workers on independent queues before caller switching; old workers remain until old work drains.

Do not add a cross-worker activation protocol. Durable queues hold work while cooperating workers start.

Type-quality requirements

The committed APIs must:

• preserve exact flow handler, dependency, condition, context, and environment inference;
• preserve the exact step-slug union through withStepQueues();
• require stepSlug only for private step-queued workers;
• perform complete synchronous generated-name checks before database access;
• keep SQL authoritative for the complete route.

Do not add a flow-slug generic or conditional string types solely for generated queue-name validation.

Deferred alias and shared-queue issues own their own type contracts if adopted.

Integration validation

The downstream validation must preserve application domain tables, transitions, claims, and terminal outcomes while replacing queue and worker mechanics.

It must report:

• orchestration code removed and retained;
• direct PGMQ references removed and retained;
• independent capacity and starvation behavior;
• success, retry, crash, ambiguity, and concurrency behavior;
• retry and recovery gaps;
• type and autocomplete quality;
• an adopt, revise, or stop recommendation for Stage 2.

An adopt result permits the stable private-queue release. A revise result adds only the missing scoped capability and repeats validation. A stop result publishes neither #650 nor #651; #647 remains part of the released 0.16.0 baseline. Any alias or shared-queue proposal still needs its own concrete justification.

Development and release sequence

Use sequential task deliveries: #650 implementation, review, and merge, then #651 implementation, review, and merge. Each deliver-task invocation owns one issue and one editing implementer at a time; do not invoke it on this epic or add concurrent editing inside one invocation. Agree on the queue-aware startup/claim contracts once, and keep #651's step selector additive. Use the repository's focused, batch, and final check cadence rather than a full release rehearsal for every internal slice.

0.16.0 baseline: lifecycle hardening + #647 (released)
  -> #650 queue identity, audit, upgrade fixture, and claim protocol
  -> #651 private per-step queues and typed workers
  -> start application validation after the first working private-queue slice
  -> close scoped gaps and finish rollout documentation
  -> finalize the queue migration and freeze the combined release candidate
  -> validate that exact candidate downstream and run final checks
  -> adopt: release
     revise: make the scoped change and repeat affected validation
     stop: remove unreleased #650/#651 changes; retain released #647
  -> merge the Changesets release PR for the stable queue release

#650 and #651 each add the appropriate Changeset, but the generated Version Packages PR remains unmerged until this sequence completes. #650 may remain operational if #651 is delayed, but it does not publish alone. If separate branches need intermediate migrations, follow the repository's temporary-migration workflow and finalize/consolidate the queue migration before exact-candidate validation. Never rewrite a released migration.

Use a coordinated package-and-database upgrade, not an old/new rolling-worker promise. #650 must test both mismatched directions against 0.16.0 and explicitly reject startup before registration/polling. The absent pre-0.16.0 three-argument function is not that compatibility check. Build on #647's worker fence and document the migration's boundary for producers, maintenance, and recovery writers.

Acceptance criteria

• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638, Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645, requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621, and Make start_tasks() apply task visibility before handlers run #656 establish truthful terminal state, dispatchable-only recovery, effective timeouts, and reliable claim visibility before queue work.
• Remove legacy flow compilation before queue identity and routing #647 removed the legacy compiler and optional worker compilation and shipped in 0.16.0.
• Persist physical queue identity for flow tasks #650 persists canonical queue identity without a speculative queue registry.
• Plain flow APIs and one-flow/one-queue routing remain unchanged for compliant slugs after a coordinated package-and-database upgrade.
• TypeScript and SQL apply identical slug restrictions; the read-only audit reports all incompatible existing definitions, and migration failure never renames or deletes them.
• Startup compatibility is tested against released 0.16.0 workers and databases, not inferred from removal of an older signature.
• Add private per-step queues with typed step workers #651 provides private per-step queues and independent worker capacity without aliases or shared registries.
• Generated names follow the lowercase 47-character contract and fail atomically with actionable errors.
• Every queue-backed task stores an immutable queue snapshot.
• Claims distinguish queued, started, terminal, clearly foreign, and fatal unsupported messages before mutation; foreign messages warn/archive without stopping valid work.
• Fatal handling commits visibility reset and HTTP restart pause before shutdown; SQL errors cannot roll those safety writes back.
• Whole-flow deletion avoids redundant archival; installed pruning functions have an explicit upgrade path.
• Generated private queue collisions cannot silently adopt external or differently referenced queues.
• Downstream validation records an explicit Stage 2 recommendation.
• Document the worker-function enable fence for production deployments #654's current fence and Add private per-step queues with typed step workers #651's per-step rollout documentation are complete before the stable worker-placement release.
• Downstream validation records the exact candidate containing Persist physical queue identity for flow tasks #650 + Add private per-step queues with typed step workers #651 on the released 0.16.0 baseline, including the final migration artifact.
• The Changesets release PR remains unmerged until Persist physical queue identity for flow tasks #650, Add private per-step queues with typed step workers #651, downstream validation, rollout docs, and final checks finish.

Out of scope

• Stable aliases.
• Explicit named or shared queues.
• Multi-flow worker registries.
• Mutable routing for an existing concrete version.
• Polling several queues from one worker instance.
• Automatic shared-queue deletion.
• Dynamic handler loading.
• Runtime input-schema enforcement.
• Manual approval or external completion in the current stages; tracked separately in Design queue-less manual completion tasks (queue: false) #661.