Example 15: Incremental Gather
What It Is
Incremental Gather is for scatter work where parent state should become useful before every clone finishes. The Cartographer folds completed stream records into insight aggregates as they arrive, so the UI can show progress without materialising the entire event stream.
The core distinction is reduce versus finalize: fold clone results as they arrive, or wait and merge once at the end.
How It Works
The scatter executor routes completed clone records into a first-class gather placement. InsightsFoldGather reads each clone result, updates bounded parent aggregates in reduce, and leaves the final finalize hook with little to do. A batch-style strategy can do the opposite: keep reduce empty and perform one final merge in finalize.
Application code chooses the timing based on product needs. Dashboards and streaming ETL usually want incremental fold; all-at-once ranking or reconciliation may prefer finalize.
Diagrams, Examples, and Outputs
DAG registration and diagram
The in-browser Cartographer is the real incremental gather example. process-stream routes completed clones to fold-insights, which folds each record into state.insights, state.journeys, and state.sampleRecords as records arrive; the UI reads those evolving aggregates while the stream is running.
Cartographer incremental gather 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 reservoirEvery GatherStrategy subclass implements the fold contract:
Run
bash
npm run docs:devWhat It Lets You Do
Incremental gather lets applications fold completed scatter clones into parent state as they finish instead of waiting for every clone to complete. Use it for streaming dashboards, large ETL jobs, and bounded-memory aggregates where the parent state should become useful during the scatter.
Code Samples
Read the snippets with the diagrams nearby so the TypeScript behavior, JSON-LD graph shape, and runtime output line up as one contract.
ts
/**
* InsightsFoldGather: stateless incremental gather strategy for the Cartographer demo.
*
* Folds each scatter clone's `state.enriched` (an EnrichedShipment) into THREE
* BOUNDED accumulators that live in parent state (via accessor), rather than in
* instance fields. Every accumulator is read, mutated, and written back via
* accessor.get/set so the pattern survives process restart: on resume, already-acked
* items' contributions remain in the checkpoint; the instance holds zero mutable state.
*
* (a) state.insights — EXACT per-region rollup (bounded: ~6-8 continent keys).
* (b) state.journeyAccumulators — BOUNDED per-journey in-progress accumulators (cap: MAX_SAMPLE_JOURNEYS).
* (c) state.journeys — FINALIZED per-journey map, written by finalize().
* (d) state.sampleRecords — CAPPED FIFO ring of recent scans (cap: MAX_SAMPLE_RECORDS).
* (e) state.errorRollup — bounded error distribution from clone capturedErrors.
*
* Memory never grows with event count. Only the journeys sample is lossy
* (first MAX_SAMPLE_JOURNEYS unique shipmentIds are tracked); region arithmetic
* is exact across all clones.
*
* Registered as 'insights-fold' at module load. The strategy instance is a
* singleton that holds NO accumulator fields; all mutable state lives in
* the parent NodeStateInterface, keyed by accessor paths.
*/
import type { GatherExecutionType, GatherRecordType } from '@studnicky/dagonizer/contracts';
import { GatherStrategies, GatherStrategy } from '@studnicky/dagonizer/core';
import type { GatherConfigType, NodeStateInterface } from '@studnicky/dagonizer/types';
import type { StateAccessorInterface } from '@studnicky/dagonizer/contracts';
import { CircularBuffer } from '@studnicky/circular-buffer';
import { EnrichedShipmentGuard, type EnrichedShipment } from '../entities/EnrichedShipment.ts';
import type { JourneyInsights, JourneyScan, RegionInsights } from '../CartographerState.ts';
import { GeoErrorRecord } from '../errors/GeoErrorRecord.ts';
import { ErrorRollup } from '../errors/ErrorRollup.ts';
// ── Module constants ───────────────────────────────────────────────────────────
const MAX_SAMPLE_JOURNEYS = 100;
const MAX_SAMPLE_RECORDS = 200;
// Per-journey scan retention cap. A journey sample keeps at most this many
// JourneyScan objects (and status entries) so memory stays O(1) regardless of
// event count: a hot shipmentId folded a million times retains a bounded scan
// list, not a million scans. Exact scalar metrics (scanCount, pathKm, epoch
// bounds, offset/timezone/jurisdiction sets) are maintained independently of
// the capped scan list, so reconstruction stays faithful for the retained
// prefix while totals remain exact.
const MAX_SCANS_PER_JOURNEY = 64;
// ── Exported accumulator type ─────────────────────────────────────────────────
/** Internal accumulator for per-journey incremental folding. Exported for CartographerState snapshot/restore. */
export interface JourneyAccumulator {
scans: JourneyScan[];
/** True scan count for this journey, maintained even when `scans` is capped. */
scanCount: number;
pathKm: number;
minEpoch: number;
maxEpoch: number;
offsets: string[];
timezones: string[];
jurisdictions: string[];
statusProgression: string[];
delivered: boolean;
/** True once order-lane facts have been captured from an etaRun scan. */
etaCaptured: boolean;
onTime: boolean;
delayHours: number;
subtotalUsdMinor: number;
shippingUsdMinor: number;
}
// ── Module-level type narrowing helpers ───────────────────────────────────────
// Named static classes per noun.verb() convention — no freestanding functions.
class RegionInsightsMap {
static is(v: unknown): v is Map<string, RegionInsights> {
return v instanceof Map;
}
}
class JourneyAccumulatorMap {
static is(v: unknown): v is Map<string, JourneyAccumulator> {
return v instanceof Map;
}
}
// ── InsightsFoldGather ────────────────────────────────────────────────────────
type SizeTierKey = 'envelope' | 'small' | 'medium' | 'large' | 'freight';
export class InsightsFoldGather extends GatherStrategy {
private static readonly sizeTierDispatch: Readonly<Record<SizeTierKey, (entry: import('../CartographerState.ts').RegionInsights) => void>> = {
'envelope': (entry) => { entry.sizeTierEnvelope++; },
'small': (entry) => { entry.sizeTierSmall++; },
'medium': (entry) => { entry.sizeTierMedium++; },
'large': (entry) => { entry.sizeTierLarge++; },
'freight': (entry) => { entry.sizeTierFreight++; },
};
private static isSizeTierKey(value: string): value is SizeTierKey {
return value === 'envelope' || value === 'small' || value === 'medium' || value === 'large' || value === 'freight';
}
readonly name = 'insights-fold';
readonly '@id' = 'urn:noocodec:node:insights-fold';
// ── initial: reset accumulators in state ─────────────────────────────────
override initial(
_config: GatherConfigType,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
accessor.set(state, 'insights', new Map<string, RegionInsights>());
accessor.set(state, 'journeyAccumulators', new Map<string, JourneyAccumulator>());
accessor.set(state, 'sampleRecords', []);
accessor.set(state, 'errorRollup', ErrorRollup.empty());
accessor.set(state, 'journeys', new Map());
}
// ── reduce: per-clone fold (batch.size === 1) ─────────────────────────────
override reduce(
_config: GatherConfigType,
batch: Parameters<GatherStrategy['reduce']>[1],
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
for (const item of batch) {
const record: GatherRecordType = item.state;
// Fold this clone's captured errors into the parent rollup FIRST — errors
// flow scatter→gather as data, independent of whether the clone produced a
// usable enriched record. A clone whose coords were out of range still
// carries its captured RangeError even if enrichment degraded.
this.foldErrors(record.cloneState, state, accessor);
const rawEnriched = accessor.get(record.cloneState, 'enriched');
if (!EnrichedShipmentGuard.is(rawEnriched) || !rawEnriched.shipmentId) continue;
const enriched: EnrichedShipment = rawEnriched;
this.foldRegion(enriched, state, accessor);
this.foldJourney(enriched, state, accessor);
this.pushSampleRing(enriched, state, accessor);
}
}
// ── finalize: build state.journeys from bounded accumulators ─────────────
override async finalize(
_config: GatherConfigType,
execution: GatherExecutionType<NodeStateInterface>,
): Promise<void> {
const built = new Map<string, JourneyInsights>();
const rawJourneyAccumulators = execution.accessor.get(execution.state, 'journeyAccumulators');
if (!JourneyAccumulatorMap.is(rawJourneyAccumulators)) return;
const journeyMap = rawJourneyAccumulators;
for (const [shipmentId, acc] of journeyMap) {
// Sort the retained (capped) scans by scanSeq (authoritative order);
// epochMs is a tiebreak. The scalar metrics below come from the exact
// accumulators, not from this bounded scan list.
const scans = [...acc.scans].sort(
(a, b) => a.scanSeq - b.scanSeq || a.epochMs - b.epochMs,
);
if (scans.length === 0) continue;
const last = scans[scans.length - 1];
if (last === undefined) continue;
// Terminal-aware status progression. The progression is capped at
// MAX_SCANS_PER_JOURNEY during folding, but `delivered` is tracked
// uncapped: a hot shipmentId can fold its terminal DELIVERED scan after
// the progression array is already full, dropping DELIVERED from the
// capped prefix while the flag stays true. Delivery is terminal, so
// restore the DELIVERED marker at the end when the flag is set but the
// capped progression lost it — bounded (+1 entry), invariant preserved.
const statusProgression = [...acc.statusProgression];
if (acc.delivered && !statusProgression.includes('DELIVERED')) {
statusProgression.push('DELIVERED');
}
// pathKm, epoch bounds, offset/timezone/jurisdiction sets, scanCount, and
// delivered are maintained exactly during folding (independent of the
// capped scan list), so totals stay faithful even when the journey folded
// more scans than MAX_SCANS_PER_JOURNEY.
const journey: JourneyInsights = {
'shipmentId': shipmentId,
'scans': scans,
'scanCount': acc.scanCount,
'pathKm': acc.pathKm,
'firstEpochMs': acc.minEpoch,
'lastEpochMs': acc.maxEpoch,
'elapsedHours': (acc.maxEpoch - acc.minEpoch) / 3_600_000,
'timezones': [...acc.timezones],
'offsets': [...acc.offsets],
'jurisdictions': [...acc.jurisdictions],
'statusProgression': statusProgression,
'lastStatus': last.status,
'lastHub': last.hub,
'delivered': acc.delivered,
'onTime': acc.onTime,
'delayHours': acc.delayHours,
'subtotalUsdMinor': acc.subtotalUsdMinor,
'shippingUsdMinor': acc.shippingUsdMinor,
};
built.set(shipmentId, journey);
}
execution.accessor.set(execution.state, 'journeys', built);
// Region insights are exact in state.insights already; finalize does not touch them.
}
// ── Private helpers ───────────────────────────────────────────────────────
/** Fold one clone's captured errors into the bounded parent rollup. */
private foldErrors(
cloneState: NodeStateInterface,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const rawErrors = accessor.get(cloneState, 'capturedErrors');
if (!GeoErrorRecord.isArray(rawErrors) || rawErrors.length === 0) return;
const rawRollup = accessor.get(state, 'errorRollup');
const rollup = ErrorRollup.is(rawRollup) ? rawRollup : ErrorRollup.empty();
for (const error of rawErrors) {
ErrorRollup.fold(rollup, error);
}
// Write the rollup back to parent state after folding. The rollup is bounded
// to O(distinct source+variant groups) regardless of event count.
accessor.set(state, 'errorRollup', rollup);
}
/** Fold one enriched scan into the per-region accumulator and update parent state. */
private foldRegion(
enriched: EnrichedShipment,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const key = enriched.geoStatus === 'water'
? 'International Waters / Maritime'
: enriched.continent;
const rawInsights = accessor.get(state, 'insights');
const regionMap: Map<string, RegionInsights> = RegionInsightsMap.is(rawInsights)
? rawInsights
: new Map<string, RegionInsights>();
let entry = regionMap.get(key);
if (entry === undefined) {
entry = {
'region': key,
'country': key,
'hub': key,
'deliveries': 0,
'exceptions': 0,
'onTimeCount': 0,
'lateCount': 0,
'totalSubtotalUsdMinor': 0,
'totalShippingUsdMinor': 0,
'totalDistanceKm': 0,
'totalDelayHours': 0,
'consentValid': 0,
'consentMissing': 0,
'consentExpired': 0,
'sizeTierEnvelope': 0,
'sizeTierSmall': 0,
'sizeTierMedium': 0,
'sizeTierLarge': 0,
'sizeTierFreight': 0,
'shipmentCount': 0,
};
regionMap.set(key, entry);
}
entry.shipmentCount++;
entry.totalSubtotalUsdMinor += enriched.subtotalUsdMinor;
entry.totalShippingUsdMinor += enriched.shippingUsdMinor;
entry.totalDistanceKm += enriched.distanceKm;
if (enriched.routing.etaRun) {
if (enriched.onTime) entry.onTimeCount++;
else {
entry.lateCount++;
entry.totalDelayHours += enriched.delayHours;
}
}
if (enriched.status === 'DELIVERED') entry.deliveries++;
if (enriched.exception) entry.exceptions++;
if (enriched.consentStatus === 'valid') entry.consentValid++;
if (enriched.consentStatus === 'missing') entry.consentMissing++;
if (enriched.consentStatus === 'expired') entry.consentExpired++;
if (InsightsFoldGather.isSizeTierKey(enriched.sizeTier)) {
InsightsFoldGather.sizeTierDispatch[enriched.sizeTier](entry);
}
// Write the internal map back to parent state after every mutation.
// The map is bounded to ~6-8 continent keys regardless of event count.
accessor.set(state, 'insights', regionMap);
}
/** Fold one enriched scan into the bounded per-journey accumulator. */
private foldJourney(
enriched: EnrichedShipment,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const id = enriched.shipmentId;
const rawJourneyAccumulators = accessor.get(state, 'journeyAccumulators');
const journeyMap: Map<string, JourneyAccumulator> = JourneyAccumulatorMap.is(rawJourneyAccumulators)
? rawJourneyAccumulators
: new Map<string, JourneyAccumulator>();
const existing = journeyMap.get(id);
const scan: JourneyScan = {
'scanSeq': enriched.scanSeq,
'epochMs': enriched.epochMs,
'localIso': enriched.localIso,
'utcOffset': enriched.utcOffset,
'timezone': enriched.timezone,
'jurisdiction': enriched.jurisdiction,
'status': enriched.status,
'hub': enriched.hub,
'region': enriched.region,
'country': enriched.country,
'lat': enriched.lat,
'lng': enriched.lng,
'legKm': enriched.legKm,
'disruptionReason': enriched.disruptionReason,
};
if (existing !== undefined) {
// Fold this scan into the existing accumulator. The scan list and status
// progression are capped at MAX_SCANS_PER_JOURNEY so a frequently-folded
// shipmentId does not grow these arrays without bound; scanCount and the
// scalar metrics below stay exact regardless of the cap.
existing.scanCount++;
if (existing.scans.length < MAX_SCANS_PER_JOURNEY) existing.scans.push(scan);
if (existing.statusProgression.length < MAX_SCANS_PER_JOURNEY) existing.statusProgression.push(enriched.status);
existing.pathKm += enriched.legKm;
if (enriched.epochMs < existing.minEpoch) existing.minEpoch = enriched.epochMs;
if (enriched.epochMs > existing.maxEpoch) existing.maxEpoch = enriched.epochMs;
if (!existing.offsets.includes(enriched.utcOffset)) existing.offsets.push(enriched.utcOffset);
if (!existing.timezones.includes(enriched.timezone)) existing.timezones.push(enriched.timezone);
if (!existing.jurisdictions.includes(enriched.jurisdiction)) existing.jurisdictions.push(enriched.jurisdiction);
if (enriched.status === 'DELIVERED') existing.delivered = true;
// Capture order-lane facts on the first etaRun scan seen for this journey.
if (enriched.routing.etaRun && !existing.etaCaptured) {
existing.etaCaptured = true;
existing.onTime = enriched.onTime;
existing.delayHours = enriched.delayHours;
existing.subtotalUsdMinor = enriched.subtotalUsdMinor;
existing.shippingUsdMinor = enriched.shippingUsdMinor;
}
accessor.set(state, 'journeyAccumulators', journeyMap);
return;
}
// New shipmentId: only start tracking if under the sample cap.
if (journeyMap.size >= MAX_SAMPLE_JOURNEYS) return;
const acc: JourneyAccumulator = {
'scans': [scan],
'scanCount': 1,
'pathKm': enriched.legKm,
'minEpoch': enriched.epochMs,
'maxEpoch': enriched.epochMs,
'offsets': [enriched.utcOffset],
'timezones': [enriched.timezone],
'jurisdictions': [enriched.jurisdiction],
'statusProgression': [enriched.status],
'delivered': enriched.status === 'DELIVERED',
// Seed order-lane facts from this scan if it is an etaRun; otherwise use
// safe defaults (non-etaRun scans have no pricing/eta data).
'etaCaptured': enriched.routing.etaRun,
'onTime': enriched.routing.etaRun ? enriched.onTime : false,
'delayHours': enriched.routing.etaRun ? enriched.delayHours : 0,
'subtotalUsdMinor': enriched.routing.etaRun ? enriched.subtotalUsdMinor : 0,
'shippingUsdMinor': enriched.routing.etaRun ? enriched.shippingUsdMinor : 0,
};
journeyMap.set(id, acc);
accessor.set(state, 'journeyAccumulators', journeyMap);
}
/** Push one enriched record into the FIFO sample ring and update parent state. */
private pushSampleRing(
enriched: EnrichedShipment,
state: NodeStateInterface,
accessor: StateAccessorInterface,
): void {
const rawSample = accessor.get(state, 'sampleRecords');
const sampleRing = CircularBuffer.create<EnrichedShipment>({
'capacity': MAX_SAMPLE_RECORDS,
'overflow': 'overwrite',
});
if (Array.isArray(rawSample)) {
for (const s of rawSample) {
if (EnrichedShipmentGuard.is(s)) sampleRing.push(s);
}
}
sampleRing.push(enriched);
const sampleRecords: EnrichedShipment[] = [];
let record = sampleRing.shift();
while (record !== undefined) {
sampleRecords.push(record);
record = sampleRing.shift();
}
accessor.set(state, 'sampleRecords', sampleRecords);
}
}
// ── Module-load registration ──────────────────────────────────────────────────
GatherStrategies.register(new InsightsFoldGather());
// GatherStrategies.resolve('insights-fold') now works in any scatter placement.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
reducehook.InsightsFoldGatherfolds each clone into bounded parent aggregates immediately after that clone completes.reduce(config, batch, state, accessor)— called once per clone (or per micro-batch) as results arrive. Override this to fold incrementally. Parent state grows after each clone completes.finalize(config, execution)— called once after all clones complete. Override this (and leavereduceas a no-op) for all-at-once processing.
The built-in map, append, collect, and partition strategies fold in reduce — parent state grows after each clone. The built-in custom strategy accumulates nothing in reduce and does its work in finalize.
The Cartographer uses the incremental path because the runnable page can stream many source events without materialising every enriched record in parent memory.
- Bounded memory. Parent state holds rollups and samples, not the full event stream.
- UI-visible progress. The browser panels update from the same aggregate state the DAG produces.
- Strategy registry. The DAG JSON-LD uses the
insights-foldstrategy key on thefold-insightsgather placement; the runnable imports the strategy module so that implementation is registered before execution.
Related Concepts
- Example 14: Gather strategies - collect vs discard: two gather strategies side-by-side
- Example 16: Scatter resume - durable-inbox checkpoint and resume across a scatter abort
- Reference: Core, GatherStrategies