feat(core): persist physical queue identity for flow tasks

.changeset/persist-queue-identity.md

@@ -0,0 +1,11 @@
+---
+'@pgflow/core': minor
+'@pgflow/dsl': minor
+'@pgflow/client': minor
+'@pgflow/edge-worker': minor
+'pgflow': minor
+---
+
+Persist physical queue identity on steps and tasks. A queued task's message identity is now `(queue_name, message_id)`, not `message_id` alone, preparing pgflow for private per-step queues while keeping one-flow/one-queue behavior.
+
+`pgflow.steps` and `pgflow.step_tasks` gain a canonical lowercase `queue_name` (snapshot at task creation), `(queue_name, message_id)` is unique per queue, and two flows can no longer share a normalized default queue. **Breaking:** `pgflow.start_tasks()` now requires the `queue_name` argument - the queue's canonical identity, `lower(flow_slug)` today - and the released three-argument form and the NULL default are gone, and `startTasks()` on `IPgflowClient`/`PgflowSqlClient` requires the queue argument as well. pgflow's own workers poll and claim through that canonical name; custom callers must pass it explicitly. There is no mixed-version rolling upgrade: stop and drain workers, pause producers and definition/maintenance/recovery writers, apply the transactional migration through Supabase's migration runner against the production database (`--linked` or `--db-url`, not the local default), replace the optional `prune_data_older_than()` helper, then deploy matching packages and workers together, restoring the exact worker `enabled` states recorded before the window (see the 0.17.0 upgrade guide). Message ids are exact decimal strings at the JavaScript boundary. Existing mixed-case queue names keep working through their original pgmq spelling - PGMQ's public message API normalizes names, so no message or queue migration is needed.

ARCHITECTURE_GUIDE.md

@@ -172,7 +172,7 @@ export default CompleteExample;
 
 **Critical Cross-Cutting Concepts**:
 
-1. **Two-Phase Polling** - Worker calls `read_with_poll()` then `start_tasks(workerId)` to prevent race conditions
+1. **Two-Phase Polling** - Worker calls `read_with_poll()` then `start_tasks(workerId, queue_name)` to prevent race conditions
 2. **Empty Array Cascade** - When `initial_tasks=0`, `cascade_complete_taskless_steps()` completes entire dependent chain in one transaction
 3. **Map Step `initial_tasks` Lifecycle**:
    - Root maps: Set at flow start from input array length
@@ -204,7 +204,7 @@ export default CompleteExample;
 2. Main loop:
    - `sendHeartbeat()` - Update status, check deprecation
    - If deprecated → exit gracefully
-   - Two-phase polling: `readMessages()` then `startTasks(workerId)`
+   - Two-phase polling: `readMessages()` then `startTasks(workerId, queueName)`
    - Execute handlers (up to `maxConcurrent` parallel)
    - `complete_task()` or `fail_task()`
 3. On shutdown:
@@ -228,8 +228,7 @@ const supabase = createClient(
 
 // Create worker with all configuration options
 const worker = createFlowWorker(supabase, MyFlow, {
-  // Queue configuration
-  queueName: 'tasks',           // Default: 'tasks'
+  // The worker polls the flow's canonical queue: lower(flow_slug)
 
   // Polling configuration
   maxPollSeconds: 2,            // Default: 2
@@ -313,7 +312,7 @@ await worker.start();
 
 **How**:
 - Phase 1: Worker calls `read_with_poll()` - reserves messages, returns `msg_id`s
-- Phase 2: Worker calls `start_tasks(flow_slug, msg_ids, workerId)` - creates `step_tasks`, returns details
+- Phase 2: Worker calls `start_tasks(flow_slug, msg_ids, worker_id, queue_name)` - creates `step_tasks`, returns details
 
 **See**:
 - Worker implementation: `/pkgs/edge-worker/src/worker/FlowWorkerLifecycle.ts`

pkgs/client/__tests__/helpers/polling.ts

@@ -53,9 +53,10 @@ export async function readAndStart<TFlow extends AnyFlow>(
     return [];
   }
 
-  // 4. Start the tasks and return the resulting rows
+  // 4. Start the tasks and return the resulting rows. The claim receives the
+  // queue this helper read from (the canonical lowercase flow slug)
   const msgIds = messages.map(m => m.msg_id);
-  const tasks = await sqlClient.startTasks(flowSlug, msgIds, workerId);
+  const tasks = await sqlClient.startTasks(flowSlug, msgIds, workerId, flowSlug.toLowerCase());
 
   return tasks;
 }

pkgs/core/README.md

@@ -288,7 +288,8 @@ SELECT * FROM pgmq.read_with_poll(
 SELECT * FROM pgflow.start_tasks(
   flow_slug => 'analyze_website',
   msg_ids => ARRAY[101, 102, 103], -- message IDs from phase 1
-  worker_id => '550e8400-e29b-41d4-a716-446655440000'::uuid
+  worker_id => '550e8400-e29b-41d4-a716-446655440000'::uuid,
+  queue_name => 'analyze_website' -- the queue's canonical name: lower(flow_slug), the exact spelling tasks store (#650)
 );
 ```
 

pkgs/core/__tests__/types/PgflowSqlClient.test-d.ts

@@ -26,7 +26,7 @@ describe('PgflowSqlClient Type Compatibility with Flow', () => {
     // Check startTasks method types
     expectTypeOf(client.startTasks).toBeFunction();
     expectTypeOf(client.startTasks).parameters.toMatchTypeOf<
-      [string, number[], string]
+      [string, string[], string, string]
     >();
     expectTypeOf(client.startTasks).returns.toEqualTypeOf<
       Promise<StepTaskRecord<typeof flow>[]>
@@ -66,19 +66,22 @@ describe('PgflowSqlClient Type Compatibility with Flow', () => {
     const client = new PgflowSqlClient<typeof flow>(sql);
 
     // Valid calls should compile
-    client.startTasks('flow_slug', [1, 2, 3], 'worker-id');
-    client.startTasks('flow_slug', [], 'worker-id');
+    client.startTasks('flow_slug', ['1', '2', '3'], 'worker-id', 'flow_slug');
+    client.startTasks('flow_slug', [], 'worker-id', 'flow_slug');
+
+    // @ts-expect-error - queueName is required (#650): no default queue fallback
+    client.startTasks('flow_slug', ['1'], 'worker-id');
 
     // @ts-expect-error - flowSlug must be string
-    client.startTasks(123, [1, 2, 3], 'worker-id');
+    client.startTasks(123, ['1', '2', '3'], 'worker-id', 'flow_slug');
 
-    // @ts-expect-error - msgIds must be number array
-    client.startTasks('flow_slug', ['1', '2', '3'], 'worker-id');
+    // @ts-expect-error - msgIds must be string array (exact decimal strings, #650)
+    client.startTasks('flow_slug', [1, 2, 3], 'worker-id', 'flow_slug');
 
     // @ts-expect-error - msgIds must be array
-    client.startTasks('flow_slug', 123, 'worker-id');
+    client.startTasks('flow_slug', 123, 'worker-id', 'flow_slug');
 
     // @ts-expect-error - workerId must be string
-    client.startTasks('flow_slug', [1, 2, 3], 123);
+    client.startTasks('flow_slug', ['1', '2', '3'], 123, 'flow_slug');
   });
 });

pkgs/core/assets/flow-lifecycle.mermaid

@@ -24,7 +24,7 @@ sequenceDiagram
     PGMQ-->>Worker: Return messages
     deactivate PGMQ
 
-    Worker->>pgflow: start_tasks(flow_slug, msg_ids, worker_id)
+    Worker->>pgflow: start_tasks(flow_slug, msg_ids, worker_id, queue_name)
     activate pgflow
     pgflow->>pgflow: Find step_tasks with matching message_ids
     pgflow->>pgflow: Mark tasks as 'started' with worker_id

pkgs/core/schemas/0030_utilities.sql

@@ -35,6 +35,23 @@ begin
 end;
 $$;
 
+create or replace function pgflow.is_valid_queue_name(
+  queue_name text
+)
+returns boolean
+language sql
+immutable
+set search_path = ''
+as $$
+  -- Mirrors pgmq.validate_queue_name() (47-character limit) and additionally
+  -- requires the canonical lowercase spelling pgflow stores (#650).
+  select
+    queue_name is not null
+    and queue_name <> ''
+    and length(queue_name) <= 47
+    and queue_name = lower(queue_name)
+$$;
+
 create or replace function pgflow.calculate_retry_delay(
   base_delay numeric,
   attempts_count int

pkgs/core/schemas/0035_function_listed_queue_name.sql

@@ -0,0 +1,47 @@
+-- Resolve a stored canonical queue name to the spelling listed in pgmq.
+--
+-- pgflow stores canonical lowercase queue names (#650). A queue created by an
+-- older pgflow release may be listed under its original mixed-case spelling.
+-- PGMQ's public message operations (send_batch, read_with_poll, set_vt,
+-- archive, delete) normalize names themselves, so message paths address the
+-- queue by its stored canonical name directly and never resolve anything.
+--
+-- Only operations that must address the queue's physical objects by their
+-- original spelling need this helper today: delete_flow_and_data (drop_queue
+-- drops metadata and tables under the listed spelling). The helper resolves
+-- the listed spelling through pgmq.list_queues() and never creates a second
+-- metadata entry. An ambiguous case-insensitive match (external damage) is
+-- rejected before any destructive work; an unlisted name is passed through
+-- unchanged so PGMQ reports the operation's own error.
+
+-- Fresh resolution against pgmq.list_queues().
+create or replace function pgflow._listed_queue_name(p_queue_name text)
+returns text
+language plpgsql
+stable
+set search_path = ''
+as $$
+declare
+  v_matches text[];
+begin
+  -- Resolve every listed spelling of the normalized name before preferring
+  -- any single match: an ambiguous pair is rejected even when one spelling
+  -- is the exact requested name (#650).
+  select array_agg(listed.queue_name order by listed.queue_name)
+  into v_matches
+  from pgmq.list_queues() as listed
+  where lower(listed.queue_name) = lower(p_queue_name);
+
+  if v_matches is null then
+    -- Not listed: pass through; PGMQ reports its own error (or no-ops)
+    return p_queue_name;
+  elsif cardinality(v_matches) > 1 then
+    raise exception
+      'queue name "%" is ambiguous: it matches listed queues %',
+      p_queue_name, v_matches
+      using errcode = 'ambiguous_alias';
+  else
+    return v_matches[1];
+  end if;
+end;
+$$;

pkgs/core/schemas/0050_tables_definitions.sql

@@ -17,6 +17,9 @@ create table pgflow.flows (
 create table pgflow.steps (
   flow_slug text not null references pgflow.flows(flow_slug),
   step_slug text not null,
+  -- Canonical queue this step's tasks are dispatched to (#650).
+  -- For this stage every step routes to the flow's default queue: lower(flow_slug).
+  queue_name text not null,
   step_type text not null default 'single',
   step_index int not null default 0,
   deps_count int not null default 0 check (deps_count >= 0),
@@ -38,6 +41,7 @@ create table pgflow.steps (
   primary key (flow_slug, step_slug),
   unique (flow_slug, step_index),  -- Ensure step_index is unique within a flow
   check (pgflow.is_valid_slug(step_slug)),
+  constraint queue_name_is_valid check (pgflow.is_valid_queue_name(queue_name)),
   check (step_type in ('single', 'map')),
   constraint opt_max_attempts_is_nonnegative check (opt_max_attempts is null or opt_max_attempts >= 0),
   constraint opt_base_delay_is_nonnegative check (opt_base_delay is null or opt_base_delay >= 0),
@@ -63,3 +67,8 @@ create table pgflow.deps (
 
 create index if not exists idx_deps_by_flow_step on pgflow.deps (flow_slug, step_slug);
 create index if not exists idx_deps_by_flow_dep on pgflow.deps (flow_slug, dep_slug);
+
+-- Two concrete flows must not address the same normalized default queue (#650).
+-- The expression index also rejects direct SQL creation of conflicting flows.
+create unique index if not exists idx_flows_normalized_slug
+on pgflow.flows (lower(flow_slug));

pkgs/core/schemas/0060_tables_runtime.sql

@@ -85,6 +85,10 @@ create table pgflow.step_tasks (
   flow_slug text not null references pgflow.flows(flow_slug),
   run_id uuid not null references pgflow.runs(run_id),
   step_slug text not null,
+  -- Snapshot of steps.queue_name taken at task creation (#650).
+  -- Runtime code never changes this value; PGMQ message IDs are queue-scoped,
+  -- so a task's message identity is (queue_name, message_id).
+  queue_name text not null,
   message_id bigint,
   task_index int not null default 0,
   status text not null default 'queued',
@@ -117,10 +121,17 @@ create table pgflow.step_tasks (
   constraint completed_at_is_after_started_at check (
     completed_at is null or started_at is null or completed_at >= started_at
   ),
-  constraint failed_at_is_after_started_at check (failed_at is null or started_at is null or failed_at >= started_at)
+  constraint failed_at_is_after_started_at check (failed_at is null or started_at is null or failed_at >= started_at),
+  constraint queue_name_is_valid check (pgflow.is_valid_queue_name(queue_name))
 );
 
-create index if not exists idx_step_tasks_message_id on pgflow.step_tasks (message_id);
+-- A message ID identifies at most one task per queue (#650).
+-- NULL message_ids (pre-dispatch rows) are not part of the identity.
+-- This index also serves queue-scoped message lookups for claims and pruning,
+-- replacing the former message_id-only index.
+create unique index if not exists idx_step_tasks_queue_message
+on pgflow.step_tasks (queue_name, message_id)
+where message_id is not null;
 create index if not exists idx_step_tasks_queued on pgflow.step_tasks (run_id, step_slug) where status = 'queued';
 create index if not exists idx_step_tasks_completed on pgflow.step_tasks (run_id, step_slug) where status = 'completed';
 create index if not exists idx_step_tasks_failed on pgflow.step_tasks (run_id, step_slug) where status = 'failed';

pkgs/core/schemas/0062_function_requeue_stalled_tasks.sql

@@ -23,6 +23,7 @@ begin
       st.step_slug,
       st.task_index,
       st.message_id,
+      st.queue_name,
       r.flow_slug,
       st.requeued_count
     from pgflow.step_tasks st
@@ -59,9 +60,11 @@ begin
     where st.run_id = tr.run_id
       and st.step_slug = tr.step_slug
       and st.task_index = tr.task_index
-    returning tr.flow_slug as queue_name, tr.message_id
+    returning tr.queue_name as queue_name, tr.message_id
   ),
-  -- Make requeued messages visible immediately (batched per queue)
+  -- Make requeued messages visible immediately (batched per queue, through
+  -- the tasks' stored queue snapshots #650; PGMQ message operations
+  -- normalize names themselves)
   visibility_reset as (
     select pgflow.set_vt_batch(
       r.queue_name,
@@ -84,10 +87,13 @@ begin
   ),
   -- Archive messages for tasks that exceeded max requeues (batched per queue)
   archived as (
-    select pgmq.archive(ta.flow_slug, array_agg(ta.message_id))
+    select pgmq.archive(
+      ta.queue_name,
+      array_agg(ta.message_id)
+    )
     from to_archive ta
     where ta.message_id is not null
-    group by ta.flow_slug
+    group by ta.queue_name
   ),
   -- Force execution of visibility_reset CTE
   _vr as (select count(*) from visibility_reset),

pkgs/core/schemas/0100_function__cascade_force_skip_steps.sql

@@ -100,13 +100,18 @@ BEGIN
         FROM skipped AS skipped_step
       )
       AND task.status IN ('queued', 'started')
-    RETURNING task.message_id
+    RETURNING task.message_id, task.queue_name
   ),
   -- ---------- Archive queued/started task messages for skipped steps ----------
+  -- Batched per stored queue route (#650)
   archived_messages AS (
-    SELECT pgmq.archive(v_flow_slug, ARRAY_AGG(task.message_id)) as result
+    SELECT pgmq.archive(
+      task.queue_name,
+      ARRAY_AGG(task.message_id)
+    ) as result
     FROM skipped_tasks AS task
     WHERE task.message_id IS NOT NULL
+    GROUP BY task.queue_name
     HAVING COUNT(task.message_id) > 0
   ),
   -- ---------- Update run counters ----------

pkgs/core/schemas/0100_function_add_step.sql

@@ -37,15 +37,17 @@ BEGIN
   FROM pgflow.steps s
   WHERE s.flow_slug = add_step.flow_slug;
 
-  -- Create the step
+  -- Create the step. queue_name records the step's resolved default route:
+  -- lower(flow_slug) for this stage (#650).
   INSERT INTO pgflow.steps (
-    flow_slug, step_slug, step_type, step_index, deps_count,
+    flow_slug, step_slug, queue_name, step_type, step_index, deps_count,
     opt_max_attempts, opt_base_delay, opt_timeout, opt_start_delay,
     required_input_pattern, forbidden_input_pattern, when_unmet, when_exhausted
   )
   VALUES (
     add_step.flow_slug,
     add_step.step_slug,
+    lower(add_step.flow_slug),
     COALESCE(add_step.step_type, 'single'),
     next_idx,
     COALESCE(array_length(add_step.deps_slugs, 1), 0),
@@ -59,7 +61,9 @@ BEGIN
     add_step.when_exhausted
   )
   ON CONFLICT ON CONSTRAINT steps_pkey
-  DO UPDATE SET step_slug = EXCLUDED.step_slug
+  DO UPDATE SET
+    step_slug = EXCLUDED.step_slug,
+    queue_name = EXCLUDED.queue_name
   RETURNING * INTO result_step;
 
   -- Insert dependencies

pkgs/core/schemas/0100_function_archive_task_message.sql

@@ -6,18 +6,19 @@ create or replace function pgflow._archive_task_message(
 returns void
 language sql
 volatile
-set search_path to ''
+set search_path = ''
 as $$
+  -- Archive through the task's stored queue snapshot (#650); PGMQ message
+  -- operations normalize names themselves.
   SELECT pgmq.archive(
-    r.flow_slug,
+    st.queue_name,
     ARRAY_AGG(st.message_id)
   )
   FROM pgflow.step_tasks st
-  JOIN pgflow.runs r ON st.run_id = r.run_id
   WHERE st.run_id = p_run_id
     AND st.step_slug = p_step_slug
     AND st.task_index = p_task_index
     AND st.message_id IS NOT NULL
-  GROUP BY r.flow_slug
+  GROUP BY st.queue_name
   HAVING COUNT(st.message_id) > 0;
 $$;