Epic: staged private per-step queues and queue identity

[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 first fixes the existing terminal-state, claim-visibility, and stalled-recovery paths that queue identity will touch.

Next, #647 removes the legacy compiler so every queue change extends one mandatory startup path. #650 can then stop treating flow_slug as queue identity without carrying temporary compiler compatibility. #651 builds generated private per-step queues on that final storage and deployment foundation.

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. Legacy compiler cleanup can start.

Stage 0 — startup-only compilation

• Remove legacy flow compilation before queue identity and routing #647 — Remove ControlPlane, migration-based flow compilation, and the obsolete FlowWorkerConfig.compilation option.

Complete #647 before #650. Queue identity and route metadata then extend only the final startup compiler, with no temporary old-worker or migration-compiler compatibility layer.

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, 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 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

#647 makes worker startup the only supported flow compiler before queue work starts. 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.

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

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 adds only the missing scoped capability and repeats validation. A stop result publishes neither #650 nor #651, while retaining #647's startup-only simplification. Any alias or shared-queue proposal still needs its own concrete justification.

Release sequence

0.15.1 hardening release #666 (done)
  -> startup-only compiler cleanup: #647
  -> database queue identity: #650
  -> private per-step queues: #651
  -> freeze one commit containing all three implementations
  -> validate that exact commit downstream
  -> adopt: continue to rollout and release
     revise: make the scoped change and repeat validation
     stop: remove #650 and #651 while retaining #647
  -> finalize #651 rollout docs (#654 is done)
  -> run migration, concurrency, package, and docs checks
  -> merge the Changesets release PR for one stable startup-only release

#647, #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. The final release intentionally requires upgraded workers because #647 removes the old three-argument startup SQL signature; use #647's fenced upgrade instructions instead of promising an old/new worker rolling upgrade.

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 removes the legacy compiler and optional worker compilation before Persist physical queue identity for flow tasks #650 starts.
• 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 for coordinated package-and-database upgrades.
• 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.
• 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 combined commit for Remove legacy flow compilation before queue identity and routing #647 + Persist physical queue identity for flow tasks #650 + Add private per-step queues with typed step workers #651 before the release decision.
• The Changesets release PR remains unmerged until Remove legacy flow compilation before queue identity and routing #647, 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.