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.

Docs & landing site · API reference · Guide

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 / WithStoreTimeout	Functional options for serialization, time, lifecycle events, and the store-write timeout.
Handle(name, fn)	Register a workflow function.
Start(ctx, name, runID, in...)	Launch a new run in its own goroutine, with an optional atomic 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. Durable when the store records it; finishes Cancelled.
Shutdown(ctx)	Park every in-flight run and refuse new work, so a graceful stop loses nothing — runs stay incomplete and resume on the next Recover.
Redrive(ctx, runID)	Reset a Stuck run to Pending after shipping a fix, so the next Recover claims it again.

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

A crash or Shutdown parks in-flight runs — they stay Running (or Pending) and resume when a later engine calls Recover; only Cancel produces Cancelled. A run the engine refuses to execute — a determinism panic, a corrupt journal, or a workflow that shrank — becomes Stuck and is excluded from recovery until Redrive re-admits it after a fix.

stateDiagram-v2
    [*] --> Pending: Start()
    Pending --> Running: exec acquires lock
    Running --> Done: workflow returns nil
    Running --> Failed: workflow returns error
    Running --> Cancelled: Cancel()
    Running --> Stuck: 💥 panic, contained to the run
    Running --> Running: 💥 crash / Shutdown + Recover()
    Pending --> Pending: 💥 crash / Shutdown + Recover()
    Stuck --> Pending: Redrive()
    Done --> [*]
    Failed --> [*]
    Cancelled --> [*]

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. A panic inside a run is contained to that run — the run becomes Stuck and the process keeps serving its other runs.
• 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

• Variables
• 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) Redrive(ctx context.Context, runID string) error
  • func (e *Engine) Shutdown(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
  • func WithStoreTimeout(d time.Duration) 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

Variables

var ErrEngineClosed = errors.New("rerun: engine is shut down")

ErrEngineClosed is returned by engine methods called after Shutdown.

var ErrRunExists = errors.New("rerun: run already exists")

ErrRunExists reports a Create for a run ID the store already has — the signal Start callers use to dedupe retried requests.

var ErrSeqConflict = errors.New("rerun: journal sequence conflict")

ErrSeqConflict reports an Append at a (run, seq) position that is already journaled: another worker owns the run (a lost lease) or the code shrank. The engine parks the run when it sees this.

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 — a doStep under the hood (Do itself would reject the reserved result tag) — 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
}

Codec is the serialization seam: how a step's result and a run's input are encoded to and decoded from the journal. The default is JSON; override it with WithCodec (for example to compress or encrypt payloads).

type Engine

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

Engine orchestrates runs: it owns the workflow registry, the swappable seams (store, codec, clock, observer), and — since runs must outlive the contexts of whoever starts them — the root context every run descends from. Only Shutdown ends that root, and it ends it with a "parked" cause so in-flight runs stop without being marked terminal.

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. When the store can record cancels durably, the request is recorded FIRST — so a crash between this call and the run's terminal write cannot resurrect the run: the next claim sees the request and finishes it Cancelled. Then, if the run is executing in this process, its context unwinds immediately. With neither a local run nor a Canceller store, there is nothing to cancel with, and that is 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) Redrive

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

Redrive resets a Stuck run to Pending so the next Recover (or, from M2, the dispatcher) claims it again — the operator's lever after shipping a fix. M1 note: this resets unconditionally and does not clear a durable cancel request, so a run that was ever cancel-requested is re-finished Cancelled at the next claim rather than re-run (start a fresh run ID in that case). M2's Inspector adds status verification and cancel-aware redrive.

func (*Engine) Shutdown

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

Shutdown parks the engine: the dispatcher-facing methods start refusing, every run's context is cancelled with errParked (runs unwind, stay incomplete in the store, and resume on a later engine), and Shutdown waits for the run goroutines to return or ctx to expire.

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.

func WithStoreTimeout

func WithStoreTimeout(d time.Duration) Opt

WithStoreTimeout bounds every journal and status write. The write context is detached from the run's cancellation on purpose: a cancel must never be able to lose the record of work that already happened. The default is 30s.

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
	Input    []byte
}

Run is the metadata for a single workflow instance. Input is the optional seed passed to Start, persisted atomically with the run's creation so a crash can never separate a run from its input.

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 is the initial state: a run created but not yet claimed and executed.
	Pending Status = iota
	// Running marks a run currently leased and executing (or replaying its journal).
	Running
	// Done is terminal: the workflow returned without error.
	Done
	// Failed is terminal: the workflow returned an error, recorded in the journal.
	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
	// Stuck marks a run the engine refuses to execute — a determinism panic, a
	// corrupt journal, a workflow that shrank. It is excluded from Incomplete so
	// nothing claims it again automatically; Redrive re-admits it after the
	// operator ships a fix. Appended so values stay stable.
	Stuck
)

func (Status) String

func (s Status) String() string

String renders the status name (Pending, Running, Done, Failed, Cancelled, Stuck) for logs and errors; an unknown value renders as Status(n).

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

Error renders the failed step's tag and message.

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.