Summary
Make queue identity explicit on flow steps and runtime tasks while preserving today's default behavior: every step still uses the concrete flow_slug queue.
This is the database foundation for later named step routing and queue-centric workers. It adds no public TypeScript routing API.
Dependencies
Problem
The SQL Core currently treats flow_slug as queue identity everywhere:
create_flow() creates a same-named PGMQ queue.
start_ready_steps() sends every task to that queue.
start_tasks() treats one argument as both flow and queue.
- completion, failure, condition, skip, retry, and stalled recovery archive or update messages through
flow_slug.
delete_flow_and_data() drops the whole same-named queue.
This stops working when one flow routes steps to different queues or one queue serves multiple flows.
PGMQ message IDs are scoped to a queue. The durable identity is:
Data model
Add resolved queue identity to definitions and immutable queue identity to tasks:
pgflow.steps
queue_name text not null
pgflow.step_tasks
queue_name text not null
message_id bigint not null
Backfill both columns from the concrete flow_slug, preserving current behavior.
Add an index or unique constraint suitable for queue message lookup:
Keep the existing task primary key:
(run_id, step_slug, task_index)
Upgrade preflight
Manual tasks do not exist before this change. A legacy step_tasks.message_id IS NULL row therefore represents state that cannot be mapped safely to a queue message.
Before any migration mutation:
- Check every existing task status for null message IDs, including terminal rows.
- Abort on any match.
- Return bounded diagnostics with affected counts and a small sample of task keys.
- Include a
HINT that points to repair guidance.
- Leave the database unchanged when the preflight fails.
Add an upgrade fixture from the previous released schema. Cover active and terminal null-message rows.
A later manual-task feature may deliberately make queue and message identity nullable under an explicit execution-mode constraint. Do not pre-build that model here.
Queue provisioning boundary
Change queue ownership as follows:
create_flow()
creates only the flow definition
add_step()
resolves queue_name
validates it
idempotently ensures that queue exists
stores queue_name
start_ready_steps()
never performs queue DDL
add_step() defaults omitted queue_name to flow_slug. Direct SQL callers and startup compilation keep today's behavior.
Both SQL-text compilation and startup shape compilation must use the same add_step() boundary while both paths remain public.
Task creation
start_ready_steps() must:
- Read each ready step's resolved
steps.queue_name.
- Group messages by queue when batching requires it.
- Send each batch to its resolved queue.
- Store the same queue name on every inserted task.
The task snapshot never changes, even if a future flow version uses another route.
Queue-aware operations
Update every PGMQ operation to use the task queue snapshot rather than the run's flow slug.
This includes:
- task claiming and visibility updates;
- completion and late-callback archives;
- retry visibility delays and exhausted-task archives;
- condition failure and skip-cascade archives;
- stalled-task recovery and permanent-stall archives.
Operations over several tasks must group by queue_name. No function may assume that all tasks in one run share one queue.
start_tasks() should accept queue identity separately from flow identity. A later worker issue will allow it to return tasks from several flows on one queue.
Stalled recovery and #621
Fix #621 while changing stalled recovery.
Use the effective timeout:
coalesce(step.opt_timeout, flow.opt_timeout)
Group visibility resets and archives by step_tasks.queue_name.
Tests must cover:
- step timeout shorter than flow timeout;
- step timeout longer than flow timeout;
- null step timeout fallback;
- queue-correct visibility reset.
Deletion and queue ownership
delete_flow_and_data() must stop dropping a queue.
A named queue may contain tasks from other flows, other versions, or non-pgflow users. Flow deletion must archive only that flow's active messages, grouped by task queue, before deleting runtime rows.
Leave queue deletion to a later explicit administrative operation. Retaining an empty queue is safer than deleting shared work.
Compatibility
This issue must make no observable routing change:
step queue = concrete flow_slug
task queue = concrete flow_slug
worker queue = concrete flow_slug
Existing Flow workers and direct SQL starts must continue to work unchanged.
Acceptance criteria
Out of scope
- Public named-routing configuration.
- Queue-centric or multi-flow workers.
- Mutable routes.
- Manual tasks and nullable manual-task queue identity.
- Automatic deletion of unused queues.
Summary
Make queue identity explicit on flow steps and runtime tasks while preserving today's default behavior: every step still uses the concrete
flow_slugqueue.This is the database foundation for later named step routing and queue-centric workers. It adds no public TypeScript routing API.
Dependencies
DeployedFlow.Problem
The SQL Core currently treats
flow_slugas queue identity everywhere:create_flow()creates a same-named PGMQ queue.start_ready_steps()sends every task to that queue.start_tasks()treats one argument as both flow and queue.flow_slug.delete_flow_and_data()drops the whole same-named queue.This stops working when one flow routes steps to different queues or one queue serves multiple flows.
PGMQ message IDs are scoped to a queue. The durable identity is:
Data model
Add resolved queue identity to definitions and immutable queue identity to tasks:
Backfill both columns from the concrete
flow_slug, preserving current behavior.Add an index or unique constraint suitable for queue message lookup:
Keep the existing task primary key:
Upgrade preflight
Manual tasks do not exist before this change. A legacy
step_tasks.message_id IS NULLrow therefore represents state that cannot be mapped safely to a queue message.Before any migration mutation:
HINTthat points to repair guidance.Add an upgrade fixture from the previous released schema. Cover active and terminal null-message rows.
A later manual-task feature may deliberately make queue and message identity nullable under an explicit execution-mode constraint. Do not pre-build that model here.
Queue provisioning boundary
Change queue ownership as follows:
add_step()defaults omittedqueue_nametoflow_slug. Direct SQL callers and startup compilation keep today's behavior.Both SQL-text compilation and startup shape compilation must use the same
add_step()boundary while both paths remain public.Task creation
start_ready_steps()must:steps.queue_name.The task snapshot never changes, even if a future flow version uses another route.
Queue-aware operations
Update every PGMQ operation to use the task queue snapshot rather than the run's flow slug.
This includes:
Operations over several tasks must group by
queue_name. No function may assume that all tasks in one run share one queue.start_tasks()should accept queue identity separately from flow identity. A later worker issue will allow it to return tasks from several flows on one queue.Stalled recovery and #621
Fix #621 while changing stalled recovery.
Use the effective timeout:
Group visibility resets and archives by
step_tasks.queue_name.Tests must cover:
Deletion and queue ownership
delete_flow_and_data()must stop dropping a queue.A named queue may contain tasks from other flows, other versions, or non-pgflow users. Flow deletion must archive only that flow's active messages, grouped by task queue, before deleting runtime rows.
Leave queue deletion to a later explicit administrative operation. Retaining an empty queue is safer than deleting shared work.
Compatibility
This issue must make no observable routing change:
Existing
Flowworkers and direct SQL starts must continue to work unchanged.Acceptance criteria
steps.queue_nameandstep_tasks.queue_nameexist and are non-null for queue-backed tasks.flow_slugqueue.(queue_name, message_id)identity throughout the SQL Core.create_flow()performs no queue DDL.add_step()resolves, provisions, and stores the default flow-slug queue.start_ready_steps()performs no queue DDL and snapshots the resolved queue on every task.requeue_stalled_tasks()uses effective step timeout and closes requeue_stalled_tasks() uses flow timeout instead of effective step timeout #621.Out of scope