Plural-Native Execution

What It Is

Dagonizer moves work through a DAG as batches. A single item is just a batch of one; branching means partitioning a batch by output port; joins mean merging pending batches at the same placement. Scatter, gather, retry, checkpoint, and reservoir behavior all build on that one contract.

This page is the mental model for execution. Once you understand that every node receives Batch<TState> and returns routed sub-batches, the rest of the engine stops looking like special cases.

How It Works

Every node receives a Batch<TState> and returns routed batches partitioned by output. A single item is represented as a batch of one. Scatter clones, reservoir batches, gather folds, checkpoint progress, and retry semantics all build on that plural-native contract.

The work-set scheduler keeps a Map<placement, Batch>. It fires the lowest-rank placement with pending items, merges returned sub-batches into downstream placements, and repeats until the graph drains or the lifecycle exits. A size-1 run follows the same path as a size-N run.

Diagrams, Examples, and Outputs

The focused runnable module exports a reservoir-backed scatter DAG. The JSON-LD shows the ScatterNode placement and reservoir policy; the Mermaid view shows that the topology is still a simple scatter-to-terminal graph.

import { DAG_CONTEXT } from '@studnicky/dagonizer';
import type { DAGType } from '@studnicky/dagonizer';

// ScoreState holds the items array fed to the scatter and the collected results.
export class ScoreState extends NodeStateBase {
  items:          unknown[] = [];
  topCandidates:  unknown[] = [];
}

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

  override async execute(batch: Batch<ScoreState>) {
    return RoutedBatch.create(NodeOutput.create('scored').output, batch);
  }
}

// DAG with a reservoir-configured scatter: items accumulate per `route` key
// before ScoreNode runs. Up to 10 items per key; partial batches flush after
// 500 ms of idle time.
export const reservoirDag: DAGType = {
  '@context':  DAG_CONTEXT,
  '@id': 'urn:noocodec:dag:plural-native-demo',
  '@type':     'DAG',
  name:        'plural-native-demo',
  version:     '1',
  entrypoints: { main: 'urn:noocodec:dag:plural-native-demo/node/batch-score' },
  nodes: [
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/batch-score',
      '@type':     'ScatterNode',
      name:        'batch-score',
      body:        { node: 'urn:noocodec:node:score' },
      source:      'items',
      itemKey:     'item',
      execution: {
        mode:       'reservoir',
        concurrency: 4,
        reservoir: {
          keyField: 'route',   // accessor path on each source item → the partition key
          capacity: 10,        // release a batch when 10 items accumulate per key
          idleMs:   500,       // flush partial batches after 500 ms idle
        },
      },
      outputs: {
        'all-success': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        partial: 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        'all-error': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        empty: 'urn:noocodec:dag:plural-native-demo/node/end',
      },
    },
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
      '@type': 'GatherNode',
      name: 'collect-top',
      sources: { 'urn:noocodec:dag:plural-native-demo/node/batch-score': {} },
      gather: {
        strategy: 'append',
        target:   'topCandidates',
      },
      outputs: { success: 'urn:noocodec:dag:plural-native-demo/node/end', error: 'urn:noocodec:dag:plural-native-demo/node/end', empty: 'urn:noocodec:dag:plural-native-demo/node/end' },
    },
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/end',
      '@type': 'TerminalNode',
      name:    'end',
      outcome: 'completed',
    },
  ],
};

plural-native reservoir scatter DAG

3 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:plural-native-demo",
  "@type": "DAG",
  "name": "plural-native-demo",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:plural-native-demo/node/batch-score"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:plural-native-demo/node/batch-score",
      "@type": "ScatterNode",
      "name": "batch-score",
      "body": {
        "node": "urn:noocodec:node:score"
      },
      "source": "items",
      "itemKey": "item",
      "execution": {
        "mode": "reservoir",
        "concurrency": 4,
        "reservoir": {
          "keyField": "route",
          "capacity": 10,
          "idleMs": 500
        }
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:plural-native-demo/node/collect-top",
        "partial": "urn:noocodec:dag:plural-native-demo/node/collect-top",
        "all-error": "urn:noocodec:dag:plural-native-demo/node/collect-top",
        "empty": "urn:noocodec:dag:plural-native-demo/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:plural-native-demo/node/collect-top",
      "@type": "GatherNode",
      "name": "collect-top",
      "sources": {
        "urn:noocodec:dag:plural-native-demo/node/batch-score": {}
      },
      "gather": {
        "strategy": "append",
        "target": "topCandidates"
      },
      "outputs": {
        "success": "urn:noocodec:dag:plural-native-demo/node/end",
        "error": "urn:noocodec:dag:plural-native-demo/node/end",
        "empty": "urn:noocodec:dag:plural-native-demo/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:plural-native-demo/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
  %% plural-native-demo (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_plural-native-demo/node/batch-score
  urn_noocodec_dag_plural-native-demo/node/batch-score[/"batch-score ▣ route ×10"/]
  urn_noocodec_dag_plural-native-demo/node/batch-score -->|all-success| urn_noocodec_dag_plural-native-demo/node/collect-top
  urn_noocodec_dag_plural-native-demo/node/batch-score -->|partial| urn_noocodec_dag_plural-native-demo/node/collect-top
  urn_noocodec_dag_plural-native-demo/node/batch-score -->|all-error| urn_noocodec_dag_plural-native-demo/node/collect-top
  urn_noocodec_dag_plural-native-demo/node/batch-score -->|empty| urn_noocodec_dag_plural-native-demo/node/end
  urn_noocodec_dag_plural-native-demo/node/collect-top{"collect-top"}
  urn_noocodec_dag_plural-native-demo/node/collect-top -->|success| urn_noocodec_dag_plural-native-demo/node/end
  urn_noocodec_dag_plural-native-demo/node/collect-top -->|error| urn_noocodec_dag_plural-native-demo/node/end
  urn_noocodec_dag_plural-native-demo/node/collect-top -->|empty| urn_noocodec_dag_plural-native-demo/node/end
  urn_noocodec_dag_plural-native-demo/node/end((("end")))
  classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
  class urn_noocodec_dag_plural-native-demo/node/batch-score reservoir

For larger runnable contexts:

• Scatter Extensions shows batch-native nodes, custom gather, direct node calls, and reservoir authoring.
• Example 14: Gather Strategies shows gather behavior against real Cartographer DAGs.
• Example 16: Scatter Resume shows durable progress over scatter batches.
• Reservoir explains keyed input batching for scatter sources.

What It Lets You Do

Use when

Use this guide when reasoning about how Dagonizer executes batches, scatter worksets, and reservoir releases. The mental model is useful before writing custom nodes, gather strategies, outcome reducers, or migration code.

Code Samples

The batch contract

A node consumes a Batch<TState> and returns a RoutedBatchType<TOutput>:

import { MonadicNode, RoutedBatch, Batch } from '@studnicky/dagonizer';
import type { NodeContextType, NodeStateInterface, RoutedBatchType, SchemaObjectType } from '@studnicky/dagonizer';

// The execute signature: consume Batch<TState>, return RoutedBatchType<TOutput, TState>.
// Items are partitioned across output ports — routing IS partitioning.
export class EchoNode extends MonadicNode<NodeStateInterface, 'out'> {
  readonly name    = 'echo';
  readonly '@id'   = 'urn:noocodec:node:echo';
  readonly outputs = ['out'] as const;
  override get outputSchema(): Record<'out', SchemaObjectType> {
    return { 'out': { 'type': 'object' } };
  }

  async execute(batch: Batch<NodeStateInterface>, _ctx: NodeContextType): Promise<RoutedBatchType<'out', NodeStateInterface>> {
    return RoutedBatch.create('out', batch);
  }
}

A Batch<TState> is an ordered collection of items, each carrying a stable id and a per-item TState. A RoutedBatchType<TOutput> is a Map<output, Batch> — the node's items partitioned across its output ports.

Routing is partitioning. Three things that look separate are one mechanism:

• per-item conditional routing — a node sends some items to needs-gdpr, others to geo-only;
• micro-batching — a node that emits batches instead of items;
• the reservoir — a node that partitions over time, buffering until a threshold.

Details for Nerds

The node taxonomy

Every node descends from one root. You pick the local implementation shape by how you want to write the work, not by what the engine does:

Shape	You implement	Use for
Batch-native MonadicNode	execute(batch, context) as one whole-batch transform	hot-path nodes that process the whole batch at once (shared LRU caches, vectorized work)
Per-item MonadicNode	execute(batch, context) with a local item loop	item-oriented logic where each state routes independently

MonadicNode is the minimum viable node — it supplies timeout / validate / destroy defaults and leaves name, outputs, outputSchema, and execute abstract. Concrete nodes must implement all four. outputSchema is an abstract getter (abstract get outputSchema(): Record<TOutput, SchemaObjectType>) that declares a per-port JSON Schema fragment describing the state delta the node writes when routing to that port; the compiler enforces its presence on every subclass. A single item is still a batch of one; per-item routing is a local loop that builds a RoutedBatchType.

The Cartographer's RouteGeoNode is the runnable example: it receives a whole batch, routes each item to has-geo or needs-geo, and returns a routed batch map for the scheduler to merge into downstream placements.

export class RouteGeoNode extends MonadicNode<CartographerState, 'has-geo' | 'needs-geo'> {
  readonly '@id' = 'urn:noocodec:node:route-geo';
  readonly 'name' = 'route-geo';
  readonly 'outputs' = ['has-geo', 'needs-geo'] as const;

  override get outputSchema(): Record<'has-geo' | 'needs-geo', SchemaObjectType> {
    return {
      'has-geo':   { 'type': 'object' },
      'needs-geo': { 'type': 'object' },
    };
  }

  override async execute(
    batch: Batch<CartographerState>,
    _context: NodeContextType,
  ): Promise<RoutedBatchType<'has-geo' | 'needs-geo', CartographerState>> {
    const acc = new Map<'has-geo' | 'needs-geo', ItemType<CartographerState>[]>();

    for (const item of batch) {
      const result = this.routeItem(item.state);
      for (const error of result.errors) {
        item.state.collectError(error);
      }
      const bucket = acc.get(result.output);
      if (bucket === undefined) {
        acc.set(result.output, [item]);
      } else {
        bucket.push(item);
      }
    }

    const routed = new Map<'has-geo' | 'needs-geo', Batch<CartographerState>>();
    for (const [output, items] of acc) {
      routed.set(output, Batch.from(items));
    }
    return routed;
  }

  private routeItem(state: CartographerState): NodeOutputType<'has-geo' | 'needs-geo'> {
    const geo = state.canonical.geo;
    // A source's pre-resolved geo only lets us skip the lookup when it actually
    // resolved a location — an 'UNK'/'Unmapped' placeholder (e.g. a ping whose
    // coords were out of range at the source) is NOT resolved, so it must run
    // the lookup path where validate-coords can reject the bad coords.
    const hasResolvedGeo =
      geo !== undefined &&
      geo.country.length > 0 &&
      geo.country !== 'UNK' &&
      geo.region.length > 0 &&
      geo.region !== 'Unmapped';

    if (hasResolvedGeo) {
      state.routing = { ...state.routing, 'geoLookupSkipped': true, 'geoLookupRun': false };
      return NodeOutput.create('has-geo');
    }
    state.routing = { ...state.routing, 'geoLookupRun': true, 'geoLookupSkipped': false };
    return NodeOutput.create('needs-geo');
  }
}

How a DAG processes a batch: the work-set walk

A DAG is substitutable with a node — given a Batch<N>, it processes it natively. The executor is a work-set scheduler:

• The work set is Map<placement, Batch> — the items waiting at each node.
• It is initialized with the input batch at the entrypoint.
• Each step fires the lowest-rank node that holds items: run it over its accumulated batch, then merge each output port's sub-batch into the downstream node's pending work. Items reaching a terminal are collected.
• Rank is a topological order over forward edges (back-edges — retry self-edges — are excluded). Lowest-rank-first means a join coalesces: it waits for all its feeders to drain, then fires once over the merged batch.

A node fires once over all the items currently at it — that is the batching. Items that branch take different next-nodes; items that converge merge at the join. For a size-1 input the work set holds one item at one node and the walk degenerates to a simple cursor — identical to single-item execution.

Retry is a flow shape. A node routes a retry output back to an earlier (or self) placement; those items re-enter that node's pending work and re-fire on a later pass, re-batched with whatever else is there. The per-item retry budget lives on state (withinRetryBudget) and bounds the loop.

The reservoir

A scatter's source can be batched by a reservoir — its keyed input-batching policy. Instead of dispatching one item at a time, it buffers items by a key and releases a Batch<N> per key when one of three triggers fires:

• capacity — the key's buffer reaches capacity;
• idle — the key has been idle for idleMs (driven by the swappable scheduler);
• complete — the source drains, flushing every partial buffer.

import { DAG_CONTEXT } from '@studnicky/dagonizer';
import type { DAGType } from '@studnicky/dagonizer';

// ScoreState holds the items array fed to the scatter and the collected results.
export class ScoreState extends NodeStateBase {
  items:          unknown[] = [];
  topCandidates:  unknown[] = [];
}

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

  override async execute(batch: Batch<ScoreState>) {
    return RoutedBatch.create(NodeOutput.create('scored').output, batch);
  }
}

// DAG with a reservoir-configured scatter: items accumulate per `route` key
// before ScoreNode runs. Up to 10 items per key; partial batches flush after
// 500 ms of idle time.
export const reservoirDag: DAGType = {
  '@context':  DAG_CONTEXT,
  '@id': 'urn:noocodec:dag:plural-native-demo',
  '@type':     'DAG',
  name:        'plural-native-demo',
  version:     '1',
  entrypoints: { main: 'urn:noocodec:dag:plural-native-demo/node/batch-score' },
  nodes: [
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/batch-score',
      '@type':     'ScatterNode',
      name:        'batch-score',
      body:        { node: 'urn:noocodec:node:score' },
      source:      'items',
      itemKey:     'item',
      execution: {
        mode:       'reservoir',
        concurrency: 4,
        reservoir: {
          keyField: 'route',   // accessor path on each source item → the partition key
          capacity: 10,        // release a batch when 10 items accumulate per key
          idleMs:   500,       // flush partial batches after 500 ms idle
        },
      },
      outputs: {
        'all-success': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        partial: 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        'all-error': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
        empty: 'urn:noocodec:dag:plural-native-demo/node/end',
      },
    },
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/collect-top',
      '@type': 'GatherNode',
      name: 'collect-top',
      sources: { 'urn:noocodec:dag:plural-native-demo/node/batch-score': {} },
      gather: {
        strategy: 'append',
        target:   'topCandidates',
      },
      outputs: { success: 'urn:noocodec:dag:plural-native-demo/node/end', error: 'urn:noocodec:dag:plural-native-demo/node/end', empty: 'urn:noocodec:dag:plural-native-demo/node/end' },
    },
    {
      '@id': 'urn:noocodec:dag:plural-native-demo/node/end',
      '@type': 'TerminalNode',
      name:    'end',
      outcome: 'completed',
    },
  ],
};

The body node then runs once over each released batch, and the gather folds the whole batch in a single reduce. With no reservoir config the dispatch unit is batch-size-1 — exactly the default behavior. See the reservoir guide.

Checkpoint and Resume

In-flight state is fully serialized. For a multi-item run, the work set (per-node item-state snapshots) is captured into state metadata on interruption and rebuilt exactly on resume. A scatter's durable inbox composes with it. Resume is byte-equivalent to an uninterrupted run; a size-1 run uses the cursor model unchanged. See checkpoint and resume.

Migrating existing nodes

Upgrading from the single-item contract? See Migrating to the batch contract — most leaf nodes change a base class and a method name, nothing more.

Related Concepts

• Migrating to the batch contract - upgrade checklist for single-item node code
• Reservoir - keyed input batching for scatter sources
• Reference: Core - Batch, RoutedBatch, gather strategy, and outcome reducer primitives
• Example 04: Scatter Scout shows fan-out over worksets.
• Example 14: Gather Strategies shows parent-state folds.
• Example 16: Scatter Resume shows durable progress over scatter batches.