Tutorial: Build a Crash-Proof AI Digest Job with Durable Steps

Build a weekly AI digest job that survives crashes: a step journal, a lock that detects dead workers, crash injection, resume, and an email that is never sent twice.

Advanced23 min readUpdated October 19, 2026

On this page· What you will build

Why this matters

Memory, state, and durable execution explains checkpoints, leases, and idempotency keys on top of Postgres. The ideas are easier to trust once you have watched them work. In this tutorial you build a job that writes a weekly product digest with one model call per update, then emails it. You will kill it in the middle, on purpose, and watch it resume without paying for finished calls again. Then you will kill it at the worst possible moment, after the email went out but before the job recorded it, and confirm that nobody gets the email twice. Everything runs on files, so you need nothing but Node.js.

What you will build#

$ CRASH=after:summarize:i3 npm run digest -- --run 2026-w42
  ran      select
  ran      summarize:i1
  ran      summarize:i2
  ran      summarize:i3
  ** simulated crash at after:summarize:i3 **
 
$ npm run digest -- --run 2026-w42
  replayed select
  replayed summarize:i1
  ...
  ran      summarize:i4
  ...
sent: 8 items, 6 model calls in this attempt, email sent (digest-2026-w42)

A step journal: an append-only file with one line per completed step, flushed to disk before the run moves on.
A lock so only one worker runs a given digest, which detects and takes over a lock left by a crashed worker.
An outbox that behaves like an email provider with idempotency keys.
Crash injection at any named point, and a resume that replays saved steps.
State across runs: the job remembers which updates earlier digests covered.

Hover or tap a component to see what it does.

Durable digest job: saved steps replay from the journal, the rest run and are saved, and the send's idempotency key keeps the email to one.

Show text description

Checkpoints make work resumable; idempotency keys make the side effects safe to repeat.

Worker: Takes over a lock only when its owner process is gone or its lease expired.
Step journal: One line per completed step. Saved steps are replayed, not re-run; a torn last line is ignored.
Model steps: One model call per step, so a crash loses at most the call in progress.
send: The side effect. Its idempotency key is derived from the run, never random.
seen.json: Which updates earlier digests covered; updated as a set union, safe to repeat.
Outbox: Like a provider with idempotency keys: a repeated send returns the first receipt.

You need Node.js 22.18 or later and an Anthropic API key.

1. Set up the project#

mkdir durable-digest && cd durable-digest
npm init -y
npm pkg set type=module
npm install @anthropic-ai/sdk
npm install -D typescript @types/node
mkdir src test data
npm pkg set scripts.digest="node --env-file=.env src/cli.ts" scripts.test="node --test" scripts.typecheck="tsc"

Copy the tsconfig.json from Foundations Tutorial 1 and add your key to .env.

2. This week's updates#

data/items.json

[
  { "id": "i1", "title": "Dark mode", "body": "Dark mode is now available on web and mobile. It follows your system setting by default and can be changed in Settings." },
  { "id": "i2", "title": "Faster search", "body": "Search results now load in under 200 ms for 95% of queries after we moved the index to a new cluster." },
  { "id": "i3", "title": "CSV export for reports", "body": "Every report can now be exported as CSV. Exports keep your filters and column order." },
  { "id": "i4", "title": "SSO for the Team plan", "body": "Single sign-on with Okta and Google Workspace is now included in the Team plan, not only Enterprise." },
  { "id": "i5", "title": "Fewer duplicate notifications", "body": "We fixed a bug that sent the same mention notification twice when a comment was edited." },
  { "id": "i6", "title": "New API rate limits", "body": "The public API now allows 600 requests per minute per token, up from 300. Burst limits are unchanged." },
  { "id": "i7", "title": "Audit log retention", "body": "Audit logs are now kept for 400 days on Enterprise, so you can review a full year plus a quarter." },
  { "id": "i8", "title": "Keyboard shortcuts", "body": "Press ? anywhere to see every keyboard shortcut. New shortcuts cover filters and bulk actions." }
]

3. The journal and the lock#

src/journal.ts

import { open, readFile, rm } from "node:fs/promises";
 
export interface StepEvent {
  key: string;
  replayed: boolean;
}
 
/**
 * An append-only step journal: one JSON line per completed step. `step(key, fn)` runs `fn`
 * once per key; on a later attempt the saved result is returned instead of running it again.
 */
export async function openJournal(file: string, onStep: (e: StepEvent) => void = () => {}) {
  const saved = new Map<string, unknown>();
  try {
    for (const line of (await readFile(file, "utf8")).split("\n")) {
      if (!line) continue;
      try {
        const record = JSON.parse(line) as { key: string; result: unknown };
        saved.set(record.key, record.result);
      } catch {
        // A crash in the middle of an append leaves a torn last line: that step did not complete.
      }
    }
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
  }
 
  const handle = await open(file, "a");
  return {
    has: (key: string) => saved.has(key),
 
    async step<T>(key: string, fn: () => Promise<T>): Promise<T> {
      if (saved.has(key)) {
        onStep({ key, replayed: true });
        return saved.get(key) as T;
      }
      const result = await fn();
      await handle.write(`${JSON.stringify({ key, result, at: new Date().toISOString() })}\n`);
      await handle.sync(); // on disk before the run moves on
      saved.set(key, result);
      onStep({ key, replayed: false });
      return result;
    },
 
    close: () => handle.close(),
  };
}
 
/**
 * One worker per run. The lock file records the owner's process ID; a lock left by a process
 * that no longer exists, or older than the lease, is taken over.
 */
export async function acquireLock(file: string, leaseMs = 10 * 60_000): Promise<{ release: () => Promise<void> }> {
  try {
    const handle = await open(file, "wx"); // fails if the file exists: creation is the lock
    await handle.write(JSON.stringify({ pid: process.pid, at: Date.now() }));
    await handle.close();
    return { release: () => rm(file, { force: true }) };
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
  }
  const owner = JSON.parse(await readFile(file, "utf8")) as { pid: number; at: number };
  const alive = (() => {
    try {
      process.kill(owner.pid, 0); // signal 0 only checks that the process exists
      return true;
    } catch {
      return false;
    }
  })();
  if (alive && owner.pid !== process.pid && Date.now() - owner.at < leaseMs) {
    throw new Error(`run is locked by process ${owner.pid} since ${new Date(owner.at).toISOString()}`);
  }
  await rm(file, { force: true }); // stale: its worker crashed or overran its lease
  return acquireLock(file, leaseMs);
}

This is the article's step() with a file instead of a table, and the same guarantees:

A step's result is written, and flushed with sync(), before the run continues. If the process dies after that line is on disk, the step is done; if it dies before, the step runs again.
A torn last line is ignored. A crash in the middle of a write is the same as a crash before it.
The lock is a file created with the wx flag, which fails if the file already exists, so two workers cannot both create it. A crashed worker never releases its lock, so the next worker checks whether the owner's process is still alive, and takes over if it is not. Across machines, the article's database lease does the same job.

4. An outbox with idempotency keys#

src/outbox.ts

import { mkdir, open, readFile } from "node:fs/promises";
import path from "node:path";
 
export interface Receipt {
  id: string;
  sent: boolean; // false: this key was already sent, nothing new went out
}
 
/**
 * A stand-in for an email provider that accepts an idempotency key (most do, as a header).
 * The first send with a key delivers; any later send with the same key returns the first receipt.
 */
export async function sendEmail(outboxDir: string, idempotencyKey: string, email: { to: string; subject: string; body: string }): Promise<Receipt> {
  await mkdir(outboxDir, { recursive: true });
  const file = path.join(outboxDir, `${idempotencyKey.replace(/[^\w-]/g, "_")}.json`);
  try {
    const handle = await open(file, "wx"); // atomic: exactly one writer can create it
    await handle.write(JSON.stringify({ ...email, key: idempotencyKey, at: new Date().toISOString() }, null, 2));
    await handle.close();
    return { id: path.basename(file, ".json"), sent: true };
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
    JSON.parse(await readFile(file, "utf8")); // it exists and is complete
    return { id: path.basename(file, ".json"), sent: false };
  }
}

With a real provider, pass the key in its idempotency header and delete this file. The contract is the same: one key, one delivery, however many times you ask.

5. The job, as steps#

src/digest.ts

import { mkdir, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
import Anthropic from "@anthropic-ai/sdk";
import { acquireLock, openJournal, type StepEvent } from "./journal.ts";
import { type Receipt, sendEmail } from "./outbox.ts";
 
export interface Item {
  id: string;
  title: string;
  body: string;
}
 
export interface DigestOptions {
  runId: string; // the same ID resumes the same run, such as 2026-w42
  dataDir: string;
  stateDir: string; // journals, locks, the outbox, and what earlier digests already covered
  to: string;
  client?: Anthropic;
  onStep?: (e: StepEvent) => void;
  crash?: (point: string) => void; // fault injection: throw or exit at a named point
}
 
const text = (m: Anthropic.Message) => m.content.flatMap((b) => (b.type === "text" ? [b.text] : [])).join("").trim();
 
async function readSeen(file: string): Promise<Set<string>> {
  try {
    return new Set(JSON.parse(await readFile(file, "utf8")) as string[]);
  } catch {
    return new Set();
  }
}
 
export async function runDigest(o: DigestOptions): Promise<{ status: "sent" | "nothing-new" | "already-done"; items: number; receipt?: Receipt; modelCalls: number }> {
  const client = o.client ?? new Anthropic();
  const crash = o.crash ?? (() => {});
  const runsDir = path.join(o.stateDir, "runs");
  const seenFile = path.join(o.stateDir, "seen.json");
  await mkdir(runsDir, { recursive: true });
 
  const lock = await acquireLock(path.join(runsDir, `${o.runId}.lock`));
  const journal = await openJournal(path.join(runsDir, `${o.runId}.jsonl`), o.onStep);
  let modelCalls = 0;
  try {
    if (journal.has("done")) return { status: "already-done", items: 0, modelCalls };
 
    // Step 1: choose the items. Saved, so a resumed run works on exactly the same list
    // even if new items arrived in the meantime.
    const items = await journal.step("select", async () => {
      const all = JSON.parse(await readFile(path.join(o.dataDir, "items.json"), "utf8")) as Item[];
      const seen = await readSeen(seenFile);
      return all.filter((i) => !seen.has(i.id));
    });
    crash("after:select");
 
    if (items.length === 0) {
      await journal.step("done", async () => true);
      return { status: "nothing-new", items: 0, modelCalls };
    }
 
    // Step 2: one model call per item, each its own step, so a crash loses at most one call.
    const summaries: string[] = [];
    for (const item of items) {
      summaries.push(
        await journal.step(`summarize:${item.id}`, async () => {
          modelCalls++;
          return text(
            await client.messages.create({
              model: "claude-haiku-5-5",
              max_tokens: 1024,
              system: "You write one-sentence summaries of product updates for a weekly customer digest. Plain words, no hype.",
              messages: [{ role: "user", content: `<update title="${item.title}">\n${item.body}\n</update>` }],
            }),
          );
        }),
      );
      crash(`after:summarize:${item.id}`);
    }
 
    // Step 3: the opening paragraph, from the summaries.
    const intro = await journal.step("compose", async () => {
      modelCalls++;
      return text(
        await client.messages.create({
          model: "claude-sonnet-5-5",
          max_tokens: 1024,
          output_config: { effort: "low" },
          system: "You write the two-sentence opening of a weekly product digest email. Mention the most useful update first.",
          messages: [{ role: "user", content: summaries.map((s, i) => `- ${items[i]!.title}: ${s}`).join("\n") }],
        }),
      );
    });
    const body = `${intro}\n\n${items.map((item, i) => `* ${item.title}: ${summaries[i]}`).join("\n")}`;
 
    // Step 4: the side effect. The idempotency key is derived from the run, so a repeat is a no-op.
    const receipt = await journal.step("send", async () => {
      const r = await sendEmail(path.join(o.stateDir, "outbox"), `digest-${o.runId}`, { to: o.to, subject: `Your weekly digest (${o.runId})`, body });
      crash("before-save:send"); // the dangerous moment: sent, but not yet journaled
      return r;
    });
 
    // Step 5: remember what this digest covered, for the next run. A set union: safe to repeat.
    await journal.step("mark-seen", async () => {
      const seen = await readSeen(seenFile);
      for (const item of items) seen.add(item.id);
      await writeFile(seenFile, JSON.stringify([...seen].sort()));
      return seen.size;
    });
    await journal.step("done", async () => true);
    return { status: "sent", items: items.length, receipt, modelCalls };
  } finally {
    await journal.close();
    await lock.release();
  }
}

The rules every step follows:

Steps run in the same order on every attempt, and their keys are stable (summarize:i4, not a counter or a timestamp). That is what lets a resumed run line its steps up with the journal.
Inputs are frozen in a step. select saves the list of items, so a resumed run summarizes the same items even if items.json changed after the crash.
One model call per step. A crash loses at most the call in progress.
Side effects carry an idempotency key derived from the run, digest-2026-w42, never a random value. A random key would be different on the retry and defeat the purpose.
Every step is safe to repeat. mark-seen is a set union, so running it twice changes nothing.

6. The CLI, with a crash switch#

src/cli.ts

import { parseArgs } from "node:util";
import { runDigest } from "./digest.ts";
 
const { values } = parseArgs({ options: { run: { type: "string" }, to: { type: "string", default: "team@example.com" } } });
if (!values.run) {
  console.error("usage: npm run digest -- --run <run-id>   (for example 2026-w42)");
  process.exit(2);
}
 
// CRASH=<point> kills the process at that point, the way a deploy or an out-of-memory kill would.
const crashAt = process.env.CRASH;
 
const result = await runDigest({
  runId: values.run,
  dataDir: "data",
  stateDir: "state",
  to: values.to,
  onStep: ({ key, replayed }) => console.log(`${replayed ? "  replayed" : "  ran     "} ${key}`),
  crash: (point) => {
    if (point === crashAt) {
      console.log(`  ** simulated crash at ${point} **`);
      process.exit(137);
    }
  },
});
 
const receipt = result.receipt ? `, email ${result.receipt.sent ? "sent" : "already sent earlier, not resent"} (${result.receipt.id})` : "";
console.log(`${result.status}: ${result.items} items, ${result.modelCalls} model calls in this attempt${receipt}`);

process.exit is a hard stop: no finally blocks run and the lock file stays behind, just as when a server is killed. That is the realistic case to test.

7. Break it#

Crash in the middle. Kill the job after the third summary:

CRASH=after:summarize:i3 npm run digest -- --run 2026-w42

  ran      select
  ran      summarize:i1
  ran      summarize:i2
  ran      summarize:i3
  ** simulated crash at after:summarize:i3 **

state/runs/ now holds the journal and a stale lock. Run the same command without the crash:

npm run digest -- --run 2026-w42

  replayed select
  replayed summarize:i1
  replayed summarize:i2
  replayed summarize:i3
  ran      summarize:i4
  ran      summarize:i5
  ran      summarize:i6
  ran      summarize:i7
  ran      summarize:i8
  ran      compose
  ran      send
  ran      mark-seen
  ran      done
sent: 8 items, 6 model calls in this attempt, email sent (digest-2026-w42)

The lock left by the dead process was taken over, the first three summaries came from the journal, and only six model calls were made: five summaries and the opening.

Crash at the worst moment. Delete state/ and kill the job right after the email goes out, before the send step is saved:

rm -rf state
CRASH=before-save:send npm run digest -- --run 2026-w42
npm run digest -- --run 2026-w42

  ...
  replayed compose
  ran      send
  ran      mark-seen
  ran      done
sent: 8 items, 0 model calls in this attempt, email already sent earlier, not resent (digest-2026-w42)

The send step ran again, because it was never journaled, but the outbox recognized the key: state/outbox/ holds exactly one email. This is the case checkpoints alone cannot fix, and the reason side effects need idempotency keys.

Run it again, and next week.

npm run digest -- --run 2026-w42   # already-done: 0 items, 0 model calls in this attempt
npm run digest -- --run 2026-w43   # nothing-new: every item was covered by 2026-w42

8. Test the failure modes#

Crashes are simulated by a crash function that throws. The tests count the model calls the client makes, so they prove that replayed steps cost nothing.

test/digest.test.ts

import assert from "node:assert/strict";
import { appendFile, mkdtemp, readdir, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import path from "node:path";
import { test } from "node:test";
import Anthropic from "@anthropic-ai/sdk";
import { runDigest } from "../src/digest.ts";
import { acquireLock, openJournal } from "../src/journal.ts";
import { sendEmail } from "../src/outbox.ts";
 
const tmp = () => mkdtemp(path.join(tmpdir(), "digest-"));
 
/** A real client whose HTTP layer answers every call with a short text, counting the calls. */
function countingClient() {
  const counter = { calls: 0 };
  const client = new Anthropic({
    apiKey: "test-key",
    maxRetries: 0,
    fetch: async () => {
      counter.calls++;
      return Response.json({
        id: `msg_${counter.calls}`,
        type: "message",
        role: "assistant",
        model: "claude-haiku-5-5",
        stop_reason: "end_turn",
        stop_sequence: null,
        usage: { input_tokens: 50, output_tokens: 10 },
        content: [{ type: "text", text: `summary ${counter.calls}` }],
      });
    },
  });
  return { client, counter };
}
 
async function fixture() {
  const dir = await tmp();
  const items = ["a", "b", "c"].map((id) => ({ id, title: `Item ${id}`, body: `Body ${id}.` }));
  await writeFile(path.join(dir, "items.json"), JSON.stringify(items));
  return { dataDir: dir, stateDir: path.join(dir, "state"), to: "team@example.com" };
}
 
class Crash extends Error {}
const crashAt = (target: string) => (point: string) => {
  if (point === target) throw new Crash(point);
};
 
test("a step runs once; later attempts replay its saved result", async () => {
  const file = path.join(await tmp(), "run.jsonl");
  let runs = 0;
  const first = await openJournal(file);
  assert.equal(await first.step("x", async () => ++runs), 1);
  await first.close();
  const second = await openJournal(file);
  assert.equal(await second.step("x", async () => ++runs), 1);
  assert.equal(runs, 1);
  await second.close();
});
 
test("a torn last line means that step did not complete", async () => {
  const file = path.join(await tmp(), "run.jsonl");
  await appendFile(file, `${JSON.stringify({ key: "a", result: 1 })}\n{"key":"b","res`);
  const journal = await openJournal(file);
  assert.equal(journal.has("a"), true);
  assert.equal(journal.has("b"), false);
  await journal.close();
});
 
test("a lock held by a live process blocks; a lock left by a dead one is taken over", async () => {
  const file = path.join(await tmp(), "run.lock");
  await writeFile(file, JSON.stringify({ pid: process.ppid, at: Date.now() })); // the test runner: alive
  await assert.rejects(acquireLock(file), /locked by process/);
  await writeFile(file, JSON.stringify({ pid: 999_999_999, at: Date.now() })); // no such process
  const lock = await acquireLock(file);
  await lock.release();
});
 
test("the outbox delivers once per idempotency key", async () => {
  const dir = await tmp();
  const email = { to: "a@example.com", subject: "s", body: "b" };
  assert.equal((await sendEmail(dir, "digest-w1", email)).sent, true);
  assert.equal((await sendEmail(dir, "digest-w1", email)).sent, false);
  assert.deepEqual(await readdir(dir), ["digest-w1.json"]);
});
 
test("a crash mid-run resumes without repeating finished model calls", async () => {
  const o = await fixture();
  const { client, counter } = countingClient();
  await assert.rejects(runDigest({ ...o, runId: "w1", client, crash: crashAt("after:summarize:b") }), Crash);
  assert.equal(counter.calls, 2); // a and b were summarized
 
  const resumed = await runDigest({ ...o, runId: "w1", client });
  assert.equal(resumed.status, "sent");
  assert.equal(resumed.modelCalls, 2); // only c and compose
  assert.equal(counter.calls, 4);
});
 
test("a crash after sending but before saving does not send twice", async () => {
  const o = await fixture();
  const { client } = countingClient();
  await assert.rejects(runDigest({ ...o, runId: "w1", client, crash: crashAt("before-save:send") }), Crash);
  const resumed = await runDigest({ ...o, runId: "w1", client });
  assert.equal(resumed.receipt?.sent, false); // the step ran again; the provider ignored the repeat
  assert.deepEqual(await readdir(path.join(o.stateDir, "outbox")), ["digest-w1.json"]);
});
 
test("a finished run is not repeated, and the next run only covers new items", async () => {
  const o = await fixture();
  const { client } = countingClient();
  assert.equal((await runDigest({ ...o, runId: "w1", client })).items, 3);
  assert.equal((await runDigest({ ...o, runId: "w1", client })).status, "already-done");
  assert.equal((await runDigest({ ...o, runId: "w2", client })).status, "nothing-new");
});

npm test   # 7 passing, no API calls

Common mistakes#

Checkpointing without idempotency. A crash between a side effect and its checkpoint repeats the side effect. Every external action needs a key the receiver deduplicates on.
Random or time-based idempotency keys. They change on retry. Derive keys from the run and the step.
Unstable step keys or order. If a resumed run names or orders its steps differently, it replays the wrong results. Keep keys stable and order deterministic.
Reading live inputs after a resume. Freeze inputs in an early step, or a resumed run works on different data than the first attempt.
Writing results without flushing. An unflushed write can be lost in a crash, and the step looks complete when it is not. Flush before moving on.
Locks with no way to recover. A crashed worker never releases its lock. Detect dead owners, or use a lease that expires.

Try it

Extend the digest job

Summarize the items in parallel, three at a time, and keep the journal correct. Which property of step keys makes that safe?
Add retries: if a summary call fails with a rate-limit or overload error, wait and retry the step, but give up and mark the run failed after five attempts.
Add a --dry-run flag that runs every step except send and prints the email instead.
Move the journal to the Postgres tables from the article, keeping step()'s signature, and run two workers at once.

Check your understanding

Key takeaways

Split long AI jobs into named steps, save each result durably before moving on, and replay saved results on resume.
Keep step keys stable and order deterministic, and freeze inputs in an early step.
Give every side effect an idempotency key derived from the run, so a repeated step is harmless.
Test durability by crashing on purpose, at the worst moments, and counting what runs again.

Next steps#

Human-in-the-loop

Next article: pause an agent for approval and resume it later.

Reliability in production

Retries, timeouts, fallbacks, and graceful degradation for LLM apps.

Finished this article? Track it to see your progress on the roadmap.

What you will build#

1. Set up the project#

2. This week's updates#

3. The journal and the lock#

4. An outbox with idempotency keys#

5. The job, as steps#

6. The CLI, with a crash switch#

7. Break it#

8. Test the failure modes#

Common mistakes#

Next steps#

Related articles