Checkpoint Persistence

What It Is

Checkpoint persistence is the storage layer for checkpoint JSON. Checkpoint owns the codec; CheckpointStore owns save, load, and delete. Together, ckpt.persist(store, key) and Checkpoint.recall(store, key) make save/resume a small adapter problem instead of a custom serialization project.

Use it when a checkpoint must survive process memory: browser parking, serverless resume, queue hand-off, crash recovery, or any execution that may continue in a different host.

How It Works

Checkpoint owns encoding and validation. CheckpointStore owns storage. ckpt.persist(store, key) serializes the checkpoint and calls the store; Checkpoint.recall(store, key) reads, parses, validates, and returns a checkpoint ready for state/store restore.

CheckpointStore is the three-method adapter contract for persistence backends. Checkpoint handles the codec (turning an ExecutionResult into a CheckpointData record and back); persistence is the application's responsibility behind this contract.

Persist + recall lifecycle

The diagram traces method invocations across the save and resume halves. It is not a Dagonizer DAG; it is a sequence over the codec API.

Diagrams, Examples, and Outputs

Persistence is not DAG topology, so this page uses a sequence diagram for the checkpoint API lifecycle and source snippets for the runnable store examples.

• Checkpoint - the codec layer (Checkpoint.capture and Checkpoint.load)
• Subclassing State - domain fields survive the round-trip via snapshotData and restoreData
• Cancellation - produce a checkpointable result by aborting an in-flight flow
• Example 23: Checkpoint Store runs the store-backed recall path.

What It Lets You Do

Use when

Use checkpoint persistence when a checkpoint must survive process memory. This is the backing-store layer for browser parking, serverless resumes, queue hand-offs, crash recovery, and any flow where Checkpoint.capture is not enough by itself.

Code Samples

API surface

Symbol	Source	Role
CheckpointStore	@studnicky/dagonizer/contracts	Adapter contract: save, load, delete
Snapshottable	@studnicky/dagonizer/contracts	Capability contract: snapshot(), restore(). Required by Checkpoint.capture and restoreStores.
StoreSnapshotType	@studnicky/dagonizer/contracts	Serialized envelope written into CheckpointData.stores
MemoryCheckpointStore	@studnicky/dagonizer/checkpoint	In-memory reference implementation (tests, demos)
ckpt.persist(store, key)	instance method	Serializes and writes via the store
Checkpoint.recall(store, key)	@studnicky/dagonizer/checkpoint	Reads, parses, validates, wraps

The contract

declare const store: CheckpointStoreInterface;
await store.save('run-42', '{"cursor":null}');
const json: string | null = await store.load('run-42');
await store.delete('run-42');
export {};

load returns null when no entry exists. Implementations handle their own concurrency, retries, and serialization details.

Schema validation on recall

Checkpoint.recall runs the JSON through Validator.checkpoint.validate(parsed) before wrapping it. Tampered or version-mismatched payloads throw ValidationError. The same goes for Checkpoint.load (which recall composes with).

Details for Nerds

Persist with ckpt.persist

// Checkpoint.capture() returns a Checkpoint instance.
// cursor !== null here because we aborted mid-run.
const checkpoint = await Checkpoint.capture('urn:noocodec:dag:count', partial);
const persisted  = checkpoint.toJson();  // → JSON string (store in DB, file, etc.)

ckpt.persist(store, key) calls store.save(key, ckpt.toJson()). One call covers serialization plus storage.

Recall with Checkpoint.recall

// Parse the persisted JSON back to an unknown value, then load into a Checkpoint.
const ckpt = Checkpoint.load(JSON.parse(persisted));

// restoreState maps the snapshot back to a typed CountingState instance.
// Consumers supply their own restore fn so the checkpoint module never
// imports domain state classes.
const { state, dagName, cursor } = ckpt.restoreState(
  CheckpointRestoreAdapter.wrap((snap) => CountingState.restore(snap)),  // rehydrates domain fields via restoreData()
);

process.stdout.write(`  restored: count=${state.count} cursor="${cursor}"\n`);

Checkpoint.recall returns null when the key is absent, or a Checkpoint instance whose restoreState yields the rehydrated state, the DAG IRI/CURIE string, the placement-IRI resume cursor, and the executed/skipped node histories.

Implementing a custom store

Implement the three methods against the backend. The store below backs the contract with a real Map — a complete, runnable implementation:

export class MapCheckpointStore implements CheckpointStoreInterface {
  readonly #entries = new Map<string, string>();

  async save(key: string, json: string): Promise<void> {
    this.#entries.set(key, json);
  }

  async load(key: string): Promise<string | null> {
    return this.#entries.get(key) ?? null;
  }

  async delete(key: string): Promise<void> {
    this.#entries.delete(key);
  }

  /** Test/inspection helper — not part of the contract. */
  get size(): number {
    return this.#entries.size;
  }
}

The same three-method pattern applies for Postgres (INSERT … ON CONFLICT/SELECT/DELETE), Redis (GET/SET/DEL), S3 (GetObject/PutObject/DeleteObject), a file system, or any other key/value store. Only the backing changes; the three methods stay identical. The contract is intentionally thin so it maps cleanly to any backing.

The custom-checkpoint-store example exercises the full save → load → delete round-trip end to end; run it with npx tsx examples/custom-checkpoint-store.ts.

Named stores and Snapshottable

Checkpoint.capture and ckpt.restoreStores both depend on the Snapshottable capability, not the full key-value Store surface. Any object that implements snapshot(): Promise<StoreSnapshotType> and restore(snapshot: StoreSnapshotType): Promise<void> participates in checkpointing. Store extends Snapshottable, so every store qualifies, but a non-KV backing (an RDF triple store, a vector index, an append-only log) can ride along in a checkpoint without implementing get/set/has/delete/update.

export class FactLog implements SnapshottableInterface {
  readonly #facts: string[] = [];

  add(fact: string): void {
    this.#facts.push(fact);
  }

  get facts(): readonly string[] {
    return this.#facts;
  }

  async snapshot(): Promise<StoreSnapshotType> {
    return {
      version: 1,
      type: 'fact-log',
      entries: this.#facts.map((fact, i) => ({ key: String(i), value: fact })),
    };
  }

  async restore(snapshot: StoreSnapshotType): Promise<void> {
    if (snapshot.type !== 'fact-log') {
      throw new Error(`Incompatible snapshot type: ${snapshot.type}`);
    }
    this.#facts.length = 0;
    for (const entry of snapshot.entries) {
      this.#facts.push(String(entry.value));
    }
  }

  async *snapshotStream(): AsyncIterable<StoreSnapshotEntryType> {
    for (let i = 0; i < this.#facts.length; i++) {
      const fact = this.#facts[i];
      if (fact !== undefined) yield { key: String(i), value: fact };
    }
  }

  async restoreStream(entries: AsyncIterable<StoreSnapshotEntryType>): Promise<void> {
    this.#facts.length = 0;
    for await (const entry of entries) {
      this.#facts.push(String(entry.value));
    }
  }
}

FactLog implements only snapshot() and restore() — no get/set/has/delete/update. Pass it to Checkpoint.capture('urn:noocodec:dag:my-dag', result, { stores: { log } }) alongside any MemoryStore, and restore it on resume with recalled.restoreStores({ log: freshLog }). The custom-checkpoint-store example runs the snapshot → restore round-trip; run it with npx tsx examples/custom-checkpoint-store.ts.

CheckpointData.stores is a required field. Checkpoint.capture always writes it: as an empty object {} when no stores are passed, or as a keyed map of StoreSnapshotType envelopes when stores are supplied. Any checkpoint payload lacking a stores field is rejected by Checkpoint.load.

Snapshot round-trip

Checkpoint.capture calls state.snapshot() and packages the result with the cursor and execution history. State subclasses that carry domain-specific fields override snapshotData() and restoreData():

export class PipelineState extends NodeStateBase {
  stage  = '';
  tally  = 0;
  trail: string[] = [];

  protected override snapshotData(): JsonObjectType {
    return { stage: this.stage, tally: this.tally, trail: [...this.trail] };
  }

  protected override restoreData(snapshot: JsonObjectType): void {
    const s = snapshot['stage'];
    if (typeof s === 'string') this.stage = s;
    const n = snapshot['tally'];
    if (typeof n === 'number') this.tally = n;
    const t = snapshot['trail'];
    if (Array.isArray(t)) this.trail = t.filter((x): x is string => typeof x === 'string');
  }
}

Lifecycle resets to pending on restore. Resume starts a fresh lifecycle run on the recovered state data.

Testing with MemoryCheckpointStore

const store1   = new MemoryCheckpointStore();
const ckpt     = await Checkpoint.capture('urn:noocodec:dag:pipeline', partial);
await ckpt.persist(store1, CHECKPOINT_KEY);

MemoryCheckpointStore exposes a read-only size getter for assertions about how many entries the store holds.

Scatter resume artefacts

Scatter placements with a source persist per-item progress under a reserved metadata key (SCATTER_PROGRESS_KEY === '__dagonizer_scatter_progress__'). When a checkpoint captures a state mid-scatter, this key carries the indices of already-completed clones so the resumed run can skip them instead of re-issuing every external call from scratch.

Three persistence-side implications:

1. The key counts toward checkpoint payload size. A 200-item scatter interrupted at clone 150 stores 150 numeric indices plus their output tags. Plan capacity in the CheckpointStore with this in mind; the payload still serialises as a single JSON document.
2. Per-batch write cadence. The dispatcher writes the progress entry once per scatter batch (not once per item). The persisted metadata is consistent with the batch boundary that was last await-ed; a crash during a batch leaves the last completed batch persisted and the in-flight batch unreported.
3. Indices are array positions in the source at resume time. If the CheckpointStore is read across processes that may rebuild state with a different source array, the resumed scatter skips by position, not by item identity. Treat the source as immutable while a scatter checkpoint is live, or clear the progress entry before calling dispatcher.resume() when the source has changed.

The reserved key piggybacks on NodeStateBase.metadata, so any CheckpointStore that already round-trips the JsonObjectType snapshot supports scatter resume with no additional adapter changes. See Checkpoint and Resume for the executable contract and index-semantics worked example.

Related Concepts

• Checkpoint - the codec layer (Checkpoint.capture and Checkpoint.load)
• Subclassing State - domain fields survive the round-trip via snapshotData and restoreData
• Cancellation - produce a checkpointable result by aborting an in-flight flow
• Example 23: Checkpoint Store
• Reference: Contracts
• Reference: Checkpoint
• Example 08: Checkpoint and Resume