feat: add private per-step queues

.changeset/private-step-queues.md

@@ -0,0 +1,11 @@
+---
+'@pgflow/core': minor
+'@pgflow/dsl': minor
+'@pgflow/edge-worker': minor
+---
+
+Add private per-step queues. Wrap a flow with `withStepQueues()` and start one `EdgeWorker` per selected `stepSlug`; pgflow derives, validates, persists, and verifies each route before dispatching work.
+
+Flow and step slugs now reject leading or trailing underscores and `__`. Flow slugs are case-insensitively unique, and step slugs are case-insensitively unique within a flow.
+
+**Breaking:** `Flow.stepOrder` is now `readonly` and frozen at construction, so mutating it (for example `push()` or `reverse()`) can no longer change a flow's shape — startup shape extraction and checked route indices must never diverge.

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, string[], string, string]
+      [string, string[], string, string, string?]
     >();
     expectTypeOf(client.startTasks).returns.toEqualTypeOf<
       Promise<StepTaskRecord<typeof flow>[]>
@@ -68,6 +68,8 @@ describe('PgflowSqlClient Type Compatibility with Flow', () => {
     // Valid calls should compile
     client.startTasks('flow_slug', ['1', '2', '3'], 'worker-id', 'flow_slug');
     client.startTasks('flow_slug', [], 'worker-id', 'flow_slug');
+    // stepSlug is the additive exact step selector (#651)
+    client.startTasks('flow_slug', ['1'], 'worker-id', 'flow_slug', 'step');
 
     // @ts-expect-error - queueName is required (#650): no default queue fallback
     client.startTasks('flow_slug', ['1'], 'worker-id');

pkgs/core/schemas/0030_utilities.sql

@@ -31,7 +31,12 @@ begin
       and slug <> ''
       and length(slug) <= 128
       and slug ~ '^[a-zA-Z_][a-zA-Z0-9_]*$'
-      and slug NOT IN ('run'); -- reserved words
+      and slug NOT IN ('run') -- reserved words
+      -- #651: '__' is reserved for pgflow-generated queue names and
+      -- boundary underscores are rejected for flows and steps alike
+      and left(slug, 1) ~ '[a-zA-Z]'
+      and right(slug, 1) ~ '[a-zA-Z0-9]'
+      and position('__' in slug) = 0;
 end;
 $$;
 

pkgs/core/schemas/0050_tables_definitions.sql

@@ -6,19 +6,25 @@ create table pgflow.flows (
   opt_max_attempts int not null default 3,
   opt_base_delay int not null default 1,
   opt_timeout int not null default 60,
+  -- Deployment metadata, persisted separately from the shape (#651):
+  -- 'flow' routes every step to lower(flow_slug), 'step' gives every step
+  -- its own generated private queue.
+  queue_mode text not null default 'flow',
   created_at timestamptz not null default now(),
   constraint slug_is_valid check (pgflow.is_valid_slug(flow_slug)),
   constraint opt_max_attempts_is_nonnegative check (opt_max_attempts >= 0),
   constraint opt_base_delay_is_nonnegative check (opt_base_delay >= 0),
-  constraint opt_timeout_is_positive check (opt_timeout > 0)
+  constraint opt_timeout_is_positive check (opt_timeout > 0),
+  constraint queue_mode_is_valid check (queue_mode in ('flow', 'step'))
 );
 
 -- Steps table - stores individual steps within 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).
+  -- Canonical queue this step's tasks are dispatched to (#650, #651):
+  -- 'flow' mode routes every step to lower(flow_slug); 'step' mode routes
+  -- each step to its generated private queue (_derive_queue_routes).
   queue_name text not null,
   step_type text not null default 'single',
   step_index int not null default 0,
@@ -68,6 +74,11 @@ 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);
 
+-- Case-only duplicate step slugs within one flow normalize to the same
+-- generated queue name and are rejected (#651).
+create unique index if not exists idx_steps_normalized_slug
+on pgflow.steps (flow_slug, lower(step_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

pkgs/core/schemas/0070_function_resolve_step_queue_name.sql

@@ -0,0 +1,55 @@
+-- Canonical per-step queue-name resolution (#651).
+--
+-- For a zero-based step index:
+--   readable = lower(flow_slug || '__' || step_slug)
+--   fallback = lower(flow_slug || '__' || step_index)
+--
+-- Resolution: readable when it fits MAX 47 characters; otherwise the
+-- actual-index fallback when it fits; otherwise the complete flow is
+-- rejected. Names are never truncated or hashed. TypeScript mirrors this
+-- resolver in @pgflow/dsl (resolveStepQueueName); vectors must stay in sync.
+create or replace function pgflow._resolve_step_queue_name(
+  p_flow_slug text,
+  p_step_slug text,
+  p_step_index int
+)
+returns text
+language plpgsql
+immutable
+set search_path = ''
+as $$
+declare
+  v_readable text := lower(p_flow_slug || '__' || p_step_slug);
+  v_fallback text := lower(p_flow_slug || '__' || p_step_index);
+  v_shortest text := lower(p_flow_slug || '__0');
+begin
+  if length(v_readable) <= 47 then
+    return v_readable;
+  end if;
+
+  if length(v_shortest) > 47 then
+    -- The flow slug cannot fit even the shortest possible index suffix.
+    raise exception
+      'Flow "%" cannot use per-step queues.',
+      p_flow_slug
+      using detail = format(
+        'The shortest required queue "%s" is %s characters; PGMQ allows at most 47.',
+        v_shortest, length(v_shortest)
+      ),
+      hint = 'Shorten the concrete flow slug or use the default single queue.';
+  end if;
+
+  if length(v_fallback) <= 47 then
+    return v_fallback;
+  end if;
+
+  raise exception
+    'Cannot derive a queue for step "%" at index % in flow "%".',
+    p_step_slug, p_step_index, p_flow_slug
+    using detail = format(
+      'The readable name is %s characters and the index fallback is %s; PGMQ allows at most 47.',
+      length(v_readable), length(v_fallback)
+    ),
+    hint = 'Shorten the concrete flow slug, shorten the step slug enough for the readable name, or use the default single queue.';
+end;
+$$;

pkgs/core/schemas/0075_function_derive_queue_routes.sql

@@ -0,0 +1,114 @@
+-- Derive the complete ordered queue route map from a shape and a queue
+-- mode (#651). This is the one canonical derivation: startup compilation is
+-- authoritative and callers may only supply a map that matches it exactly.
+--
+-- Returns an ordered jsonb array [{stepSlug, queueName}] by shape
+-- ordinality (zero-based step_index). In 'step' mode every name is resolved
+-- through _resolve_step_queue_name, validated with the installed
+-- pgmq.validate_queue_name(), and checked for duplicates. In 'flow' mode
+-- every step routes to lower(flow_slug).
+create or replace function pgflow._derive_queue_routes(
+  p_flow_slug text,
+  p_shape jsonb,
+  p_queue_mode text
+)
+returns jsonb
+language plpgsql
+-- Volatile like the pgmq.validate_queue_name() call it performs: an
+-- IMMUTABLE label on a function that runs PGMQ validation would mislead
+-- the planner about what the call can do.
+volatile
+set search_path = ''
+as $$
+declare
+  v_step jsonb;
+  v_step_slug text;
+  v_step_index int;
+  v_queue_name text;
+  v_routes jsonb := '[]'::jsonb;
+  v_seen_normalized_steps text[] := '{}';
+  v_seen_step_slugs text[] := '{}';
+  v_seen_queues text[] := '{}';
+  v_seen_queue_steps text[] := '{}';
+begin
+  if p_queue_mode not in ('flow', 'step') then
+    raise exception 'Unknown queue mode "%".', p_queue_mode
+      using hint = 'Queue mode must be ''flow'' or ''step''.';
+  end if;
+
+  if p_queue_mode = 'step' and jsonb_array_length(coalesce(p_shape->'steps', '[]'::jsonb)) = 0 then
+    raise exception
+      'Flow "%" cannot use per-step queues: it has no steps.',
+      p_flow_slug
+      using detail = 'Per-step queue mode requires at least one step.',
+      hint = 'Add a step or keep the flow on the default single queue.';
+  end if;
+
+  -- Validate the concrete flow slug and every step slug before route
+  -- resolution, collision checks, or PGMQ work: an invalid slug is a
+  -- definition error that outranks every queue concern, so a shape with
+  -- both problems reports the slug, not the queue.
+  if not pgflow.is_valid_slug(p_flow_slug) then
+    raise exception
+      'Flow slug "%" is not valid.',
+      p_flow_slug
+      using detail = 'Slugs are 1-128 characters of letters, digits, and single underscores, must start with a letter, cannot end with an underscore, cannot contain ''__'', and cannot be the reserved word ''run''.',
+      hint = 'Fix the flow slug in the flow definition.';
+  end if;
+
+  for v_step in select * from jsonb_array_elements(coalesce(p_shape->'steps', '[]'::jsonb))
+  loop
+    v_step_slug := v_step->>'slug';
+
+    if not pgflow.is_valid_slug(v_step_slug) then
+      raise exception
+        'Step slug "%" in flow "%" is not valid.',
+        v_step_slug, p_flow_slug
+        using detail = 'Slugs are 1-128 characters of letters, digits, and single underscores, must start with a letter, cannot end with an underscore, cannot contain ''__'', and cannot be the reserved word ''run''.',
+        hint = 'Fix the step slug in the flow definition.';
+    end if;
+  end loop;
+
+  for v_step in select * from jsonb_array_elements(coalesce(p_shape->'steps', '[]'::jsonb))
+  loop
+    v_step_slug := v_step->>'slug';
+    v_step_index := jsonb_array_length(v_routes);
+
+    -- Validate normalized step identity before route resolution. Long
+    -- case-only variants can resolve to distinct index fallbacks, so route
+    -- collisions alone cannot detect this invalid definition.
+    if lower(v_step_slug) = any(v_seen_normalized_steps) then
+      raise exception
+        'Steps "%" and "%" in flow "%" conflict case-insensitively.',
+        v_seen_step_slugs[array_position(v_seen_normalized_steps, lower(v_step_slug))],
+        v_step_slug,
+        p_flow_slug
+        using detail = 'Step slugs must be unique case-insensitively.',
+        hint = 'Rename one of the colliding steps.';
+    end if;
+    v_seen_normalized_steps := v_seen_normalized_steps || lower(v_step_slug);
+    v_seen_step_slugs := v_seen_step_slugs || v_step_slug;
+
+    if p_queue_mode = 'step' then
+      v_queue_name := pgflow._resolve_step_queue_name(p_flow_slug, v_step_slug, v_step_index);
+      perform pgmq.validate_queue_name(v_queue_name);
+    else
+      v_queue_name := lower(p_flow_slug);
+    end if;
+
+    if p_queue_mode = 'step' and v_queue_name = any(v_seen_queues) then
+      raise exception
+        'Steps "%" and "%" in flow "%" both resolve to queue "%".',
+        v_seen_queue_steps[array_position(v_seen_queues, v_queue_name)], v_step_slug, p_flow_slug, v_queue_name
+        using detail = 'Generated queue names must be unique per flow.',
+        hint = 'Step slugs must be unique case-insensitively; rename one of the colliding steps.';
+    end if;
+
+    v_seen_queues := v_seen_queues || v_queue_name;
+    v_seen_queue_steps := v_seen_queue_steps || v_step_slug;
+    v_routes := v_routes || jsonb_build_object('stepSlug', v_step_slug, 'queueName', v_queue_name);
+  end loop;
+
+  return v_routes;
+end;
+$$;

pkgs/core/schemas/0076_function_assert_step_queue_available.sql

@@ -0,0 +1,90 @@
+-- Shared step-queue route preflight (#651): one canonical ownership and
+-- listing check for every step-mode queue name, so startup verification and
+-- creation cannot diverge.
+--
+-- Callers must already hold the normalized-flow advisory lock
+-- (pg_advisory_xact_lock(1, hashtext(lower(flow_slug)))).
+--
+-- Rejects:
+-- - a queue name routed to by another concrete flow's steps, or defaulted
+--   to by another flow-mode flow (cross-flow reference);
+-- - an ambiguous case-insensitive match among listed PGMQ queues
+--   (external damage), even when this flow's definition owns the route.
+--
+-- Allows one exact listed queue only when the existing definition of this
+-- exact flow owns that route (idempotent reuse). A name that is neither
+-- listed nor owned is left for the caller to create.
+--
+-- Returns true when the caller must create the queue.
+create or replace function pgflow._assert_step_queue_available(
+  p_flow_slug text,
+  p_queue_name text
+)
+returns boolean
+language plpgsql
+volatile
+set search_path = ''
+as $$
+declare
+  v_owner_flow_slug text;
+  v_listed text[];
+begin
+  -- A name derived or referenced by another concrete flow is rejected
+  select s.flow_slug into v_owner_flow_slug
+  from pgflow.steps as s
+  where s.queue_name = p_queue_name
+    and lower(s.flow_slug) <> lower(p_flow_slug)
+  limit 1;
+
+  if v_owner_flow_slug is null then
+    select f.flow_slug into v_owner_flow_slug
+    from pgflow.flows as f
+    where f.queue_mode = 'flow'
+      and lower(f.flow_slug) = p_queue_name
+      and lower(f.flow_slug) <> lower(p_flow_slug)
+    limit 1;
+  end if;
+
+  if v_owner_flow_slug is not null then
+    raise exception
+      'cannot create flow "%": queue "%" is already used by another flow ("%")',
+      p_flow_slug, p_queue_name, v_owner_flow_slug
+      using detail = 'Generated per-step queue names must belong to exactly one concrete flow.',
+      hint = 'Use a different concrete flow slug, or drop the conflicting definition.';
+  end if;
+
+  -- Ambiguous normalized matches among listed queues are external damage
+  select array_agg(listed.queue_name order by listed.queue_name)
+  into v_listed
+  from pgmq.list_queues() as listed
+  where lower(listed.queue_name) = p_queue_name;
+
+  if v_listed is not null and cardinality(v_listed) > 1 then
+    raise exception
+      'queue "%" matches multiple listed PGMQ queues (%)',
+      p_queue_name, v_listed
+      using detail = 'An ambiguous case-insensitive match is external damage.',
+      hint = 'Resolve the duplicate queue spellings manually, then retry.';
+  end if;
+
+  -- Owned by an existing definition of this exact flow: reuse idempotently.
+  -- A verified definition owns every derived route, so its one exact listed
+  -- queue is allowed here.
+  if exists (
+    select 1 from pgflow.steps as s
+    where s.flow_slug = p_flow_slug and s.queue_name = p_queue_name
+  ) then
+    return false;
+  end if;
+
+  if v_listed is not null then
+    raise exception
+      'cannot create flow "%": queue "%" is already listed in PGMQ and not owned by this flow',
+      p_flow_slug, p_queue_name
+      using detail = 'A missing definition must not adopt an already listed queue.',
+      hint = 'Drop the conflicting queue or use a different concrete flow slug.';
+  end if;
+
+  return true;
+end;
+$$;

pkgs/core/schemas/0100_function_add_step.sql

@@ -21,6 +21,13 @@ DECLARE
   result_step pgflow.steps;
   next_idx int;
 BEGIN
+  -- Serialize with ensure_flow_compiled and every other definition writer on
+  -- the same normalized flow identity (#651): the max(step_index)+1 read and
+  -- the insert below must not interleave with a concurrent compilation of
+  -- the same flow. Re-entrant under ensure_flow_compiled's lock; taken
+  -- before any table access.
+  PERFORM pg_advisory_xact_lock(1, hashtext(lower(add_step.flow_slug)));
+
   -- Validate map step constraints
   -- Map steps can have either:
   --   0 dependencies (root map - maps over flow input array)
@@ -37,6 +44,25 @@ BEGIN
   FROM pgflow.steps s
   WHERE s.flow_slug = add_step.flow_slug;
 
+  -- add_step stays flow-only (#651): a step-mode definition is provisioned
+  -- exclusively by the complete-route compilation path. An incremental
+  -- add_step on a step-mode flow would bypass complete-route preflight and
+  -- persist a route to a queue compilation never created, so it is rejected;
+  -- recompile the complete definition instead. Repeated add_step calls on a
+  -- flow-mode flow never reset a persisted route: it stays lower(flow_slug).
+  IF EXISTS (
+    SELECT 1
+    FROM pgflow.flows AS f
+    WHERE f.flow_slug = add_step.flow_slug
+      AND f.queue_mode = 'step'
+  ) THEN
+    RAISE EXCEPTION
+      'Flow "%" uses per-step queues: steps cannot be added incrementally.',
+      add_step.flow_slug
+      USING detail = 'A step-mode definition is provisioned only by complete compilation.',
+      hint = 'Recompile the complete flow definition with every step instead.';
+  END IF;
+
   -- Create the step. queue_name records the step's resolved default route:
   -- lower(flow_slug) for this stage (#650).
   INSERT INTO pgflow.steps (
@@ -62,8 +88,7 @@ BEGIN
   )
   ON CONFLICT ON CONSTRAINT steps_pkey
   DO UPDATE SET
-    step_slug = EXCLUDED.step_slug,
-    queue_name = EXCLUDED.queue_name
+    step_slug = EXCLUDED.step_slug
   RETURNING * INTO result_step;
 
   -- Insert dependencies