Checkpointed Tasks

A checkpointed task persists intermediate results while a Task Worker executes a TaskRun. If the task fails and is retried, the worker can reuse those results instead of repeating work that has already been checkpointed.

Checkpointing is useful when a single task performs several operations, such as calling multiple APIs or invoking tools in an agent loop. It provides lightweight durable execution inside a task without turning each operation into a separate workflow node.

Availability

Checkpointed tasks were introduced in LittleHorse 0.15.0. The Java checkpoint logging API used in the example below is available starting in 0.16.0.

The runnable examples below use SDK version 1.3.0 and require a compatible LittleHorse Server.

Concepts

A Checkpoint stores a result associated with a TaskRun. Its ID contains the TaskRunId and a checkpoint number, starting at zero. Checkpoints belong to the TaskRun, so their results can be reused across its retry attempts.

Execution and Replay

The Java, Python, Go, TypeScript, and .NET Task Worker SDKs provide a checkpoint method on the worker context. In Java, WorkerContext.executeAndCheckpoint() accepts a callback and its result type. Each SDK's checkpoint method does one of two things:

1. If a prior attempt persisted a checkpoint at this position, the SDK retrieves and deserializes its value without executing the callback.
2. Otherwise, the SDK executes the callback, persists its return value as a new checkpoint, and returns that value to the task method.

On retry, the task method starts again from the beginning. It does not resume at a saved line of code. Code outside checkpoint callbacks runs again, while callbacks with persisted results are skipped.

For example, if a task saves a checkpoint and then fails later in the task method, the next attempt reuses the saved result without executing that checkpoint callback again.

info

Checkpointing does not configure retries. Configure retries on the task node in your WfSpec, just as you would for any other task. See Exception Handling.

Replay-Safe Task Logic

Checkpoints are matched by the order of checkpoint calls, not by callback name or arguments. Keep that order consistent across attempts of the same TaskRun.

• Put work that should be skipped on replay inside the callback, rather than computing it before calling the checkpoint method.
• Return the data needed by later code. On replay, the SDK restores the return value, not changes the callback made to in-memory state.
• Use result types that the SDK can serialize and deserialize.
• Keep branching and loop behavior consistent when replaying existing checkpoints. A branch based on a fresh random value or an external read can change which callback receives a saved result.
• Take care when deploying changes to workers that may retry existing tasks. Inserting, removing, or reordering checkpoint calls, or changing their result types, can make existing checkpoints incompatible with the new code.

Checkpoints and Idempotency

Checkpointing reduces duplicate work, but does not guarantee exactly-once execution of external side effects. An API call can succeed and the worker can fail before its result is persisted. A retry will then execute that callback again.

In Practice

• Run the Java example.
• Run the Python example.
• Run the Go example.
• Run the TypeScript example.
• Run the C# example.

Define the Task Method

The following task simulates sending a greeting and checkpoints a receipt. It deliberately fails after the checkpoint on attempt zero. On retry, it restores the receipt without executing the callback again. The console print represents the greeting operation; this example does not call an external service.

Java

import io.littlehorse.sdk.worker.LHTaskMethod;
import io.littlehorse.sdk.worker.WorkerContext;

public class CheckpointTasksWorker {

    @LHTaskMethod("checkpointed-side-effect")
    public String checkpointedSideEffect(String name, WorkerContext context) {
        int attemptNumber = context.getAttemptNumber();
        System.out.printf(
                "metadata: wfRun=%s taskRun=%s attempt=%d%n",
                context.getWfRunId(),
                context.getTaskRunId().getTaskGuid(),
                attemptNumber);

        String sideEffectReceipt = context.executeAndCheckpoint(
                checkpointContext -> {
                    checkpointContext.log("sending the greeting side effect");
                    System.out.println("SIDE EFFECT: send greeting for " + name);
                    return "receipt-for-" + name;
                },
                String.class);

        System.out.println("after checkpoint on attempt " + attemptNumber);
        if (attemptNumber == 0) {
            throw new RuntimeException("Intentional post-checkpoint failure on the first attempt.");
        }

        return sideEffectReceipt + "; completed on attempt " + attemptNumber;
    }
}

The SDK supplies the WorkerContext to the task method. String.class tells the SDK how to deserialize each saved result. Logs written with checkpointContext.log() are persisted with the checkpoint.

Python

from littlehorse.worker import CheckpointContext, WorkerContext


async def greeting(name: str, context: WorkerContext) -> str:
    attempt_number = context.attempt_number
    print(
        f"metadata: wfRun={context.wf_run_id.id} "
        f"taskRun={context.task_run_id.task_guid} attempt={attempt_number}"
    )

    def create_greeting(checkpoint_context: CheckpointContext) -> str:
        checkpoint_context.log("sending the greeting side effect")
        print("SIDE EFFECT: send greeting for " + name)
        return "receipt-for-" + name

    side_effect_receipt = await context.execute_and_checkpoint(create_greeting)

    print("after checkpoint on attempt " + str(attempt_number))
    if attempt_number == 0:
        raise RuntimeError("Intentional post-checkpoint failure on the first attempt.")

    return side_effect_receipt + "; completed on attempt " + str(attempt_number)

The SDK supplies WorkerContext to the task. Await execute_and_checkpoint(), but pass a synchronous callback, not an async def function. Logs written with checkpoint_context.log() are persisted with the checkpoint.

Go

package main

import (
    "fmt"

    "github.com/littlehorse-enterprises/littlehorse/sdk-go/littlehorse"
)

func Greeting(name string, context *littlehorse.WorkerContext) (string, error) {
    attemptNumber := context.GetAttemptNumber()
    fmt.Printf("metadata: wfRun=%s taskRun=%s attempt=%d\n",
    context.GetWfRunId().GetId(), context.ScheduledTask.TaskRunId.GetTaskGuid(), attemptNumber)

    sideEffectReceipt, err := context.ExecuteAndCheckpoint(func(checkpointContext *littlehorse.CheckpointContext) (interface{}, error) {
    checkpointContext.Log("sending the greeting side effect")
    fmt.Println("SIDE EFFECT: send greeting for " + name)
    return "receipt-for-" + name, nil
    })
    if err != nil {
    return "", err
    }

    fmt.Println("after checkpoint on attempt", attemptNumber)
    if attemptNumber == 0 {
    return "", fmt.Errorf("Intentional post-checkpoint failure on the first attempt.")
    }

    return fmt.Sprintf("%s; completed on attempt %d", sideEffectReceipt.(string), attemptNumber), nil
}

The SDK supplies *littlehorse.WorkerContext to the task. ExecuteAndCheckpoint() returns a value and an error; this callback returns a string, so the task asserts the restored value as string. Logs written with checkpointContext.Log() are persisted with the checkpoint.

TypeScript

import { WorkerContext } from "littlehorse-client";

async function greeting(name: string, context: WorkerContext): Promise<string> {
    const attemptNumber = context.getAttemptNumber();
    console.log(
        `metadata: wfRun=${context.getWfRunId()?.id} taskRun=${context.getTaskRunId()?.taskGuid} attempt=${attemptNumber}`,
    );

    const sideEffectReceipt = await context.executeAndCheckpoint<string>((checkpointContext) => {
        checkpointContext.log("sending the greeting side effect");
        console.log("SIDE EFFECT: send greeting for " + name);
        return "receipt-for-" + name;
    });

    console.log("after checkpoint on attempt " + attemptNumber);
    if (attemptNumber === 0) {
        throw new Error("Intentional post-checkpoint failure on the first attempt.");
    }

    return sideEffectReceipt + "; completed on attempt " + attemptNumber;
}

The SDK supplies WorkerContext as the final task argument. Await executeAndCheckpoint(); its callback can return a value or a promise. The generic type describes the expected result but does not perform runtime validation. Logs written with checkpointContext.log() are persisted with the checkpoint.

C#

using System;
using System.Threading.Tasks;
using LittleHorse.Sdk.Worker;

public class CheckpointTaskWorker
{
    [LHTaskMethod("checkpoint-greet-dotnet")]
    public Task<string> Greeting(string name, LHWorkerContext context)
    {
        int attemptNumber = context.AttemptNumber;
        Console.WriteLine(
            $"metadata: wfRun={context.WfRunId.Id} taskRun={context.TaskRunId.TaskGuid} attempt={attemptNumber}");

        string? sideEffectReceipt = context.ExecuteAndCheckpoint<string>(checkpointContext =>
        {
            checkpointContext.Log("sending the greeting side effect");
            Console.WriteLine("SIDE EFFECT: send greeting for " + name);
            return "receipt-for-" + name;
        });

        Console.WriteLine("after checkpoint on attempt " + attemptNumber);
        if (attemptNumber == 0)
        {
            throw new InvalidOperationException("Intentional post-checkpoint failure on the first attempt.");
        }

        return Task.FromResult(sideEffectReceipt + "; completed on attempt " + attemptNumber);
    }
}

The SDK supplies LHWorkerContext to the task. .NET task methods with an output return Task<T>; the checkpoint callbacks themselves are synchronous. ExecuteAndCheckpoint<string>() specifies the type used to deserialize the saved value. Logs written with checkpointContext.Log() are persisted with the checkpoint.

Register and start the worker as described in Tasks and Task Workers.

Configure Retries in the Workflow

Inside your workflow definition, schedule the task with retries:

Java

WfRunVariable name = wf.declareStr("name").required();
wf.execute("checkpointed-side-effect", name).withRetries(2);

In Java, WfRunVariable is from io.littlehorse.sdk.wfsdk.

Python

name = wf.declare_str("name").required()
wf.execute("checkpoint-greet-python", name, retries=2)

Go

name := wf.DeclareStr("name").Required()
wf.Execute("checkpoint-greet-go", name).WithRetries(2)

TypeScript

const name = wf.declareStr("name").required();
wf.execute("checkpoint-greet-typescript", name).withRetries(2);

C#

WfRunVariable name = wf.DeclareStr("name").Required();
wf.Execute("checkpoint-greet-dotnet", name).WithRetries(2);

Here, wf is the workflow's WorkflowThread. Each linked runnable example includes the complete workflow and worker setup.

Observe Replay

Start the worker using the command in your language's README, then run:

lhctl run checkpoint-tasks name obiwan

Look for these lines in the worker terminal:

SIDE EFFECT: send greeting for obiwan
after checkpoint on attempt 0
after checkpoint on attempt 1

On attempt zero, the callback prints SIDE EFFECT, saves receipt-for-obiwan as checkpoint zero, and the task then fails intentionally. On attempt one, the SDK retrieves that receipt without executing the callback. The after checkpoint line is outside the callback, so it prints on both attempts.

The successful task output is receipt-for-obiwan; completed on attempt 1. There is only one checkpoint. The workflow ends after the task, the receipt remains available in the TaskRun output.

Checkpoints or Workflow Nodes?

Use workflow nodes to represent the steps and decisions in your business process. For example, verifying a customer's identity, approving an application, and notifying the customer are distinct business steps that belong in the WfSpec.

Use checkpoints for the technical work needed to carry out one of those steps. Within an identity-verification task, for example, checkpoints can persist results from individual API calls so that retries do not repeat calls whose results have already been saved. The workflow still sees one task node and consumes the task's final output; checkpointing does not add nodes to the WfSpec.

This keeps the business process visible in the workflow while leaving implementation details inside the Task Worker. Business-level waits, such as waiting for an external verification result, belong in the workflow using External Events or other workflow primitives, rather than keeping a Task Worker occupied.

API Reference

The Task Worker SDK manages checkpoint persistence and retrieval. The underlying API includes:

• Checkpoint: the persisted value, logs, and creation metadata.
• CheckpointId: the TaskRunId and checkpoint number.
• PutCheckpoint: creates a checkpoint for a running task attempt.
• GetCheckpoint: retrieves a checkpoint by ID.