durable

README

durable-go

Lightweight, embeddable durable task execution for Go.

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

Releases follow Semantic Versioning; see the latest release.

Features

• Engine API — NewEngine, RegisterTask, RunTask, RunStep in a single package.
• Memoized steps — completed steps replay from the journal; they are not run again.
• Pending steps — return ErrStepPending and complete later via CompleteStep (human approval, webhooks).
• Fire-and-forget runs — RunTask returns immediately; TaskRun.Get waits; RunID() is available at once.
• 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.

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.

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", func(ctx context.Context) (string, error) {
            return chargeCard(in)
        }).Get(ctx)
        if err != nil {
            return OrderOutput{}, err
        }
        shipped, err := durable.RunStep(ctx, s, "ship", func(ctx context.Context) (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", func(ctx context.Context) (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. An external caller completes it with the token from StepToken():

approval, err := durable.RunStep(ctx, s, "approve", func(ctx context.Context) (Approval, error) {
    token := s.StepToken()
    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"})

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, reloadInput(t))
    go func() { _, _ = run.Get(ctx) }()
}

Task inputs are not persisted. Pass the same input when resuming.

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).

Also:

• Unique step IDs — one stable string per step (literals or fmt.Sprintf("step-%d", i)). Reusing an ID panics.
• Sequential RunStep calls — concurrent calls on the same StepRunner panic. Fan out work, then persist results sequentially.
• JSON results — step and task outputs must be JSON-marshalable.
• Same runID to resume — inputs are not persisted; pass the same input to RunTask when resuming.

Examples

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

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

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

Development

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

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

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

License

Apache 2.0

Disclaimer

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

Documentation

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", func(ctx context.Context) (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

• Variables
• func CompleteStep[O any](ctx context.Context, e *Engine, token string, result O) error
• func RegisterTask[I, O any](e *Engine, taskID string, task Task[I, O], opts ...TaskOption) error
• type Engine
  • func NewEngine(ctx context.Context, dataDir string, opts ...EngineOption) (*Engine, error)
  • func (e *Engine) Close() error
  • func (e *Engine) DeleteTask(ctx context.Context, taskID string) error
  • func (e *Engine) DeleteTaskRun(ctx context.Context, taskID, runID string) error
  • func (e *Engine) GetTask(ctx context.Context, taskID, runID string) (TaskInfo, bool, error)
  • func (e *Engine) ListTasks(ctx context.Context, statuses ...TaskStatus) ([]TaskInfo, error)
  • func (e *Engine) LoadSteps(ctx context.Context, taskID, runID string) ([]StepRecord, error)
• type EngineOption
  • func WithAutoPurge(age time.Duration, interval ...time.Duration) EngineOption
  • func WithLockTimeout(d time.Duration) EngineOption
  • func WithLogger(l *slog.Logger) EngineOption
  • func WithMaxRetries(n int) EngineOption
  • func WithTimeout(d time.Duration) EngineOption
• type ReadOnlyEngine
  • func NewReadOnlyEngine(dataDir string, opts ...ReadOnlyOption) (*ReadOnlyEngine, error)
  • func (r *ReadOnlyEngine) Close() error
  • func (r *ReadOnlyEngine) GetTask(ctx context.Context, taskID, runID string) (TaskInfo, bool, error)
  • func (r *ReadOnlyEngine) ListTasks(ctx context.Context, statuses ...TaskStatus) ([]TaskInfo, error)
  • func (r *ReadOnlyEngine) LoadSteps(ctx context.Context, taskID, runID string) ([]StepRecord, error)
• type ReadOnlyOption
  • func WithROLockTimeout(d time.Duration) ReadOnlyOption
  • func WithROLogger(l *slog.Logger) ReadOnlyOption
• type RunOption
  • func WithRunMaxRetries(n int) RunOption
  • func WithRunTimeout(d time.Duration) RunOption
• type StepOption
  • func WithStepMaxRetries(n int) StepOption
  • func WithStepTimeout(d time.Duration) StepOption
• type StepRecord
• type StepRun
  • func RunStep[O any](ctx context.Context, s *StepRunner, stepID string, ...) *StepRun[O]
  • func (r *StepRun[O]) Get(ctx context.Context) (O, error)
  • func (r *StepRun[O]) StepID() string
• type StepRunner
  • func (s *StepRunner) Logger() *slog.Logger
  • func (s *StepRunner) RunID() string
  • func (s *StepRunner) StepSeq() int
  • func (s *StepRunner) StepToken() string
  • func (s *StepRunner) TaskID() string
• type StepStatus
• type Task
• type TaskFunc
  • func Func[I, O any](fn func(ctx context.Context, s *StepRunner, in I) (O, error)) TaskFunc[I, O]
  • func (f TaskFunc[I, O]) Exec(ctx context.Context, s *StepRunner, input I) (O, error)
• type TaskInfo
• type TaskOption
  • func WithName(name string) TaskOption
  • func WithTag(k, v string) TaskOption
  • func WithTaskMaxRetries(n int) TaskOption
  • func WithTaskTimeout(d time.Duration) TaskOption
• type TaskRun
  • func RunTask[I, O any](ctx context.Context, e *Engine, taskID string, runID string, input I, ...) *TaskRun[O]
  • func (r *TaskRun[O]) Get(ctx context.Context) (O, error)
  • func (r *TaskRun[O]) RunID() string
  • func (r *TaskRun[O]) Status() TaskStatus
• type TaskStatus

Variables

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 the run
	// until CompleteStep delivers a result for that step. The engine writes
	// StepStatusWaiting and blocks the run goroutine.
	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")
)

Functions

func CompleteStep

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.

func RegisterTask

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. 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

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

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) Close

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.

func (*Engine) DeleteTask

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

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) GetTask

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

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) LoadSteps

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

LoadSteps returns all StepRecords for a run ordered by Seq. Used to inspect progress. Includes waiting, completed, and failed steps.

type EngineOption

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

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

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

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

func (r *ReadOnlyEngine) Close() error

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

func (*ReadOnlyEngine) GetTask

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

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) LoadSteps

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

LoadSteps returns all StepRecords for a run ordered by Seq.

type ReadOnlyOption

type ReadOnlyOption func(*readOnlyConfig)

ReadOnlyOption configures NewReadOnlyEngine.

func WithROLockTimeout

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

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

type RunOption func(*runConfig)

RunOption configures a single RunTask call.

func WithRunMaxRetries

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

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 StepOption

type StepOption func(*stepConfig)

StepOption configures a single RunStep call.

func WithStepMaxRetries

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

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.

type StepRecord

type StepRecord struct {
	StepID      string
	Seq         int
	InputHash   string // unused in v1; RunStep has no input parameter
	Result      []byte
	Error       string
	PanicTrace  string
	Status      StepStatus
	StartedAt   time.Time
	CompletedAt time.Time
}

StepRecord is the persistent checkpoint for one memoised step.

type StepRun

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

StepRun is the handle returned by RunStep. Get returns immediately — RunStep itself is synchronous (including the wait on ErrStepPending).

func RunStep

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

RunStep executes fn as a memoised checkpoint. On a completed cache hit, fn is not called. On a miss, fn runs synchronously and the result is persisted. stepID must be unique within the run — a duplicate panics. Concurrent calls on the same StepRunner panic.

func (*StepRun[O]) Get

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

Get returns the step result. ctx is reserved for future async support; the result is already available because RunStep waits before returning.

func (*StepRun[O]) StepID

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. It is not safe for concurrent use.

func (*StepRunner) Logger

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

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

func (*StepRunner) RunID

func (s *StepRunner) RunID() string

RunID returns the runID of the current run.

func (*StepRunner) StepSeq

func (s *StepRunner) StepSeq() int

StepSeq returns the number of steps executed so far. Zero before the first RunStep call; incremented after each RunStep returns (including cache hits).

func (*StepRunner) StepToken

func (s *StepRunner) StepToken() string

StepToken returns an opaque token encoding taskID, runID, and the current stepID. Must be called inside a RunStep function — panics if currentStepID is empty. Pass the token to an external caller (DB, email, webhook URL) so they can later call CompleteStep.

func (*StepRunner) TaskID

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 (
	// 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 input type; O is the output type.

type TaskFunc

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

TaskFunc adapts a plain function to 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.

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. Input is intentionally omitted — callers own input persistence and reload it on recovery.

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

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

func WithTaskTimeout(d time.Duration) TaskOption

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

type TaskRun

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

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. Pass an empty runID to resume the oldest active run for taskID, or to generate a new ULID if none is active. Pass an existing runID to resume; a completed or failed run returns the stored result without spawning a goroutine.

func (*TaskRun[O]) Get

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

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

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"
)