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

[jumski]

Outcome

Separate flow behavior, physical queue identity, worker capacity, version aliases, and shared queue dispatch in stages without breaking the current one-flow/one-queue experience.

The first delivery target is intentionally smaller than general queue routing:

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. Custom queue names and multi-flow sharing remain a later stage.

Why staged delivery

The SQL Core must first fix the existing skip, 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 new queue-aware operation.

Once queue identity and ownership are correct everywhere, pgflow can expose private per-step queues without solving shared-worker registries, version coverage, and draining at the same time.

A downstream integration then decides whether aliases and larger shared-queue work are worth building.

Ownership model

Flow
  DAG, handlers, conditions, and execution semantics

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

DeployedFlow
  stable alias membership for one concrete plain or step-queued flow

Postgres queue registry
  canonical physical queue identity and private/shared ownership

Step definition
  resolved physical queue for future tasks

Task
  concrete flow slug and immutable queue snapshot

Plain flow worker
  one concrete flow and its default queue

Step worker
  one concrete flow and one private step queue

Shared queue worker
  one physical queue and a complete registry of concrete flows

Compatibility defaults

plain Flow
  alias = flow.slug
  every step queue = flow.slug
  worker queue = flow.slug
  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().

Staged issues

Correctness prerequisites

• Merge PR Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649 so skipped paths terminalize their task rows.
• Fix requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 so stalled recovery uses the effective step timeout.
• Make start_tasks() apply task visibility before handlers run #656 — Make start_tasks() apply task visibility before handlers run.

Release these as independent hardening patches before queue-identity implementation.

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 and ownership for flow tasks #650 — Persist canonical queue identity and ownership on the database side.

This stage changes no routing. 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

• community-threads#27 — Compare one concrete StepQueuedFlow with generated private classification and delivery queues.

The current downstream issue still requires aliases, named shared queues, and multi-flow workers. Revise that issue before treating it as this gate. Until then, it validates Stage 4 rather than Stage 2.

The corrected integration must measure whether private per-step queues remove meaningful application-owned queue and worker code while preserving domain correctness. Do not make #648 or #652 prerequisites.

Stage 3 — stable aliases

• Add stable flow aliases through DeployedFlow startup deployment #648 — Add alias-first starts and DeployedFlow after the private-queue design is proven.

Aliases compose with both plain and step-queued flows. Generated queues continue to use concrete slugs.

Stage 4 — explicit shared queues

• Add explicit shared queues with multi-flow dispatch #652 — Add explicit step-to-queue maps and queue workers with complete multi-flow registries.

Start this stage only after the integration records an adopt result and a real need for grouping or sharing.

Required rollout documentation

• Document the worker-function enable fence for production deployments #654 — Document manifest-driven in-place replacement and new-version drain as separate production sequences.

Land #654 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 the downstream recommendation, but it does not change queue architecture.

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.

Queue ownership

Generated default and per-step queues are private pgflow resources. Compilation rejects a physical queue that is unowned or owned by another concrete flow.

Explicit queues introduced by #652 are shared/non-owned resources and are never dropped with one flow.

Physical queue identity is canonical and case-safe because PGMQ lowercases table names.

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, never aliases. Existing tasks never move when an alias changes.

Worker coverage

One worker polls one queue. Private step workers support one exact (flow_slug, step_slug) pair. Shared queue workers support a complete set of pairs.

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 explicit old/new function manifest.

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

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

Type-quality requirements

The final APIs must:

• preserve exact flow handler, dependency, condition, context, and environment inference;
• preserve exact step-slug unions through withStepQueues() and defineDeployedFlow();
• require stepSlug only for private step-queued workers;
• preserve explicit queue-name literals for future shared-worker autocomplete;
• reject missing and unknown routing keys;
• perform compile-time queue-name checks where literal information exists;
• perform complete synchronous runtime checks before database access.

Do not add a flow-slug generic solely for generated queue-name length validation.

Integration validation

The downstream spike 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.

An adopt result permits alias and shared-queue follow-up work. A revise result creates only the missing scoped capability. A stop result leaves private per-step queues independently useful.

Release sequence

hardening patches: PR #649 + #621 + #656
  -> breaking startup-only compilation release: #647
  -> internal queue-identity PR: #650
  -> private per-step queue PR: #651
  -> exact combined snapshot
  -> corrected downstream integration gate
  -> one stable #650 + #651 release after migration and concurrency fixtures
  -> separate alias release: #648
  -> shared queues only after measured demand: #652

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

Acceptance criteria

• Existing skip, recovery, and claim-visibility correctness fixes land before Persist physical queue identity and ownership 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 and ownership for flow tasks #650 makes canonical physical queue identity, exact PGMQ metadata spelling, and ownership explicit everywhere.
• 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 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.
• Private queue collisions cannot silently adopt external or differently owned queues.
• Document the worker-function enable fence for production deployments #654 lands before the first stable worker-placement release.
• The corrected integration spike records an explicit recommendation before shared-queue work starts.
• Add stable flow aliases through DeployedFlow startup deployment #648 uses explicit compare-and-swap alias activation with safe version drain.
• Add explicit shared queues with multi-flow dispatch #652 adds explicit shared queues and registry-aware adoption without another task-identity migration.
• The staged release sequence uses an exact snapshot before stable Persist physical queue identity and ownership for flow tasks #650 + Add private per-step queues with typed step workers #651.

Out of scope

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