Example 16: Scatter Resume
What It Is
Scatter Resume is for long fan-out work that can stop halfway through. The Cartographer streams source events through a scatter, aborts in the middle of the run, checkpoints progress, and resumes without replaying items that already completed.
The runtime does this with a durable inbox. Pulled items are recorded before execution, acknowledged items are recorded after execution, and checkpoint/resume uses those two sets to decide what is safe to skip and what must be retried.
How It Works
The engine records scatter progress under SCATTER_PROGRESS_KEY in state metadata. Pulled items enter the inbox before body execution. Completed items move to ackedResults. A checkpoint preserves both structures. Resume drains inbox items first because their completion is uncertain, skips acked items because they already finished, and then continues pulling the remaining source.
Application code still defines an ordinary scatter. The resume behavior belongs to the dispatcher and checkpoint state, so the DAG stays readable: source, body, gather, and routes remain the pieces you author.
Diagrams, Examples, and Outputs
DAG registration and diagram
The JSON-LD graph is a normal scatter; the resume behavior comes from checkpoint metadata captured while that scatter is running. The Cartographer owns this runnable path through cartographerResumeDAG and CartographerResumableScenario.
Cartographer scatter resume 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 scatter engine uses a durable-inbox model to survive crashes:
- When an item is pulled from the source it enters the
inbox(persisted in state metadata underSCATTER_PROGRESS_KEY). - When the body completes successfully the item leaves the inbox and moves to
ackedResults. - On abort,
Checkpoint.captureincludes both inbox andackedResults. - On resume, inbox items are reprocessed first (they may not have finished), then remaining source items continue. Acked items are never re-executed.
The resumable scenario fires an abort after a fixed number of stream completions so the abort happens inside the running scatter. This is deterministic and credential-free.
Items 0–(ABORT_AFTER-1): run, ack, accumulate in ackedResults.
Item ABORT_AFTER: node body fires abort → scatter exits before next pull.
Items after abort: never pulled. Resume runs them fresh.Watch: execLog shows different labels for Run 1 vs Run 2 items. The union of both logs covers all items with no duplicates.
Run
bash
npx tsx examples/the-cartographer/runCartographer.ts --streamWhat It Lets You Do
Scatter resume lets applications restart long fan-out work without re-running completed items. Use it when clone bodies call external APIs, model providers, or expensive transforms and a crash, abort, or timeout should replay only uncertain work.
This is the scatter version of exactly-once-at-the-DAG-boundary thinking: node bodies still need to be idempotent, but the engine gives the application a durable record of which source items are already done.
Code Samples
The DAG snippet shows the scatter shape. The scenario snippet shows the deterministic abort, checkpoint capture, resume call, and execution log used by the runnable Cartographer demo.
ts
/**
* cartographerResumeDAG: streaming-resume variant of cartographerDAG.
*
* Identical topology but WITHOUT reservoir on process-stream.
* Per-item dispatch (ScatterWorkerPool path) allows the run-level abort signal
* to fire between item pulls, giving a non-zero StreamCursor.resumeAfter(state,
* 'process-stream') value on abort. Used by CartographerResumableScenario only.
*
* Topology (same as cartographerDAG):
* 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 cartographerResumeDAG: DAGType = new DAGBuilder(CARTOGRAPHER_RESUME_DAG_IRI, '1.0')
.gather(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_RESUME_DAG_IRI),
{ 'strategy': 'source-intake' },
{
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'process-stream'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'process-stream'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'done'),
},
)
.scatter(
CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'process-stream'),
'sources',
{ 'dag': STREAM_EVENT_DAG_IRI },
{
'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'fold-insights'),
'partial': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'fold-insights'),
'all-error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'fold-insights'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'summarize'),
},
{
'itemKey': 'source-payload',
'execution': { 'mode': 'item', 'concurrency': 16 },
},
)
.gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'fold-insights'), {
[CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'process-stream')]: {},
}, { 'strategy': 'insights-fold' }, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'summarize'),
'error': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'summarize'),
'empty': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'summarize'),
})
.node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'summarize'), summarizeInsights, {
'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'done'),
})
.terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'done'), { outcome: 'completed' })
.entrypoints({
'position-ping': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
'facility-scan': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
'sensor-reading': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
'customs-event': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
'delivery-confirmation': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_RESUME_DAG_IRI, 'intake-gather'),
})
.build();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 inbox. The scatter pull loop persists each pulled item to the
inboxbefore dispatching the body. The inbox survives process interruption viaCheckpoint.capture. ackedResultsdeduplication. After a body completes, the engine moves the item from inbox toackedResults. On resume, the engine skips any item whose key is already inackedResults— no re-execution.- Inbox reprocessing. Items that were in the inbox at checkpoint time (pulled but not yet acked) are re-run first on resume. The body is idempotent by design.
SCATTER_PROGRESS_KEY. The constant under which the scatter engine stores the inbox and acked-results index instate.metadata. Inspect it after a checkpoint capture to see the progress snapshot.- No reservoir in the resume DAG. The runnable resume variant uses per-item dispatch so abort can leave a meaningful stream cursor.
Related Concepts
- Example 08: Checkpoint and Resume - checkpoint lifecycle: capture, persist, recall, resume
- Example 23: Checkpoint store - MemoryCheckpointStore persist / recall round-trip
- Example 04: Scatter Scout - scatter mechanics: source, body DAG, gather placement, reduce
- Reference: Checkpoint