Example 35: Stream Resume Cursor
What It Is
Stream Resume Cursor is the streaming counterpart to scatter resume. The Cartographer aborts a streaming scatter, reads StreamCursor.resumeAfter(...), recreates the source stream from a durable cursor, and resumes without duplicate processing.
The source must be regenerable and ordered. Given that, Dagonizer can reconnect the resumed scatter to the right point in the stream.
How It Works
The scatter stores durable progress in checkpoint state. On resume, StreamCursor.resumeAfter(...) converts that progress into a cursor for the stream producer. The producer regenerates the same ordered source, skips the consumed prefix, and yields only remaining items into the resumed scatter.
This divides responsibility cleanly: the scatter records what it has safely pulled and acknowledged; the producer knows how to rebuild the ordered stream after that point.
Diagrams, Examples, and Outputs
DAG registration and diagram
The Cartographer resume DAG uses the same streaming topology as the main DAG, but the scatter runs in item mode so abort can land between pulls. StreamCursor.resumeAfter(state, 'process-stream') reads the durable pull count from checkpoint state and feeds that cursor back into CartographerSourceIntake.mergedFor(state, cursor).
Cartographer resumable stream 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-resume",
"@type": "DAG",
"name": "dag:cartographer-resume",
"version": "1.0",
"entrypoints": {
"position-ping": "urn:noocodec:dag:cartographer-resume/node/intake-gather",
"facility-scan": "urn:noocodec:dag:cartographer-resume/node/intake-gather",
"sensor-reading": "urn:noocodec:dag:cartographer-resume/node/intake-gather",
"customs-event": "urn:noocodec:dag:cartographer-resume/node/intake-gather",
"delivery-confirmation": "urn:noocodec:dag:cartographer-resume/node/intake-gather"
},
"nodes": [
{
"@id": "urn:noocodec:dag:cartographer-resume/node/intake-gather",
"@type": "GatherNode",
"name": "dag:cartographer-resume/node/intake-gather",
"sources": {
"urn:noocodec:dag:cartographer-resume/entrypoint/position-ping": {},
"urn:noocodec:dag:cartographer-resume/entrypoint/facility-scan": {},
"urn:noocodec:dag:cartographer-resume/entrypoint/sensor-reading": {},
"urn:noocodec:dag:cartographer-resume/entrypoint/customs-event": {},
"urn:noocodec:dag:cartographer-resume/entrypoint/delivery-confirmation": {}
},
"gather": {
"strategy": "source-intake"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer-resume/node/process-stream",
"error": "urn:noocodec:dag:cartographer-resume/node/process-stream",
"empty": "urn:noocodec:dag:cartographer-resume/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer-resume/node/process-stream",
"@type": "ScatterNode",
"name": "dag:cartographer-resume/node/process-stream",
"source": "sources",
"body": {
"dag": "urn:noocodec:dag:stream-event"
},
"outputs": {
"all-success": "urn:noocodec:dag:cartographer-resume/node/fold-insights",
"partial": "urn:noocodec:dag:cartographer-resume/node/fold-insights",
"all-error": "urn:noocodec:dag:cartographer-resume/node/fold-insights",
"empty": "urn:noocodec:dag:cartographer-resume/node/summarize"
},
"itemKey": "source-payload",
"reducer": "aggregate",
"execution": {
"mode": "item",
"concurrency": 16
}
},
{
"@id": "urn:noocodec:dag:cartographer-resume/node/fold-insights",
"@type": "GatherNode",
"name": "dag:cartographer-resume/node/fold-insights",
"sources": {
"urn:noocodec:dag:cartographer-resume/node/process-stream": {}
},
"gather": {
"strategy": "insights-fold"
},
"outputs": {
"success": "urn:noocodec:dag:cartographer-resume/node/summarize",
"error": "urn:noocodec:dag:cartographer-resume/node/summarize",
"empty": "urn:noocodec:dag:cartographer-resume/node/summarize"
}
},
{
"@id": "urn:noocodec:dag:cartographer-resume/node/summarize",
"@type": "SingleNode",
"name": "dag:cartographer-resume/node/summarize",
"node": "urn:noocodec:node:summarize",
"outputs": {
"success": "urn:noocodec:dag:cartographer-resume/node/done"
}
},
{
"@id": "urn:noocodec:dag:cartographer-resume/node/done",
"@type": "TerminalNode",
"name": "dag:cartographer-resume/node/done",
"outcome": "completed"
}
]
}Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
%% dag:cartographer-resume (v1.0)
entry_position-ping(["position-ping"])
entry_position-ping --> urn_noocodec_dag_cartographer-resume/node/intake-gather
entry_facility-scan(["facility-scan"])
entry_facility-scan --> urn_noocodec_dag_cartographer-resume/node/intake-gather
entry_sensor-reading(["sensor-reading"])
entry_sensor-reading --> urn_noocodec_dag_cartographer-resume/node/intake-gather
entry_customs-event(["customs-event"])
entry_customs-event --> urn_noocodec_dag_cartographer-resume/node/intake-gather
entry_delivery-confirmation(["delivery-confirmation"])
entry_delivery-confirmation --> urn_noocodec_dag_cartographer-resume/node/intake-gather
urn_noocodec_dag_cartographer-resume/node/intake-gather{"dag_cartographer-resume/node/intake-gather"}
urn_noocodec_dag_cartographer-resume/node/intake-gather -->|success| urn_noocodec_dag_cartographer-resume/node/process-stream
urn_noocodec_dag_cartographer-resume/node/intake-gather -->|error| urn_noocodec_dag_cartographer-resume/node/process-stream
urn_noocodec_dag_cartographer-resume/node/intake-gather -->|empty| urn_noocodec_dag_cartographer-resume/node/done
urn_noocodec_dag_cartographer-resume/node/process-stream[/"dag:cartographer-resume/node/process-stream"/]
urn_noocodec_dag_cartographer-resume/node/process-stream -->|all-success| urn_noocodec_dag_cartographer-resume/node/fold-insights
urn_noocodec_dag_cartographer-resume/node/process-stream -->|partial| urn_noocodec_dag_cartographer-resume/node/fold-insights
urn_noocodec_dag_cartographer-resume/node/process-stream -->|all-error| urn_noocodec_dag_cartographer-resume/node/fold-insights
urn_noocodec_dag_cartographer-resume/node/process-stream -->|empty| urn_noocodec_dag_cartographer-resume/node/summarize
urn_noocodec_dag_cartographer-resume/node/fold-insights{"dag_cartographer-resume/node/fold-insights"}
urn_noocodec_dag_cartographer-resume/node/fold-insights -->|success| urn_noocodec_dag_cartographer-resume/node/summarize
urn_noocodec_dag_cartographer-resume/node/fold-insights -->|error| urn_noocodec_dag_cartographer-resume/node/summarize
urn_noocodec_dag_cartographer-resume/node/fold-insights -->|empty| urn_noocodec_dag_cartographer-resume/node/summarize
urn_noocodec_dag_cartographer-resume/node/summarize["dag:cartographer-resume/node/summarize"]
urn_noocodec_dag_cartographer-resume/node/summarize -->|success| urn_noocodec_dag_cartographer-resume/node/done
urn_noocodec_dag_cartographer-resume/node/done((("dag:cartographer-resume/node/done")))The graph shows the resumable scatter boundary. The code proves exactly-once behavior by comparing a full baseline fold to an interrupted-and-resumed fold.
Run
bash
npx tsx examples/the-cartographer/runCartographer.ts --streamWhat It Lets You Do
Stream resume cursors let applications restart a streaming scatter from the durable pull position instead of replaying already-folded items. Use this when a source can regenerate an ordered stream and the DAG must avoid duplicate processing after abort or crash.
This fits file cursors, event feeds, generated datasets, and model-produced work streams where the source can be reconstructed deterministically from a cursor.
Code Samples
The intake helper rebuilds the same ordered stream and skips the acknowledged prefix:
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;
}
}
}The intake gather is the DAG-visible convergence point before the resumable scatter:
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());The CLI scenario aborts, resumes, and compares fingerprints:
ts
/** Fixed event count for the interrupted+resume run pair. */
const RESUME_EVENT_COUNT = 40;
/**
* Number of scatter item completions (aggregate-event inside process-stream)
* after which the interrupted run aborts. cartographerResumeDAG has no reservoir,
* so items are dispatched one-at-a-time (ScatterWorkerPool path) and the abort
* signal fires between pulls — giving a non-zero StreamCursor value.
*/
const ABORT_AFTER_ITEMS = 8;
/**
* InsightsFingerprint: deterministic canonical digest of a regional insights Map.
*
* Sorts entries by region → country → hub and emits all numeric fields of each
* RegionInsights plus the string keys. JSON.stringify over the sorted plain array
* gives a stable string suitable for equality comparison.
*/
class InsightsFingerprint {
private constructor() { /* static-only */ }
static of(insights: Map<string, RegionInsights>): string {
const rows = [...insights.values()].sort((a, b) => {
const byRegion = a.region.localeCompare(b.region);
if (byRegion !== 0) return byRegion;
const byCountry = a.country.localeCompare(b.country);
if (byCountry !== 0) return byCountry;
return a.hub.localeCompare(b.hub);
});
const normalized = rows.map((r) => ({
'region': r.region,
'country': r.country,
'hub': r.hub,
'deliveries': r.deliveries,
'exceptions': r.exceptions,
'onTimeCount': r.onTimeCount,
'lateCount': r.lateCount,
'totalSubtotalUsdMinor': r.totalSubtotalUsdMinor,
'totalShippingUsdMinor': r.totalShippingUsdMinor,
'totalDistanceKm': r.totalDistanceKm,
'totalDelayHours': r.totalDelayHours,
'consentValid': r.consentValid,
'consentMissing': r.consentMissing,
'consentExpired': r.consentExpired,
'sizeTierEnvelope': r.sizeTierEnvelope,
'sizeTierSmall': r.sizeTierSmall,
'sizeTierMedium': r.sizeTierMedium,
'sizeTierLarge': r.sizeTierLarge,
'sizeTierFreight': r.sizeTierFreight,
'shipmentCount': r.shipmentCount,
}));
return JSON.stringify(normalized);
}
}
/**
* CartographerResumableScenario: self-contained abort→cursor→resume verification.
*
* Uses `cartographerResumeDAG` (no reservoir) so abort fires mid-scatter, leaving
* acked items in the checkpoint and un-pulled items un-acked.
*
* Baseline — Full streaming pass over all RESUME_EVENT_COUNT items (no abort).
* Produces the reference InsightsFingerprint.
* Step A — Interrupted run: abort after ABORT_AFTER_ITEMS aggregate-event
* completions; read durable cursor from checkpoint.
* Step B — Resume: restore from firstState.snapshot() (carries accumulator +
* checkpoint) and supply the remainder through CartographerSourceIntake.
* Assert cursor > 0 and resumeResult.cursor === null (completed).
* Proof — Compare InsightsFingerprint of resumed state to baseline fingerprint.
* Equal → exactly-once; unequal → throw with full diff.
*/
class CartographerResumableScenario {
private constructor() { /* static-only */ }
/** Register cartographerResumeBundle bundles onto a fresh ObservedCartographer. */
static #buildResumeDispatcher(services: CartographerServices): ObservedCartographer {
const d = new ObservedCartographer({});
d.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
d.registerBundle(orderEnrichmentBundle);
d.registerBundle(gdprComplianceBundle);
d.registerPlugin(normalizeSourcesPlugin);
d.registerBundle(ingestSourceBundle);
d.registerBundle(cartographerResumeBundle);
return d;
}
static async run(
_dispatcher: ObservedCartographer,
services: CartographerServices,
logger: ConsoleLogger,
_eventCount: number,
): Promise<void> {
logger.info('CartographerResumableScenario', 'run', `Starting streamed-resume verification (${RESUME_EVENT_COUNT} events, abort after ${ABORT_AFTER_ITEMS})`);
// ── Baseline: full streaming pass (no abort) ─────────────────────────────
// Runs the same producer + same event count through the same DAG without
// interruption. Produces the reference accumulator for the exactly-once proof.
const baselineDispatcher = CartographerResumableScenario.#buildResumeDispatcher(services);
const baselineState = new CartographerState();
baselineState.useStreamingSource = true;
baselineState.eventCount = RESUME_EVENT_COUNT;
baselineState.streamCount = RESUME_EVENT_COUNT;
await baselineDispatcher.execute('urn:noocodec:dag:cartographer-resume', baselineState);
const baselineFingerprint = InsightsFingerprint.of(baselineState.insights);
logger.info('CartographerResumableScenario', 'baseline', `Baseline streamed run folded ${baselineState.insights.size} region(s).`);
// ── Step A: Interrupted run ──────────────────────────────────────────────
// AbortingCartographer fires abort after ABORT_AFTER_ITEMS aggregate-event
// completions inside process-stream. cartographerResumeDAG has no reservoir,
// so the ScatterWorkerPool checks abort between item pulls — giving cursor > 0.
const interruptAc = new AbortController();
const abortingDispatcher = new AbortingCartographer({}, interruptAc, ABORT_AFTER_ITEMS);
abortingDispatcher.registerBundle(GeoSourceResolveDAG.build(services.ipGeolocator, services.addressGeocoder));
abortingDispatcher.registerBundle(orderEnrichmentBundle);
abortingDispatcher.registerBundle(gdprComplianceBundle);
abortingDispatcher.registerPlugin(normalizeSourcesPlugin);
abortingDispatcher.registerBundle(ingestSourceBundle);
abortingDispatcher.registerBundle(cartographerResumeBundle);
const firstState = new CartographerState();
firstState.useStreamingSource = true;
firstState.eventCount = RESUME_EVENT_COUNT;
firstState.streamCount = RESUME_EVENT_COUNT;
let interruptedCursor: string | null = null;
try {
const interruptedResult = await abortingDispatcher.execute(
'cartographer-resume', firstState, { 'signal': interruptAc.signal },
);
interruptedCursor = interruptedResult.cursor;
} catch (err) {
if (!(err instanceof DAGError && err.code === 'EXECUTION_ERROR')) throw err;
}
// Read the durable stream cursor from the interrupted checkpoint.
const cursor = StreamCursor.resumeAfter(firstState, 'process-stream');
logger.info(
'CartographerResumableScenario', 'interrupted',
`Interrupted after ${ABORT_AFTER_ITEMS} items. execution cursor='${String(interruptedCursor)}' stream cursor=${cursor}`,
);
process.stdout.write(`ASSERT cursor > 0: ${cursor > 0 ? 'PASS' : 'FAIL'} (cursor=${cursor})\n`);
if (cursor === 0) {
throw new Error('CartographerResumableScenario: cursor is 0 — checkpoint not preserved after abort');
}
// ── Step B: Resume ───────────────────────────────────────────────────────
// Restore from the interrupted snapshot — this is the faithful cross-process
// restart path: the partial insights accumulator AND the SCATTER_PROGRESS_KEY
// checkpoint are both carried by CartographerState.restore(firstState.snapshot()).
// Acked items (below the watermark) already contributed to state.insights and
// are NOT replayed by the engine; the accumulator carry ensures their folds
// survive. Un-acked items in the durable inbox are replayed by the engine.
const resumeDispatcher = CartographerResumableScenario.#buildResumeDispatcher(services);
const resumeState = CartographerState.restore(firstState.snapshot());
resumeState.useStreamingSource = true;
resumeState.eventCount = RESUME_EVENT_COUNT;
resumeState.streamCount = RESUME_EVENT_COUNT;
// Supply the remainder: the top-level intake gather is already complete in
// the restored checkpoint, so resume enters process-stream directly.
resumeState.sources = CartographerSourceIntake.mergedFor(resumeState, cursor);
const resumeResult = await resumeDispatcher.resume('urn:noocodec:dag:cartographer-resume', resumeState, 'process-stream');
logger.info(
'CartographerResumableScenario', 'resume',
`Resume complete. cursor=${String(resumeResult.cursor)} (expected null)`,
);
process.stdout.write(`ASSERT resume completed: ${resumeResult.cursor === null ? 'PASS' : 'FAIL'} (cursor=${String(resumeResult.cursor)})\n`);
if (resumeResult.cursor !== null) {
throw new Error(`CartographerResumableScenario: resume did not complete (cursor='${String(resumeResult.cursor)}')`);
}
// ── Exactly-once proof: compare resumed fingerprint to baseline ───────────
// The fingerprint encodes ALL numeric fields for every region, sorted
// deterministically. Equal → every acked fold was carried (not lost, not
// double-counted); unequal → the accumulator carry is broken.
const resumeFingerprint = InsightsFingerprint.of(resumeState.insights);
const exactlyOnce = resumeFingerprint === baselineFingerprint;
process.stdout.write(`ASSERT exactly-once (resume insights == baseline insights): ${exactlyOnce ? 'PASS' : 'FAIL'}\n`);
if (!exactlyOnce) {
throw new Error(
`CartographerResumableScenario: exactly-once violated — resumed insights differ from baseline.\n` +
` baseline: ${baselineFingerprint}\n` +
` resumed: ${resumeFingerprint}`,
);
}
// ── Shipment-count cross-check ────────────────────────────────────────────
// Grand total of shipmentCount across all regions must be identical between
// the resumed run and the baseline (gross undercount / double-count guard).
let baselineTotal = 0;
for (const r of baselineState.insights.values()) baselineTotal += r.shipmentCount;
let resumeTotal = 0;
for (const r of resumeState.insights.values()) resumeTotal += r.shipmentCount;
const countMatch = resumeTotal === baselineTotal;
process.stdout.write(`ASSERT shipment-count (resume=${resumeTotal} == baseline=${baselineTotal}): ${countMatch ? 'PASS' : 'FAIL'}\n`);
if (!countMatch) {
throw new Error(
`CartographerResumableScenario: shipment-count mismatch — resumed total (${resumeTotal}) != baseline (${baselineTotal})`,
);
}
process.stdout.write('Streamed resume: COMPLETE. Exactly-once verified.\n');
}
}Details for Nerds
- Durable pull cursor. The cursor is based on scatter items durably pulled and acknowledged, not on producer push count.
- Deterministic replay. The producer regenerates the ordered stream and skips the consumed prefix.
- No duplicate fold. The resumed run’s regional-insight fingerprint matches the uninterrupted baseline.
- Same browser graph shape. Resume changes execution mode and cursor state, not the embedded DAG assembly model.
Related Concepts
- Example 34: Intake stream source - Cartographer source-intake gather stream assembly
- Example 16: Scatter resume - the same Cartographer resume DAG from the scatter perspective
- Streaming Producers - full streaming API: driven, fanIn, resumable, DagStreamProducer