Make flow workers queue-centric with multi-flow dispatch

[jumski]

Summary

Make flow workers queue-centric: one worker polls one queue and dispatches tasks through a registry of concrete Flow or DeployedFlow versions.

This lets several flows share a queue and lets different steps of one flow use workers with different concurrency, connection, and timeout settings.

Dependencies

• Parent epic: Epic: DeployedFlow aliases, queue routing, and queue-centric workers #653.
• Add stable flow aliases through DeployedFlow startup deployment #648 provides DeployedFlow and stable concrete-version aliases.
• Persist queue identity on flow steps and tasks #650 persists queue identity on definitions and tasks.
• Add typed static queue routing to DeployedFlow #651 compiles static DeployedFlow routing before workers consume named queues.

Public API

Keep the existing single-flow default:

EdgeWorker.start(flow)

Add a registry form for queue-centric workers:

EdgeWorker.start([deployedV1, deployedV2], {
  queueName: 'classify_threads',
  maxConcurrent: 1,
})

Another worker may consume another route from the same deployments:

EdgeWorker.start([deployedV1, deployedV2], {
  queueName: 'deliver_slack_alerts',
  maxConcurrent: 5,
})

Rules:

• One worker instance polls exactly one queue.
• EdgeWorker.start(flow) infers the existing flow.slug queue.
• Registry or named-routing forms require an explicit queue when it cannot be inferred safely.
• queueName autocompletes from queue literals declared by the supplied deployments.
• A worker carries complete deployment definitions, not partial per-worker routing maps.

Startup lifecycle

Before worker registration and polling:

1. Normalize plain flows to default deployments.
2. Reject duplicate concrete flow slugs.
3. Compile or verify every deployment.
4. Derive the exact (flow_slug, step_slug) pairs routed to the configured queue.
5. Check active, draining, executable, and recoverable coverage for that queue.
6. Register the worker against that queue.

Compilation and activation may commit before worker registration. If later startup fails, durable tasks remain queued for recovery.

The startup banner should show one queue and each concrete flow's independent compilation and activation state.

Queue-centric claiming

Change claiming semantics from flow-scoped to queue-scoped:

start_tasks(queue_name, message_ids, worker_id, supported_task_pairs)

The exact SQL signature may differ, but it must separate:

• the queue that owns the PGMQ message;
• the concrete flow recorded on the task;
• the worker's supported (flow_slug, step_slug) pairs for that queue.

A claim may return tasks from several concrete flows. Each returned task includes its concrete flow_slug, step_slug, and immutable queue snapshot.

Validate exact flow-step pairs before any task mutation. A flow-level allowlist is insufficient because one concrete flow may route different steps to different queue workers.

Use (queue_name, message_id) for matching. Never infer queue identity from flow_slug.

Dispatch

Build a registry keyed by concrete flow slug:

flow_slug -> Flow handlers and type metadata

For each claimed task:

1. Find the concrete flow.
2. Find the step handler on that flow.
3. Build the existing typed handler input and context.
4. Execute through the existing completion and failure APIs.

Keep domain logic, dependency resolution, retries, task state, and result aggregation in their existing owners.

Unsupported-flow safety

A worker must not partially start a batch that contains unsupported concrete flows.

Required behavior:

1. Detect unsupported (flow_slug, step_slug) pairs before task status changes, or make the claim transaction fail atomically.
2. Reset visibility for every message in the complete read batch.
3. Emit one fatal configuration error with queue name, message IDs, and unsupported flow-step pairs, without message bodies.
4. Request worker shutdown.
5. Leave tasks claimable by a correctly configured worker without consuming a task attempt.

Do not archive or fail an unsupported task. It is a deployment configuration error, not a handler failure. Do not add worker-rejection headers or custom PGMQ filtering: every live consumer of one queue must support every message on that queue.

Coverage checks

At startup, validate coverage for:

• every active alias target with a step routed to the configured queue;
• every concrete version with a started run that may later reach that queue;
• every executable or recoverable task snapshot on that queue.

Fail startup when the registry lacks any required (flow_slug, step_slug) pair. Include bounded diagnostics and repair guidance.

The check reduces rollout errors but does not remove the runtime unsupported-flow guard. State can change after startup.

Version draining

Alias activation redirects new runs but does not remove old handlers.

Use the existing production deployment fence for every affected queue worker function:

1. Record which pgflow.worker_functions rows are enabled.
2. Set those functions to enabled = false so supervision cannot restart old code.
3. Deprecate their live workers and wait until all have stopped.
4. Deploy every affected function while it remains disabled; each new registry includes both old and new concrete versions.
5. Re-enable only the functions recorded in step 1.

The first new worker may compile and activate the new concrete version before another queue worker starts. This is safe: old consumers have stopped, and tasks for the not-yet-running queue wait durably. Normal rollout does not require an explicit activation call or cross-worker readiness protocol.

For a shared queue, the safe registry is:

EdgeWorker.start([deployedV1, deployedV2], {
  queueName: 'classify_threads',
})

Remove V1 only after no started V1 run can later reach that queue and no executable or recoverable V1 task remains there.

Worker registration and monitoring

pgflow.workers.queue_name remains the worker subscription identity.

Worker functions may support several flows. Monitoring, heartbeat, deprecation, and restart logic must not assume one worker row equals one concrete flow.

Keep worker configuration queue-scoped:

• concurrency;
• batch size;
• database connections;
• polling intervals;
• initial read visibility.

Step retry and timeout configuration remains flow/step configuration in the SQL Core.

Acceptance criteria

• Existing EdgeWorker.start(flow) behavior remains source compatible.
• A worker can accept a registry of concrete flows or deployments and poll one explicit queue.
• Worker queueName autocomplete derives from supplied routing literals.
• Startup rejects duplicate concrete slugs and incomplete active, draining, executable, or recoverable flow-step coverage.
• Every deployment compiles or verifies before worker registration.
• Queue polling and task claiming use queue identity independently of flow identity.
• One claim may return and dispatch tasks from several concrete flows.
• Dispatch selects handlers by concrete flow slug and step slug.
• Unsupported flow-step batches make no partial task-state changes, reset complete-batch visibility, consume no attempt, and stop the worker.
• The documented production rollout disables affected functions, stops old workers, deploys all replacements, and re-enables the prior function set.
• Old and new concrete versions coexist in every affected worker registry until old runs and tasks drain.
• Worker monitoring remains queue-centric and supports multi-flow registries.
• Startup and task logs identify queue, concrete flow, and step without message bodies.
• Tests cover shared queues, separate step queues, fenced multi-function rollout, activation before every new worker starts, version drain, startup coverage, unsupported-flow races, and backward compatibility.

Out of scope

• Polling several queues from one worker instance.
• Dynamic handler loading.
• Mutable queue routes.
• Automatic old-version removal.
• Manual tasks and external completion.