Skip to main content

Online edition for BLOGE 0.9.8-RC1 · facts verified 2026-09-15 · 中文

Chapter 13 — Durable Execution

Promise: By the end of this chapter you will understand how BLOGE persists execution state so that a graph can survive process restarts, restore its accepted checkpoints, and continue on another eligible worker when the graph definition and durable stores can be reconstructed.

Learning Goals​

  1. Explain what "durable execution" means in BLOGE — checkpoint-based persistence of node outputs, suspend contexts, loop snapshots, and execution identity so a graph can resume after a crash.
  2. Identify the runtime stores that make durability possible: ExecutionStore, ExecutionCheckpointStore, WaitStore, and WorkItemStore.
  3. Wire in-memory durable stores into a GraphEngine and run a suspend → signal → resume flow across two engine instances.
  4. Describe how cold recovery works: execution identity → graph registry lookup → checkpoint reload → continue.
  5. Recognise the checkpoint types (NODE_OUTPUT, LOOP_SNAPSHOT, FOREACH_PROGRESS, SUSPEND_CONTEXT, PARTIAL_RESULT) and when each is written.

Prerequisites​

Source Examples​

FileWhat it shows
RuntimeGraphEngineDurabilityTest.javaCross-process signal using runtime stores; two engine instances share state
RuntimeGraphRegistryRecoveryTest.javaCold recovery from a published graph definition; hash-mismatch rejection; automatic expired-claim recovery
PaymentWaitExample.javaEnd-to-end suspend → publishEvent → resume lifecycle with runtime stores
payment-wait.blogeDSL await block with event correlation, timeout, and branching
RuntimeLoopOperatorDurabilityTest.javaLoop snapshot checkpoint and resume-from-iteration

Why This Matters​

Everything you've learned so far — nodes, dependencies, branches, resilience, waiting — assumes the JVM stays alive. In production that assumption fails:

  • A process crashes mid-execution. Without durability, every in-flight node output is lost and the workflow must restart from scratch.
  • A suspend can last minutes, hours, or days (waiting for a payment callback, a manager approval). Keeping a thread blocked that long wastes resources and is destroyed by any deploy.
  • Scaling out means multiple worker processes. A graph that suspended on Worker A must be resumable on Worker B without Worker A ever coming back.

Durable execution solves all three. The engine writes checkpoints to a persistent store as nodes complete, and writes wait records when a node suspends. On resume — whether triggered by a signal, a timer, or crash recovery — any engine instance can reload those checkpoints and continue from the exact frontier where execution paused.

Key insight: Durability in BLOGE is not "event sourcing" or "replay". The engine skips nodes whose checkpoints it accepts and schedules the remaining frontier. An external effect that happened before its checkpoint was committed can still be uncertain; idempotency and reconciliation remain application responsibilities.

Mental Model​

Think of durable execution as a bookmark in a recipe book:

Diagram: 13-durable-execution figure 1

When the payment event arrives, any engine instance can:

  1. Load the ExecutionInstance from the ExecutionStore.
  2. Reload the NODE_OUTPUT checkpoints for the three completed nodes.
  3. Resolve the WAIT_SIGNAL wait record.
  4. Resume from awaitPayment → branch → fulfillOrder.

The four pillars of this model:

PillarStoreWhat it records
IdentityExecutionStoreExecution lifecycle, status, lease, recovery attempts
CheckpointsExecutionCheckpointStoreNode outputs, loop snapshots, suspend contexts, foreach progress
WaitsWaitStoreWhy the execution is paused (signal, timer, task, event)
Work itemsWorkItemStorePending actions: timer-due, event-matched, task-resume

The execution lifecycle as a state diagram:

Diagram: 13-durable-execution figure 2

The bloge-runtime-spi Module​

Before any code, name the seam. The durable execution layer is split into two modules:

  • bloge-runtime-spi — interfaces only. ExecutionStore, ExecutionCheckpointStore, WaitStore, TimerService, GraphRegistryStore, LeaseStore, AuditJournalStore, TaskInboxStore, plus the typed events those stores emit.
  • bloge-durable — engine plumbing that consumes those interfaces: the recovery loop, hash check, checkpoint codec, and the in-memory implementations used in tests.

Database-backed implementations live in their own modules (bloge-durable-mybatis for MyBatis/JDBC). The split means: if you only need to plug in a custom backing store, depend on bloge-runtime-spi alone — you do not have to bring in the engine artifact.

<dependency>
  <groupId>com.leanowtech</groupId>
  <artifactId>bloge-runtime-spi</artifactId>
</dependency>

First Working Example​

This example is extracted from RuntimeGraphEngineDurabilityTest. It demonstrates a graph that suspends on one engine instance and completes on another — exactly the "cross-process signal" pattern you need in production.

Step 1 — Create the stores and the graph​

// Shared durable stores (in production these would be database-backed)
var executionStore    = new InMemoryExecutionStore();
var checkpointStore   = new InMemoryExecutionCheckpointStore();
var waitStore         = new InMemoryWaitStore();
var timerService      = new InMemoryTimerService();

// A suspendable operator that parks execution
SuspendableOperator<Void, String> waitOp = (in, ctx) ->
        OperatorResult.suspend("ck", null, Duration.ofMinutes(10));
Operator<Object, String> completeOp = (in, ctx) -> "C";

var graphBuilder = new GraphBuilder("runtime-cross-process");
var graph = graphBuilder
        .suspendNode("w", waitOp)
        .node("c", completeOp).dependsOn("w")
        .build();

CountDownLatch suspended = new CountDownLatch(1);
AtomicReference<String> executionId = new AtomicReference<>();

ExecutionListener listener = new ExecutionListener() {
    @Override
    public void onGraphStart(String graphName, GraphContext ctx) {
        executionId.set((String) ctx.get(ReservedKeys.EXECUTION_ID));
    }

    @Override
    public void onNodeSuspended(String graphName, String nodeId, String suspendKey) {
        if ("w".equals(nodeId)) {
            suspended.countDown();
        }
    }
};

Step 2 — Execute until suspension on Engine 1​

var engine1 = GraphEngine.builder()
        .registry(new DefaultOperatorRegistry())
        .executionStore(executionStore)
        .executionCheckpointStore(checkpointStore)
        .waitStore(waitStore)
        .timerService(timerService)
        .listeners(List.of(listener))
        .build();

// The source test runs executeWithOperators() on a background thread so it can
// observe suspension and then resume from a second engine instance.
Thread runThread = Thread.ofVirtual().start(() -> {
    try {
        engine1.executeWithOperators(graph, null, graphBuilder.operators());
    } catch (Exception ignored) {
    }
});

// The execution is now SUSPENDED in the store
assertTrue(suspended.await(5, TimeUnit.SECONDS));
assertNotNull(executionId.get());
assertEquals(ExecutionStatus.SUSPENDED,
    executionStore.get(executionId.get()).orElseThrow().status());

Step 3 — Resume on a different Engine 2​

var engine2 = GraphEngine.builder()
        .registry(new DefaultOperatorRegistry())
        .executionStore(executionStore)          // same stores
        .executionCheckpointStore(checkpointStore)
        .waitStore(waitStore)
        .timerService(timerService)
        .build();

// Signal resumes the suspended node and continues to "c"
engine2.signal(graph, executionId.get(), "w", "resume-data");

// Execution completes
assertEquals(ExecutionStatus.COMPLETED,
    executionStore.get(executionId.get()).orElseThrow().status());
assertEquals("C",
    CheckpointCodec.DEFAULT.deserialize(
        checkpointStore.load(executionId.get(), CheckpointType.NODE_OUTPUT, "c")
            .orElseThrow().payload()));
runThread.join(1_000);

Engine 1 and Engine 2 are completely independent objects. They share nothing except the durable stores. This is the fundamental production pattern: one process suspends, another resumes.

Cold Recovery: What Survives Process Death​

Warm resume is easy to overestimate because the original engine still holds objects in memory. A real durable claim needs Engine 1 to disappear, Engine 2 to load the same executionId, and completed work to remain completed.

Diagram: cold recovery and unknown effect boundary

Follow loan-42, not an engine object​

MomentDurable factRuntime consequence
Engine 1 checkpointsfetchApplication and checkCredit completedTheir outputs can be restored
Process stopsHeap objects disappearIn-memory references prove nothing
Engine 2 claims loan-42Graph hash and checkpoint matchOnly unfinished nodes become schedulable
Recovery completesNew checkpoint records later workSame logical execution continues

The identity belongs to the execution record, not to either engine instance. The graph definition is loaded again and checked against retained state before unfinished work is dispatched.

The effect can still be UNKNOWN​

Suppose Engine 1 called approveLoan, the lender committed the approval, and the process died before the completion checkpoint. Engine 2 sees an unfinished node. The checkpoint cannot tell whether the external effect happened.

Recovery therefore needs the Chapter 6 contract: stable idempotency key, reconciliation read, and an explicit policy for retry or manual repair. Durable execution prevents completed engine work from being scheduled again when a completion was retained; it does not manufacture exactly-once effects across an uncoordinated external system.

Break It Apart​

Checkpoint types​

The engine writes different checkpoint types at different lifecycle moments. These are defined in CheckpointType:

TypeWhen writtenPurpose
NODE_OUTPUTEach node completesSerialized output — skipped on resume
SUSPEND_CONTEXTAn operator returns OperatorResult.suspend()Suspend key, partial output, graph context snapshot
PARTIAL_RESULTAlongside SUSPEND_CONTEXTIntermediate output visible while suspended
LOOP_SNAPSHOTAfter each loop iterationcompletedIterations + carryStateJson so loops resume mid-way
FOREACH_PROGRESSAfter each sequential foreach itemPer-item progress so foreach can skip finished items
EVENT_CORRELATIONWhen an await registers a matcherCorrelation key so events route to the right execution
SESSION_PHASE_SNAPSHOT / SESSION_ROUND_SNAPSHOTSession phase/round transitionsSession-specific state (Chapter 14 — Multi-Turn Sessions)

Execution lifecycle states​

ExecutionStatus tracks where an execution is in its lifecycle:

  • RUNNING → nodes are actively executing
  • SUSPENDED → at least one node returned OperatorResult.suspend()
  • COMPLETED → all terminal nodes finished successfully
  • FAILED → a node failed after all resilience attempts
  • FAILED_RECOVERY → automatic crash recovery exceeded maxRecoveryAttempts
  • CANCELLED / TERMINATED → explicitly stopped

Wait types​

WaitType records why an execution is suspended:

  • WAIT_SIGNAL — waiting for engine.signal() with a matching key
  • WAIT_TIMER — a timeout-based auto-resume
  • WAIT_EVENT — waiting for engine.publishEvent() with a correlation match
  • WAIT_TASK — waiting for a user-task completion
  • WAIT_PHASE_TIMEOUT — a session phase-level timeout fired
  • WAIT_ROUND_TIMEOUT — a session round-level timeout fired
  • WAIT_RETRY_BACKOFF — a delayed retry scheduled as a durable work item

If sessions are new to you, Chapter 14 — Multi-Turn Sessions unpacks how phase/round timeouts and session snapshots fit on top of the same durable machinery. State-machine durability takes a different path: StateMachineCheckpoint bridges through the generic execution-checkpoint SPI instead of introducing a new core checkpoint type, which you will meet in Chapter 15 — State Machines.

The DurableManager bridge​

DurableManager is the internal bridge that the GraphEngine uses to coordinate writes across all stores. You don't call it directly — the engine calls it when:

  • A graph starts → initExecution()
  • A node completes → saveNodeCompletion()
  • A node suspends → writes suspend context plus the corresponding wait/timer records
  • A loop iteration finishes → saveLoopSnapshot()
  • A resume completes → cleanup suspend context + wait records

Common Trap​

❌ Assuming checkpoint data keeps its original Java type​

When you suspend and later resume, checkpoint payloads are serialized to JSON and deserialized back. If your operator output was a PaymentEvent record, after deserialization it will be a Map<String, Object> unless you use a typed CheckpointCodec.

// FRAGILE — will ClassCastException after resume
PaymentEvent event = results.get("awaitPayment", PaymentEvent.class);

// SAFE — handle both fresh-run and checkpoint-restored shapes
Object raw = results.getRaw("awaitPayment");
String txnId;
if (raw instanceof PaymentEvent pe) {
    txnId = pe.transactionId();
} else if (raw instanceof Map<?,?> m) {
    txnId = (String) m.get("transactionId");
}

This pattern appears throughout the PaymentWaitExample.

Fix: Either use getRaw() with pattern matching, or wire a JacksonCheckpointCodec with typed deserialization into GraphEngine.builder().checkpointCodec(...).

What Goes Wrong​

Checkpoint deserialization failure after schema change​

You deploy version 2 of an operator that changes the output type from record OrderOutput(String orderId, int total) to record OrderOutput(String orderId, BigDecimal total, String currency).

An in-flight execution was checkpointed with the v1 schema. When recovery loads the checkpoint:

Recovery error: cannot deserialize the checkpoint for node 'createOrder'
  because required field 'currency' is missing from the stored payload.

Fix: Make schema changes backward-compatible (add optional fields, don't remove required ones). For breaking changes, drain in-flight executions before deploying, or implement a CheckpointCodec that handles version migration.

Guided Rewrite​

Open RuntimeGraphRegistryRecoveryTest. The test publishesDslGraphAndRecoversColdSignalWithoutCallerGraph demonstrates cold recovery — resuming without the caller passing a Graph object:

// Engine 1: execute with a DSL graph — the engine publishes the
// GraphDefinition to graphRegistryStore automatically
GraphEngine engine1 = GraphEngine.builder()
        .registry(registry)
        .executionStore(executionStore)
        .executionCheckpointStore(checkpointStore)
        .waitStore(waitStore)
        .graphRegistryStore(graphRegistryStore)        // ← enables cold recovery
        .addGraphDefinitionCodec(codec)                // ← how to decode DSL
        .timerService(timerService)
        .build();

engine1.execute(graph, new GraphContext(Map.of("sessionId", "SESSION-001")));
// Graph suspends at "wait" node

// Engine 2: signal WITHOUT passing the graph object
GraphEngine engine2 = GraphEngine.builder()
        .registry(recoveryRegistry)
        .executionStore(executionStore)
        .executionCheckpointStore(checkpointStore)
        .waitStore(waitStore)
        .graphRegistryStore(graphRegistryStore)
        .addGraphDefinitionCodec(codec)
        .timerService(timerService)
        .build();

// The engine loads the graph from graphRegistryStore using the
// execution's graphVersion + graphHash binding
engine2.signal(executionId, "wait", Map.of("message", "resume"));

Questions to consider:

  1. What happens if the graph definition changes between suspend and resume? The engine compares the graphHash stored in ExecutionIdentity with the hash of the published GraphDefinition. If they don't match, recovery fails with a "hash mismatch" error. This is tested in the rejectsColdRecoveryWhenPublishedGraphHashChanges test case.

  2. What about automatic crash recovery? When RecoveryConfig is enabled and an execution's lease expires (the worker crashed), another engine instance can automatically claim the expired execution and re-execute it. The recoveryAttempts counter is incremented each time. If the budget is exhausted, the status transitions to FAILED_RECOVERY.

  3. Where are graph definitions stored? In the GraphRegistryStore, which persists graphName + graphVersion + graphHash + source. In production this is backed by the bd_graph_definition database table.

Graph Definition Source​

When the engine publishes a graph to the GraphRegistryStore, it needs to know where the definition came from — a DSL file on disk, a Java builder call, or a registry import. The GraphDefinitionSource interface (com.leanowtech.bloge.core.runtime.registry.GraphDefinitionSource) captures that provenance:

public interface GraphDefinitionSource {
    String sourceId();          // unique identifier for this source
    String sourceType();        // e.g. "dsl-file", "java-builder", "registry"
    Optional<String> content(); // raw source content if available
}

Attach a source when building a graph for registry publication:

var source = new FileBasedGraphDefinitionSource(
        "file:///graphs/order-process.bloge", dslContent);

Graph graph = Graph.builder("orderProcess")
    .definitionSource(source)
    // ... nodes and edges ...
    .build();

engine.publishGraphDefinition(graph);

The three fields serve different recovery needs:

FieldPurpose
sourceId()Uniquely identifies the origin — a file path, a classpath resource, or a registry URI
sourceType()Tells the registry how the definition was produced ("dsl-file", "java-builder", "registry")
content()Optionally stores the raw DSL text so the definition can be reconstructed without the original file

Without a GraphDefinitionSource, the registry stores the compiled graph structure but loses the trail back to the original artifact. For production systems that need audit trails or multi-cluster replication, always attach a source.

Checkpoint Codec SPI​

By default the engine serializes checkpoint payloads as JSON. For many workflows this is fine, but production systems sometimes need:

  • Custom compression to reduce storage costs for high-volume workflows
  • Encryption for checkpoints that contain sensitive data (PII, financial records)
  • Compatibility layers that handle schema migration across deployments

The CheckpointCodec SPI (com.leanowtech.bloge.core.checkpoint.CheckpointCodec) lets you replace the default serialization:

public interface CheckpointCodec {
    byte[] encode(ExecutionCheckpoint checkpoint);
    ExecutionCheckpoint decode(byte[] data);
}

Wire it into the engine builder:

GraphEngine engine = GraphEngine.builder()
    .registry(registry)
    .checkpointCodec(new JsonCheckpointCodec())  // custom serialization
    .build();

A typed codec also eliminates the Map<String, Object> problem described in the Common Trap section above — if your codec knows the target types, restored checkpoints come back as proper Java records instead of raw maps.

Operator Fingerprint Mismatch Policy​

Cold recovery still rejects a graph-definition hash mismatch. A separate question appears when a completed node has a saved Operator fingerprint but the registered Operator implementation now has another fingerprint. VersionMismatchPolicy controls only that checkpoint decision:

PolicyBehaviour
WARNRestore the checkpoint and log the Operator mismatch; this is the default.
RERUNSkip that checkpoint so the node is scheduled again.
FAILAbort resume with OperatorVersionMismatchException.

Wire an explicit choice into the engine builder:

GraphEngine engine = GraphEngine.builder()
    .registry(registry)
    .versionMismatchPolicy(VersionMismatchPolicy.FAIL)
    .build();

Choose by effect semantics, not environment labels. RERUN is safe only when the node can repeat without duplicating an irreversible effect. WARN accepts old output under new code and therefore needs an explicit compatibility decision. FAIL stops automatic recovery and leaves migration or operator intervention outside the engine.

Production warning — schema migration is a separate version axis

Treat the dialect-specific Flyway migrations packaged in bloge-durable-mybatis as authority. /actuator/bloge/schema reports UP_TO_DATE, PENDING, or FAILED; apply required schema migrations before deploying binaries that depend on them. In RC1, V25 only migrates legacy session enum values. Recovery leases were introduced by V17, the audit journal table by V9, and lease fencing by V26. Appendix H keeps the operational migration checklist outside this narrative chapter.

Brain Check​

  1. What are the four runtime stores that enable durable execution? (ExecutionStore, ExecutionCheckpointStore, WaitStore, WorkItemStore.)

  2. What happens to already-completed nodes when an engine resumes a suspended execution? (They are skipped. The engine loads their NODE_OUTPUT checkpoints and populates the NodeResults without re-executing the operators.)

  3. Why does the engine reject a cold recovery when the graph hash changes? (Because the graph structure may have changed — new nodes, removed edges, different branching — so the checkpointed node outputs may not correspond to the current graph. Resuming would produce undefined behaviour.)

  4. What ExecutionStatus does an execution transition to when automatic recovery exceeds maxRecoveryAttempts? (FAILED_RECOVERY — a terminal status.)

  5. If a SuspendableOperator returns OperatorResult.suspend("key", partial, Duration.ofMinutes(5)), how many wait records are created? (Two: one WAIT_SIGNAL for the manual signal resume path, and one WAIT_TIMER for the 5-minute timeout auto-resume path.)

  6. Design question: Your durable workflow processes insurance claims that take 2–4 weeks. During that time, you will deploy multiple code updates. What is your strategy for ensuring in-flight executions survive deployments? (Keep operator output schemas backward-compatible. Use GraphRegistryStore to version graphs so recovery can find the correct graph definition. Design compensation operators to be idempotent so that a recovered execution re-running a node is safe. Test recovery in staging with long-running executions before shipping.)

Lab​

Goal: Build a two-phase durable payment flow and simulate a crash between phases.

  1. Create a graph with three nodes:

    • createOrder — a normal operator that returns an order ID.
    • awaitPayment — a SuspendableOperator that suspends with OperatorResult.suspend("payment:" + orderId, null, Duration.ofSeconds(30)).
    • fulfillOrder — depends on awaitPayment, returns a shipment ID.

  2. Wire runtime stores (use the in-memory implementations):

    var executionStore    = new InMemoryExecutionStore();
    var checkpointStore   = new InMemoryExecutionCheckpointStore();
    var waitStore         = new InMemoryWaitStore();

  3. Execute the graph on engine1. Verify that:

    • The execution status is SUSPENDED.
    • A NODE_OUTPUT checkpoint exists for createOrder.
    • A SUSPEND_CONTEXT checkpoint exists for awaitPayment.
    • Wait records contain both WAIT_SIGNAL and WAIT_TIMER.

  4. Create engine2 with the same stores but a fresh DefaultOperatorRegistry. Signal the suspended node:

    engine2.signal(graph, executionId, "awaitPayment", paymentData);

  5. Assert that:

    • The execution completes.
    • fulfillOrder has a NODE_OUTPUT checkpoint.
    • All wait records have been cleaned up.

Stretch: Enable GraphRegistryStore and try calling engine2.signal() without passing the graph object — use only the execution ID.

Experiment acceptance card​

  • Expected and observed: A new process takes over using identity, definition, and checkpoint.
  • Failure and recovery: Change the definition version; recover with a match or explicit migration.
  • Proof boundary: Proves engine cold recovery, not known external-effect state.
  • Exercise contract: Interrupted checkpoint; change one version; deliver accepted/rejected receipts; stop when completed nodes do not rerun and UNKNOWN stays separate.

Recap​

  • Durable execution means persisting node outputs, suspend contexts, wait records, and execution identity so a graph survives crashes and can be resumed by any engine instance.
  • The four runtime stores — ExecutionStore, ExecutionCheckpointStore, WaitStore, and WorkItemStore — are the foundation. Use in-memory implementations for development; MyBatis-backed implementations for production.
  • On resume, the engine loads checkpoints and skips completed nodes. It does not replay already-finished work.
  • Cold recovery uses GraphRegistryStore to reload the graph definition from its graphVersion + graphHash binding — no caller-supplied Graph needed.
  • The engine rejects recovery when the graph hash changes, preventing checkpoint/graph mismatches.
  • Automatic crash recovery uses lease expiry and RecoveryConfig to let healthy workers claim expired executions, with a recoveryAttempts budget.
  • Checkpoint payloads are JSON-serialized. Use getRaw() with pattern matching or a custom CheckpointCodec to handle type differences between fresh runs and restored executions.

Next Step​

In Chapter 14 — Multi-Turn Sessions you will move from durable one-shot graphs to longer-lived interactions. The same checkpointing machinery you learned here becomes the substrate for phase/round sessions, and then for explicit state machines in Chapter 15.

Coding Agent: Open the versioned task guide.