duraturo

README

duraturo

Durable execution for existing Go code: wrap your functions, run a worker, done.

Report Bug · Go Docs

   

Beta. duraturo is pre-1.0 and under active development. Interfaces are stabilizing but may change between minor versions, and the Postgres and Redis adapters are young. Crash recovery and graceful shutdown are correct today: a cancelled worker persists in-flight outcomes and hands interrupted runs back without burning retry budget. Workflow bodies are single-goroutine in v1; distributed activities, forking and child runs, and durable timers are on the roadmap.

Features

• Wrap, don't rewrite: Activity, Step, and Event wrap the code you have. Outside a run every wrapped function is a plain Go call, so tests, scripts, and codebases mid-migration keep working unchanged. Adoption is monotone.
• Interfaces first, no owned schema: pkg/ledger.Ledger and pkg/queue.Queue are pure interfaces, the root module has zero third-party dependencies, and the in-memory implementations are complete single-process systems.
• Your tables, validated, never migrated: the Postgres adapter maps onto tables you already own via a declarative Mapping. Validate introspects the live schema and errors with guidance; RecommendedDDL returns suggested DDL and never executes it.
• Postgres-only works: pgqueue (FOR UPDATE SKIP LOCKED) runs the queue in the same database. Redis (redisqueue: a ZSET lease queue plus a Redis Streams delta log) is a throughput upgrade, not a requirement, and is disposable: the janitor rebuilds it from the ledger.
• Activities are the ABI: record identity is name#occurrence (nested calls scoped under their parent's key), so code rolls forward across releases with no version, patch, or pinning APIs; checkpoints stay valid when other, differently-named calls are inserted, removed, or reordered; divergence fails loudly (ErrNonDeterministic); a breaking input/output change is a rename (charge-payment.v2).
• Fenced attempts, budgeted failures: (runID, attempt) fences every claim, with the attempt owned by the queue and incremented at every claim across the run's whole lineage; retry budget is a separate failure counter that moves only when an execution genuinely fails, so waiting and shutdown interruptions are free. The first terminal write wins; heartbeats never touch the ledger. The happy path costs two run-level ledger writes plus one per record.
• Waiting is native: an Event is a record that does not exist yet, so the run parks off the queue at zero cost. Resume is one record write plus one enqueue: Client.Signal, or a row in your own table with the janitor as backstop.
• Streaming deltas: activities Emit progress persisted on the queue side, split by attempt and seal dividers. Checkpointed activities never re-stream; losing deltas loses observability history, never correctness.
• Errors that mean it: retryable failures burn retry budget with replay-cheap retries, NonRetryable memoizes the failure itself and surfaces it as *run.RecordedError on every attempt including the first, panics are retryable, MaxAttempts bounds failed executions (waiting never counts, so parked runs are never poisoned), and IdempotencyKey(ctx) hands downstream systems the key that makes at-least-once effective-once.

Installation

go get github.com/urmzd/duraturo

The adapters are separate modules, so their dependencies never touch yours until you choose one:

go get github.com/urmzd/duraturo/adapters/postgres
go get github.com/urmzd/duraturo/adapters/redis

Quick Start

You have an order pipeline (from examples/quickstart):

func process(ctx context.Context, o Order) (Receipt, error) {
	chargeID, err := charge(ctx, o)
	if err != nil {
		return Receipt{}, err
	}
	if _, err := reserve(ctx, o); err != nil {
		return Receipt{}, err
	}
	if _, err := confirm(ctx, o); err != nil {
		return Receipt{}, err
	}
	return Receipt{OrderID: o.ID, ChargeID: chargeID, Amount: o.Amount, IssuedAt: time.Now()}, nil
}

It works, until the process dies between the charge and the email. Wrap the pieces once, at package level:

var (
	chargePayment    = duraturo.Activity("charge-payment", charge)
	reserveInventory = duraturo.Activity("reserve-inventory", reserve)
	sendConfirmation = duraturo.Activity("send-confirmation", confirm)
	processOrder     = duraturo.Activity("process-order", process)
)

func process(ctx context.Context, o Order) (Receipt, error) {
	chargeID, err := chargePayment.Call(ctx, o)
	if err != nil {
		return Receipt{}, err
	}
	if _, err := reserveInventory.Call(ctx, o); err != nil {
		return Receipt{}, err
	}
	if _, err := sendConfirmation.Call(ctx, o); err != nil {
		return Receipt{}, err
	}
	// Step records inline non-determinism: the first value is the value forever.
	issuedAt, err := duraturo.Step(ctx, "issued-at",
		func(context.Context) (time.Time, error) { return time.Now(), nil })
	if err != nil {
		return Receipt{}, err
	}
	return Receipt{OrderID: o.ID, ChargeID: chargeID, Amount: o.Amount, IssuedAt: issuedAt}, nil
}

Then wire a ledger, a queue, and a worker. The worker can be your own process:

// Ledger: your Postgres, your tables. duraturo only validates, never migrates.
pool, _ := pgxpool.New(ctx, "postgres://duraturo:duraturo@localhost:5432/duraturo")
lgr, err := pgledger.New(pool, pgledger.DefaultMapping())
if err != nil {
	log.Fatal(err)
}
if err := lgr.Validate(ctx); err != nil {
	log.Fatal(err)
}

// Queue: disposable flow. Wiping Redis loses no runs; they live in the ledger.
rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
q := redisqueue.New(rdb, "quickstart")

c := duraturo.New(lgr, q)

// The worker is a pull loop, not a service: run it on a goroutine.
w := worker.New(lgr, q)
go w.Run(ctx)

receipt, err := duraturo.Exec(ctx, c, processOrder,
	Order{ID: "1042", Email: "ada@example.com", Amount: 2400},
	duraturo.WithRunID("order-1042")) // the run ID is the submit idempotency key

Now kill the worker after charge-payment records. The lease lapses, the next claim takes attempt N+1, and the workflow re-executes from the top: charge-payment returns instantly from the ledger instead of executing (no double charge), reserve-inventory likewise, and execution resumes for real at the first unrecorded call. The quickstart demos exactly this: it crashes its own worker mid-confirmation and lets a second worker finish the order.

docker compose up -d --wait   # Postgres 17 + Redis 8; the init hook applies the suggested schema
make demo                     # or: cd examples/quickstart && go run .

Bring Your Own Tables

DefaultMapping targets the suggested duraturo_runs / duraturo_records tables, but a mapping can point at tables you already have, including tables shared with other rows:

lgr, err := pgledger.New(pool, pgledger.Mapping{
	Runs: pgledger.RunsMap{
		Table:    "orders",
		RunID:    "order_ref",
		Status:   "workflow_status",
		Envelope: "workflow_state",
	},
	Records: pgledger.RecordsMap{
		Table:    "order_events",
		RunID:    "order_ref",
		Key:      "event_key",
		Envelope: "payload",
		// Scope: this table also holds non-duraturo rows; see only ours.
		Scope: pgledger.Scope{Column: "source", Value: "duraturo"},
		// Defaults: fill your own NOT NULL columns on insert.
		Defaults: func(rec run.Record) map[string]any {
			return map[string]any{"tenant_id": "acme"}
		},
	},
})
if err != nil {
	log.Fatal(err)
}
if err := lgr.Validate(ctx); err != nil {
	log.Fatal(err) // reports exactly what is missing; RecommendedDDL suggests the fix
}

Two validation rules worth knowing up front: mapped identifiers are lower_snake_case only, and the uniqueness the ledger depends on must be a plain unique index or unique constraint (the runs table on its run-ID column, the records table on its run-ID and key pair). Partial or expression unique indexes are rejected, because the adapter's ON CONFLICT writes cannot use them.

Wait for a Human

Waiting is the absence of a record. An Event with no record parks the run: settled off the queue, still pending, costing nothing. Anything that writes the record and enqueues the run resumes it:

// In the workflow: park until someone approves.
approval, err := duraturo.Event[Approval](ctx, "approval")

// Anywhere else, any process: approve and wake the run.
err = c.Signal(ctx, "order-1042", "approval", Approval{By: "ada"})

Signal is sugar for "write the event record, enqueue the run". A row inserted into your own table plus an enqueue does the same, and the janitor re-enqueues parked runs as a backstop. Parking costs nothing: a run can wait days and resume with its full retry budget intact. Call Event from the workflow body, not inside a nested activity: record keys are parent-scoped, and Signal targets top-level event keys.

Stream Progress

// Inside an activity: emit deltas.
_ = duraturo.Emit(ctx, progress{Stage: "charged"})

// Anywhere: catch-up, then live tail.
stream, _ := c.Watch(ctx, "order-1042", "")
for {
	d, _, err := stream.Recv(ctx)
	if err != nil {
		break
	}
	fmt.Printf("%s\n", d.Payload)
}

Deltas live on the queue side behind the optional DeltaLog capability (Redis Streams in the redis adapter), split by attempt and seal dividers so consumers can discard superseded partial output. Checkpointed activities never re-stream.

Concepts

Concept	What it is
ledger	Durable truth: runs and their memoized records, in tables you already own (pkg/ledger)
queue	Flow and clock: delivery of run IDs under fenced, time-bounded claims; disposable, rebuildable from the ledger (pkg/queue)
worker	A pull loop with no inbound surface: claims, replays to the frontier, writes one outcome per claim (pkg/worker)

Examples

Example	Description
quickstart	The hero, runnable: an order pipeline on Postgres + Redis that survives its own worker being killed mid-run
chaos	Cross-process kill test: SIGKILL a worker binary mid-activity and prove recorded activities never re-execute

Documentation

• Architecture overview: the three concepts, the package DAG, modules, adapters, capabilities
• Replay design: the replay algorithm, the claim protocol, and the failure-mode table
• AGENTS.md: AI-facing conventions, commands, extension guide
• Contributing: development workflow and commit convention

License

Apache-2.0

Documentation

Overview

Package duraturo makes existing Go code durable without rewrites.

Three concepts, nothing else:

• ledger — durable truth: runs and their memoized records, stored in tables the caller already owns (pkg/ledger; adapters map, never migrate)
• queue — flow and clock: delivery of run IDs under fenced, time-bounded claims (pkg/queue); disposable, rebuildable from the ledger
• worker — a pull loop, not a service (pkg/worker); it can be the submitting process itself or something else entirely

Wrap a function with Activity, wrap non-determinism with Step, run a worker. On crash or retry the workflow re-executes from the top: calls with recorded results return them instantly, the first unrecorded call executes for real. Runs replay any number of times, or error out.

Outside a run every wrapped function is a plain Go call — code keeps working unchanged in tests, scripts, and codebases mid-migration.

Index

• func Emit(ctx context.Context, payload any) error
• func Event[O any](ctx context.Context, name string) (O, error)
• func Exec[I, O any](ctx context.Context, c *Client, a *ActivityFn[I, O], in I, opts ...StartOption) (O, error)
• func IdempotencyKey(ctx context.Context) string
• func InRun(ctx context.Context) bool
• func NonRetryable(err error) error
• func PriorDeltas(ctx context.Context) ([][]byte, error)
• func Step[O any](ctx context.Context, name string, fn func(context.Context) (O, error)) (O, error)
• type ActivityFn
  • func Activity[I, O any](name string, fn func(context.Context, I) (O, error)) *ActivityFn[I, O]
  • func ActivityIn[I, O any](reg *run.Registry, name string, fn func(context.Context, I) (O, error)) *ActivityFn[I, O]
  • func (a *ActivityFn[I, O]) Call(ctx context.Context, in I) (O, error)
  • func (a *ActivityFn[I, O]) CallKeyed(ctx context.Context, key string, in I) (O, error)
  • func (a *ActivityFn[I, O]) Name() string
• type Client
  • func New(lgr ledger.Ledger, q queue.Queue, opts ...ClientOption) *Client
  • func (c *Client) Signal(ctx context.Context, runID, name string, payload any) error
  • func (c *Client) Watch(ctx context.Context, runID string, from queue.Cursor) (queue.Stream, error)
• type ClientOption
  • func WithCodec(c run.Codec) ClientOption
  • func WithDefaultMaxAttempts(n int) ClientOption
• type DecodeError
  • func (e *DecodeError) Error() string
  • func (e *DecodeError) Unwrap() error
• type Handle
  • func HandleFor[O any](c *Client, runID string) *Handle[O]
  • func Start[I, O any](ctx context.Context, c *Client, a *ActivityFn[I, O], in I, opts ...StartOption) (*Handle[O], error)
  • func (h *Handle[O]) Result(ctx context.Context) (O, error)
  • func (h *Handle[O]) RunID() string
• type RunFailedError
  • func (e *RunFailedError) Error() string
• type RunInfo
  • func FromContext(ctx context.Context) (RunInfo, bool)
• type StartOption
  • func WithMaxAttempts(n int) StartOption
  • func WithParent(runID string) StartOption
  • func WithRunID(id string) StartOption
• type Void

Functions

func Emit

func Emit(ctx context.Context, payload any) error

Emit appends a delta to the run's stream, tagged with the activity currently executing. Deltas are advisory flow — sealed or superseded by checkpoints, streamed to watchers, never truth. Outside a run, or when the queue has no DeltaLog capability, Emit is a no-op.

func Event

func Event[O any](ctx context.Context, name string) (O, error)

Event is wait-as-absence: it returns the named event record's payload if one exists, and parks the run if not. There is no suspend mechanism — an event is simply a record that doesn't exist yet. The run stops at its frontier (the worker settles the queue item; the run stays pending) and resumes when anything writes the record and enqueues the run: a Client.Signal call, a row inserted into your own table plus an enqueue, or the janitor's backstop poll.

The returned run.ErrParked must propagate: return it up like any error.

Call Event from the workflow body (the function handed to Start), not from inside a nested activity: record keys are parent-scoped, and Client.Signal targets top-level "event:<name>#<k>" keys. An event awaited inside a nested activity needs its prefixed record written by hand.

func Exec

func Exec[I, O any](ctx context.Context, c *Client, a *ActivityFn[I, O], in I, opts ...StartOption) (O, error)

Exec is Start + Result: synchronous composition.

func IdempotencyKey

func IdempotencyKey(ctx context.Context) string

IdempotencyKey returns "{runID}:{recordKey}" for the activity currently executing — a stable key to hand downstream systems (payment providers, mail APIs) so duraturo's at-least-once side effects become effective-once where the callee deduplicates. Empty outside a wrapped call.

func InRun

func InRun(ctx context.Context) bool

InRun reports whether ctx is executing under a run.

func NonRetryable

func NonRetryable(err error) error

NonRetryable marks an error terminal: the failure is recorded (memoized) and the run fails without burning further attempts. Re-exported from pkg/run for the common case.

func PriorDeltas

func PriorDeltas(ctx context.Context) ([][]byte, error)

PriorDeltas returns the payloads the currently-executing activity emitted on previous attempts — raw material for provider-side resume (for example, prompt-prefix continuation of an interrupted stream). The read is exposed; nothing is applied automatically.

func Step

func Step[O any](ctx context.Context, name string, fn func(context.Context) (O, error)) (O, error)

Step records inline non-determinism: time, random values, generated IDs, one-off reads. Inside a run it executes once ever — the first recorded value is the value forever. Outside a run it just runs fn.

at, err := duraturo.Step(ctx, "completed-at",
    func(context.Context) (time.Time, error) { return time.Now(), nil })

Give distinct names to steps whose program order may vary: steps carry no input hash, so a same-name reorder cannot be detected the way it is for activities.

Types

type ActivityFn

type ActivityFn[I, O any] struct {
	// contains filtered or unexported fields
}

ActivityFn is a durable function handle. Declare once at package level — the Go analogue of a decorator:

var ChargePayment = duraturo.Activity("charge-payment", chargePayment)

The name is the correlation contract: records belong to it, releases roll forward against it, and a breaking input/output change means a new name ("charge-payment.v2") — the one deliberate act.

func Activity

func Activity[I, O any](name string, fn func(context.Context, I) (O, error)) *ActivityFn[I, O]

Activity wraps fn as a named durable function and registers it in run.DefaultRegistry. Duplicate names panic at init.

func ActivityIn

func ActivityIn[I, O any](reg *run.Registry, name string, fn func(context.Context, I) (O, error)) *ActivityFn[I, O]

ActivityIn registers into a specific registry — test isolation, or fleets serving disjoint activity sets.

func (*ActivityFn[I, O]) Call

func (a *ActivityFn[I, O]) Call(ctx context.Context, in I) (O, error)

Call executes the activity: memoized inside a run, a plain function call outside one.

func (*ActivityFn[I, O]) CallKeyed

func (a *ActivityFn[I, O]) CallKeyed(ctx context.Context, key string, in I) (O, error)

CallKeyed executes the activity under an explicit record key. Explicit keys are order-independent — the escape hatch for calls whose program order may vary (keyed fan-out) — and become part of the idempotency key handed downstream.

func (*ActivityFn[I, O]) Name

func (a *ActivityFn[I, O]) Name() string

Name returns the registered activity name.

type Client

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

Client submits runs and observes them. It is the producer side of the two interfaces; a worker embedded in the same process consumes from the same pair.

func New

func New(lgr ledger.Ledger, q queue.Queue, opts ...ClientOption) *Client

New composes a duraturo client from any Ledger and Queue implementation. Nothing is created, nothing is migrated: the implementations own storage.

func (*Client) Signal

func (c *Client) Signal(ctx context.Context, runID, name string, payload any) error

Signal writes an event record and wakes the run: the sugar form of "insert the row, enqueue the run". It targets the first unclaimed occurrence of name, so repeated signals satisfy successive Event calls.

func (*Client) Watch

func (c *Client) Watch(ctx context.Context, runID string, from queue.Cursor) (queue.Stream, error)

Watch streams a run's deltas (catch-up then live tail). It requires the queue to have the DeltaLog capability.

type ClientOption

type ClientOption func(*Client)

ClientOption configures a Client.

func WithCodec

func WithCodec(c run.Codec) ClientOption

WithCodec overrides the default JSON codec. Client and worker must agree; the recorded ContentType makes a mismatch fail loudly.

func WithDefaultMaxAttempts

func WithDefaultMaxAttempts(n int) ClientOption

WithDefaultMaxAttempts sets the retry budget applied to runs that don't set their own (default 5).

type DecodeError

type DecodeError struct {
	Name string
	Err  error
}

DecodeError reports a payload that would not round-trip through the codec — a recorded output that no longer decodes into today's type, or an unserializable value. Always terminal: retrying deterministic corruption is a hazard, not resilience. Fix: versioned names for breaking type changes ("charge-payment.v2").

func (*DecodeError) Error

func (e *DecodeError) Error() string

func (*DecodeError) Unwrap

func (e *DecodeError) Unwrap() error

type Handle

type Handle[O any] struct {
	// contains filtered or unexported fields
}

Handle observes one run.

func HandleFor

func HandleFor[O any](c *Client, runID string) *Handle[O]

HandleFor re-attaches to an existing run — after a process restart, or from a different process entirely.

func Start

func Start[I, O any](ctx context.Context, c *Client, a *ActivityFn[I, O], in I, opts ...StartOption) (*Handle[O], error)

Start durably submits a run of a: ledger Accept, then queue Enqueue. A nil error means accepted-and-queued — "acknowledged means durable". Start is a free function because Go methods cannot introduce type parameters.

func (*Handle[O]) Result

func (h *Handle[O]) Result(ctx context.Context) (O, error)

Result blocks until the run is terminal and returns its output, polling the ledger with backoff. A failed run returns *RunFailedError. Backends with the ledger.RunGetter capability are polled with a point read; others fall back to Load (which also prefetches the record history every poll — implement GetRun on custom ledgers).

func (*Handle[O]) RunID

func (h *Handle[O]) RunID() string

RunID returns the run's ID.

type RunFailedError

type RunFailedError struct {
	RunID   string
	Message string
}

RunFailedError is the terminal failure of a run, surfaced by Handle.Result.

func (*RunFailedError) Error

func (e *RunFailedError) Error() string

type RunInfo

type RunInfo struct {
	RunID   string
	Attempt int
}

RunInfo identifies the run a context is executing under.

func FromContext

func FromContext(ctx context.Context) (RunInfo, bool)

FromContext returns the run identity, if ctx is executing under a run.

type StartOption

type StartOption func(*startCfg)

StartOption configures one submission.

func WithMaxAttempts

func WithMaxAttempts(n int) StartOption

WithMaxAttempts overrides the run's retry budget: the maximum number of FAILED executions (retryable errors, crashes discovered by lease expiry) before the run is terminally failed. Waiting never counts — park/resume cycles and janitor backstop polls are free.

func WithParent

func WithParent(runID string) StartOption

WithParent records lineage (forks and child runs read it; informational in v1).

func WithRunID

func WithRunID(id string) StartOption

WithRunID supplies the run ID — which doubles as the submit idempotency key. Starting an existing ID is a no-op returning a handle to the existing run; natural keys ("order-1234") make resubmission safe by construction.

type Void

type Void = struct{}

Void is the output type for effect-only activities: the (input, output) shape stays uniform, effect-only activities return Void{}.