Example 17: Async Scatter Source
What It Is
Async Scatter Source lets a DAG process input that arrives over time instead of forcing the application to build a complete array before execution starts. The Cartographer uses this for generated event streams: the scatter pulls from an AsyncIterable only when worker capacity is available.
The runtime normalises Array, Iterable, and AsyncIterable sources to one pull interface. Your DAG still declares a normal scatter; the source value determines whether items are already available or produced lazily.
How It Works
The scatter executor normalises arrays, iterables, and async iterables into one pull interface. It calls iterator.next() only when the configured concurrency pool has capacity. That means a producer cannot outrun the slowest workers by more than the pool window, and the same gather/reducer contract applies after each clone completes.
For application authors, the important knob is still scatter concurrency. With concurrency: 2, the pull loop keeps at most two body executions in flight and asks the async source for the next item only when one finishes.
Diagrams, Examples, and Outputs
DAG registration and diagram
The DAG shape is standard scatter; the source path resolves to an AsyncIterable at runtime. The Cartographer is the in-browser runnable for this principle: five intake entrypoints feed intake-gather, the gather writes a merged async stream to state.sources, and process-stream pulls only as worker capacity opens.
Cartographer async-source scatter DAG
5 placements{
"@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:cartographer",
"@type": "DAG",
"name": "dag:cartographer",
"version": "1.0",
"entrypoints": {
"position-ping": "urn:noocodec:dag:cartographer/node/intake-gather",
"facility-scan": "urn:noocodec:dag:cartographer/node/intake-gather",
"sensor-reading": "urn:noocodec:dag:cartographer/node/intake-gather",
"customs-event": "urn:noocodec:dag:cartographer/node/intake-gather",
"delivery-confirmation": "urn:noocodec:dag:cartographer/node/intake-gather"
},
"nodes": [
{
"@id": "urn:noocodec:dag:cartographer/node/intake-gather",
"@type": "GatherNode",
"name": "dag:cartographer/node/intake-gather",
"sources": {
"urn:noocodec:dag:cartographer/entrypoint/position-ping": {},
"urn:noocodec:dag:cartographer/entrypoint/facility-scan": {},
"urn:noocodec:dag:cartographer/entrypoint/sensor-reading": {},
"urn:noocodec:dag:cartographer/entrypoint/customs-event": {},
"urn:noocodec:dag:cartographer/entrypoint/delivery-confirmation": {}
},
"gather": {
"strategy": "source-intake"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/process-stream",
"error": "urn:noocodec:dag:cartographer/node/process-stream",
"empty": "urn:noocodec:dag:cartographer/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/process-stream",
"@type": "ScatterNode",
"name": "dag:cartographer/node/process-stream",
"source": "sources",
"body": {
"dag": "urn:noocodec:dag:stream-event"
},
"outputs": {
"all-success": "urn:noocodec:dag:cartographer/node/fold-insights",
"partial": "urn:noocodec:dag:cartographer/node/fold-insights",
"all-error": "urn:noocodec:dag:cartographer/node/fold-insights",
"empty": "urn:noocodec:dag:cartographer/node/summarize"
},
"itemKey": "source-payload",
"reducer": "aggregate",
"execution": {
"mode": "reservoir",
"concurrency": 16,
"reservoir": {
"keyField": "eventType",
"capacity": 1000
}
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/fold-insights",
"@type": "GatherNode",
"name": "dag:cartographer/node/fold-insights",
"sources": {
"urn:noocodec:dag:cartographer/node/process-stream": {}
},
"gather": {
"strategy": "insights-fold"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/summarize",
"error": "urn:noocodec:dag:cartographer/node/summarize",
"empty": "urn:noocodec:dag:cartographer/node/summarize"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/summarize",
"@type": "SingleNode",
"name": "dag:cartographer/node/summarize",
"node": "urn:noocodec:node:summarize",
"outputs": {
"success": "urn:noocodec:dag:cartographer/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer/node/done",
"@type": "TerminalNode",
"name": "dag:cartographer/node/done",
"outcome": "completed"
}
]
}Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
%% dag:cartographer (v1.0)
entry_position-ping(["position-ping"])
entry_position-ping --> urn_noocodec_dag_cartographer/node/intake-gather
entry_facility-scan(["facility-scan"])
entry_facility-scan --> urn_noocodec_dag_cartographer/node/intake-gather
entry_sensor-reading(["sensor-reading"])
entry_sensor-reading --> urn_noocodec_dag_cartographer/node/intake-gather
entry_customs-event(["customs-event"])
entry_customs-event --> urn_noocodec_dag_cartographer/node/intake-gather
entry_delivery-confirmation(["delivery-confirmation"])
entry_delivery-confirmation --> urn_noocodec_dag_cartographer/node/intake-gather
urn_noocodec_dag_cartographer/node/intake-gather{"dag_cartographer/node/intake-gather"}
urn_noocodec_dag_cartographer/node/intake-gather -->|success| urn_noocodec_dag_cartographer/node/process-stream
urn_noocodec_dag_cartographer/node/intake-gather -->|error| urn_noocodec_dag_cartographer/node/process-stream
urn_noocodec_dag_cartographer/node/intake-gather -->|empty| urn_noocodec_dag_cartographer/node/done
urn_noocodec_dag_cartographer/node/process-stream[/"dag:cartographer/node/process-stream ▣ eventType ×1000"/]
urn_noocodec_dag_cartographer/node/process-stream -->|all-success| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|partial| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|all-error| urn_noocodec_dag_cartographer/node/fold-insights
urn_noocodec_dag_cartographer/node/process-stream -->|empty| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights{"dag_cartographer/node/fold-insights"}
urn_noocodec_dag_cartographer/node/fold-insights -->|success| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights -->|error| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/fold-insights -->|empty| urn_noocodec_dag_cartographer/node/summarize
urn_noocodec_dag_cartographer/node/summarize["dag:cartographer/node/summarize"]
urn_noocodec_dag_cartographer/node/summarize -->|success| urn_noocodec_dag_cartographer/node/done
urn_noocodec_dag_cartographer/node/done((("dag:cartographer/node/done")))
classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
class urn_noocodec_dag_cartographer/node/process-stream reservoirThe scatter source field accepts Array, Iterable, or AsyncIterable. The engine normalises all three to the same AsyncIterator interface internally. The pull loop only calls iterator.next() when a worker slot is free — giving true backpressure: the generator yields no more than concurrency items ahead of the slowest worker.
The browser runner executes the same JSON-LD DAG. Data-type entrypoint labels target intake-gather directly; source-intake opens those per-type async streams and the scatter consumes the merged stream at bounded concurrency.
Run
bash
npm run docs:devWhat It Lets You Do
Async scatter sources let applications process streams without materialising the full input collection first. Use this when source data arrives from a generator, file cursor, network feed, channel, or producer DAG and the host needs backpressure.
The benefit is memory and pacing control. A large import, telemetry stream, or model-produced candidate stream can feed the same scatter/gather machinery as a small array without changing the downstream nodes.
Code Samples
CartographerSourceIntake creates one async stream per event type. SourceIntakeGather receives entrypoint-label records and merges the matching streams into state.sources. The DAG snippet shows that the gather is the intake point and the scatter placement does not need a separate streaming-specific node type.
ts
/**
* sourceIntake: stream construction helpers for Cartographer's open intake gather.
*
* The intake gather is entrypoint-driven: each data-type entrypoint is a
* canonical entrypoint IRI, and the gather/source helpers open the matching
* typed source stream directly.
*/
import type { CartographerState } from '../CartographerState.ts';
import type { SourcePayload } from '../entities/SourcePayload.ts';
import type { CanonicalEventVariant } from '../entities/index.ts';
import { EventStreamSource } from '../services/EventStreamSource.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { GatherRecordType } from '@studnicky/dagonizer/contracts';
import type { NodeStateInterface } from '@studnicky/dagonizer/types';
type CartographerEventType = CanonicalEventVariant['eventType'];
type CartographerIntakeState = NodeStateInterface & Pick<CartographerState, 'eventConfig' | 'streamCount'>;
class SourcePayloadStream {
private constructor() { /* static-only */ }
static async *roundRobin(streams: readonly AsyncIterable<SourcePayload>[]): AsyncIterable<SourcePayload> {
const active = streams.map((stream) => stream[Symbol.asyncIterator]());
while (active.length > 0) {
for (let i = 0; i < active.length;) {
const step = await active[i]?.next();
if (step === undefined || step.done === true) {
active.splice(i, 1);
continue;
}
yield step.value;
i++;
}
}
}
static async *skip(stream: AsyncIterable<SourcePayload>, count: number): AsyncIterable<SourcePayload> {
let skipped = 0;
for await (const item of stream) {
if (skipped < count) {
skipped++;
continue;
}
yield item;
}
}
static async *empty(): AsyncIterable<SourcePayload> {
return;
}
}
export class CartographerSourceIntake {
private constructor() { /* static-only */ }
static isState(state: NodeStateInterface): state is CartographerIntakeState {
const eventConfig = Reflect.get(state, 'eventConfig');
return eventConfig !== null
&& typeof eventConfig === 'object'
&& typeof Reflect.get(state, 'streamCount') === 'number';
}
static streamFor(state: CartographerIntakeState, eventType: CartographerEventType): AsyncIterable<SourcePayload> {
const totalCount = state.streamCount > 0 ? state.streamCount : undefined;
return CartographerSourceIntake.filterType(
EventStreamSource.streamTyped(state.eventConfig, totalCount),
eventType,
);
}
static mergedFor(state: CartographerState, resumeAfter: number = 0): AsyncIterable<SourcePayload> {
const streams = CARTOGRAPHER_IRIS.intakeEventTypes.map((eventType) =>
CartographerSourceIntake.streamFor(state, eventType),
);
const merged = SourcePayloadStream.roundRobin(streams);
return resumeAfter > 0 ? SourcePayloadStream.skip(merged, resumeAfter) : merged;
}
static mergeRecords(
records: readonly GatherRecordType[],
state: NodeStateInterface,
): AsyncIterable<SourcePayload> {
if (!CartographerSourceIntake.isState(state)) return SourcePayloadStream.empty();
const streams: AsyncIterable<SourcePayload>[] = [];
for (const source of CARTOGRAPHER_IRIS.intakeEventTypes) {
const record = records.find((candidate) => CartographerSourceIntake.sourceType(candidate.source) === source);
if (record === undefined) continue;
streams.push(CartographerSourceIntake.streamFor(state, source));
}
return SourcePayloadStream.roundRobin(streams);
}
private static sourceType(source: string): CartographerEventType | null {
const marker = '/entrypoint/';
const index = source.indexOf(marker);
if (index < 0) return null;
const label = decodeURIComponent(source.slice(index + marker.length));
return CARTOGRAPHER_IRIS.intakeEventTypes.includes(label as CartographerEventType)
? (label as CartographerEventType)
: null;
}
private static async *filterType(
stream: AsyncIterable<SourcePayload>,
eventType: CartographerEventType,
): AsyncIterable<SourcePayload> {
for await (const item of stream) {
if (item.eventType === eventType) yield item;
}
}
}ts
export class SourceIntakeGather extends GatherStrategy {
readonly name = 'source-intake';
readonly '@id' = 'urn:noocodec:node:source-intake';
override initial(
_config: GatherConfigType,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
accessor.set(state, 'sources', []);
}
override reduce(
_config: GatherConfigType,
batch: Parameters<GatherStrategy['reduce']>[1],
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const records: GatherRecordType[] = [];
for (const item of batch) records.push(item.state);
accessor.set(state, 'sources', CartographerSourceIntake.mergeRecords(records, state));
}
}
GatherStrategies.register(new SourceIntakeGather());ts
/**
* cartographerDAG: multi-entry source intake gather over raw source payloads.
*
* Five canonical entrypoint IRIs target intake-gather directly. The
* source-intake gather opens those per-type streams and merges them into
* state.sources.
* The processing scatter reads that merged stream at concurrency 16, runs
* stream-event per item, and folds completed
* clone state through insights-fold. Memory is O(1) regardless of event count.
*
* Topology:
* 5 data-type entrypoints → gather('intake-gather', source-intake)
* → scatter('process-stream', 'sources', { dag: 'stream-event' }, concurrency: 16)
* → gather('fold-insights', strategy: insights-fold)
* → summarize → done
*/
export const cartographerDAG: DAGType = new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')
.gather(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
{ 'strategy': 'source-intake' },
{
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
},
)
.scatter(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
'sources',
{ 'dag': STREAM_EVENT_DAG_IRI },
{
'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'partial': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'all-error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
},
{
'itemKey': 'source-payload',
'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': 1000 } },
},
)
.gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
}, { 'strategy': 'insights-fold' }, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
})
// Pass-through in the streaming path (insights-fold already populated
// state.insights, state.journeys, and state.sampleRecords). Falls back
// to the records-based fold for non-streaming callers.
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'), summarizeInsights, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
})
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'), { outcome: 'completed' })
.entrypoints({
'position-ping': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'facility-scan': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'sensor-reading': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'customs-event': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
})
.build();Details for Nerds
AsyncIterableas scatter source. Any async generator or async-iterable value is a valid scatter source. The engine calls.next()lazily on each tick of the concurrency pool.- Bounded-concurrency backpressure. With
concurrency=2, at most two clones run simultaneously. The pull loop does not calliterator.next()until a slot frees, capping how far ahead the generator runs. Array sources follow the same discipline — "eagerly available" only affects when data is produced, not the concurrency semantics. - Resumability note. An
AsyncIterableon state is not captured byCheckpoint.capture()— generators are not JSON-serialisable. After an abort with an async source, the resume call must re-provide the generator at the continuation position. Cartographer does that throughCartographerSourceIntake.mergedFor(state, cursor). - Runnable source. The Cartographer Stream panel is fed by the same async source path as the DAG.
Related Concepts
- Example 04: Scatter Scout - scatter mechanics: source, body DAG, gather placement, reduce
- Example 16: Scatter resume - durable inbox: resumability across abort with async sources
- Example 15: Incremental gather - incremental gather: fold results as clones complete