Cancellation

What It Is

Cancellation flows through the standard Web AbortSignal API. The dispatcher accepts a caller signal and an optional deadlineMs; both compose into the signal every node receives as context.signal.

The goal is not to crash the run. The goal is to stop safely, return a structured ExecutionResult, preserve interruptedAt, and keep a cursor when the run can be resumed.

How It Works

The dispatcher composes all cancellation inputs into one signal and passes it through NodeContextType. Nodes and runtime helpers propagate that signal into adapters, tools, schedulers, and retry policies. When it fires, the dispatcher records interruptedAt, keeps cursor when resume is possible, and returns a normal ExecutionResult.

Cancellation flows through the standard Web AbortSignal API. The dispatcher accepts two optional fields in the execute() and resume() options object: a caller-supplied signal and a deadlineMs budget. Internally the two compose via Signal.compose({ signal, deadlineMs }) and the result lands on context.signal for every node.

Diagrams, Examples, and Outputs

Example 06 runs a slow DAG twice: once with a caller abort and once with a deadline. The topology is intentionally small so the lifecycle result is easy to read:

Example 06 cancellation DAG

2 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:slow-dag",
  "@type": "DAG",
  "name": "slow-dag",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:slow-dag/node/slow"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:slow-dag/node/slow",
      "@type": "SingleNode",
      "name": "slow",
      "node": "urn:noocodec:node:slow",
      "outputs": {
        "success": "urn:noocodec:dag:slow-dag/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:slow-dag/node/end",
      "@type": "TerminalNode",
      "name": "end",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% slow-dag (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_slow-dag/node/slow
  urn_noocodec_dag_slow-dag/node/slow["slow"]
  urn_noocodec_dag_slow-dag/node/slow -->|success| urn_noocodec_dag_slow-dag/node/end
  urn_noocodec_dag_slow-dag/node/end((("end")))

• Retry - RetryPolicy.run honors context.signal so retries abort cleanly
• Checkpoint and Resume - abort and persist the cursor; resume continues from that point
• Observability - onError fires when an abort or deadline interrupts a node
• Example 06: Cancellation - runnable AbortController and deadlineMs example

What It Lets You Do

Use when

Use cancellation when a caller must stop work safely: a browser tab closes, an HTTP request disconnects, a queue lease expires, or a host-level deadline fires. The goal is to interrupt execution with a structured lifecycle result and a resumable cursor, not to let nodes throw arbitrary errors.

Code Samples

The snippets below show the call-site options, the cancellation-aware node, and the cursor checks Example 06 asserts.

signal and deadlineMs

Example 06 runs the same slow DAG twice: once with a caller AbortController and once with a dispatcher deadline.

Caller-controlled abort:

// (a) Caller-controlled abort: fires after 25 ms
const ctl    = new AbortController();
const aState = new NodeStateBase();
setTimeout(() => ctl.abort(new Error('user pressed cancel')), 25);
const aResult = await dispatcher.execute('urn:noocodec:dag:slow-dag', aState, { "signal": ctl.signal });

Dispatcher deadline (fires automatically after the budget):

// (b) Dispatcher deadline: fires after 25 ms automatically
const bState  = new NodeStateBase();
const bResult = await dispatcher.execute('urn:noocodec:dag:slow-dag', bState, { "deadlineMs": 25 });

Both produce a non-null result.cursor and a non-null result.interruptedAt; the only difference is the discriminator on interruptedAt.reason ('abort' versus 'timeout').

NodeContextType

Nodes receive the composed signal in the context argument and must propagate it into every IO call to be cancellable. The Example 06 node wires context.signal into its delay primitive:

export class SlowNode extends MonadicNode<NodeStateBase, 'success'> {
  readonly name = 'slow';
  readonly '@id' = 'urn:noocodec:node:slow';
  readonly outputs = ['success'] as const;
  override get outputSchema(): Record<'success', SchemaObjectType> {
    return { 'success': { 'type': 'object' } };
  }

  override async execute(batch: Batch<NodeStateBase>, context: NodeContextType) {
    // Wrap the delay in a manual Promise that listens for abort. If the node
    // ignores context.signal, cancellation would not take effect until the
    // current node finishes, even if the signal fires.
    //
    // When composing multiple cancellation concerns (e.g. a per-operation
    // timeout plus the dispatcher's signal), prefer `Signal.compose(...)`.
    // Do not manually chain listeners.
    await new Promise<void>((resolve, reject) => {
      const t = setTimeout(resolve, 5_000);
      context.signal.addEventListener(
        'abort',
        () => { clearTimeout(t); reject(context.signal.reason); },
        { "once": true },
      );
    });
    return RoutedBatch.create(NodeOutput.create('success').output, batch);
  }
}

context also carries context.dagName and context.nodeName for logging and correlation. Treat them as observability fields; the DAG document @id and placement @id remain the execution identity.

Detecting abort inside a node

export class BatchProcessNode extends MonadicNode<NodeStateBase, 'success'> {
  readonly name = 'batch-process';
  readonly '@id' = 'urn:noocodec:node:batch-process';
  readonly outputs = ['success'] as const;
  override get outputSchema(): Record<'success', SchemaObjectType> {
    return { 'success': { 'type': 'object' } };
  }

  override async execute(batch: Batch<NodeStateBase>, context: NodeContextType) {
    const items = ['alpha', 'beta', 'gamma', 'delta', 'epsilon'];
    for (const batchItem of batch) {
      for (const item of items) {
        if (context.signal.aborted) break;        // check between iterations
        await this.processItem(batchItem.state, item, context.signal); // propagate to every IO call
      }
    }
    return RoutedBatch.create(NodeOutput.create('success').output, batch);
  }

  /** Simulate per-item IO; propagates the signal so the item-level wait aborts. */
  private async processItem(state: NodeStateBase, item: string, signal: AbortSignal): Promise<void> {
    await new Promise<void>((resolve, reject) => {
      const t = setTimeout(resolve, 200);
      signal.addEventListener('abort', () => { clearTimeout(t); reject(signal.reason); }, { once: true });
    });
    state.setMetadata('lastProcessedItem', item);
  }
}

A node that ignores context.signal runs to completion even after the signal fires. The dispatcher stops the iterator once the current node returns, but the in-flight node body still races to finish on its own.

After cancellation

Once the signal fires:

• The iterator stops without starting the next node.
• result.cursor holds the placement IRI that would have run next. Pass it to dispatcher.resume() to continue from that point.
• result.state.lifecycle.variant is 'cancelled' (caller signal) or 'timed_out' (deadline).

// Check the cursor and lifecycle after a cancelled run. The cursor holds the
// name of the next node that would have run; pass it to resume() to continue.
if (aResult.cursor !== null) {
  process.stdout.write(`paused at node: ${aResult.cursor}\n`);
  process.stdout.write(`lifecycle: ${aState.lifecycle.variant}\n`); // 'cancelled'
}

interruptedAt

When a flow exits via abort or timeout, result.interruptedAt carries structured cancellation telemetry: { nodeName: string; reason: 'abort' | 'timeout' }.

result.interruptedAt is null on clean exits (completed, terminal-reached, configuration error, node throw without abort). When a signal aborts the run or a deadline expires, it carries the node that was current when the signal fired and the discriminant:

// interruptedAt carries the node name and the abort discriminant.
// It is null on clean exits (completed, terminal reached, node throw without abort).
if (aResult.interruptedAt !== null) {
  process.stdout.write(`interrupted at node: ${aResult.interruptedAt.nodeName}\n`);
  process.stdout.write(`reason: ${aResult.interruptedAt.reason}\n`); // 'abort' or 'timeout'
}
if (bResult.interruptedAt !== null) {
  process.stdout.write(`deadline interrupted at: ${bResult.interruptedAt.nodeName}\n`);
  process.stdout.write(`reason: ${bResult.interruptedAt.reason}\n`); // 'timeout'
}

reason: 'timeout' is set when the abort reason is a TimeoutError (either the run-level deadlineMs deadline or a per-node timeoutMs budget). reason: 'abort' is set when the caller-supplied signal fired with any other reason.

Signal composition

The dispatcher uses Signal.compose(...) to merge cancellation concerns. Callers can do the same before passing a single signal in:

// Compose a user abort and a request-scoped deadline into a single signal
// before passing it to execute().
const userAbortController = new AbortController();
const combined = Signal.compose({
  'signal':     userAbortController.signal,
  'deadlineMs': 10_000,
});

const cState  = new NodeStateBase();
const cResult = await dispatcher.execute('urn:noocodec:dag:slow-dag', cState, { signal: combined });
userAbortController.abort(new Error('request scope ended')); // clean up

This is equivalent to passing both as signal plus deadlineMs. Pick whichever form fits the call site.

Details for Nerds

Runtime contract

Cancellation is cooperative at the node boundary. The dispatcher can stop before starting the next placement, but it cannot interrupt arbitrary synchronous work inside an already-running node. A node that performs IO or waits should pass context.signal into the underlying API, adapter, scheduler, or retry helper.

Related Concepts

• Retry - RetryPolicy.run honors context.signal so retries abort cleanly
• Checkpoint and Resume - abort and persist the cursor; resume continues from that point
• Observability - onError fires when an abort or deadline interrupts a node
• Example 06: Cancellation - runnable AbortController and deadlineMs example
• @studnicky/signal, Signal
• Reference, Contracts, ExecuteOptionsType