durable

package module
v0.1.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 16, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

README

durable-go

CI Security Release Go Reference License

Lightweight, embeddable durable task execution for Go.

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

Releases follow Semantic Versioning; see the latest release.

Features

  • Engine API — NewEngine, RegisterTask, RunTask, RunStep in a single package.
  • Async fan-out — RunStep starts work and returns immediately, matching RunTask; call several before Get-ing any of them.
  • First-of-N — StepRun.Done() exposes a read-only channel so you can select across handles and react to whichever finishes first.
  • Memoized steps — completed steps replay from the journal; they are not run again. Failed steps replay their stored error instead of re-running fn.
  • Pending steps — return ErrStepPending and complete later via CompleteStep (human approval, webhooks). Suspends only that step's Get, not the whole task — sibling steps keep running.
  • Cancellation — CancelRun persists a durable cancel signal and cancels the run's ctx immediately if it is executing in this process; every RunStep call, including on a later resume, fails fast with ErrRunCancelled instead of re-running fn.
  • Fire-and-forget runs — RunTask returns immediately; TaskRun.Get waits; RunID() is available at once.
  • Step observability — GetStep / LoadSteps for current state, WatchSteps for a live/historical event stream (including STARTED).
  • Timeouts and retries — engine / task / run / step options. Retries default to 0 (opt-in).
  • Panic recovery — task and step panics are recorded and returned as errors.
  • Auto-purge — optional background cleanup of old completed and failed runs.
  • Flexible execution — tasks as durable.Func closures or structs with Exec.

Pre-1.0. v0.1.0–v0.1.3 are retracted. Start from v0.1.4. Journals from v0.1.3 still load; v0.1.0–v0.1.2 are not compatible — upgrade those only between clean runs.

Why durable-go

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

durable-go takes a zero-infra, in-process approach: a single Go library with a filesystem journal running inside your application process. Instead of replaying entire function call graphs from an external orchestrator, durable-go memoizes individual step results. On resume the task runs again from the top; completed steps return the cached result. There is no replay-determinism sandbox. The journal is files, not SQLite: no schema migrations when the library changes, and no connection/busy-lock handling for a single-process writer. See Use cases for more places where durable-go is a perfect fit.

One writer per dataDir. NewEngine takes an exclusive OS flock on <dataDir>/.lock. Do not open the same directory from two writer processes. Another process can open the same directory with NewReadOnlyEngine (shared lock).

Install

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

Go 1.26.5+. No infrastructure required.

Quick Start

e, err := durable.NewEngine(ctx, "./data", durable.WithLogger(logger))
if err != nil { ... }
defer e.Close()

err = durable.RegisterTask(e, "process-order", durable.Func(
    func(ctx context.Context, s *durable.StepRunner, in OrderInput) (OrderOutput, error) {
        charged, err := durable.RunStep(ctx, s, "charge", in, func(ctx context.Context, in OrderInput) (string, error) {
            return chargeCard(in)
        }).Get(ctx)
        if err != nil {
            return OrderOutput{}, err
        }
        shipped, err := durable.RunStep(ctx, s, "ship", charged, func(ctx context.Context, charged string) (string, error) {
            return scheduleShip(charged)
        }).Get(ctx)
        if err != nil {
            return OrderOutput{}, err
        }
        return OrderOutput{Result: shipped}, nil
    },
))

run := durable.RunTask[OrderInput, OrderOutput](ctx, e, "process-order", "", input)
storeRunID(run.RunID())      // available immediately before Get
output, err := run.Get(ctx)  // block for result

Full example: examples/func-task/.

Struct-based tasks

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

type Job struct {
    DB   *Database
    Mail Mailer
}

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

durable.RegisterTask(e, "notify", &Job{DB: db, Mail: mailer})
run := durable.RunTask[string, string](ctx, e, "notify", "", "42")
out, err := run.Get(ctx)

Full example: examples/struct-task/.

Pending steps

A step suspends itself by returning ErrStepPending. This blocks only that step's Get — sibling steps started before it keep running. An external caller completes it with the token from StepToken(ctx):

approval, err := durable.RunStep(ctx, s, "approve", struct{}{}, func(ctx context.Context, _ struct{}) (Approval, error) {
    token := s.StepToken(ctx)
    sendEmail("manager@co.com", token)
    return Approval{}, durable.ErrStepPending
}).Get(ctx)

// webhook / CLI / another goroutine:
durable.CompleteStep(ctx, e, token, Approval{By: "manager@co.com"})
Fan-out and first-of-N

RunStep starts work and returns immediately — the same shape as RunTask. Start several steps before calling Get on any of them to run them concurrently, then join:

charge := durable.RunStep(ctx, s, "charge", order, func(ctx context.Context, order Order) (string, error) {
    return chargeCard(order)
})
notify := durable.RunStep(ctx, s, "notify", order, func(ctx context.Context, order Order) (string, error) {
    return sendReceipt(order)
})
chargeResult, err := charge.Get(ctx)
if err != nil {
    return Output{}, err
}
notifyResult, err := notify.Get(ctx)

To react to whichever of several steps finishes first, select on Done() instead of blocking on Get:

a := durable.RunStep(ctx, s, "provider-a", req, callProviderA)
b := durable.RunStep(ctx, s, "provider-b", req, callProviderB)
select {
case <-a.Done():
    result, err := a.Get(ctx)
case <-b.Done():
    result, err := b.Get(ctx)
}

Full example: examples/fanout/.

Cancellation

CancelRun persists a durable cancel signal (reusing the same journal plumbing as CompleteStep) and, if the run is executing in this process, cancels the ctx delivered to that run's Task.Exec and every in-flight RunStep call:

run := durable.RunTask[OrderInput, OrderOutput](ctx, e, "process-order", "", input)
// ... later, from another goroutine or request:
if err := e.CancelRun(ctx, "process-order", run.RunID()); err != nil {
    // ErrRunAlreadyFinished if the run already completed or failed.
}
_, err := run.Get(ctx) // errors.Is(err, durable.ErrRunCancelled)
  • Immediate for a running process. A step already blocked on ctx.Done() (e.g. inside waitForSignal after ErrStepPending, or a step function that itself selects on ctx) unblocks right away.
  • Durable across a crash. The cancel signal is written to the journal before this call returns. If the process crashes before the in-process cancellation above takes effect, the next RunTask for that taskID/runID cancels ctx before Task.Exec is invoked at all — every subsequent RunStep call, including a cached/replayed one, returns ErrRunCancelled immediately without running fn.
  • Cooperative, like any Go ctx. RunStep.Get/RunTask.Get return promptly regardless of whether the step's goroutine has exited, but the engine's drain (Close, and any run reaching a terminal state) waits for it to actually return — see rule 4 in Writing tasks.
  • The run ends up StatusFailed (the same terminal status used for Close and timeouts) with TaskInfo.Error equal to durable.ErrRunCancelled.Error(), so callers can tell a deliberate cancel apart from another failure.
Resume

Register tasks after every NewEngine, then resume active runs. Pass the saved runID (or "" to resume the oldest Running/Waiting run for that taskID). Completed steps replay from the journal.

durable.RegisterTask(e, "process-order", ...)
pending, _ := e.ListTasks(ctx, durable.StatusRunning, durable.StatusWaiting)
for _, t := range pending {
    run := durable.RunTask[OrderInput, OrderOutput](ctx, e, t.TaskID, t.RunID, OrderInput{})
    go func() { _, _ = run.Get(ctx) }()
}

RunTask writes input.json on first start. The same runID reloads it; the input argument is ignored.

GetStep and LoadSteps return current state — the latest record per step, in-memory-map order (not sorted). WatchSteps returns history and live updates instead: every STARTED, WAITING, COMPLETED, and FAILED event in the order it was written, either from the beginning (fromOffset 0) or resuming past a previously-seen Offset / ByteOffset. Cancelling the watch does not stop the run. A watch that falls behind may have events dropped (logged as a warning) rather than blocking the run — reconnect with the last Offset/ByteOffset you saw to catch up.

Full example: examples/resume/.

Writing tasks

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

  1. Side effects in RunStep. Do not call an API, write to a database, or publish to a queue in the task body. Wrap that work in durable.RunStep.
  2. Non-deterministic values in RunStep. Do not use time.Now(), UUIDs, or random values in the task body to choose a step ID or a branch. Generate them inside a RunStep so resume sees the same result.
  3. Idempotent steps. A crash can re-run a step after the side effect already happened. Charging a card or sending mail must be safe to do twice (or no-op).
  4. Check ctx to be cancellable. CancelRun, WithStepTimeout, and engine Close only cancel ctx — they cannot forcibly stop a step function. Select on ctx.Done() in any loop or long-running step body, and pass ctx to ctx-aware calls (http.NewRequestWithContext, a database/sql *Context method, etc.). A step that never checks ctx keeps running in the background past cancellation/timeout/Close, and the engine still waits for it to actually return before the run reaches a terminal state.

Also:

  • Unique step IDs — one stable string per step (literals or fmt.Sprintf("step-%d", i)). Reusing an ID panics.
  • Step IDs are the resume key — never rename one. The journal matches records by stepID string only. Renaming a step between deploys orphans the old result: on the next run fn executes again under the new name as if it had never run. Treat a stepID like a database column name, not a display label.
  • Concurrent RunStep calls are safe. Start several steps before Get-ing any of them to fan out; join with Get or select on Done(). A duplicate stepID within one run still panics.
  • One stepID is reserved. RunStep panics if stepID is "\x00cancel" — it is reserved internally for CancelRun's durable signal. Any human-readable stepID you would actually choose is unaffected.
  • JSON results — step and task outputs must be JSON-marshalable. Step input in must be too.
  • One task input — I is a single value, not variadic args. Bundle multiple fields in one struct; a task with no payload uses struct{} and struct{}{}.
  • One step input — in is a single value, not variadic args. Bundle multiple fields in one struct. No payload: struct{} and struct{}{}. Stored for inspect.
  • Bump step version on the next deploy — resume returns the cached result if stepID is unchanged, even when fn or in changed. If this step’s code or params change and in-flight runs must re-execute it, set WithStepVersion to a new string ("1" → "2") or rename the stepID (charge → charge-v2). Same version (or no version) = cache. Side effects on re-run are the caller’s problem (idempotent steps).
  • Same runID to resume — input.json is reloaded; you do not need to pass the original input again.

Examples

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

Example What it shows
examples/resume/ Crash after step 2, resume from cache
examples/func-task/ Closure-style durable.Func
examples/struct-task/ Struct task with injected deps, retries, timeout
examples/fanout/ Concurrent RunStep, Get-all join, ErrStepPending
examples/yaml-task/ YAML file as one task; each YAML step is a RunStep 
# from repo root
go run ./examples/resume/
go run ./examples/func-task/
go run ./examples/struct-task/
go run ./examples/fanout/
go run ./examples/yaml-task/

Inspect CLI

durable-inspect is a read-only viewer for a journal (task list / task get / step list / step get). --dir / -d wins over DURABLE_DIR.

go install github.com/agenticenv/durable-go/cmd/durable-inspect@latest
durable-inspect -d ./data task list

Commands, flags, lookup by name or ID, and the writer-lock behavior: see cmd/durable-inspect/README.md.

Use cases

Match this table to your app. If your work is one process plus a local journal, durable-go is a fit.

Use case Why this library
Single-process agent (in-process loop, agent CLI) Memoize each LLM / tool step so a crash resumes the same run; gate tools with ErrStepPending; inspect progress with WatchSteps / NewReadOnlyEngine. This is how agent-sdk-go uses it by default.
Ops CLI (migrate, import, backup, deploy) Re-run the same command after a crash — completed steps skip; a second process can inspect the journal read-only while the job runs.
Daemon / sidecar on one box On boot, ListTasks + RunTask(runID) resumes in-flight work; webhooks call CompleteStep; CancelRun stops a live run and survives a crash.
Cron / batch / ETL on one machine Skip already-fetched extracts; fan-out parallel source/transform steps and join with Get, or take the first success with select on Done().
Provisioning / install scripts Sequence create/configure/verify as steps so a failed run does not recreate what already succeeded.
Single-process service (orders, payments, reports) Charge, ship, notify as memoized steps; wait on manager approval or a webhook without blocking sibling work; race multiple providers with first-of-N.

Development

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

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

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

License

Apache 2.0

Disclaimer

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

Documentation

Overview

Package durable provides embeddable durable task execution for a single OS process. Register typed tasks, run memoised steps, and resume from a local journal after a crash. A step may return ErrStepPending and complete later via CompleteStep. 

e, err := durable.NewEngine(ctx, "./data")
if err != nil { ... }
defer e.Close()

_ = durable.RegisterTask(e, "process-order", durable.Func(
    func(ctx context.Context, s *durable.StepRunner, in OrderInput) (OrderOutput, error) {
        charged, err := durable.RunStep(ctx, s, "charge", in, func(ctx context.Context, in OrderInput) (string, error) {
            return chargeCard(in)
        }).Get(ctx)
        if err != nil {
            return OrderOutput{}, err
        }
        return OrderOutput{Result: charged}, nil
    },
))

run := durable.RunTask[OrderInput, OrderOutput](ctx, e, "process-order", "", input)
output, err := run.Get(ctx)

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrTaskNotRegistered is returned by RunTask.Get when the taskID has not
	// been registered on this Engine. The registry is in-memory only and must
	// be rebuilt after every NewEngine call.
	ErrTaskNotRegistered = errors.New("durable: task not registered")

	// ErrTaskAlreadyRegistered is returned by RegisterTask when the same
	// taskID is registered twice on one Engine.
	ErrTaskAlreadyRegistered = errors.New("durable: task already registered")

	// ErrRunAlreadyFinished is returned by CompleteStep when the target step
	// is already completed or the run is in a terminal state (completed or failed).
	ErrRunAlreadyFinished = errors.New("durable: run already completed or failed")

	// ErrRunActive is returned by DeleteTaskRun and DeleteTask when the target run
	// (or any run under the task) is currently executing, including while
	// blocked in StatusWaiting.
	ErrRunActive = errors.New("durable: run is currently executing")

	// ErrInvalidRunID is returned when a runID contains path separators or
	// parent-directory references that would escape the data directory.
	ErrInvalidRunID = errors.New("durable: invalid run ID")

	// ErrEngineLocked is returned by NewEngine and NewReadOnlyEngine when
	// dataDir is already held by another engine instance (same process or
	// another OS process) and the lock cannot be acquired before the timeout.
	ErrEngineLocked = errors.New("durable: dataDir locked by another engine")

	// ErrStepPending is returned from a step function to suspend that step
	// until CompleteStep delivers a result for it. The engine writes
	// StepStatusWaiting and blocks that step's Get — not the whole task.
	// Sibling steps started before this one keep running; call Get on them
	// independently or select on their Done channels.
	ErrStepPending = errors.New("durable: step pending external completion")

	// ErrInvalidToken is returned by CompleteStep when the token cannot be
	// decoded into taskID, runID, and stepID.
	ErrInvalidToken = errors.New("durable: invalid step token")

	// ErrRunCancelled is the error stored on a run (TaskInfo.Error, as its
	// string form) and returned by RunStep/Get after CancelRun. It marks
	// the run as StatusFailed for the same reason engine Close and task/run
	// timeouts already do — but with a distinct message so callers can
	// tell a deliberate CancelRun apart from a generic ctx cancellation or
	// deadline. Because TaskInfo.Error is a plain string, matching after a
	// resume requires comparing against ErrRunCancelled.Error(), not
	// errors.Is.
	ErrRunCancelled = errors.New("durable: run was cancelled")
)

Functions

func CompleteStep added in v0.1.2 

func CompleteStep[O any](ctx context.Context, e *Engine, token string, result O) error

CompleteStep delivers result to a suspended step identified by token. The payload is appended as a SignalEntry (durable) before the in-process waiter is signalled, so a crash after this call still resumes on the next RunTask. A second call with the same token is a no-op once the SignalEntry is on disk. Returns ErrRunAlreadyFinished if the step or run is already terminal, and ErrInvalidToken if the token cannot be decoded.

The run's status transition back to StatusRunning is owned by the waiting step itself (see leaveWaiting), not by CompleteStep — with several steps possibly waiting at once, only the step that actually resumes knows whether a sibling is still pending.

Concurrent CompleteStep calls for the same token are serialised on a per-(taskID,runID,stepID) lock distinct from the run-execution lock (which is held for the entire run, including while blocked waiting — locking it here would deadlock). A CompleteStep that loses a race with the run reaching a terminal state may append an orphan SignalEntry that is never read; this is harmless and does not corrupt the journal.

func RegisterTask added in v0.1.2 

func RegisterTask[I, O any](e *Engine, taskID string, task Task[I, O], opts ...TaskOption) error

RegisterTask stores taskID → (closure + config) in the in-memory registry. Must be called before RunTask. I is the single task input type (one struct of fields, or struct{} if the task has no payload). Re-registering the same taskID returns ErrTaskAlreadyRegistered. The registry is not persisted; call this again after every NewEngine. taskID must not contain path separators or ':'.

Types

type Engine added in v0.1.2 

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

Engine is the process-level entry point for durable task execution. Create one per dataDir via NewEngine. Safe for concurrent use.

Always call Close to cancel in-flight runs, release the exclusive flock, stop the purger, and close journal file handles.

func NewEngine added in v0.1.2 

func NewEngine(ctx context.Context, dataDir string, opts ...EngineOption) (*Engine, error)

NewEngine opens or creates dataDir, acquires an exclusive flock on <dataDir>/.lock, and initialises the in-memory task registry. Fails with ErrEngineLocked if another Engine or ReadOnlyEngine holds the directory.

func (*Engine) CancelRun added in v0.1.3 

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

CancelRun requests cancellation of a run. It persists a durable cancel signal — reusing the SignalEntry/CompleteStep journal plumbing via a reserved signal ID — so the intent survives a crash: on the next RunTask for this taskID/runID, the run's ctx is cancelled before Task.Exec is invoked, and every RunStep call fails fast with ErrRunCancelled instead of re-running fn. If the run is currently executing in this process, its ctx is also cancelled immediately.

Cancelling ctx only signals a step function; it cannot forcibly stop one. RunStep.Get and RunTask.Get both return promptly regardless — they select on ctx.Done() independently of whether the step's goroutine has exited. But the engine still waits for that goroutine to actually return before the run reaches a terminal state or Close/compactJournal can safely proceed (see Close). If the step function never checks ctx and never returns, that goroutine runs to completion in the background and the run stays non-terminal until it does — Close called afterward would block on it too. Write step functions so they select on ctx instead of running unconditionally to completion.

The run ends up StatusFailed (the same terminal status used for engine Close and task/run timeouts) with TaskInfo.Error set to ErrRunCancelled.Error(), so callers can distinguish a deliberate cancel from another failure by comparing that string.

Returns ErrRunAlreadyFinished if the run is already StatusCompleted or StatusFailed, or an error if taskID/runID is invalid or the run does not exist. Idempotent: calling it more than once on the same run is a no-op after the first call's signal is durably written.

func (*Engine) Close added in v0.1.2 

func (e *Engine) Close() error

Close cancels in-flight runs, waits for them and the auto-purger to finish writing, then releases the exclusive flock and journal handles. Safe to call more than once.

Cancelling ctx only signals — it does not forcibly stop a step function. Close waits for every in-flight RunStep goroutine to actually return (see StepRunner.waitInFlight) so a step's own journal append can never race compactJournal or a second Engine opening the same dataDir. If a step function never checks ctx and never returns, Close blocks forever and the exclusive flock on dataDir is never released. Write step functions so they select on ctx (or pass it to ctx-aware calls like http.NewRequestWithContext) instead of running unconditionally to completion.

func (*Engine) DeleteTask added in v0.1.2 

func (e *Engine) DeleteTask(ctx context.Context, taskID string) error

DeleteTask removes all runs under taskID. Destructive — use for a full wipe only. Returns ErrRunActive if any run under taskID is currently executing (Running or Waiting in-process).

func (*Engine) DeleteTaskRun added in v0.1.2 

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

DeleteTaskRun removes one run directory and its journal. No-op if not found. Returns ErrRunActive if the run is currently executing, including while blocked in StatusWaiting — runLocks is held for that entire duration.

func (*Engine) GetStep added in v0.1.3 

func (e *Engine) GetStep(ctx context.Context, taskID, runID, stepID string) (StepRecord, bool, error)

GetStep returns the latest StepRecord for one stepID in O(1) after one journal load — (zero, false, nil) if the run has no record for that stepID yet (including one stuck mid-execution: an unfinished STARTED step never appears here — see loadJournal).

func (*Engine) GetTask added in v0.1.2 

func (e *Engine) GetTask(ctx context.Context, taskID, runID string) (TaskInfo, bool, error)

GetTask returns a single run's metadata. (zero, false, nil) if not found.

func (*Engine) ListTasks added in v0.1.2 

func (e *Engine) ListTasks(ctx context.Context, statuses ...TaskStatus) ([]TaskInfo, error)

ListTasks returns TaskInfo records. Zero args returns every status. Pass explicit statuses to filter, e.g. ListTasks(ctx, StatusRunning, StatusWaiting) for recovery after a restart.

func (*Engine) LoadInput added in v0.1.4 

func (e *Engine) LoadInput(ctx context.Context, taskID, runID string) ([]byte, bool, error)

LoadInput returns the JSON-encoded task input written on first RunTask (one I value). (nil, false, nil) if input.json is missing.

func (*Engine) LoadSteps added in v0.1.2 

func (e *Engine) LoadSteps(ctx context.Context, taskID, runID string) ([]StepRecord, error)

LoadSteps returns all StepRecords for a run. Used to inspect progress. Includes waiting, completed, and failed steps. Order is not sorted or otherwise guaranteed — it reflects map iteration order internally. Use WatchSteps if you need append order or history.

func (*Engine) WatchSteps added in v0.1.3 

func (e *Engine) WatchSteps(ctx context.Context, taskID, runID string, fromOffset int, fromByteOffset int64) (<-chan StepEvent, error)

WatchSteps streams step lifecycle events for a run: STARTED, WAITING, COMPLETED, FAILED — every append, in journal order. fromOffset (event count) or fromByteOffset (file position) skip already-seen history; fromByteOffset takes priority when non-zero (pass 0 for fromOffset if you only have a byte offset). Already-written events after that point are sent first, then each new event as it is persisted. The channel closes when ctx is cancelled or the engine closes; cancelling the watch does not stop the run. A slow consumer may miss live events — a warning is logged and delivery continues; reconnect with the last Offset/ByteOffset seen to catch up from the journal.

type EngineOption added in v0.1.2 

type EngineOption func(*engineConfig)

EngineOption configures NewEngine.

func WithAutoPurge 

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

WithAutoPurge starts a background goroutine that deletes Completed and Failed runs whose UpdatedAt is older than age. The optional interval controls how often the purger runs; it defaults to one hour. Running and Waiting runs are never purged.

func WithLockTimeout added in v0.1.2 

func WithLockTimeout(d time.Duration) EngineOption

WithLockTimeout sets how long NewEngine waits for the exclusive flock. Default is 2 seconds.

func WithLogger 

func WithLogger(l *slog.Logger) EngineOption

WithLogger sets the slog.Logger used for task and step lifecycle events. If nil or omitted, a discard logger is used.

func WithMaxRetries 

func WithMaxRetries(n int) EngineOption

WithMaxRetries sets the engine-wide default for task-level retries (re-invoking the task closure). Default is 0 — retries are opt-in so non-idempotent work is not silently repeated. Overridden by WithTaskMaxRetries and WithRunMaxRetries.

func WithTimeout 

func WithTimeout(d time.Duration) EngineOption

WithTimeout sets the engine-wide default task deadline. Zero (default) means no timeout. Overridden by WithTaskTimeout and WithRunTimeout.

type ReadOnlyEngine added in v0.1.2 

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

ReadOnlyEngine is a compile-time-restricted view of a dataDir. It acquires a shared flock so multiple readers can coexist. It cannot register or run tasks. Safe to open from a separate CLI process with no task registration.

func NewReadOnlyEngine added in v0.1.2 

func NewReadOnlyEngine(dataDir string, opts ...ReadOnlyOption) (*ReadOnlyEngine, error)

NewReadOnlyEngine acquires a shared flock on <dataDir>/.lock. Multiple readers coexist. Returns ErrEngineLocked after the lock timeout if a writer holds the exclusive lock.

func (*ReadOnlyEngine) Close added in v0.1.2 

func (r *ReadOnlyEngine) Close() error

Close releases the shared flock. Safe to call more than once.

func (*ReadOnlyEngine) GetStep added in v0.1.3 

func (r *ReadOnlyEngine) GetStep(ctx context.Context, taskID, runID, stepID string) (StepRecord, bool, error)

GetStep returns the latest StepRecord for one stepID. Same semantics as Engine.GetStep.

func (*ReadOnlyEngine) GetTask added in v0.1.2 

func (r *ReadOnlyEngine) GetTask(ctx context.Context, taskID, runID string) (TaskInfo, bool, error)

GetTask returns a single run's metadata. (zero, false, nil) if not found.

func (*ReadOnlyEngine) ListTasks added in v0.1.2 

func (r *ReadOnlyEngine) ListTasks(ctx context.Context, statuses ...TaskStatus) ([]TaskInfo, error)

ListTasks returns TaskInfo records with the same filter semantics as Engine.ListTasks.

func (*ReadOnlyEngine) LoadInput added in v0.1.4 

func (r *ReadOnlyEngine) LoadInput(ctx context.Context, taskID, runID string) ([]byte, bool, error)

LoadInput returns the JSON-encoded task input written on first RunTask (one I value). Same semantics as Engine.LoadInput.

func (*ReadOnlyEngine) LoadSteps added in v0.1.2 

func (r *ReadOnlyEngine) LoadSteps(ctx context.Context, taskID, runID string) ([]StepRecord, error)

LoadSteps returns all StepRecords for a run ordered by Seq.

type ReadOnlyOption added in v0.1.2 

type ReadOnlyOption func(*readOnlyConfig)

ReadOnlyOption configures NewReadOnlyEngine.

func WithROLockTimeout added in v0.1.2 

func WithROLockTimeout(d time.Duration) ReadOnlyOption

WithROLockTimeout sets how long NewReadOnlyEngine waits for the shared flock. Default is 2 seconds. Named distinctly from WithLockTimeout because both option types live in the same package.

func WithROLogger added in v0.1.2 

func WithROLogger(l *slog.Logger) ReadOnlyOption

WithROLogger sets the slog.Logger for the read-only engine. Named distinctly from WithLogger because both option types live in the same package.

type RunOption added in v0.1.2 

type RunOption func(*runConfig)

RunOption configures a single RunTask call.

func WithRunMaxRetries added in v0.1.2 

func WithRunMaxRetries(n int) RunOption

WithRunMaxRetries overrides the task and engine task-level retry count for this run. An explicit 0 disables a non-zero parent default.

func WithRunTimeout added in v0.1.2 

func WithRunTimeout(d time.Duration) RunOption

WithRunTimeout overrides the task and engine timeout for this run. An explicit 0 disables a non-zero parent timeout.

type StepEvent added in v0.1.3 

type StepEvent struct {
	StepRecord
	Offset     int
	ByteOffset int64
}

StepEvent is one journal entry delivered by WatchSteps: STARTED (running), WAITING, COMPLETED, or FAILED — every append, in journal order (never last-write-wins). Offset is the 1-based position of this event across the run's whole journal (steps and signals share one counter); ByteOffset is the file position immediately after it. Reconnect with either value as fromOffset/fromByteOffset to resume a watch without missing or repeating events; byteOffset skips a full rescan, offset does not.

type StepOption added in v0.1.2 

type StepOption func(*stepConfig)

StepOption configures a single RunStep call.

func WithStepMaxRetries added in v0.1.2 

func WithStepMaxRetries(n int) StepOption

WithStepMaxRetries sets how many times this step's function is re-invoked on a non-panic, non-ErrStepPending error. Default is 0 (one attempt). Does not inherit from task or engine retry settings.

func WithStepTimeout added in v0.1.2 

func WithStepTimeout(d time.Duration) StepOption

WithStepTimeout is an inner bound: the step deadline is min(task deadline, step timeout). It cannot extend past the task deadline. The deadline only cancels the step's ctx — it does not forcibly stop fn. A fn that never checks ctx keeps running past the deadline in the background, and the engine (Close, a later CancelRun's drain, etc.) still waits for it to actually return.

func WithStepVersion added in v0.1.4 

func WithStepVersion(v string) StepOption

WithStepVersion tags this step so a later deploy can opt into re-running it. Resume still returns the cached result when the stored version equals v. If this step's code or params change and in-flight runs must re-execute it, pass a new v ("1" → "2") or rename the stepID. Empty v is a no-op (cache by stepID only). Side effects on re-run are the caller's problem.

type StepRecord 

type StepRecord struct {
	StepID      string
	Version     string
	Input       []byte
	Result      []byte
	Error       string
	PanicTrace  string
	Status      StepStatus
	StartedAt   time.Time
	CompletedAt time.Time
}

StepRecord is the latest persistent checkpoint for one memoised step.

type StepRun added in v0.1.2 

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

StepRun is the handle returned by RunStep. RunStep starts work and returns immediately; Get waits for the result, Done reports readiness without blocking so callers can select across several handles (first-of-N).

func RunStep added in v0.1.2 

func RunStep[I, O any](ctx context.Context, s *StepRunner, stepID string, in I, fn func(ctx context.Context, in I) (O, error), opts ...StepOption) *StepRun[O]

RunStep starts fn as a memoised checkpoint and returns immediately. Get waits for the result. On a completed or failed cache hit, fn is not called. On a miss, fn runs on a goroutine and the result is persisted. in is a single JSON-marshalable value (one struct of fields, or struct{} if the step has no payload); it is stored on the record for inspect. stepID must be unique within the run — a duplicate panics. Concurrent calls on the same StepRunner are safe: start several steps, then Get them in any order, or select on their Done channels for first-of-N.

Step IDs are the resume key: never rename a stepID once a run has started. A renamed step is treated as a new, unrelated step — the old result is orphaned and fn runs again under the new name. To re-run the same stepID after a code or param change, pass WithStepVersion with a new string.

func (*StepRun[O]) Done added in v0.1.3 

func (r *StepRun[O]) Done() <-chan struct{}

Done reports readiness. It is closed once the step completes or fails — on a cache hit it is already closed when RunStep returns. Use it in a select across multiple StepRun handles to react to whichever finishes first, then call Get to retrieve that handle's result or error.

func (*StepRun[O]) Get added in v0.1.2 

func (r *StepRun[O]) Get(ctx context.Context) (O, error)

Get blocks until the step completes or ctx is cancelled. The result is already available on a cache hit.

func (*StepRun[O]) StepID added in v0.1.2 

func (r *StepRun[O]) StepID() string

StepID returns the stepID this handle corresponds to.

type StepRunner 

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

StepRunner is scoped to a single run and provides RunStep, StepToken, and run-context accessors. Concurrent RunStep calls are safe: fan out work by calling RunStep multiple times before Get-ing any handle.

func (*StepRunner) Logger added in v0.1.2 

func (s *StepRunner) Logger() *slog.Logger

Logger returns the engine logger pre-scoped with taskID and runID.

func (*StepRunner) RunID added in v0.1.2 

func (s *StepRunner) RunID() string

RunID returns the runID of the current run.

func (*StepRunner) StepToken added in v0.1.2 

func (s *StepRunner) StepToken(ctx context.Context) string

StepToken returns an opaque token encoding taskID, runID, and the current stepID. Must be called inside a RunStep function with that function's ctx (the stepID is carried on ctx, not on the StepRunner, so concurrent steps each get their own token). Pass the token to an external caller so they can later call CompleteStep.

func (*StepRunner) TaskID added in v0.1.2 

func (s *StepRunner) TaskID() string

TaskID returns the taskID of the current run.

type StepStatus 

type StepStatus string

StepStatus is the lifecycle state of a single step checkpoint. 

const (
	// StepStatusRunning means the step function is currently executing.
	// Watch-only: it is superseded by a terminal or waiting status once the
	// step finishes, and is never the record used to replay RunStep/GetStep —
	// an unfinished StepStatusRunning after a crash is treated as missing
	// (fn re-runs), not stuck.
	StepStatusRunning StepStatus = "running"
	// StepStatusWaiting means the step is suspended and awaiting CompleteStep.
	StepStatusWaiting StepStatus = "waiting"
	// StepStatusCompleted means the step succeeded and its result is cached.
	StepStatusCompleted StepStatus = "completed"
	// StepStatusFailed means the step returned an error or panicked.
	StepStatusFailed StepStatus = "failed"
)

type Task 

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

Task is the execution contract for a durable task. Implementations should treat any work with external side effects as a RunStep; non-deterministic logic outside of RunStep calls may not be replayed correctly. I is the single input type (not variadic args); O is the output type. Bundle several fields in one struct. A task with no payload uses I = struct{} and RunTask(..., struct{}{}).

type TaskFunc 

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

TaskFunc adapts a plain function to Task[I, O].

func Func 

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

Func wraps a plain function as a Task, triggering Go's generic type inference so callers do not need to specify type parameters explicitly. in I is the single task input; see Task and RunTask.

func (TaskFunc[I, O]) Exec 

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

Exec implements Task[I, O] for TaskFunc.

type TaskInfo 

type TaskInfo struct {
	TaskID      string            `json:"task_id"`
	RunID       string            `json:"run_id"`
	Name        string            `json:"name"`
	Tags        map[string]string `json:"tags"`
	Status      TaskStatus        `json:"status"`
	Error       string            `json:"error"`
	PanicTrace  string            `json:"panic_trace"`
	CreatedAt   time.Time         `json:"created_at"`
	StartedAt   time.Time         `json:"started_at"`
	CompletedAt time.Time         `json:"completed_at"`
	UpdatedAt   time.Time         `json:"updated_at"`
}

TaskInfo is the persistent metadata for a single run. The task input is stored separately in input.json (see RunTask), not on this struct.

type TaskOption 

type TaskOption func(*taskConfig)

TaskOption configures RegisterTask.

func WithName 

func WithName(name string) TaskOption

WithName sets a human-readable label stored on TaskInfo. It does not affect uniqueness or execution.

func WithTag 

func WithTag(k, v string) TaskOption

WithTag attaches an arbitrary key-value annotation to TaskInfo. Call multiple times to set multiple tags.

func WithTaskMaxRetries added in v0.1.2 

func WithTaskMaxRetries(n int) TaskOption

WithTaskMaxRetries overrides the engine default for task-level retries (re-invoking the whole closure). Nil-vs-set is tracked so an explicit 0 disables a non-zero engine default.

func WithTaskTimeout added in v0.1.2 

func WithTaskTimeout(d time.Duration) TaskOption

WithTaskTimeout overrides the engine default task deadline. An explicit 0 disables a non-zero engine default.

type TaskRun added in v0.1.2 

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

TaskRun is the handle returned by RunTask. RunID is available immediately; Get blocks until the run reaches a terminal state.

func RunTask added in v0.1.2 

func RunTask[I, O any](ctx context.Context, e *Engine, taskID string, runID string, input I, opts ...RunOption) *TaskRun[O]

RunTask starts or resumes a run in a background goroutine and returns immediately. input is one typed value I — not variadic args. Put several fields on one struct; a task with no payload uses struct{} and struct{}{}. On first start the value is written to input.json. The same runID reloads that file and ignores the input argument. Pass an empty runID to resume the oldest active run for taskID, or to generate a new ULID if none is active. A completed or failed run returns the stored result without spawning a goroutine.

func (*TaskRun[O]) Get added in v0.1.2 

func (r *TaskRun[O]) Get(ctx context.Context) (O, error)

Get blocks until the run completes or ctx is cancelled. The typed result is cached after the first successful wait so later calls do not re-decode.

func (*TaskRun[O]) RunID added in v0.1.2 

func (r *TaskRun[O]) RunID() string

RunID returns the resolved runID. Empty if the taskID was not registered or the runID was invalid — check Get for the error.

func (*TaskRun[O]) Status added in v0.1.2 

func (r *TaskRun[O]) Status() TaskStatus

Status returns the current run status without blocking.

type TaskStatus 

type TaskStatus string

TaskStatus is the lifecycle state of a run. 

const (
	// StatusRunning means the run is actively executing.
	StatusRunning TaskStatus = "running"
	// StatusWaiting means at least one step is awaiting CompleteStep.
	StatusWaiting TaskStatus = "waiting"
	// StatusCompleted means the run finished successfully.
	StatusCompleted TaskStatus = "completed"
	// StatusFailed means the run terminated with an error or recovered panic.
	StatusFailed TaskStatus = "failed"
)

Source Files

View all Source files

Directories

Path Synopsis
cmd
durable-inspect command
Command durable-inspect is a read-only viewer for a durable-go journal.
 Command durable-inspect is a read-only viewer for a durable-go journal.
examples
fanout command
Package main demonstrates durable-go's async RunStep fan-out: several steps run concurrently, joined with Get; a first-of-N select reacts to whichever of two steps finishes first; one step suspends with ErrStepPending and is resumed by an external CompleteStep call.
 Package main demonstrates durable-go's async RunStep fan-out: several steps run concurrently, joined with Get; a first-of-N select reacts to whichever of two steps finishes first; one step suspends with ErrStepPending and is resumed by an external CompleteStep call.
func-task command
Package main demonstrates the functional / closure style of durable-go.
 Package main demonstrates the functional / closure style of durable-go.
resume command
Package main demonstrates durable-go's core value: crash recovery and step replay.
 Package main demonstrates durable-go's core value: crash recovery and step replay.
struct-task command
Package main demonstrates the struct / method-receiver style of durable-go.
 Package main demonstrates the struct / method-receiver style of durable-go.
yaml-task command
Package main runs a YAML workflow file as one durable-go task.
 Package main runs a YAML workflow file as one durable-go task.
Package durablepb contains the protobuf-generated wire types for journal.log entries.
 Package durablepb contains the protobuf-generated wire types for journal.log entries.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.