Persist physical queue identity and ownership for flow tasks

[jumski]

Summary

Separate physical queue identity from concrete flow identity throughout the SQL Core while preserving today's one-flow/one-queue behavior.

This issue provides the complete database foundation for private per-step queues in #651 and explicit shared queues in #652. It adds no public routing API.

Dependencies

• Parent epic: Epic: staged per-step queues, aliases, and shared queue routing #653.
• Land after Remove ControlPlane and migration-based flow compilation #647 makes startup compilation the only deployment path.
• Merge PR Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #649 first because it changes the same failure and skip paths.
• Land requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621's effective-timeout fix first because it changes queue-aware recovery.
• Land Make start_tasks() apply task visibility before handlers run #656's start_tasks() visibility-execution fix first because this issue changes the same claim path.

Keep those fixes in separate releases. This issue consumes their corrected behavior rather than rebasing old queue assumptions over them.

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 references pgflow.queues(queue_name)

pgflow.step_tasks
  queue_name text not null references pgflow.queues(queue_name)
  message_id bigint null

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.

Queue ownership registry

Persist queue ownership separately from step and task state:

pgflow.queues
  queue_name text primary key
  pgmq_name text not null
  ownership private | shared
  owner_flow_slug text null references pgflow.flows(flow_slug)

queue_name is the lowercase canonical physical identity. pgmq_name preserves the exact spelling in pgmq.meta, including mixed-case legacy names, and is used for metadata-sensitive PGMQ operations such as drop_queue().

Constraints:

queue_name = lower(queue_name)
queue_name = lower(pgmq_name)
length(queue_name) <= 47
private ownership -> owner_flow_slug is not null
shared ownership  -> owner_flow_slug is null
unique lower(pgmq_name)

Index owner_flow_slug; PostgreSQL does not add an index for the foreign key automatically.

The model must distinguish:

private queue
  owned by one concrete pgflow flow
  may be dropped with that flow

shared queue
  no exclusive flow owner
  never dropped with one flow

Only private queues are created by this issue. #652 later enables explicit shared queues without changing task identity again. Do not add alias columns, route history, owner_step_slug, or multi-flow registries here.

Requirements:

• reject two registry entries that address the same lowercased PGMQ tables;
• backfill each existing flow-slug queue as private and owned by that concrete flow;
• preserve each legacy queue's exact pgmq.meta.queue_name spelling in pgmq_name;
• reject an existing PGMQ queue that has no matching pgflow ownership record instead of silently adopting it;
• allow an existing matching private owner during idempotent startup;
• reject a queue owned by another concrete flow.

A migration preflight must reject legacy aliases such as distinct Orders and orders metadata rows because they address the same physical tables.

Provisioning boundary

The complete compiler preflight is the primary provisioning boundary:

compile complete definition
  -> resolve every required queue before mutation
  -> register or verify every owner
  -> validate and idempotently provision queues
  -> create the flow and steps

create_flow()
  creates only the flow definition

add_step()
  defaults omitted queue_name to lower(flow_slug)
  calls the same internal registration function
  stores queue_name

start_ready_steps()
  performs no queue DDL

A plain flow with zero steps still provisions its default private queue for backward compatibility. Direct internal SQL calls must use the same queue-registration function. #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 signature. It resolves the flow's canonical default queue and delegates to the queue-aware boundary.

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 may add an exact step selector without changing queue identity again.

Deletion

delete_flow_and_data() must:

1. archive or remove active messages using each task's queue snapshot;
2. delete runtime rows and step definitions while retaining the flow identity row;
3. drop only private queues owned by that concrete flow;
4. pass the exact registry pgmq_name to metadata-sensitive PGMQ deletion;
5. remove each private registry row only after PGMQ deletion succeeds;
6. delete the concrete flow identity row last;
7. never drop a shared or unowned queue.

This order respects ownership foreign keys and supports fenced local destructive recompilation without stale private queues or pgmq.meta rows.

Upgrade preflight

Take the migration lock and explicit table locks before inspection so concurrent pgflow or PGMQ registration cannot invalidate the checked state.

Before mutation:

• detect case-insensitive physical queue aliases in pgmq.meta;
• confirm each metadata row has its expected queue, archive, and sequence objects;
• detect duplicate non-null (queue_name, message_id) identities;
• reject active PGMQ messages without matching pgflow task identities;
• confirm every legacy flow queue has unambiguous private ownership;
• reject queues whose ownership cannot be established;
• return bounded counts and sample keys with repair hints;
• leave the database unchanged on failure.

Backfill queue_name = lower(flow_slug) for existing steps and tasks, including tasks whose message_id is null. Preserve the exact legacy PGMQ spelling in pgmq_name.

Add a mixed-case PGMQ 1.5.1 migration fixture that creates, migrates, archives, drops, and recreates a camelCase queue. Check metadata, queue tables, archive tables, and sequences after every operation.

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. Registry and task identity always use lowercase canonical names.

Support this rolling-upgrade matrix:

new database + old plain worker
  -> supported

new database + new plain worker
  -> supported

old database + new queue-aware worker
  -> fail before registration or polling

Plain flow workers and starts remain source-compatible. This stage must remain operational if #651 is delayed; it is not a stable release by itself.

Acceptance criteria

• pgflow.queues persists canonical identity, exact PGMQ metadata spelling, and private/shared ownership.
• Registry constraints and an owner_flow_slug index enforce the ownership model.
• Existing flow queues backfill as private queues owned by their concrete flows.
• steps.queue_name and step_tasks.queue_name are non-null foreign keys to the registry.
• step_tasks.message_id remains nullable.
• Non-null queue messages use unique (queue_name, message_id) identity.
• Queue registration rejects unowned, differently owned, malformed, missing-object, and case-alias collisions.
• 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 the default queue through the shared registration boundary.
• 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.
• JavaScript represents PGMQ message IDs without precision loss.
• Flow deletion uses exact PGMQ metadata spelling and drops only queues privately owned by that flow.
• Upgrade preflight locks before inspection and leaves the database unchanged on any failure.
• Previous-version fixtures cover normal rows, null IDs, duplicate identities, active unmatched messages, missing PGMQ objects, and mixed-case metadata.
• Old and new plain workers pass the rolling-upgrade matrix.
• Existing one-flow/one-queue behavior passes unchanged end to end.

Out of scope

• Per-step queue naming or workers.
• Public custom queue routing.
• Shared queue consumption.
• Multi-flow worker registries.
• Mutable routes.
• Automatic deletion of shared queues.
• The requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621 effective-timeout fix.