durable

README

durable-go

Lightweight, embeddable durable task execution for Go.

durable-go lets you define typed tasks, run memoized steps, and persist progress so work can resume safely after failures or restarts. Useful for any Go app that needs reliable, resumable workflows without a heavy orchestration framework.

Releases follow Semantic Versioning; see the latest release.

Features

• Typed tasks — generic Run / Task with input and output types.
• Memoized steps — completed steps replay from the store; they are not run again.
• In-process — no cluster or workflow server; one process, one store.
• Pluggable persistence — Store interface; journal-per-task filesystem store included.
• Timeouts and retries — WithTimeout, WithMaxRetries on the task handle.
• Panic recovery — task and step panics are recorded and returned as errors.
• Auto-purge — optional background cleanup of old completed and failed records.
• Flexible execution — tasks as durable.Func closures or structs with Exec.

Why durable-go

Most durable-execution frameworks require external infrastructure—such as a dedicated workflow server or a Postgres database—and enforce strict code execution models like replay determinism.

durable-go takes a zero-infra, in-process approach: a single Go library with a filesystem-backed journal store running inside your application process. Instead of replaying entire function call graphs from an external orchestrator, durable-go memoizes individual step results in your store. On resume the task runs again from the top; completed steps return the cached result. There is no replay-determinism sandbox.

Single-process only. The built-in journal store is designed for use within one OS process. Do not share the store directory across multiple processes or pods — concurrent appends from separate processes corrupt the journal. For distributed workloads, implement the Store interface backed by a server-mode database of your choice.

Install

go get github.com/agenticenv/durable-go@latest

Go 1.26.5+. No infrastructure required. No external dependencies — the included journal store writes to the local filesystem.

Quick Start

import (
    "context"

    durable "github.com/agenticenv/durable-go"
    "github.com/agenticenv/durable-go/store/journal"
)

// errors omitted for brevity
store, _ := journal.NewJournalStore("./durable-data")
defer store.Close()

client, _ := durable.NewClient(context.Background(), store)
defer client.Close()

handle := client.NewTask("job-42", durable.WithName("Example job"))

out, _ := durable.Run(context.Background(), handle, "hello", durable.Func(
    func(ctx context.Context, s *durable.StepRunner, in string) (string, error) {
        greet, err := durable.Step(ctx, s, "greet", func(ctx context.Context) (string, error) {
            return in + " world", nil
        })
        if err != nil {
            return "", err
        }
        return durable.Step(ctx, s, "upper", func(ctx context.Context) (string, error) {
            return greet, nil
        })
    },
))
_ = out

Full example: examples/func-task/.

Struct-based tasks

For services with injected dependencies, implement Exec on a struct and pass it to Run:

type Job struct {
    DB    *Database
    Mail  Mailer
}

func (j *Job) Exec(ctx context.Context, s *durable.StepRunner, id string) (string, error) {
    return durable.Step(ctx, s, "notify", func(ctx context.Context) (string, error) {
        return j.Mail.Send(ctx, id)
    })
}

out, _ := durable.Run(ctx, handle, "42", &Job{DB: db, Mail: mailer})

Full example: examples/struct-task/.

Resume

Call Run again with the same NewTask ID and the same input. Completed steps replay from the store. After a crash, leftover records show as StatusRunning in ListTasks; you still call Run — the library does not auto-resume.

To detect zombie running tasks from a previous crash, use durable.ListStaleTasks:

stale, err := durable.ListStaleTasks(ctx, store, 10*time.Minute)
// stale contains tasks with status=running not updated in the last 10 minutes

Full example: examples/resume/.

Writing tasks

Follow these when you write a task. On resume, the task runs again from the top; completed steps are reused, not re-executed.

1. Side effects in Step. Do not call an API, write to a database, or publish to a queue in the task body. Wrap that work in durable.Step.
2. Non-deterministic values in Step. Do not use time.Now(), UUIDs, or random values in the task body to choose a step ID or a branch. Generate them inside a Step so resume sees the same result.
3. Idempotent steps. A crash can re-run a step after the side effect already happened. Charging a card or sending mail must be safe to do twice (or no-op).

Also:

• Unique step IDs — one stable string per step (literals or deterministic keys). Reusing an ID returns the first completed result.
• JSON results — step outputs must be JSON-marshalable.
• Same task ID to resume — task inputs are not persisted; pass the same input to Run when resuming. Do not call Run concurrently for the same ID.

Examples

Runnable examples in examples/ — see examples/README.md for setup and run instructions.

Example	What it shows
examples/resume/	Crash after step 2, resume from cache
examples/func-task/	Closure-style durable.Func
examples/struct-task/	Struct task with injected deps, retries, timeout

# from repo root
go run ./examples/resume/
go run ./examples/func-task/
go run ./examples/struct-task/

Development

See CONTRIBUTING.md for setup, workflow, and guidelines. Project policies: SECURITY.md · CODE_OF_CONDUCT.md

Quick commands (requires Task): task check | task test | task lint | task fmt | task tidy | task test-coverage

Coverage reports (PR and default branch) are on Codecov. Run task test-coverage locally to produce coverage.out and coverage.html.

License

Apache 2.0

Disclaimer

This project is provided "as is" under the Apache License 2.0. You are responsible for how you persist and handle task data, including secrets and personally identifiable information in step outputs. For security issues, follow SECURITY.md.

Documentation

Index

• func Run[I, O any](ctx context.Context, h *TaskHandle, input I, task Task[I, O]) (O, error)
• func Step[O any](ctx context.Context, s *StepRunner, stepID string, ...) (O, error)
• type Client
  • func NewClient(ctx context.Context, store Store, opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) NewTask(id string, opts ...TaskOption) *TaskHandle
• type Option
  • func WithAutoPurge(age time.Duration, interval ...time.Duration) Option
  • func WithLogger(logger *slog.Logger) Option
• type StepRecord
• type StepRunner
• type StepStatus
• type Store
• type Task
• type TaskFunc
  • func Func[I, O any](fn func(ctx context.Context, s *StepRunner, in I) (O, error)) TaskFunc[I, O]
  • func (f TaskFunc[I, O]) Exec(ctx context.Context, s *StepRunner, input I) (O, error)
• type TaskHandle
  • func (h *TaskHandle) ID() string
• type TaskInfo
  • func ListStaleTasks(ctx context.Context, store Store, staleSince time.Duration) ([]TaskInfo, error)
• type TaskOption
  • func WithMaxRetries(n int) TaskOption
  • func WithName(name string) TaskOption
  • func WithTag(k, v string) TaskOption
  • func WithTimeout(d time.Duration) TaskOption
• type TaskStatus

Functions

func Run

func Run[I, O any](ctx context.Context, h *TaskHandle, input I, task Task[I, O]) (O, error)

Run executes the given task using the execution handle h.

On first run, a new TaskInfo record is persisted and the task closure is executed. On a subsequent run with the same h.ID(), all previously completed steps are loaded from the store in a single batch and replayed from cache without re-execution.

The engine recovers panics from the task closure, records the panic value and stack in TaskInfo.PanicTrace, marks the task StatusFailed, and returns an error.

Concurrent Run calls for different task IDs are safe. Concurrent calls for the same task ID result in undefined behaviour; callers must serialise per task ID.

func Step

func Step[O any](ctx context.Context, s *StepRunner, stepID string, fn func(ctx context.Context) (O, error)) (O, error)

Step executes fn as a memoised, fault-tolerant step within the parent task.

stepID must be unique and stable within the task across runs. Derive it from static literals or a deterministic counter — never from runtime data that may change between executions (e.g. timestamps, random values).

Cache hit: if a StepRecord with StepStatusCompleted exists for (taskID, stepID), its Result is unmarshalled and returned immediately without calling fn.

Cache miss: fn is executed. On success the result is marshalled, persisted as StepStatusCompleted, and returned. On error the step is persisted as StepStatusFailed and the error is returned; subsequent replays will re-execute fn.

Panics in fn are recovered, persisted as StepStatusFailed with PanicTrace populated, and returned as a non-nil error to the caller.

Step is not safe for concurrent use within a single StepRunner.

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is the process-level entry point for durable task execution. Create one per process via NewClient and share it across goroutines. Client is safe for concurrent use.

Always call Close (or defer it) when the Client is no longer needed to stop the background auto-purger goroutine and release associated resources.

durable-go is designed for single-process use only. Do not share the underlying store across multiple OS processes or pods.

func NewClient

func NewClient(ctx context.Context, store Store, opts ...Option) (*Client, error)

NewClient initialises a Client bound to the given store. Returns an error if store is nil. If WithAutoPurge is provided, a background goroutine is started that periodically purges old completed/failed task records. Call Close to stop it.

client, err := durable.NewClient(ctx, store)
// with auto-purge:
client, err := durable.NewClient(ctx, store, durable.WithAutoPurge(24*time.Hour))

func (*Client) Close

func (c *Client) Close() error

Close stops the background auto-purger goroutine if it is running. It is safe to call Close more than once. Idiomatic usage:

client, err := durable.NewClient(ctx, cfg)
if err != nil { ... }
defer client.Close()

func (*Client) NewTask

func (c *Client) NewTask(id string, opts ...TaskOption) *TaskHandle

NewTask creates a TaskHandle bound to this client's store. id must be unique per logical task execution; reusing the same id on a subsequent Run replays previously completed steps from the store.

type Option

type Option func(*config)

Option is a functional option applied when constructing a Client via NewClient.

func WithAutoPurge

func WithAutoPurge(age time.Duration, interval ...time.Duration) Option

WithAutoPurge enables periodic purging of completed and failed task records older than age. The optional interval controls how frequently the purger runs (defaults to 1 hour). Pass a single duration for the most common case:

durable.WithAutoPurge(24 * time.Hour)

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger sets a custom slog.Logger for the Client. All task and step lifecycle events are emitted through this logger at the appropriate level (Debug for normal flow, Info for completions, Warn for retries, Error for failures and panics). The log level is determined entirely by the handler attached to logger — pass a handler with a higher min-level to suppress verbose output. If nil or not provided, a no-op logger is used and nothing is emitted.

type StepRecord

type StepRecord struct {
	// StepID is the caller-supplied, task-scoped identifier for this step.
	StepID string
	// Seq is the zero-based execution order of this step within its parent task.
	// Used to validate replay ordering and detect sequence divergence.
	Seq int
	// InputHash is an optional hash of the step's input used for cache-busting
	// when the same StepID is reused with different inputs across runs.
	InputHash string
	// Result is the JSON-marshalled output value of a successfully completed step.
	// Non-nil only when Status == StepStatusCompleted.
	Result []byte
	// Error is the serialised error string when Status == StepStatusFailed.
	Error string
	// PanicTrace holds the recovered panic value and stack when the step panicked.
	PanicTrace string
	// Status is the current lifecycle state of this step.
	Status StepStatus
	// StartedAt is the wall-clock time when the step function was invoked.
	StartedAt time.Time
	// CompletedAt is the wall-clock time when the step reached a terminal state.
	CompletedAt time.Time
}

StepRecord is the persistent checkpoint for a single memoized step. Once Status == StepStatusCompleted, Result is immutable and will be replayed verbatim on subsequent task runs without re-executing the step function.

type StepRunner

type StepRunner struct {
	// contains filtered or unexported fields
}

StepRunner is scoped to a single task execution and provides the Step primitive. It is not safe for concurrent use; do not share a StepRunner across goroutines.

type StepStatus

type StepStatus string

StepStatus represents the lifecycle state of a single Step within a Task.

const (
	// StepStatusPending indicates the step has been registered but not yet executed.
	StepStatusPending StepStatus = "pending"
	// StepStatusCompleted indicates the step executed successfully and its result is cached.
	StepStatusCompleted StepStatus = "completed"
	// StepStatusFailed indicates the step execution returned an error or panicked.
	StepStatusFailed StepStatus = "failed"
)

type Store

type Store interface {
	// SaveTask persists or updates a TaskInfo record.
	// Must be an upsert: if a record with task.ID already exists, all mutable
	// fields are overwritten. Returns an error if the store is unavailable.
	SaveTask(ctx context.Context, task TaskInfo) error

	// GetTask retrieves a TaskInfo by its ID.
	// Returns (info, true, nil) when found, (zero, false, nil) when not found,
	// and (zero, false, err) on store errors.
	GetTask(ctx context.Context, taskID string) (TaskInfo, bool, error)

	// ListTasks returns all recorded TaskInfo records ordered by CreatedAt descending.
	ListTasks(ctx context.Context) ([]TaskInfo, error)

	// DeleteTask removes a task record and all associated step records atomically.
	// Must be a no-op (not an error) when the task does not exist.
	DeleteTask(ctx context.Context, taskID string) error

	// PurgeTasks deletes all task records (and their steps) with the given status
	// whose UpdatedAt is strictly before the cutoff time.
	// Returns the number of task records deleted.
	PurgeTasks(ctx context.Context, status TaskStatus, before time.Time) (int64, error)

	// SaveStep persists or updates a StepRecord under the given taskID.
	// Must be an upsert keyed on (taskID, step.StepID).
	SaveStep(ctx context.Context, taskID string, step StepRecord) error

	// LoadStep retrieves a single StepRecord by (taskID, stepID).
	// Returns (record, true, nil) when found, (zero, false, nil) when not found,
	// and (zero, false, err) on store errors.
	LoadStep(ctx context.Context, taskID, stepID string) (StepRecord, bool, error)

	// LoadSteps returns all StepRecords for a task ordered by Seq ascending.
	// Used during task replay to pre-populate the step cache in a single round-trip.
	// Must return a non-nil empty slice when no steps exist.
	LoadSteps(ctx context.Context, taskID string) ([]StepRecord, error)

	// ListStepIDs returns only the step IDs for a task ordered by Seq ascending.
	// Prefer LoadSteps when full records are needed.
	ListStepIDs(ctx context.Context, taskID string) ([]string, error)
}

Store is the persistence contract for durable task and step state. All implementations must be safe for concurrent use by multiple goroutines. Implementations must treat writes as upserts (idempotent on ID conflict).

durable-go's built-in store (store/journal) is designed for single-process use only. Do not share a store across multiple OS processes or pods.

type Task

type Task[I, O any] interface {
	// Exec performs the task logic. ctx is cancelled when the task timeout elapses.
	// s is the StepRunner bound to this execution; wrap all memoised work in Step calls.
	// Panics are recovered by the engine, recorded in TaskInfo.PanicTrace, and
	// surfaced as an error to the Run caller.
	Exec(ctx context.Context, s *StepRunner, input I) (O, error)
}

Task is the execution contract for a durable task. Implementations should treat any work with external side effects as a Step; non-deterministic logic outside of Step calls may not be replayed correctly. I is the input type; O is the output type.

type TaskFunc

type TaskFunc[I, O any] func(ctx context.Context, s *StepRunner, input I) (O, error)

TaskFunc adapts a plain function to the Task[I, O] interface, enabling inline closures.

func Func

func Func[I, O any](fn func(ctx context.Context, s *StepRunner, in I) (O, error)) TaskFunc[I, O]

Func wraps a plain function as a Task, triggering Go's generic type inference so callers do not need to specify type parameters explicitly.

func (TaskFunc[I, O]) Exec

func (f TaskFunc[I, O]) Exec(ctx context.Context, s *StepRunner, input I) (O, error)

Exec implements Task[I, O] for TaskFunc.

type TaskHandle

type TaskHandle struct {
	// contains filtered or unexported fields
}

TaskHandle is a bound execution handle for a uniquely identified task. Created via Client.NewTask and passed to Run. Safe to store and reuse across Run calls; a non-terminal task reuses its existing step cache.

func (*TaskHandle) ID

func (h *TaskHandle) ID() string

ID returns the unique task identifier associated with this handle.

type TaskInfo

type TaskInfo struct {
	// ID is the unique, caller-supplied task identifier.
	ID string
	// Name is a human-readable label set via WithName. May be empty.
	Name string
	// Tags are arbitrary key-value annotations for filtering and observability.
	Tags map[string]string
	// Status is the current lifecycle state of the task.
	Status TaskStatus
	// Error is the serialised error message when Status == StatusFailed.
	Error string
	// PanicTrace holds the recovered panic value and stack when the task panicked.
	// Empty string if no panic occurred.
	PanicTrace string
	// CreatedAt is the wall-clock time when the task record was first persisted.
	CreatedAt time.Time
	// StartedAt is the wall-clock time when Run began executing the task closure.
	StartedAt time.Time
	// CompletedAt is the wall-clock time when the task reached a terminal state.
	CompletedAt time.Time
	// UpdatedAt is the wall-clock time of the most recent record mutation.
	UpdatedAt time.Time
}

TaskInfo is the persistent metadata record for a single task execution. It is written on task start and updated on completion or failure.

func ListStaleTasks

func ListStaleTasks(ctx context.Context, store Store, staleSince time.Duration) ([]TaskInfo, error)

ListStaleTasks returns all tasks with StatusRunning whose UpdatedAt is older than staleSince ago. These tasks were most likely left in the running state by a previous process crash and were never resumed.

Callers can inspect the returned tasks and decide whether to re-run them (by calling Run with the same task ID, which will replay completed steps) or mark them failed via DeleteTask and starting fresh.

stale, err := durable.ListStaleTasks(ctx, store, 10*time.Minute)

type TaskOption

type TaskOption func(*taskConfig)

TaskOption is a functional option applied when constructing a TaskHandle.

func WithMaxRetries

func WithMaxRetries(n int) TaskOption

WithMaxRetries sets the number of additional execution attempts on failure. Defaults to 0 (no retries). On each retry the task closure is re-invoked; already-completed steps are replayed from cache without re-executing.

func WithName

func WithName(name string) TaskOption

WithName sets a human-readable display name on the task record. Does not affect uniqueness or execution behaviour.

func WithTag

func WithTag(k, v string) TaskOption

WithTag attaches an arbitrary key-value annotation to the task record. Call multiple times to set multiple tags. Useful for grouping and observability.

func WithTimeout

func WithTimeout(d time.Duration) TaskOption

WithTimeout sets a hard execution deadline for the task. The context passed to the task closure and all child steps is cancelled after this duration. Zero (default) means no timeout is applied.

type TaskStatus

type TaskStatus string

TaskStatus represents the lifecycle state of a Task.

const (
	// StatusPending indicates the task has been registered but not yet started.
	StatusPending TaskStatus = "pending"
	// StatusRunning indicates the task is actively executing.
	StatusRunning TaskStatus = "running"
	// StatusCompleted indicates the task finished successfully.
	StatusCompleted TaskStatus = "completed"
	// StatusFailed indicates the task terminated with an error or recovered panic.
	StatusFailed TaskStatus = "failed"
)