Epic: staged per-step queues, aliases, and shared queue routing

[jumski]

Outcome

Deliver private per-step queues in stages without breaking the current one-flow/one-queue experience:

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 SQL Core must first fix the existing terminal-state, claim-visibility, and stalled-recovery paths that queue identity will touch. It can then stop treating flow_slug as queue identity without rebasing older correctness bugs across every queue-aware operation.

Once physical identity is correct everywhere, pgflow can expose generated private per-step queues without prebuilding shared-queue ownership, multi-flow registries, alias activation, or another deployment 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.

Bundle #638, #645, #621, and #656 into the 0.15.1 hardening release tracked by #666 before queue-identity implementation. #645, #621, and #650 change the same stalled-recovery path, so their implementation order remains important.

Stage 0 — one deployment path

• Remove ControlPlane and migration-based flow compilation #647 — Remove ControlPlane, migration-based flow compilation, compilation: false, and production allowDataLoss.

Worker startup becomes the only supported compiler and structural verifier. Compilation succeeds before worker-function tracking or registration. Local destructive recompilation must quiesce old workers first.

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).

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 remain unchanged.

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.

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.
• Add private per-step queues with typed step workers #651 — Document complete per-step replacement, new-version queue coverage, and old-version drain.

Land both documentation requirements 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 may change a downstream recommendation, but it does not change queue architecture or block the committed stages.

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

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 concrete-slug lock.

Compilation finishes before worker-function tracking and worker registration. Production mismatches never use destructive recompilation. Local destructive recompilation first drains old runtime workers.

There is no migration-based flow compiler after #647. 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.

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

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 and must appear in monitoring.

Message classification and unexpected work

Every claim classifies the complete batch before mutation:

queued exact task -> claim
started exact task -> benign duplicate visibility; consume no attempt
terminal exact task -> archive idempotently
missing, wrong-queue, or unsupported route -> fatal unsupported work

If any message is unsupported, claim none, reset the complete batch, persistently pause HTTP restart, emit one fatal error without message bodies, and stop. Never classify a visible still-started task 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 creates only the missing scoped capability. A stop result leaves the current one-flow/one-queue behavior unchanged. Any alias or shared-queue proposal still needs its own concrete justification.

Release sequence

hardening release #666: #638/#649 + #645/#663 + #621/#664 (done) + #656
  -> breaking startup-only compilation release: #647
  -> internal queue-identity PR: #650
  -> private per-step queue PR: #651
  -> exact combined snapshot
  -> downstream Stage 2 validation
  -> #654 and #651 rollout documentation complete
  -> one stable #650 + #651 release after migration and concurrency fixtures

Do not publish #650 alone as a stable release, but keep it operational if #651 is delayed. The combined stable release requires the old/new plain-worker compatibility matrix.

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 Persist physical queue identity for flow tasks #650.
• Remove ControlPlane and migration-based flow compilation #647 establishes startup-only compilation, compile-before-registration, and fenced local recompilation.
• Persist physical queue identity for flow tasks #650 persists canonical queue identity without a speculative queue registry.
• Plain one-flow/one-queue behavior remains source-compatible across the old/new worker matrix.
• 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, and unsupported messages before mutation.
• Generated private queue collisions cannot silently adopt external or differently referenced queues.
• 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 land before the stable worker-placement release.
• Downstream validation records an explicit Stage 2 recommendation.
• The staged release sequence uses an exact combined snapshot before stable Persist physical queue identity for flow tasks #650 + Add private per-step queues with typed step workers #651.

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.