Persist physical queue identity for flow tasks

[jumski]

Summary

Separate physical queue identity from concrete flow identity throughout the SQL Core while preserving one-flow/one-queue routing. This issue also introduces stricter shared slug validation and a coordinated upgrade from 0.16.0.

This issue provides the task-identity foundation for private per-step queues in #651. A future #652 may reuse that identity, but it owns any metadata required for explicit shared queues. This issue adds no public routing API and no queue registry.

Dependencies

• Parent epic: Epic: staged private per-step queues and queue identity #653.
• 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 shipped in 0.15.1; preserve their terminalization, lock-order, timeout, recovery, and visibility behavior.
• Remove legacy flow compilation before queue identity and routing #647 is complete: refactor: make worker startup the only flow compilation path #672 implemented startup-only compilation and Version Packages #674 released it in 0.16.0. The baseline worker already calls the two-argument ensure_flow_compiled() function.

Implement queue identity only through the startup compiler retained by #647. create_flow() and add_step() remain database building blocks, but compileFlow(), ControlPlane, and pgflow compile no longer exist.

Deliver this issue before #651, using one normal deliver-task invocation for implementation, review, and PR delivery. Establish the queue-capable startup and claim contracts here so #651 adds mode/routes and exact step selection without replacing a temporary protocol. Do not add per-step workers or routing in this issue.

Current problem

The SQL Core currently uses flow_slug as queue identity in every path:

• flow creation provisions a same-named PGMQ queue;
• ready tasks are sent to that queue;
• task claiming treats one argument as both flow and queue;
• completion, failure, retry, skip, recovery, pruning, and deletion infer the queue from the flow.

PGMQ message IDs are queue-scoped. The durable identity is:

(queue_name, message_id)

A message ID without its queue is not globally meaningful.

Data model

Add resolved queue identity to definitions and immutable queue identity to runtime tasks:

pgflow.steps
  queue_name text not null

pgflow.step_tasks
  queue_name text not null
  message_id bigint null

Queue names stored by pgflow are lowercase canonical physical identities and must satisfy the existing 47-character compatibility limit.

Keep the task primary key:

(run_id, step_slug, task_index)

Add a partial unique index suitable for queue-message lookup:

create unique index ...
on pgflow.step_tasks (queue_name, message_id)
where message_id is not null;

message_id remains nullable. Existing cleanup tests deliberately cover tasks without a queue message; this issue must not redefine that state.

Do not add queue_name to runs or step_states. Do not add a pgflow.queues registry solely for generated private queues.

Deferred manual completion

#661 explores queue-less tasks completed by trusted external code. This issue intentionally models only queue-backed execution and keeps its queue identity non-null.

That choice is scoped to #650. It is not a rejection or final storage decision for #661. If the manual-task design advances, it must explicitly reconcile this model through nullable identity, an execution mode, or another migration. #661 is not a prerequisite for this issue.

Shared slug validation

Apply the same rules to flow and step slugs in TypeScript and SQL:

• reject a leading or trailing _;
• reject __ anywhere;
• allow single internal underscores and camelCase;
• preserve existing allowed characters, no-leading-digit, length, and reserved-word rules;
• reject case-only duplicate flow slugs, and case-only duplicate step slugs within each flow.

__ belongs exclusively to generated queue names. Case-insensitive uniqueness must hold atomically, including concurrent compilation and direct database building-block calls. Preserve the exact accepted slug spelling; do not lowercase concrete flow or step identities.

These rules let #651 use readable lower(flow_slug || '__' || step_slug) names without ambiguous separators. Numeric index fallbacks cannot equal readable step names because slugs cannot start with a digit. Do not add special internal/ghost-step naming flags or alternate validation paths before that feature exists.

This is a breaking validation change for plain and step-queued flows. The audit and migration requirements below must cover existing definitions, including definitions without active runs. No automatic rename or deletion is allowed.

Generated private queue ownership

The current stages create only deterministic private queues. Their persisted flow definition is sufficient ownership evidence:

plain flow
  generated default queue = lower(flow_slug)

step-queued flow in #651
  generated queues = the complete persisted step route

A generated queue may be provisioned or reused only when one of these conditions holds:

• the physical queue is absent and the compiler creates it while persisting the definition;
• an existing concrete definition has the exact matching persisted default or step route and the PGMQ objects are valid.

For a missing concrete definition, an existing physical queue with the generated name is a collision. Reject it instead of silently adopting it. Also reject a generated name already derived or referenced by another concrete flow.

Do not duplicate PGMQ metadata spelling in pgflow. Resolve the exact pgmq.meta.queue_name at metadata-sensitive operations such as drop_queue(). Require exactly one case-insensitive metadata match and reject aliases such as distinct Orders and orders rows because they address the same physical tables.

This derived model is intentionally limited to generated private queues. If #652 is adopted, it must add only the metadata needed for explicit shared queues and adoption without changing task identity again.

Keep the flow lock and make ownership inspection/provisioning/drop race-safe at every entry point, including add_step(), local recompilation, and deletion. Reject case aliases atomically. External PGMQ queues can still collide with valid generated names: a queue that appears after preflight must not be silently adopted by an idempotent create call. Use the minimum necessary database constraints and locking; no queue registry or global lock solely for ambiguous slug concatenation.

Provisioning boundary

The complete compiler preflight is the primary provisioning boundary:

compile complete definition
  -> resolve every required canonical queue name
  -> validate all names, collisions, PGMQ metadata, and physical objects before mutation
  -> persist the flow identity
  -> provision missing generated queues atomically
  -> create the steps with the complete route

create_flow()
  creates only the flow definition

add_step()
  defaults omitted queue_name to lower(flow_slug)
  uses the same generated-queue validation and provisioning path
  stores queue_name

start_ready_steps()
  performs no queue DDL

A plain flow with zero steps still provisions its generated default queue for backward compatibility. #651 rejects withStepQueues() for an empty flow.

Task creation

start_ready_steps() must:

1. Read each ready step's steps.queue_name.
2. Group messages by queue where batching needs it.
3. Send each batch to its resolved queue.
4. Store the same queue name on every inserted task.

The task snapshot never changes after insertion.

Queue-aware operations

Every PGMQ operation must use the task queue snapshot rather than the run's flow slug. This includes:

• claiming and visibility changes;
• completion and late-callback archival;
• retries and exhausted-task archival;
• condition failures and skip cascades;
• stalled-task recovery and permanent-stall handling;
• maintenance pruning;
• flow deletion.

Operations over several tasks must group by queue_name.

Separate the worker subscription from handler identity in claiming. The final claim boundary must carry at least:

queue_name
flow_slug
message_ids
worker_id

Preserve a compatibility wrapper for the existing plain-worker SQL claim signature. It resolves the flow's canonical default queue and delegates to the queue-aware boundary. This wrapper does not promise startup compatibility with released 0.16.0 workers; the startup contract below rejects mismatched workers.

Represent PGMQ bigint message IDs as decimal strings at the JavaScript boundary. Cast them to bigint[] only in SQL calls; do not rely on unsafe JavaScript number precision.

#651 adds an exact step selector to this protocol without changing queue identity or rebuilding batch classification.

Complete batch classification

Look up each read message by (queue_name, message_id) and validate the worker subscription before judging the message body. Inspect the complete batch before mutation:

exact eligible queued task -> claim
exact started task -> defer without another attempt
exact terminal 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 malformed-looking body is not grounds to archive a matching live task. Check any available pgflow task address before treating a message as foreign; uncertain identity is fatal rather than silently discarded. Envelope classification is not runtime validation of application inputs.

When no fatal message exists, one transaction may claim eligible tasks, defer started tasks, and archive terminal or clearly foreign messages. Warnings include queue/message IDs but not bodies. Document that applications must not send directly into pgflow-owned queues.

If any message is fatal, claim none and perform no partial task mutation or archival. Commit visibility reset for the complete read batch and the persistent HTTP restart pause, then return an explicit fatal outcome. The worker emits one fatal error without bodies and stops. Do not raise a SQL error after reset/pause and roll those writes back. Keep ordinary database/network errors retryable; add the terminal-result path to the worker/poller here rather than rebuilding it in #651.

Preserve the current margins: claim visibility uses coalesce(step.opt_timeout, flow.opt_timeout) + 2 seconds; stalled recovery uses the same effective timeout plus 30 seconds. A visible started message is benign. Defer it relative to its existing recovery deadline, not a fresh full timeout after every read.

Deletion

delete_flow_and_data() must perform the following in one transaction:

1. retain and lock the concrete identity while collecting and validating its complete private route, task snapshots, exact PGMQ metadata spelling, and physical objects;
2. acquire relevant runtime locks in the established parent/run, step, task, then queue order, consistently with callbacks and recovery;
3. delete runtime rows and step definitions while retaining the flow identity and captured validated queue names;
4. drop only those private queues and their archives;
5. delete the concrete flow identity row last.

Do not archive or individually delete messages immediately before dropping the whole private queue and archive. That extra work preserves no history and introduces queue-before-task lock hazards. Task snapshots must still agree with the validated private route before rows disappear.

A missing, ambiguous, malformed, or differently referenced queue must fail safely instead of dropping an uncertain physical resource. Failure rolls back the deletion. If #652 later introduces explicit shared queues, it must retain their queues and perform flow-specific message cleanup instead.

Optional pruning-function upgrade

The documented pgflow.prune_data_older_than(interval) function is manually installed, not part of default migrations. Updating the source snippet does not update installed copies.

Update the snippet and its tests to use task queue snapshots, enumerate persisted private routes for archive pruning, and preserve task-before-queue lock order. The upgrade audit reports whether this optional function is installed. Provide copyable instructions to replace the stock installed definition before step-queued work begins; flag customized definitions for manual adaptation rather than overwriting them automatically.

Do not promote the optional destructive helper into the managed schema for this feature. Add an upgrade scenario with an existing installed helper, not only fresh-database tests.

Read-only upgrade audit and transactional migration

Provide one copyable read-only SQL report that runs against 0.16.0 before the upgrade. It must identify affected flow/step definitions and reasons, including:

• leading/trailing _, embedded __, and case-only duplicate identities, even when a flow has no active runs;
• the steps/tasks to backfill, including null message IDs;
• missing, ambiguous, or malformed generated PGMQ objects and conflicting ownership;
• duplicate queue-message identities and active messages without matching task identities in pgflow-owned queues;
• the installed optional pruning function that needs replacement or adaptation.

Return every incompatible slug with its exact flow/step name and reason. For large task/message problems, use bounded counts and sample keys with repair hints. The audit performs no repair, rename, archival, or deletion. A clean result is a point-in-time report, not authorization to skip migration checks.

The migration must recheck under explicit locks before mutation. Document the lock set/order and a bounded lock_timeout; account for producers calling start_flow(), definition/deletion calls, maintenance, and recovery cron as well as worker callbacks. The HTTP worker fence alone does not stop those writers. Use a documented maintenance fence or tested lock-and-wait behavior so concurrent activity cannot invalidate preflight. Inspect queue contents/objects only for pgflow-owned candidates; inspect the global namespace as needed for collisions, not as a general audit of unrelated application queues.

Reject incompatible existing slugs, case aliases, duplicate identities, unmatched active messages, missing objects, and uncertain ownership with actionable errors. Leave the database unchanged on failure. Do not automatically rename or delete definitions, runs, or queues to make the upgrade pass.

Generate structural/function changes from schema source with Atlas, but explicitly preserve the migration-only preflight and data-backfill sections that a schema diff cannot infer. In one transaction, perform locked preflight, add columns in a backfillable form, backfill queue_name = lower(flow_slug) on existing steps and tasks, then enforce non-null constraints/indexes and install the final functions. Null message_id rows receive queue identity too. Refresh the Atlas checksum after the required data-migration edits.

Add a 0.16.0-to-queue upgrade fixture while retaining the existing 0.15.0 hardening fixture. Cover both successful backfill and atomic rejection with populated old tables. Include mixed-case PGMQ 1.5.1 metadata: create, migrate, archive, drop, and recreate a camelCase queue; check metadata, queue/archive tables, and sequences. Resolve exact spelling from pgmq.meta rather than persisting a duplicate copy.

Compatibility

This issue makes no routing change:

step queue = lower(concrete flow_slug)
task queue = lower(concrete flow_slug)
worker queue = the same canonical physical queue

A mixed-case worker argument still reaches the same PGMQ tables. Persisted step and task identity always use lowercase canonical names.

The released baseline is 0.16.0, whose workers already use the two-argument startup function. Removing the earlier three-argument overload does not reject these workers. Establish an explicit queue-capable startup signature/handshake here and keep it extensible for #651, without a temporary old-worker compatibility layer.

Support this coordinated-upgrade matrix:

new database + new plain worker with compliant slugs
  -> supported

new database + released 0.16.0 worker
  -> rejected at startup before registration or polling

0.16.0 database + new queue-aware worker
  -> rejected at startup before registration or polling

Keep the plain Flow and EdgeWorker.start(flow, config) call signatures and default routing. Do not claim unconditional source compatibility: the stricter slug rules reject some previously valid definitions. Users run the audit and resolve incompatible definitions before the migration, without automatic data deletion or renaming.

Build on #647's fenced package-and-database upgrade instructions, including this migration's producer/maintenance boundary and optional pruning-function replacement. Do not promise an old/new rolling-worker upgrade. #650 may remain operational if #651 is delayed, but it is not a stable release by itself; #653 owns the combined validation and release gate.

Acceptance criteria

• steps.queue_name and step_tasks.queue_name store lowercase canonical queue identity without a new queue registry.
• step_tasks.message_id remains nullable.
• Non-null queue messages use unique (queue_name, message_id) identity.
• Existing steps and tasks backfill to lower(flow_slug).
• TypeScript and SQL share the no-leading/trailing-underscore and no-double-underscore slug rules while preserving camelCase and single internal underscores.
• Case-only duplicate flows and steps within one flow are rejected atomically; no speculative internal-step bypass exists.
• Generated queue validation rejects silent adoption, cross-flow collisions, malformed objects, and case aliases.
• An exact persisted definition may verify and reuse its generated queues idempotently.
• Metadata-sensitive operations resolve the exact spelling from pgmq.meta and reject zero or multiple matches.
• The complete compiler preflight provisions the default queue for an empty plain flow.
• create_flow() and start_ready_steps() perform no queue DDL.
• add_step() resolves and provisions the default queue through the shared generated-queue path.
• start_ready_steps() snapshots each resolved queue.
• Every visibility, archive, retry, skip, recovery, pruning, and deletion path uses task queue snapshots.
• Multi-task operations group by queue.
• Complete batch classification is implemented here: queued, started, terminal, clearly foreign, and fatal unsupported work have distinct outcomes.
• Clearly foreign untracked messages warn/archive without blocking valid work; apparently genuine or ambiguous unsupported work is preserved.
• Fatal batches commit reset/pause without task mutations or archival, return an explicit outcome, and stop the worker without a retry loop or message-body logging.
• Visibility and recovery retain their distinct +2/+30-second margins.
• JavaScript represents PGMQ message IDs without precision loss.
• Flow deletion validates task snapshots/private ownership, preserves runtime-row-before-queue lock order, and drops only that flow's queues without redundant per-message archival.
• The optional manually installed pruning function has a queue-aware replacement, upgrade guidance, and an installed-helper test.
• A read-only 0.16.0 audit reports exact incompatible slugs, case conflicts, affected data, queue hazards, and installed pruning without changing anything.
• Upgrade preflight locks before inspection, accounts for producers/maintenance/recovery, uses bounded lock waits, and leaves the database unchanged on any failure.
• Migration-only preflight/backfill precedes final constraints and never silently renames or deletes incompatible existing definitions.
• A new 0.16.0 upgrade fixture covers normal rows, null IDs, incompatible slugs, duplicate identities, active unmatched messages, missing PGMQ objects, and mixed-case metadata; the old hardening fixture remains.
• New workers pass against the migrated database, while either mismatched package/database direction fails before registration or polling.
• No test, fixture, type, or runtime path reintroduces compileFlow(), ControlPlane, pgflow compile, or the removed startup signature.
• Existing one-flow/one-queue execution passes end to end for upgraded applications with compliant slugs; the breaking validation and startup boundary are documented.

Out of scope

• A general queue registry.
• Persisted shared-queue ownership or adoption metadata.
• Per-step queue naming or workers.
• Public custom queue routing.
• Shared queue consumption.
• Multi-flow worker registries.
• Mutable routes.
• Automatic deletion of shared queues.
• Manual tasks or external completion, tracked separately in Design queue-less manual completion tasks (queue: false) #661.
• The requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 effective-timeout fix.