Dagonizer
Dagonizer

Reservoir ​

What It Is ​

A reservoir is scatter's keyed input-batching policy. Instead of dispatching one source item per body invocation, a reservoir buffers source items by key and releases bounded Batch<N> chunks when capacity, idle time, or source completion says it is time.

Use it when the scatter body should process micro-batches: provider APIs that accept batches, grouped aggregation, streaming fan-out with bounded memory, or resumable work where reissuing already-acknowledged items is unacceptable.

How It Works ​

The scatter buffers source items by key until a release condition is met: size, time, flush, or source completion. It then dispatches a batch for that key through the normal scatter body. Checkpoint metadata tracks released and pending work so resume can avoid duplicates.

A reservoir is a scatter's keyed input-batching policy. Without it, a scatter dispatches one source item per body invocation (batch-size-1). With it, the scatter buffers source items by a key and releases a Batch<N> per key — so the body node runs once over N items, the gather folds the whole batch in a single reduce, and throughput amortizes over the batch.

The reservoir is not a new placement: it is execution.reservoir — one of the two execution.mode variants on a ScatterNode (the other is 'item', the non-reservoir default). execution.concurrency, when set alongside mode: 'reservoir', gates concurrently in-flight batches — the same Semaphore concept mode: 'item' uses for concurrently in-flight items, applied at batch instead of item granularity. There is no throttle field in reservoir mode: a per-item Throttle does not compose with a batch dispatch unit whose size varies with capacity/idle/flush triggers; the schema structurally forbids the combination. See ScatterNode execution policy for the full field reference.

Diagrams, Examples, and Outputs ​

The supporting reservoir example exports the exact ScatterNode configuration. The JSON-LD shows execution.mode: "reservoir"; Mermaid shows that it is still one scatter placement in the DAG.

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

What It Lets You Do ​

Use when ​

Use a reservoir when scatter input is a stream but the body should process keyed micro-batches instead of one item at a time. This is for throughput, batching APIs, grouped aggregation, and resumable fan-out with bounded memory.

Code Samples ​

The source below is the reservoir placement from the same DAG rendered above.

Details for Nerds ​

Configuration ​

ts
/**
 * DAG showing a reservoir-configured scatter: items are batched by `route`
 * before the score node runs. The reservoir holds up to 10 items per partition
 * key; a partial batch flushes after 500 ms of idle time.
 */
export const reservoirDag: DAGType = {
  '@context': DAG_CONTEXT,
  '@id': RESERVOIR_DEMO_DAG_IRI,
  '@type':    'DAG',
  name:       'reservoir-demo',
  version:    '1',
  entrypoints: { main: placement('batch-score') },
  nodes: [
    {
      '@id': placement('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': 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('batch-score')]: {} },
      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',
    },
  ],
};

The partition key is String(accessor.get(item, keyField)). The reservoir requires a node body (a { node } body, not a sub-DAG or container body) — the node processes the released Batch<N> directly.

Release triggers ​

A key's buffer releases as one batch when any of three triggers fires:

  • capacity — the buffer reaches capacity items. The primary trigger; bounds memory.
  • idle — idleMs is set and the key receives no new item for that long. Driven by the engine's swappable Scheduler, so it is deterministic under VirtualScheduler in tests. Bounds latency for sparse keys.
  • complete — the source drains; every non-empty buffer flushes as a final partial batch. Always on.

Exactly-once and crash safety ​

The reservoir inherits the scatter's durable inbox. Each pulled item enters the inbox before it is buffered (at-least-once at the source). A released batch acks atomically: the gather reduce folds the whole batch once, all N items leave the inbox, and the checkpoint is written. On resume the buffers are rebuilt from the inbox grouped by key, so no item is lost or double-folded. See checkpoint and resume.

When to use it ​

  • Throughput — amortize per-item dispatch and bridge overhead over a batch.
  • Micro-batching at a decision point — "emit the batch of events that need GDPR review" is a keyed reservoir on the route.
  • Bounded memory on a stream — a 1k–1M-event source releases fixed-size batches with backpressure instead of materializing everything.

Visualization ​

A reservoir-configured scatter renders a distinct glyph. The Mermaid renderer labels it ▣ <keyField> ×<capacity> and assigns a reservoir class; the Cytoscape renderer adds a dag-reservoir class and a reservoir data field. A application animation layer drives the live per-key fill from observer buffer-size deltas. See visualization.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky