Distributed Checkpointing & Recovery

Long-running agent workflows, financial reconciliations that take 40 minutes, document analysis pipelines that process thousands of pages, multi-day research workflows, cannot afford to restart from scratch when a container is evicted, a pod crashes, or a dependency times out. Distributed checkpointing converts these failures from full restarts into partial replays, bounding the re-execution cost to the work done since the last checkpoint. LangGraph’s checkpointer abstraction makes this architectural pattern first-class.

Why Agent Workflows Need Durable State

An agent workflow’s state, which nodes have completed, what each node produced, which tool calls have been made, lives entirely in memory during execution. When the process hosting that workflow terminates for any reason (graceful shutdown, OOM kill, node failure, deployment), that state is gone.

For a 5-minute workflow, a full restart is annoying. For a 45-minute workflow with $0.80 of LLM spend and a 1-hour SLA, a full restart at minute 38 means delivering the result 38 minutes late, spending $0.80 again, and potentially triggering an SLA breach. At scale, 500 concurrent workflows, a 2% infrastructure failure rate: you are looking at 10 full workflow restarts per hour, each costing money and SLA margin.

Checkpointing solves this by writing a serialized snapshot of the workflow’s state to durable storage at defined intervals. On recovery, the workflow loads the most recent checkpoint and resumes execution from that point. The re-execution cost is bounded by the work done since the last checkpoint, not the work done since the beginning of the workflow.

flowchart TD
    N1["Node 1\n(fetch data)"] -->|"checkpoint"| CP1[("Checkpoint 1\nPostgreSQL")]
    CP1 --> N2["Node 2\n(validate)"]
    N2 -->|"checkpoint"| CP2[("Checkpoint 2\nPostgreSQL")]
    CP2 --> N3["Node 3\n(analyze)"]
    N3 -->|"CRASH"| FAIL["Pod Evicted"]
    FAIL --> REC["Recovery Process"]
    REC --> CP2
    CP2 -->|"resume from\nCheckpoint 2"| N3R["Node 3\n(re-execute)"]
    N3R --> N4["Node 4\n(report)"]

    style N1 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style N2 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style N3 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style N3R fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style N4 fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style CP1 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style CP2 fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style FAIL fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style REC fill:#fff7ed,stroke:#f59e0b,color:#0F172A

LangGraph PostgreSQL Checkpointer Setup

LangGraph’s checkpointer abstraction supports multiple backends: in-memory (development only), SQLite (single-instance), and PostgreSQL (production). PostgreSQL is the production choice: it is durable, queryable, replicable, and supports concurrent writers from multiple workflow instances.

import asyncpg
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define workflow state: every field must be serializable to JSON.
# LangGraph snapshots this entire dict at each checkpoint.
class ReconciliationState(TypedDict):
    thread_id: str
    transactions: list[dict]
    validated_count: int
    # Annotated with operator.add makes this field appendable across nodes
    # each node's output is merged into the list rather than replacing it.
    anomalies: Annotated[list[dict], operator.add]
    report_sections: Annotated[list[str], operator.add]
    current_step: str

async def build_checkpointed_workflow():
    # PostgreSQL connection pool for the checkpointer.
    # Use a dedicated schema ("checkpoints") to isolate checkpoint tables
    # from application tables and simplify access control.
    pool = await asyncpg.create_pool(
        dsn="postgresql://agent_user:password@pg-primary:5432/agent_db",
        min_size=2,
        max_size=10,
    )
    checkpointer = AsyncPostgresSaver(pool, schema_name="checkpoints")
    # Create checkpoint tables if they don't exist: idempotent, safe to call on startup.
    await checkpointer.setup()

    # Build the graph. Checkpointing is transparent: LangGraph saves state
    # automatically after each node completes, before the next node starts.
    graph = StateGraph(ReconciliationState)
    graph.add_node("fetch_transactions", fetch_transactions_node)
    graph.add_node("validate_transactions", validate_transactions_node)
    graph.add_node("detect_anomalies", detect_anomalies_node)
    graph.add_node("generate_report", generate_report_node)

    graph.set_entry_point("fetch_transactions")
    graph.add_edge("fetch_transactions", "validate_transactions")
    graph.add_edge("validate_transactions", "detect_anomalies")
    graph.add_edge("detect_anomalies", "generate_report")
    graph.add_edge("generate_report", END)

    # Compile with checkpointer: every node boundary now writes a checkpoint.
    return graph.compile(checkpointer=checkpointer)

async def run_or_resume_workflow(thread_id: str, initial_input: dict):
    workflow = await build_checkpointed_workflow()
    config = {
        "configurable": {
            # thread_id is the checkpoint partition key
            # all checkpoints for one workflow instance share this ID.
            # Using a stable, task-derived ID (not a random UUID) makes
            # recovery deterministic: the same task always maps to the same checkpoint.
            "thread_id": thread_id,
        }
    }
    # LangGraph checks for an existing checkpoint under this thread_id.
    # If one exists, the workflow resumes from the last saved node.
    # If none exists, execution starts from the entry point.
    # The caller does not need to know which path is taken.
    result = await workflow.ainvoke(initial_input, config=config)
    return result

Checkpoint Granularity Tradeoffs

Every-node checkpointing (the LangGraph default) is the right choice for workflows where each node makes an LLM call or external API call, the cost of re-execution exceeds the cost of the checkpoint write. PostgreSQL INSERT latency at p99 is under 5ms on a well-configured instance; an LLM call costs 50-500ms and real money.

For workflows with cheap, fast nodes (pure data transformations, in-memory computations), checkpointing every N nodes reduces write overhead without meaningfully increasing re-execution cost. Implement this by wrapping your graph in conditional checkpoint logic that skips the write for lightweight nodes.

The practical rule: if re-executing a node costs more than $0.01 or takes more than 1 second, checkpoint after it. If it costs less and takes less, it is safe to batch checkpoint writes.

Idempotency for Tool Calls During Replay

Checkpointing ensures workflow state is recovered. But when a node is re-executed after recovery, any side effects it causes, API calls, emails sent, database writes, external service triggers, will be executed again. This is correct for read operations (re-fetching data produces the same result) but dangerous for write operations (re-sending an email produces a duplicate).

import hashlib
from functools import lru_cache

def make_idempotency_key(thread_id: str, node_name: str, call_index: int) -> str:
    """
    Generate a stable, unique key for a specific tool call within a specific workflow run.
    thread_id + node_name + call_index uniquely identifies one tool invocation.
    Using SHA-256 gives a fixed-length key safe for use as a database primary key or
    HTTP Idempotency-Key header.
    """
    raw = f"{thread_id}:{node_name}:{call_index}"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

async def send_notification_idempotent(
    thread_id: str,
    node_name: str,
    call_index: int,
    recipient: str,
    message: str,
    idempotency_store,  # Redis or PostgreSQL table tracking used keys
) -> dict:
    key = make_idempotency_key(thread_id, node_name, call_index)

    # Check if this exact call has already been completed.
    # If yes, return the cached result: do NOT re-send the notification.
    existing = await idempotency_store.get(key)
    if existing:
        return existing["result"]

    # First time this key is seen: execute the side effect.
    result = await send_email(recipient, message)

    # Store the result so future replays return the same value.
    await idempotency_store.set(key, {"result": result}, ttl_seconds=86400 * 7)
    return result

Cross-Region Replication and Retention Policies

flowchart TD
    PRI["PostgreSQL Primary\n(us-east-1)"] -->|"streaming replication"| REP["PostgreSQL Replica\n(us-west-2)"]
    PRI --> CHK["Checkpoint Store\n(workflow snapshots)"]
    REP --> DR["Disaster Recovery\n(failover target)"]
    CHK -->|"retention policy\n(delete after 30 days)"| ARCH["Archive\n(S3 Glacier)"]
    DR -->|"on failover"| WF["Workflow resumes\nfrom replica"]

    style PRI fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style REP fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style CHK fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style DR fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style ARCH fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style WF fill:#f0fdf9,stroke:#0D9488,color:#0F172A

For disaster recovery scenarios, an entire cloud region becoming unavailable, checkpoint data must be replicated cross-region. PostgreSQL streaming replication to a standby in a second region provides RPO (Recovery Point Objective) measured in seconds. On failover, workflows that were mid-execution in the failed region can resume from the replica’s checkpoint data in the recovery region.

Checkpoint retention requires a deliberate policy. Completed workflow checkpoints are not useful after the workflow succeeds and its results have been persisted to your application database. Delete completed workflow checkpoints after 30 days (or immediately after verification, if storage cost is a concern). Checkpoints for in-flight workflows should never be deleted automatically, use a lock-based mechanism to prevent the retention job from deleting checkpoints with an active lock.

Distributed checkpointing, fault-tolerant design, RBAC, zero-trust security, and audit trails together form the enterprise-grade foundation for multi-agent systems that organizations can depend on in production. The patterns in this module are not theoretical, they are the difference between a system that impresses in a demo and one that runs reliably at 3 AM when an on-call engineer is not watching.