Use terminal events to cooperatively abort in-flight worker tasks

[jumski]

Idea

Explore how pgflow workers can cooperatively abort handlers that are already executing after their step or run becomes terminal.

This is separate from persisted task-state fixes:

• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638 makes task rows truthful when a map step is skipped;
• Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645 covers unfinished task rows on failed runs;
• this issue concerns JavaScript handlers that a worker already started before the database terminalized their step or run.

Keep this issue open as a design and prototype task. Do not block #638 on it.

Problem

Map tasks execute concurrently. A typical skip race is:

map input: [A, B, C]

A: running
B: running
C: queued

A exhausts retries
    -> map step becomes skipped
    -> C's message is archived, so C never starts
    -> B's message is archived, but B's handler is already executing
    -> run may complete before B returns

The database correctly ignores B's late completion or failure callback. It cannot undo side effects that B performed before returning:

• an email may be sent;
• an external API may be called;
• a database write may commit;
• compute time may continue after the run is terminal.

For pure or idempotent handlers this is mainly wasted work. For non-idempotent external effects, it can violate user expectations about a skipped or failed workflow.

Existing building blocks

Handler signal

Handlers already receive context.shutdownSignal through the DSL BaseContext.

On current main, createFlowWorker populates it from platformAdapter.shutdownSignal, and ExecutionController passes the shared worker abort signal to each executor. The handler-facing API therefore exists, but the current signal is wired to worker shutdown rather than a selectively controlled task cancellation source.

A future design should decide whether to:

• compose a task-specific controller with worker shutdown while preserving the existing context field;
• introduce a more accurately named task cancellation signal;
• or treat the existing signal as a general execution-abort signal and document the broader semantics.

Avoid a breaking context API unless it provides a concrete benefit.

Terminal Realtime events

Core SQL already emits Supabase Realtime broadcasts through realtime.send, including:

step:skipped
step:failed
run:failed
run:completed

The existing application-facing topic is:

pgflow:run:<run_id>

@pgflow/client demonstrates broadcast subscription, stabilization delay, reconnect handling, and database snapshot refresh.

Worker dependencies

@pgflow/edge-worker already depends on @supabase/supabase-js and has a service-role Supabase client in its platform resources. Realtime support would not require adding the Supabase client dependency.

The client adapter itself is not currently exported from @pgflow/client, and its UI state-tracking behavior is not necessarily the right abstraction for workers. Treat it as a reference implementation rather than adding an edge-worker-to-client dependency by default.

Required semantics

Cancellation can only be cooperative.

queued task:  prevent it from starting reliably
running task: request abort and let abort-aware code stop
completed side effect: cannot be undone
late callback: ignore it and preserve terminal orchestration state

A task handler must pass the signal to abort-aware operations such as fetch, or call throwIfAborted() at safe checkpoints. JavaScript cannot forcibly stop an arbitrary promise, CPU loop, database transaction, or external side effect.

Do not kill the whole worker to cancel one task. A worker can execute unrelated tasks concurrently, potentially from other runs.

Architectural constraint: no per-task subscriptions

Starting a Realtime subscription or PostgreSQL listener for every task is not acceptable.

It would add:

• task startup latency;
• subscription churn;
• cleanup and reconnect races;
• resource usage proportional to active tasks;
• complex interactions with future channel customization.

The listener must belong to the worker. Tasks should only register and unregister an in-memory cancellation target.

Conceptually:

worker starts
    -> establish one cancellation transport
    -> start polling work

task starts
    -> activeTasks[(run_id, step_slug, task_index)] = AbortController

event arrives
    -> find matching entries
    -> abort matching controllers

task settles
    -> remove registry entry

Multiple workers can listen to the same cancellation stream. Each worker ignores events that do not match its local active registry.

Channel configuration boundary

Internal worker control routing must remain separate from user-facing observability channels.

Future users may want custom Realtime channels per flow, step, or run. Worker correctness must not depend on those options. A misconfigured public channel must not make workers miss cancellation.

Recommended separation:

public observability topics:
  application-facing
  potentially configurable

internal control topic:
  worker-facing
  fixed and predictable
  not configurable

If Realtime is selected, a likely internal topic is:

pgflow:flow:<flow_slug>:control

A flow worker would subscribe once at startup. Existing per-run broadcasts would continue for clients. Terminal SQL paths would dual-publish a small internal control payload containing the scope and identifiers needed for local routing.

Example payloads:

{
  "event_type": "cancel:step",
  "flow_slug": "invoiceFlow",
  "run_id": "...",
  "step_slug": "sendEmails",
  "reason": "step_skipped"
}

{
  "event_type": "cancel:run",
  "flow_slug": "invoiceFlow",
  "run_id": "...",
  "reason": "run_failed"
}

Do not expose task-specific channel names. Routing belongs in the payload and the worker's active registry.

Option A: flow-wide Supabase Realtime control topic

Shape

• One Supabase client/Realtime socket per worker platform adapter.
• One fixed flow-control channel per flow worker.
• One in-memory registry for all active tasks in that worker.
• Core SQL sends control broadcasts in the same terminal-state transaction as existing events.

Supabase channels multiplex over the client's Realtime WebSocket. This avoids one socket per task, but every worker still opens and maintains a WebSocket.

Advantages

• Reuses Supabase infrastructure already required by edge workers.
• Works across process and Supabase Edge runtimes without a dedicated PostgreSQL session.
• Reuses existing terminal event knowledge and payload conventions.
• Low cancellation latency under a healthy connection.
• Naturally broadcasts to every worker that may own a matching task.

Problems and caveats

• Realtime delivery is ephemeral; disconnected workers can miss events.
• Subscription acknowledgment does not immediately guarantee backend routing. @pgflow/client currently uses a stabilization delay for this Supabase behavior.
• One Realtime socket per worker may be material when many flow workers run simultaneously.
• New control broadcasts add traffic in addition to existing per-run events.
• Channel authentication, private/public topic behavior, service-role use, and cleanup need review.
• A Realtime outage must not leak channels, crash handlers, or silently claim guaranteed cancellation.

Race-free startup and reconnect requirement

Do not copy a snapshot-before-subscribe sequence. It can miss a terminal transition between the snapshot and successful subscription.

The safer sequence is:

subscribe to fixed control topic
wait for routing stabilization
read authoritative statuses for currently active tasks
start or resume normal monitoring

Events received during or after the snapshot are idempotent. If an event happened before routing became active, the post-subscription snapshot observes the terminal database state.

At initial worker startup, establish monitoring before polling the first task. On reconnect, reconcile all locally active task keys in one batched database query.

Failure policy to decide

Options:

1. Fail open: continue handlers if Realtime is unavailable, log degraded cancellation, and rely on database callback guards.
2. Fail closed: stop accepting new tasks until control monitoring is healthy.
3. Hybrid: continue existing handlers but pause new task starts during an extended control-channel outage.

Fail-open behavior preserves current availability but weakens cancellation. Fail-closed behavior makes workflow execution depend on Realtime availability. The issue needs an explicit decision and documentation.

Option B: PostgreSQL LISTEN/NOTIFY

Shape

A worker establishes one long-lived listener, not one listener per task:

LISTEN pgflow_cancellations;

Terminal SQL paths send compact routing payloads with pg_notify. Every worker receives the notification and filters it against its local active registry.

A per-flow channel could reduce irrelevant notifications, but dynamic SQL channel names and lifecycle management need review. One fixed channel with flow_slug in the payload is simpler but broadcasts every cancellation to every listener.

Advantages

• Native PostgreSQL mechanism with low latency.
• Very small payloads and no Supabase Realtime server dependency.
• A single listener can serve every active task in a process.
• Notifications commit with the surrounding transaction.

Problems and caveats

• LISTEN requires a persistent session and does not work reliably through transaction pooling.
• Each worker or process may consume a dedicated database connection.
• Edge-worker connection limits may make one extra persistent connection expensive.
• Reconnects can miss notifications and still need authoritative database reconciliation.
• Process and Supabase Edge runtime support need separate prototypes.
• PostgreSQL notification payloads are size-limited, although cancellation payloads should fit easily.
• A global channel can send high irrelevant traffic to workers for other flows.

The difficulty is session ownership and portability, not per-task listener creation.

Option C: periodic batched database reconciliation

Shape

A worker keeps the same active registry but runs a periodic query for the distinct (run_id, step_slug) keys represented there. It aborts tasks whose run or step is terminal.

Advantages

• Database state is authoritative; no event can be permanently missed.
• No WebSocket and no dedicated LISTEN connection.
• Works through existing SQL access in all worker runtimes.
• Straightforward recovery after worker pauses or transient network failures.

Problems and caveats

• Adds recurring database queries per worker.
• Cancellation latency equals the reconciliation interval.
• Large worker fleets can create steady polling load.
• The check must run independently from task-slot availability; a worker with every slot occupied cannot wait for a slot before checking cancellation.
• Short intervals improve latency but increase load.

A prototype should measure one batched query per worker at realistic concurrency and interval settings.

Possible hybrid

Realtime or LISTEN/NOTIFY can provide fast hints while periodic or reconnect-time reconciliation supplies correctness after missed events.

A hybrid offers the strongest behavior but owns both event and polling complexity. Do not choose it without evidence that one transport plus reconnect reconciliation is insufficient.

Active execution registry requirements

Regardless of transport, the worker needs a registry that supports:

• registration before handler execution;
• lookup by run;
• lookup by run and step;
• removal in a finally path;
• idempotent abort;
• composition with worker shutdown;
• no retention after task settlement;
• isolation between unrelated runs and steps.

Potential ownership locations include ExecutionController, a dedicated worker cancellation coordinator, or the flow-worker composition root. Avoid adding an abstraction until a prototype shows which component already owns enough lifecycle information.

Event-to-abort rules

At minimum:

step:skipped  -> abort active tasks with matching run_id and step_slug
run:failed    -> abort every active task with matching run_id
run:completed -> abort any unexpectedly active task with matching run_id

step:failed normally precedes a failed run; run-level cancellation is sufficient for unrelated branches. Duplicated step and run events must remain harmless.

A future user-facing run-cancel event can use the same run-scoped mechanism, but that API is outside this issue.

Database reconciliation rules

For each active execution, cancellation is required when:

runs.status IN ('failed', 'completed')
OR step_states.status IN ('failed', 'completed', 'skipped')

Whether step_states.status = 'completed' should abort a physically late sibling needs careful review. A correctly completed map should have no active sibling handlers, but a defensive abort may still be appropriate.

The query should batch distinct keys and return only terminal scopes. Do not issue one database query per task.

Required prototype questions

Before selecting a transport, measure or prove:

1. How many workers and Realtime sockets a realistic deployment creates.
2. Whether a Supabase Edge worker can keep the control subscription stable for its full lifetime.
3. Whether process and Edge runtimes can support a dedicated LISTEN session without pooler or connection-budget problems.
4. The database cost and cancellation latency of one batched reconciliation query per worker.
5. How worker shutdown, transport reconnect, and task-specific abort signals compose without leaks.

Required tests for an eventual implementation

Matching cancellation

Start two sibling map handlers and make one exhaust retries. Assert the other abort-aware handler receives an abort request after the step becomes skipped.

Isolation

Keep tasks active for:

• another step in the same run;
• another run of the same flow;
• another flow where supported.

A step-scoped cancellation must abort only the matching run and step. A run-scoped cancellation must not affect another run.

Queued work

Assert queued siblings never start after the step becomes skipped or the run fails. This remains primarily a core SQL/message-archive guarantee.

Missed event recovery

Disconnect or delay the cancellation transport, terminalize the step or run, reconnect, and prove the authoritative snapshot aborts the active handler.

Duplicate events

Deliver the same step and run cancellation several times. Assert one effective abort, no error, and no registry leak.

Cleanup

After the last local task for a run or flow settles, assert listener references and registry entries are removed according to the chosen transport lifecycle.

Worker shutdown

Trigger worker shutdown and task cancellation concurrently. Assert the handler sees an aborted signal, unrelated cleanup completes, and no unhandled rejection occurs.

Transport outage

Assert the documented fail-open, fail-closed, or hybrid policy. Logs and metrics must state degraded cancellation accurately.

Acceptance criteria for a future implementation

• No task creates its own Realtime or PostgreSQL listener.
• Internal cancellation routing is fixed and independent of configurable observability channels.
• Step cancellation targets only matching active tasks.
• Run cancellation targets only the matching run.
• Worker shutdown still aborts every local task.
• Missed notifications are reconciled from authoritative database state.
• Duplicate events and reconnects are idempotent.
• Listener and controller lifecycle has no leaks.
• The behavior is documented as cooperative cancellation.
• Existing callback guards remain the final database safety boundary.
• Connection, query, and cancellation-latency costs are measured before selecting the default transport.

Out of scope

• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638's persisted skipped-task fix.
• The failed-run cancelled database status in Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645.
• Hard-killing a process, isolate, or worker thread.
• Undoing external side effects already committed.
• Making handlers automatically idempotent.
• User-configurable internal worker control channels.
• A public cancel-run API.

Related

• Skip paths archive sibling messages but never terminalize sibling step_tasks rows, leaving status='started' tasks on completed runs #638 — terminalize task rows under skipped steps.
• Failed runs leave unfinished sibling tasks queued or started instead of cancelled #645 — terminalize unfinished task rows on failed runs.
• pkgs/client/src/lib/SupabaseBroadcastAdapter.ts — existing Realtime subscription reference.
• pkgs/edge-worker/src/flow/createFlowWorker.ts — current handler context and signal wiring.
• pkgs/edge-worker/src/core/ExecutionController.ts — current concurrent execution lifecycle.
• pkgs/edge-worker/src/flow/StepTaskExecutor.ts — current handler execution and callback behavior.