rerun

README

██████╗ ███████╗██████╗ ██╗   ██╗███╗   ██╗
██╔══██╗██╔════╝██╔══██╗██║   ██║████╗  ██║
██████╔╝█████╗  ██████╔╝██║   ██║██╔██╗ ██║
██╔══██╗██╔══╝  ██╔══██╗██║   ██║██║╚██╗██║
██║  ██║███████╗██║  ██║╚██████╔╝██║ ╚████║
╚═╝  ╚═╝╚══════╝╚═╝  ╚═╝ ╚═════╝ ╚═╝  ╚═══╝

⟲ Durable execution for Go  ·  crash · restart · resume

A multi-step process that runs to completion — even when the machine crashes halfway through and restarts hours later. It resumes from where it left off instead of starting over.

Do · Sleep · Recover — that's the whole API.

---

The thirty-second pitch

Completed steps are replayed from a journal rather than re-executed, so with idempotent steps there are no double charges, no skipped steps, and no half-finished state left behind.

It's the guarantee you'd otherwise reach for Temporal to get — without the cluster. No servers, no queues, no YAML. rerun is the core idea, not the platform around it: under 1000 lines of engine, a pure-Go SQLite default (zero CGO via modernc.org/sqlite), and zero third-party dependencies in the core.

e.Handle("checkout", func(w *rerun.W) error {
	txn, err := rerun.Do(w, "charge", func(ctx context.Context) (string, error) {
		return chargeCard(ctx, order) // runs once, ever — even across crashes
	})
	if err != nil {
		return err
	}

	rerun.Sleep(w, 24*time.Hour) // durable: survives restarts, skipped on replay

	_, err = rerun.Do(w, "receipt", func(ctx context.Context) (any, error) {
		return nil, sendReceipt(ctx, txn)
	})
	return err
})

If the process dies after charge but before receipt, the next boot's Recover replays the journal: charge returns its stored result without re-charging, the Sleep is skipped because its deadline already passed, and execution continues into receipt.

New to durable execution? docs/durable-execution.md is a concept-to-code tour — the why, the one rule, and how every idea maps to the source.

How it works

A workflow is an ordinary Go function. Each step is wrapped in Do, which makes a single decision — have I already done this?

flowchart TD
    A(["Do(w, tag, fn)"]) --> B{Replaying<br/>and a journal entry<br/>exists for this tag?}
    B -->|yes| C[Return the journaled result<br/>skip the work entirely]
    B -->|no| D[Execute fn live]
    D --> E[Marshal &amp; append result<br/>to the journal]
    E --> F[Notify observer]
    C --> G([Advance to next step])
    F --> G

Recovery is just running the function again. Steps that completed before the crash replay instantly from the journal; the first step without an entry executes for real; everything after it runs forward normally. The function is written once, as if crashes didn't exist, and the engine makes it crash-proof by recording and replaying.

The thing you persist is not which step you reached, but the result every completed step produced. With the results in hand, recovery isn't guesswork.

Mental models

Four ways to hold the idea in your head. Pick whichever one sticks.

1 · The journal is the source of truth, the code is a cursor. Your function isn't the state — the journal is. The function is just a pointer that walks the journal forward, executing only past its end.

flowchart LR
    subgraph J["Journal (durable)"]
      direction LR
      s0["seq 0<br/>create-account<br/>✓ acct_1"]
      s1["seq 1<br/>charge<br/>✓ txn_2"]
      s2["seq 2<br/>·"]
    end
    s0 --> s1 --> s2
    cursor(["▲ cursor — replay up to here,<br/>then run live from here on"]) -.-> s2

2 · A crash just rewinds the tape; replay fast-forwards it. Restarting re-runs the function from the top, but completed steps return instantly from the journal — no card re-charged, no email re-sent — until execution reaches the first step the crash never recorded.

sequenceDiagram
    autonumber
    participant W as Workflow fn
    participant E as Engine
    participant J as Journal
    Note over W,J: First run
    W->>E: Do("charge")
    E->>J: append txn_2
    W--xE: 💥 crash before "receipt"
    Note over W,J: Recover() → run the same fn again
    W->>E: Do("charge")
    E->>J: entry exists → return txn_2 (no live charge)
    W->>E: Do("receipt")
    E->>J: no entry → execute live, append, finish ✓

3 · Do is the membrane between two worlds. Inside a Do, the messy non-deterministic world — clocks, randomness, network — runs once and gets frozen into the journal. Outside, the workflow body must be a deterministic skeleton, replayable byte-for-byte.

flowchart LR
    subgraph D["Deterministic skeleton (replayable)"]
      direction TB
      ctrl["control flow<br/>step order &amp; tags"]
    end
    subgraph N["Non-deterministic world"]
      direction TB
      io["time · random · network · I/O"]
    end
    ctrl -- "Do(tag, fn)" --> io
    io -- "result journaled once" --> ctrl

4 · One goroutine per run, parked for free. Each run drives a plain top-to-bottom function on its own goroutine. A Sleep or a slow step just blocks that goroutine cheaply, so thousands of long-lived workflows sit idle at once without a thread each.

Install

go get github.com/sylvester-francis/rerun

The core engine needs only Go 1.18+ (generics power the type-safe Do[T]). The bundled pure-Go SQLite and Postgres backends pull the module's minimum to Go 1.25; if you need a lower floor, use the in-memory store or your own Store and those backends never enter your build.

Quick start

package main

import (
	"context"
	"time"

	"github.com/sylvester-francis/rerun"
	"github.com/sylvester-francis/rerun/sqlite"
)

func main() {
	ctx := context.Background()

	// Swap sqlite.New(path) for postgres.New(dsn) to run across processes —
	// the workflow code below is identical against any Store.
	e := rerun.New(sqlite.New("rerun.db"))

	e.Handle("checkout", func(w *rerun.W) error {
		orderID, _ := rerun.Input[string](w) // the seed passed to Start, replayed on recovery

		txn, err := rerun.Do(w, "charge", func(ctx context.Context) (string, error) {
			return chargeCard(ctx, orderID)
		})
		if err != nil {
			return err
		}

		if err := rerun.Sleep(w, 24*time.Hour); err != nil {
			return err
		}

		_, err = rerun.Do(w, "receipt", func(ctx context.Context) (any, error) {
			return nil, sendReceipt(ctx, txn)
		})
		return err
	})

	e.Recover(ctx)                                  // resume anything mid-flight before this boot
	e.Start(ctx, "checkout", "run-1", "order-4711") // new run "run-1" seeded with an order ID
}

The API surface

The whole surface a user touches — small because the idea is small:

Symbol	What it's for
New(store, ...Opt)	Build an engine over a Store.
WithCodec / WithClock / WithObserver	Functional options for serialization, time, and lifecycle events.
Handle(name, fn)	Register a workflow function.
Start(ctx, name, runID, in...)	Launch a new run in its own goroutine, with an optional journaled input.
Recover(ctx)	Re-launch every incomplete run after a restart.
Do[T](w, tag, fn)	Run a step once; return its journaled value on replay.
DoTimeout[T](w, tag, d, fn)	A Do with a per-step deadline; the timeout outcome is journaled.
Sleep(w, d)	A durable delay that survives restarts and is skipped on replay.
Retry[T](w, tag, policy, fn)	Retry a step with durable backoff; each attempt is its own journaled step.
Input[T](w) / Return[T](w, v) / Result[T](ctx, e, id)	Pass a value into a run and hand one back, both journaled.

For multi-run and multi-process workloads, a few more primitives build on the same journal:

Symbol	What it's for
Wait[T](w, name) / Deliver(ctx, runID, name, v)	Block on an external event (an approval, a webhook) and journal it, so it survives a crash. Needs a Signaler store.
Version(w, changeID, min, max)	Pin an in-flight run to its original code path so a deploy that changes the workflow doesn't break it.
Cancel(ctx, runID)	Stop a run — instantly if it's in this process, or cross-process (eventual) with WithCancelPoll + a Canceller store. It finishes Cancelled.

Everything else is an interface a backend implements, or an internal detail.

The one rule: determinism

Replay matches journaled steps to your code by tag and order. Your workflow must issue the same sequence of Do/Sleep calls, with the same tags, every time it runs with the same inputs. Anything non-deterministic — the clock, a random number, a read whose result steers a branch — must be captured inside a Do so its value is journaled and replayed, not recomputed.

// ✗ Trap: the branch can differ between the original run and the replay.
if time.Now().Hour() < 12 {
	rerun.Do(w, "morning", ...)
} else {
	rerun.Do(w, "evening", ...)
}

// ✓ Fix: journal the decision, then branch on the recorded value.
morning, _ := rerun.Do(w, "is-morning", func(ctx context.Context) (bool, error) {
	return time.Now().Hour() < 12, nil
})
if morning {
	rerun.Do(w, "morning", ...)
} else {
	rerun.Do(w, "evening", ...)
}

The engine enforces this rather than hope for it: if a Do presents a tag that doesn't match the journal at that position, it panics with the exact position and both tags — a determinism bug fails loudly at the first divergence instead of quietly producing wrong results.

Errors are results too. A step that returns "card declined" on the first run reproduces that same error on replay — it does not re-run the charge hoping for a different answer.

Patterns fall out of the primitive

Do is the only primitive; useful shapes are just loops and conditionals over it.

Retry — each attempt is its own journaled step:

func chargeWithRetry(w *rerun.W) (string, error) {
	var last error
	for attempt := 0; attempt < 3; attempt++ {
		txn, err := rerun.Do(w, fmt.Sprintf("charge:attempt-%d", attempt),
			func(ctx context.Context) (string, error) { return charge(ctx) })
		if err == nil {
			return txn, nil
		}
		last = err
	}
	return "", fmt.Errorf("charge failed after retries: %w", last)
}

On replay the whole attempt sequence is reproduced from the journal without calling the processor again.

Beyond the basics

The single-process engine extends to the hard problems of durable execution without changing its shape — each is the same insight again (anything nondeterministic becomes a journaled step), and each ships with a runnable example.

Problem	Mechanism	Example
Multi-process execution	A non-blocking try-lock lease on Guarder; the postgres backend uses pg_try_advisory_lock on a dedicated connection for exactly-once dispatch across machines.	examples/workers
Durable timers	Sleep journals the absolute deadline; on recovery it waits only the remainder, recomputed from the journal.	examples/durabletimer
Signals & external events	Wait[T] is a Do whose value arrives from an external mailbox (Signaler); Deliver deposits it, and it survives a crash because it's journaled.	examples/signals
Versioning across deploys	Version journals the code-path version so in-flight runs replay their original branch while new runs take the new one.	examples/versioning

Design

Dependencies point one way only — inward, toward abstractions. The engine knows nothing about SQLite; SQLite knows nothing about the engine. Both meet at the Store interface.

flowchart TD
    User["Your workflow code"] --> Engine

    subgraph Core["rerun — engine + interfaces (zero third-party deps)"]
      direction TB
      Engine["Engine<br/><i>orchestrates runs</i>"]
      W["W<br/><i>replays the journal</i>"]
      Engine --> W
      Engine -. "uses" .-> Store([Store])
      Engine -. "uses" .-> Codec([Codec])
      Engine -. "uses" .-> Clock([Clock])
      Engine -. "uses" .-> Observer([Observer])
    end

    Store -.->|implemented by| SQLite["sqlite<br/><i>pure-Go, single node</i>"]
    Store -.->|implemented by| Mem["internal/memstore<br/><i>in-memory default</i>"]
    Store -.->|implemented by| PG["postgres<br/><i>multi-process lease</i>"]
    Store -.->|implemented by| Own["your backend<br/><i>…</i>"]
    Codec -.->|default| JSON["jsonCodec"]
    Clock -.->|default| Wall["wall clock"]
    Observer -.->|default| Noop["no-op"]

Arrows are dependencies. Nothing inside Core points outward at a concrete implementation — adapters depend on the interfaces, never the reverse. The Postgres backend was added exactly this way: a new package satisfying Store, zero lines of engine source changed. A protobuf codec or a metrics observer slots in the same way.

Backends and extensions never import the engine; the engine imports no adapter. Each piece changes for exactly one reason:

• Engine orchestrates runs (one goroutine per run — thousands park cheaply on a Sleep).
• W replays the journal.
• Store persists logs and run metadata.
• Codec serializes step results (JSON by default).
• Clock tells time (wall clock by default; injectable for tests).
• Observer receives lifecycle events for logging and metrics (no-op by default).

Pluggable storage, segmented by need

Store composes three small interfaces so consumers depend only on what they use:

Interface	Methods	Who needs it
Writer	Create, Append, Finish	the hot path
Reader	LoadLogs, Incomplete	a monitoring dashboard, read-only
Guarder	Acquire	mutual exclusion; a test double makes it a no-op

Any Store is drop-in — SQLite, Postgres, or in-memory — and the engine can't tell the difference. The contract suite ships as a public package (storetest) so anyone can write a backend and prove it correct against the same bar the in-tree backends meet:

func TestSQLiteStore(t *testing.T) {
	storetest.RunStoreContract(t, func() rerun.Store {
		return sqlite.New(t.TempDir() + "/test.db")
	})
}

Configuration

e := rerun.New(store,
	rerun.WithCodec(myCodec),
	rerun.WithClock(myClock),
	rerun.WithObserver(myObserver),
)

Run lifecycle

Every run moves through four states. Recover re-launches exactly those left in Running or Pending when the process died.

stateDiagram-v2
    [*] --> Pending: Start()
    Pending --> Running: exec acquires lock
    Running --> Done: workflow returns nil
    Running --> Failed: workflow returns error
    Running --> Running: 💥 crash + Recover()
    Pending --> Pending: 💥 crash + Recover()
    Done --> [*]
    Failed --> [*]

When it panics vs. returns an error

• Panics for programmer errors that can't be recovered: non-determinism (tag mismatch), journal corruption, duplicate workflow registration.
• Errors for operational failures: the store is unavailable, the context is cancelled, a step returns an error.

Repository layout

rerun/
├── store.go            Run, Log, Status, the Store/Writer/Reader/Guarder interfaces
├── codec.go            serialization seam, JSON by default
├── clock.go            time seam, wall clock by default
├── hooks.go            Observer seam, no-op by default
├── engine.go           the engine: registry, New, Handle, options
├── workflow.go         Do, replayStep, liveStep, Sleep
├── run.go              Start, exec, Recover
├── input.go            Input[T], the run's journaled seed
├── signal.go           Signaler, Deliver, Wait[T] — steps whose value comes from outside
├── version.go          Version — pin in-flight runs across a deploy
├── errors.go           StepError, so errors survive replay
├── internal/memstore.go   in-process store (the default and a test aid)
├── storetest/             importable Store contract suite for any backend
├── sqlite/sqlite.go       persistent single-node backend (pure Go, zero CGO)
├── postgres/postgres.go   multi-process backend (advisory-lock lease)
├── tools/mutate/          dependency-free mutation tester
└── examples/              skeleton · recover · durablesleep · capstone
                           workers · durabletimer · signals · versioning

Testing

go test -race ./...   # the whole suite, race-clean
go vet ./...
make mutate           # mutation testing: known faults are killed
make pg-test          # Postgres store contract against an ephemeral container (needs Docker)

The same Store contract runs against all three backends — in-memory, SQLite (a real file), and Postgres (a real database) — so the durability claim is proven, not asserted.

Guarantees & non-goals

rerun is v0.x and unstable — the API may change between minor versions.

What it guarantees: runs are durable and resumable. Side effects are at-least-once, and exactly-once only when your steps are idempotent. A step repeats only if the process dies in the narrow window after its side effect runs and before its journal entry commits — which is why production steps (a card charge, an email) are written to be idempotent. Any claim of plain "exactly once" would be dishonest.

Non-goals (deliberately not provided):

• No distributed scheduler. A sleeping run parks a cheap goroutine; that scales to thousands, not to a durable-timer service polling millions. Incomplete plus a due-before query is where that would attach — a planned v0.2.
• Not a Temporal replacement. rerun is the core idea — journal and replay — not the platform (UI, namespaces, cross-language SDKs, activity workers) around it.

Retries with durable backoff (Retry), per-step timeouts (DoTimeout), typed results (Return/Result), durable signals, and run cancellation — in-process, or cross-process (eventual) via WithCancelPoll — are all built in. See docs/using-rerun.md.

Documentation

• docs/using-rerun.md — how to wire rerun into a real application: input/result, retries, timeouts, cancellation, signals, choosing a backend, and the rules you must follow.
• docs/durable-execution.md — a concept-to-code tour of durable execution and this codebase; the why behind the how.
• CONTRIBUTING.md · SECURITY.md · CHANGELOG.md
• API reference on pkg.go.dev.

License

Apache License 2.0 © 2026 Sylvester Francis

Derivative works must retain the attribution notices in NOTICE, per Section 4(d) of the license.

Documentation

Overview

Package rerun is a lightweight durable execution library. Workflows survive crashes, restarts, and deploys by journaling every step to a pluggable store; on recovery the engine replays the journal, skipping completed steps, and resumes exactly where it left off.

The whole API fits on a napkin. Do runs a step once and journals its result; Sleep is a durable delay; Recover resumes every unfinished run after a restart. A workflow is an ordinary Go function.

e := rerun.New(sqlite.New("rerun.db"))
e.Handle("signup", func(w *rerun.W) error {
	id, _ := rerun.Do(w, "create-account", func(ctx context.Context) (string, error) {
		return createAccount(ctx)
	})
	rerun.Sleep(w, time.Hour)
	rerun.Do(w, "welcome-email", func(ctx context.Context) (struct{}, error) {
		return struct{}{}, sendEmail(ctx, id)
	})
	return nil
})
e.Start(ctx, "signup", "user-42")
// ...and after a restart, in the same process:
e.Recover(ctx)

Determinism

A workflow body must be deterministic in its control flow and its step tags. Anything nondeterministic — the current time, a random number, a read whose result steers a branch — must be captured inside a Do so its value is journaled and replayed, never recomputed. On replay, a tag that diverges from the journal panics, loudly and early, because a silent divergence corrupts every later step.

Guarantees

rerun is durable and resumable; at-least-once for side effects; exactly-once only when steps are idempotent. A step repeats only if the process dies in the narrow window after its side effect runs and before its journal entry commits, which is why production steps are written to be idempotent. See the README's Non-goals for what rerun deliberately does not do.

Backends

Persistence is the Store seam, and swapping a backend changes no engine code: internal.MemStore for tests and single-process programs, package sqlite for a single persistent node, package postgres for multi-process execution behind an advisory-lock lease. Any Store that passes storetest.RunStoreContract is a drop-in.

Index

• func Do[T any](w *W, tag string, fn func(context.Context) (T, error)) (T, error)
• func DoTimeout[T any](w *W, tag string, d time.Duration, fn func(context.Context) (T, error)) (T, error)
• func ExpBackoff(base, max time.Duration) func(int) time.Duration
• func FixedBackoff(d time.Duration) func(int) time.Duration
• func Input[T any](w *W) (T, error)
• func Result[T any](ctx context.Context, e *Engine, runID string) (T, error)
• func Retry[T any](w *W, tag string, p RetryPolicy, fn func(context.Context) (T, error)) (T, error)
• func Return[T any](w *W, v T)
• func Sleep(w *W, d time.Duration) error
• func Version(w *W, changeID string, min, max int) int
• func Wait[T any](w *W, name string) (T, error)
• type Canceller
• type Clock
• type Codec
• type Engine
  • func New(s Store, opts ...Opt) *Engine
  • func (e *Engine) Cancel(ctx context.Context, runID string) error
  • func (e *Engine) Deliver(ctx context.Context, runID, name string, payload any) error
  • func (e *Engine) Handle(name string, fn Func)
  • func (e *Engine) Recover(ctx context.Context) error
  • func (e *Engine) Start(ctx context.Context, workflow, runID string, in ...any) error
• type Func
• type Guarder
• type Log
• type Observer
• type Opt
  • func WithCancelPoll(d time.Duration) Opt
  • func WithClock(c Clock) Opt
  • func WithCodec(c Codec) Opt
  • func WithObserver(o Observer) Opt
• type Reader
• type RetryPolicy
• type Run
• type Signaler
• type Status
  • func (s Status) String() string
• type StepError
  • func (e *StepError) Error() string
• type Store
• type W
• type Writer

Functions

func Do

func Do[T any](w *W, tag string, fn func(context.Context) (T, error)) (T, error)

Do runs a step exactly once. On the first run it executes fn, journals the result, and returns it; on replay it returns the journaled result without executing fn. It panics if the tag diverges from the journal: that means the workflow is no longer deterministic, and every later positional match would be meaningless.

func DoTimeout

func DoTimeout[T any](w *W, tag string, d time.Duration, fn func(context.Context) (T, error)) (T, error)

DoTimeout is Do with a per-step deadline: fn runs under a context that cancels after d. Whatever fn returns — a value, its own error, or a deadline error if it honors the context — is journaled as the step's outcome, so replay is deterministic and a timed-out step is never silently retried into a success.

func ExpBackoff

func ExpBackoff(base, max time.Duration) func(int) time.Duration

ExpBackoff doubles the wait each attempt, starting at base and capped at max.

func FixedBackoff

func FixedBackoff(d time.Duration) func(int) time.Duration

FixedBackoff waits the same duration before every retry.

func Input

func Input[T any](w *W) (T, error)

Input returns the value passed to Start for this run, decoded as T. It is journaled as the run's seed at Start, so replay sees the same value every time; a run started without input yields the zero value.

func Result

func Result[T any](ctx context.Context, e *Engine, runID string) (T, error)

Result reads the value a run recorded with Return, decoded as T. If the run failed, it returns the terminal error the workflow returned; if the run has recorded no result yet (still running, or never called Return), it returns an error saying so. It reads through the store, so it works from any process.

func Retry

func Retry[T any](w *W, tag string, p RetryPolicy, fn func(context.Context) (T, error)) (T, error)

Retry runs fn until it succeeds or MaxAttempts is reached. Each attempt is its own journaled step (tag:attempt-N), and the wait between attempts uses the durable Sleep — so the backoff itself survives a crash, and on replay the whole attempt-and-wait sequence is reproduced from the journal without calling fn again. It returns the first success, or the last error wrapped after the final attempt.

func Return

func Return[T any](w *W, v T)

Return records the run's result. Call it once, at a deterministic point (typically the workflow's last act). It is journaled like any step — an ordinary Do under the hood — so replay reproduces it without re-running.

func Sleep

func Sleep(w *W, d time.Duration) error

Sleep is a durable delay. It journals an absolute deadline as an instant step, then waits only the time still remaining until that deadline, recomputed on every run. A restart before the deadline waits out the remainder; a restart after it returns immediately. The wait itself is never journaled — it is a function of the deadline and the clock, so it is always recomputed, never stored.

func Version

func Version(w *W, changeID string, min, max int) int

Version pins a run to a code path so in-flight runs survive a deploy that changes the workflow. On the first run it journals max, the newest version this build knows; on replay it returns the journaled value, so a run started before the change keeps taking its original branch while new runs take the new one. Branch on the returned int, and give every change its own changeID.

It panics if the journaled version falls outside [min, max]: that means a rollback to code too old to honor a version some run already committed to, and failing loud beats silently taking the wrong branch.

func Wait

func Wait[T any](w *W, name string) (T, error)

Wait blocks the workflow until a signal named name arrives, then journals and returns its payload. On replay it returns the journaled payload without waiting, so a workflow that waited three days for an approval resumes with the approval already in hand and never waits again. It is an ordinary Do whose function happens to block on a mailbox instead of computing — which is why the received value is journaled and replayed like any other step result.

Types

type Canceller

type Canceller interface {
	RequestCancel(ctx context.Context, runID string) error
	CancelRequested(ctx context.Context, runID string) (bool, error)
}

Canceller is an optional Store capability: a durable record that a run should be cancelled. A backend that implements it lets Cancel reach a run executing in another process (see WithCancelPoll). Like Signaler, it is kept separate so the core Store contract stays minimal — a store without it supports only in-process cancellation.

type Clock

type Clock interface {
	Now() time.Time
	After(d time.Duration) <-chan time.Time
}

Clock abstracts time so workflows can be driven by a fake clock in tests.

type Codec

type Codec interface {
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

type Engine

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

Engine orchestrates runs: it owns the workflow registry and the swappable seams (store, codec, clock, observer).

func New

func New(s Store, opts ...Opt) *Engine

New builds an Engine over a Store, applying defaults (JSON codec, wall clock, no-op observer) before any options.

func (*Engine) Cancel

func (e *Engine) Cancel(ctx context.Context, runID string) error

Cancel stops a run. If the run is executing in this process, its context is cancelled immediately, so a parked Sleep or a ctx-respecting step unwinds and the run finishes Cancelled. Otherwise, if the store is a Canceller, the request is recorded durably and a worker running the run elsewhere picks it up on its next poll (see WithCancelPoll) — cancellation is then eventual. With neither a local run nor a Canceller store, it returns an error.

func (*Engine) Deliver

func (e *Engine) Deliver(ctx context.Context, runID, name string, payload any) error

Deliver records an external event for a run. Call it from outside the workflow — an HTTP handler, a webhook, another service. Delivery may arrive before the workflow reaches its Wait; the event queues in the mailbox until Wait consumes it, so it is never lost.

func (*Engine) Handle

func (e *Engine) Handle(name string, fn Func)

Handle registers a workflow body under a name. It panics on a duplicate: two workflows sharing a name is a build-time programmer error, not a runtime condition to recover from.

func (*Engine) Recover

func (e *Engine) Recover(ctx context.Context) error

Recover relaunches every incomplete run through exec, resuming workflows a process restart left in flight.

func (*Engine) Start

func (e *Engine) Start(ctx context.Context, workflow, runID string, in ...any) error

Start creates a Pending run and launches it in its own goroutine. It returns as soon as the run is durably created, so the caller is never blocked on execution and one process can drive many thousands of runs. An optional input is journaled as the run's seed and read back inside the workflow with Input.

type Func

type Func func(w *W) error

Func is a registered workflow body.

type Guarder

type Guarder interface {
	Acquire(ctx context.Context, runID string) (release io.Closer, acquired bool, err error)
}

Guarder leases a run to one worker. Acquire is a non-blocking try-lock: acquired is false when another worker already holds the run, so recovery across many processes never replays the same run twice. A single-process store that never contends may always report acquired.

type Log

type Log struct {
	Seq     int
	Tag     string
	Payload []byte
	Err     string
	At      time.Time
}

Log is one immutable journal entry: the result a completed step produced. Persisting the result a step produced — not merely the fact that a step ran — is what lets recovery replay a step instead of executing it a second time.

type Observer

type Observer interface {
	OnStart(r Run)
	OnStep(runID string, l Log)
	OnFinish(runID string, s Status)
}

Observer receives lifecycle events for logging and metrics. Events are passed by value so observers cannot mutate engine state.

type Opt

type Opt func(*Engine)

Opt configures an Engine at construction time.

func WithCancelPoll

func WithCancelPoll(d time.Duration) Opt

WithCancelPoll enables cross-process cancellation: each running run checks the store for a cancel request every d (the store must implement Canceller). Zero, the default, disables the check so a run parks for free and only in-process Cancel applies. Cross-process cancellation is eventual, within one interval.

func WithClock

func WithClock(c Clock) Opt

WithClock overrides the time seam.

func WithCodec

func WithCodec(c Codec) Opt

WithCodec overrides the serialization seam.

func WithObserver

func WithObserver(o Observer) Opt

WithObserver overrides the observability seam.

type Reader

type Reader interface {
	LoadLogs(ctx context.Context, runID string) ([]Log, error)
	Incomplete(ctx context.Context) ([]Run, error)
}

Reader is the cold path: loading a journal to replay it and finding the runs a restart left unfinished.

type RetryPolicy

type RetryPolicy struct {
	MaxAttempts int
	Backoff     func(attempt int) time.Duration
}

RetryPolicy configures Retry: how many attempts to make and how long to wait before each retry (Backoff is called with the zero-based attempt that just failed). A nil Backoff means retry immediately.

type Run

type Run struct {
	ID       string
	Workflow string
	Status   Status
	Created  time.Time
}

Run is the metadata for a single workflow instance.

type Signaler

type Signaler interface {
	PushSignal(ctx context.Context, runID, name string, payload []byte) error
	PopSignal(ctx context.Context, runID, name string) (payload []byte, ok bool, err error)
}

Signaler is an optional Store capability: a per-run mailbox for external events. A backend that implements it enables Wait and Deliver. Keeping it a separate interface — not a method on Store — is interface segregation: a backend with no use for signals simply does not implement it.

type Status

type Status int

Status is a run's lifecycle state.

const (
	Pending Status = iota
	Running
	Done
	Failed
	// Cancelled is terminal, like Done and Failed. It is appended after Failed
	// so the existing values keep their numbers in journals and databases.
	Cancelled
)

func (Status) String

func (s Status) String() string

type StepError

type StepError struct {
	Tag string
	Msg string
}

StepError carries a failed step's tag and message so the failure survives being journaled and is reconstructed identically on replay.

func (*StepError) Error

func (e *StepError) Error() string

type Store

type Store interface {
	Writer
	Reader
	Guarder
}

Store is the whole persistence seam. It is split three ways so a consumer can depend on only the face it uses — a dashboard on Reader, a lease manager on Guarder, a log shipper on Writer — and so any backend is a drop-in the engine cannot tell apart.

type W

type W struct {
	RunID string
	// contains filtered or unexported fields
}

W is the handle threaded through a workflow. It carries the loaded journal and a cursor so Do can tell replay from live execution.

type Writer

type Writer interface {
	Create(ctx context.Context, r Run) error
	Append(ctx context.Context, runID string, l Log) error
	Finish(ctx context.Context, runID string, s Status) error
}

Writer is the hot path: creating a run and appending step results as they complete.