Dagonizer
Dagonizer

Scatter Extensions ​

What It Is ​

Scatter Extensions shows how an application adds domain-specific scatter behavior without changing dispatcher code. The CLI example installs TopNGatherStrategy and ThresholdReducer, then references both by registry key from a normal scatter DAG.

This is the extension seam behind custom gather and reducer policy: register the implementation once, then keep DAG assembly declarative.

How It Works ​

GatherStrategies.register(...) and OutcomeReducers.register(...) install custom implementations into global registries. A scatter placement references them through gather.strategy and reducer. The dispatcher resolves those names at execution time.

Diagrams, Examples, and Outputs ​

The DAG below is the same reservoirDag exported by the CLI example. It shows the custom gather strategy and reducer as graph configuration, not code branches inside the dispatcher.

scatter-extensions reservoir 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:reservoir-demo",
  "@type": "DAG",
  "name": "reservoir-demo",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:reservoir-demo/node/batch-score"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:reservoir-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:reservoir-demo/node/collect-top",
        "partial": "urn:noocodec:dag:reservoir-demo/node/collect-top",
        "all-error": "urn:noocodec:dag:reservoir-demo/node/collect-top",
        "empty": "urn:noocodec:dag:reservoir-demo/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:reservoir-demo/node/collect-top",
      "@type": "GatherNode",
      "name": "collect-top",
      "sources": {
        "urn:noocodec:dag:reservoir-demo/node/batch-score": {}
      },
      "gather": {
        "strategy": "top-n",
        "target": "topCandidates"
      },
      "outputs": {
        "success": "urn:noocodec:dag:reservoir-demo/node/end",
        "error": "urn:noocodec:dag:reservoir-demo/node/end",
        "empty": "urn:noocodec:dag:reservoir-demo/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:reservoir-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
  %% reservoir-demo (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_reservoir-demo/node/batch-score
  urn_noocodec_dag_reservoir-demo/node/batch-score[/"batch-score ▣ route ×10"/]
  urn_noocodec_dag_reservoir-demo/node/batch-score -->|all-success| urn_noocodec_dag_reservoir-demo/node/collect-top
  urn_noocodec_dag_reservoir-demo/node/batch-score -->|partial| urn_noocodec_dag_reservoir-demo/node/collect-top
  urn_noocodec_dag_reservoir-demo/node/batch-score -->|all-error| urn_noocodec_dag_reservoir-demo/node/collect-top
  urn_noocodec_dag_reservoir-demo/node/batch-score -->|empty| urn_noocodec_dag_reservoir-demo/node/end
  urn_noocodec_dag_reservoir-demo/node/collect-top{"collect-top"}
  urn_noocodec_dag_reservoir-demo/node/collect-top -->|success| urn_noocodec_dag_reservoir-demo/node/end
  urn_noocodec_dag_reservoir-demo/node/collect-top -->|error| urn_noocodec_dag_reservoir-demo/node/end
  urn_noocodec_dag_reservoir-demo/node/collect-top -->|empty| urn_noocodec_dag_reservoir-demo/node/end
  urn_noocodec_dag_reservoir-demo/node/end((("end")))
  classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
  class urn_noocodec_dag_reservoir-demo/node/batch-score reservoir

Run ​

bash
npx tsx examples/scatter-extensions.ts

What It Lets You Do ​

Scatter extensions let applications add domain-specific gather and reducer behavior without changing dispatcher code. Use them when built-in strategies do not match the merge policy or success threshold your DAG needs.

GatherStrategies.register and OutcomeReducers.register install custom plugins into the global registries. A scatter placement then references them by strategy/reducer key. No dispatcher or DAG document changes are needed beyond the gather.strategy and reducer keys.

This example registers two plugins:

  • TopNGatherStrategy — collects the top-3 candidates by score from each clone's state into state.topCandidates.
  • ThresholdReducer — gates 'success' on ≥ 75% of clones returning 'success'; routes 'partial' below that threshold.

Code Samples ​

ts
/**
 * scatter-extensions: registers and uses the custom TopNGatherStrategy and
 * ThresholdReducer from dags/scatter-extensions.ts in a live scatter DAG.
 *
 * Importing dags/scatter-extensions.ts triggers the two side-effect
 * GatherStrategies.register / OutcomeReducers.register calls that install the
 * plugins into the global registries. A scatter DAG then references them by
 * name ('top-n' gather strategy, 'threshold-75' reducer).
 *
 * The worker node writes a scored { title, score } candidate to each clone's
 * state. TopNGatherStrategy collects the top-3 by score into state.topCandidates.
 * ThresholdReducer gates success on >= 75% of clones returning 'success'.
 *
 * DAG definition (GatherStrategy + OutcomeReducer registrations, ScoreNode, RankingState): examples/dags/scatter-extensions.ts
 *
 * Run: npx tsx examples/scatter-extensions.ts
 */

// Import triggers the registry.register calls for 'top-n' and 'threshold-75'.
import {
  RankingState,
  ScoreNode,
} from './dags/scatter-extensions.js';

import {
  DAG_CONTEXT,
  Dagonizer,
  GatherStrategies,
  OutcomeReducers,
} from '@studnicky/dagonizer';
import type { DAGType } from '@studnicky/dagonizer';

// ── DAG: scatter over items, gather with 'top-n', reduce with 'threshold-75' ─

const TOP_N_RANKING_DAG_IRI = 'urn:noocodec:dag:top-n-ranking';
const placement = (placementIdentifier: string): string => `${TOP_N_RANKING_DAG_IRI}/node/${encodeURIComponent(placementIdentifier)}`;

const dag: DAGType = {
  '@context':  DAG_CONTEXT,
  '@id': TOP_N_RANKING_DAG_IRI,
  '@type':     'DAG',
  name:        'top-n-ranking',
  version:     '1',
  entrypoints: { main: placement('rank-all') },
  nodes: [
    {
      '@id': placement('rank-all'),
      '@type':      'ScatterNode',
      name:         'rank-all',
      body:         { node: 'urn:noocodec:node:score' },
      source:       'items',
      itemKey:      'item',
      execution: { mode: 'item', concurrency: 5 },
      reducer:      'threshold-75',    // custom OutcomeReducer: >= 75% success
      outputs: {
        'all-success': placement('collect-top'),
        partial: placement('collect-top'),
        'all-error': placement('collect-top'),
        empty: placement('end'),
      },
    },
    {
      '@id': placement('collect-top'),
      '@type': 'GatherNode',
      name: 'collect-top',
      sources: { [placement('rank-all')]: {} },
      gather: {
        strategy: 'top-n',
        target:   'topCandidates',
      },
      outputs: { success: placement('end'), error: placement('end'), empty: placement('end') },
    },
    {
      '@id': placement('end'),
      '@type':   'TerminalNode',
      name:      'end',
      outcome:   'completed',
    },
  ],
};

// ── Verify plugins registered before dispatch ─────────────────────────────────

process.stdout.write('\n=== scatter-extensions: custom GatherStrategy + OutcomeReducer ===\n\n');
process.stdout.write(`Registered gather strategies: ${GatherStrategies.list().join(', ')}\n`);
process.stdout.write(`Registered outcome reducers:  ${OutcomeReducers.list().join(', ')}\n\n`);

// ── Run ──────────────────────────────────────────────────────────────────────

const dispatcher = new Dagonizer<RankingState>();
dispatcher.registerNode(new ScoreNode());
dispatcher.registerDAG(dag);

const state = new RankingState();
// Items with varying title lengths → varying scores
state.items = [
  'A',
  'Medium length title',
  'This is a longer candidate title',
  'Short',
  'Another medium item here',
  'x',
];

await dispatcher.execute(TOP_N_RANKING_DAG_IRI, state);

process.stdout.write(`topCandidates (top-3 by score):\n`);
for (const c of state.topCandidates) {
  process.stdout.write(`  { title: "${c.title}", score: ${String(c.score)} }\n`);
}

process.stdout.write('\nLesson: GatherStrategies.register + OutcomeReducers.register install\n');
process.stdout.write('        extension keys globally; DAG topology still routes by placement IRI.\n');

Details for Nerds ​

  • GatherStrategies.register(strategy). Installs a custom GatherStrategy implementation into the global registry under strategy.name. Any gather configuration whose strategy field matches the key uses this implementation.
  • Custom GatherStrategy. Implement reduce(config, batch, state, accessor) (called per clone as it completes) and override finalize(config, execution) for all-at-once processing after every clone is done. The TopNGatherStrategy accumulates nothing in reduce and sorts by score in finalize, keeping only the top 3 and merging into state.topCandidates.
  • OutcomeReducers.register(reducer). Installs a custom OutcomeReducer into the global registry under reducer.name. A scatter placement's reducer field references that key.
  • Custom OutcomeReducer. Receives the array of clone outcomes ('success' / 'error' / 'empty') and returns the routing token for the parent's output port. ThresholdReducer computes the success ratio and routes 'success' at ≥ 75%, 'partial' below.
  • Side-effect registration. Importing examples/dags/scatter-extensions.ts triggers the register calls. The convention is to register in the module that defines the plugin, so any application that imports the module gets the plugin installed automatically.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky