Help us shape the future of Go! Take the 2026 Go Developer Survey

durable

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 17, 2026 License: MIT Imports: 26 Imported by: 0

README

Durable Task Runner

Durable Task Runner is a Go library for background jobs that need to survive worker crashes. It stores a sequence of activities in PostgreSQL, tracks each attempt, and recovers unfinished work when a worker disappears.

Use it for jobs such as imports or account provisioning where losing progress between steps would leave work unfinished. Your application supplies Go handlers; the engine stores progress and coordinates workers through PostgreSQL.

It supports bounded retries, execution deadlines, cooperative cancellation, explicit compensation, and versioned handlers. It includes operator inspection and optional telemetry. See supported environments and limits.

Try it

You need Go 1.26.9 or later and Docker with Compose. PostgreSQL 17 runs locally through Compose.

git clone --branch v1.0.0 https://github.com/KleitonBarone/durable-task-runner.git
cd durable-task-runner
go run ./cmd/dev up
go run ./cmd/dev example

These commands also work in PowerShell. If Docker is installed only inside WSL, the development command uses the Ubuntu distribution automatically.

The example runs fake customer welcome steps, prints the timeline and result, then shuts down. It creates no real accounts or email. Other modes run separate worker and scheduler processes.

Stop PostgreSQL with go run ./cmd/dev down. Its development volume is preserved.

Use it in an application

Start with the complete Go example and the integration guide.

  1. Declare typed activities and give each step a stable key in a linear plan.
  2. Register the definition and open an engine connected to PostgreSQL.
  3. Apply the embedded migrations, then start an execution with an optional deduplication key.
  4. Register typed activity handlers and run workers.
  5. Run one scheduler for recovery. Read prior-step results in handlers and inspect outcomes through public APIs.

Your application owns runtime configuration and lifecycle. PostgreSQL coordinates the processes.

Install the tagged module:

go get github.com/KleitonBarone/durable-task-runner@v1.0.0

The integration guide includes separate worker and scheduler entry points.

What survives a crash?

The engine commits the workflow and its full activity plan before workers can claim work. Completing a step and making its successor ready share a transaction. State changes and their history events commit together.

Workers hold expiring leases and heartbeat while handling an activity. The scheduler recovers abandoned work within its attempt budget. Old lease tokens cannot acknowledge replacement attempts.

Activity delivery is at least once. A worker can perform a side effect and crash before recording success, so another worker may repeat it. Your handlers must use application-level idempotency or deduplication. Handler metadata includes a stable effect key for this purpose. The engine does not provide exactly-once side effects.

A handler error fails its workflow unless its step opts into bounded retries. The default budget is three claims, including crashes. Attempts default to five minutes and workflows to 24 hours; heartbeats cannot extend these limits. See retry and deadline options. The worker continues to other jobs. The last step's output is the workflow result. Versioned handlers let old and new executions coexist. Follow the upgrade guide before deploying incompatible changes.

Applications can cancel an execution. A committed request stops forward progress and asks handlers to stop; it does not undo external effects. Optional compensation runs declared cleanup for completed steps in reverse order and records cleanup failures separately.

Development

Run all checks with go run ./cmd/dev check. It starts disposable PostgreSQL, runs formatting, static analysis, unit and integration tests, kills and recovers real worker processes, runs the example and a separate consumer module, and removes its test containers.

Read CONTRIBUTING.md for focused tests, race checks, and debugging. See the roadmap and release evidence for scope and verification.

License

MIT

Documentation

Overview

Package durable runs persisted linear workflows using PostgreSQL and application-owned Go workers and a scheduler.

Declare typed steps with DefineActivity, build a deterministic plan with Define, and Register it before calling Start. Start commits the complete plan before any step becomes claimable. New uses an application-owned pgx pool; Open owns its pool. Apply Migrate explicitly before running workers.

A Worker executes one handler at a time and heartbeats its expiring lease. A Scheduler recovers abandoned attempts and promotes persisted retry waits. Production ownership uses PostgreSQL time through SystemClock. ManualClock controls persisted time and runtime tickers in deterministic tests.

Delivery is at least once. A handler can perform an external effect and crash before acknowledgement. Use Metadata's stable effect key with application-owned idempotency. Cancellation fences engine progress and requests cooperative handler shutdown. Compensation runs explicitly declared cleanup for completed steps; neither operation promises to undo arbitrary external effects.

Keep workflow and handler name/version/schema identities immutable. Existing executions continue their stored plans. ListWorkflows and the other List methods provide bounded, redacted live pages. Raw result/query methods can expose payloads. PurgeTerminal removes terminal details while preserving keyed-start deduplication and its original input indefinitely.

See the repository's integration and operations guides for a complete application, supported environments, deployment, and recovery procedures.

Index

Examples

Constants

View Source 
const MaxPayloadBytes = 1 << 20

MaxPayloadBytes limits each workflow input, step input, and step output before persistence. JSON uses encoding/json and PostgreSQL jsonb semantics.

Variables

View Source 
var (
	ErrNoActivity          = errors.New("no activity is ready")
	ErrLeaseLost           = errors.New("activity lease is no longer valid")
	ErrUnknownWorkflow     = errors.New("workflow definition is not registered")
	ErrNotFound            = errors.New("execution or step not found")
	ErrInvalidInput        = errors.New("invalid input")
	ErrPayloadTooLarge     = errors.New("JSON payload exceeds limit")
	ErrResultUnavailable   = errors.New("step result is not available")
	ErrTypeMismatch        = errors.New("activity type does not match persisted contract")
	ErrStartConflict       = errors.New("start key already used with different input")
	ErrWorkerBusy          = errors.New("worker instance is already running")
	ErrHandlerUnresponsive = errors.New("activity handler did not stop within shutdown grace period")
	ErrExecutionDeadline   = errors.New("execution deadline exceeded")
	ErrIncompatibleSchema  = errors.New("database schema is incompatible with this binary")
	ErrExecutionExpired    = errors.New("execution was removed by retention")
)

Functions

func Migrate 

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate checks the immutable migration catalog and atomically applies pending SQL under an exclusive transaction lock. The pool's role must have DDL rights.

func Permanent 

func Permanent(err error) error

Permanent excludes this error from an opted-in handler retry policy.

func Register 

func Register[Input any](registry *Registry, definition Definition[Input]) error

Register adds an immutable workflow name/version and rejects duplicate entries.

func ValidateActivityTransition 

func ValidateActivityTransition(from, to ActivityState) error

ValidateActivityTransition checks scheduling, retry, and terminal step transitions.

func ValidateAttemptTransition 

func ValidateAttemptTransition(from, to AttemptState) error

ValidateAttemptTransition allows one running attempt to reach one terminal outcome.

func ValidateLeaseTransition 

func ValidateLeaseTransition(from, to LeaseState) error

ValidateLeaseTransition permits active ownership to be released or expired once.

func ValidateWorkflowTransition 

func ValidateWorkflowTransition(from, to WorkflowState) error

ValidateWorkflowTransition rejects illegal transitions, including reopening terminal work.

Types

type Activity 

type Activity struct {
	Contract              string
	Version               string
	InputSchema           string
	OutputSchema          string
	Kind                  ActivityKind
	CompensatesActivityID *string
	Compensation          *CompensationPlan
	ID                    string
	WorkflowID            string
	Position              int
	StepKey               string
	Name                  string
	Input                 json.RawMessage
	Output                json.RawMessage
	State                 ActivityState
	AttemptCount          int
	CreatedAt             time.Time
	UpdatedAt             time.Time
	CompletedAt           *time.Time
	RetryPolicy           RetryPolicy
	AvailableAt           time.Time
	AttemptTimeout        time.Duration
	LastError             string
}

Activity is a persisted forward or cleanup step. ListActivities omits payloads.

type ActivityDefinition 

type ActivityDefinition[Input, Output any] struct {
	// contains filtered or unexported fields
}

ActivityDefinition ties one durable name to its Go input and output types. Keep names and types compatible while executions using them are in progress.

func DefineActivity 

func DefineActivity[Input, Output any](name string) ActivityDefinition[Input, Output]

DefineActivity declares typed step inputs, handler outputs, and result reads.

Example 
package main

import (
	"fmt"
	"log"
	"time"

	durable "github.com/KleitonBarone/durable-task-runner"
)

func main() {
	double := durable.DefineActivity[int, int]("double").
		WithVersion("1", "integer/1", "integer/1")
	step, err := double.Step("double-input", 21,
		durable.WithAttemptTimeout(time.Minute),
		durable.WithRetry(durable.RetryPolicy{MaxAttempts: 3, RetryErrors: true,
			InitialInterval: time.Second, MaximumInterval: time.Minute}))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(step.Name, step.Key, string(step.Input))
}

Output:
double double-input 21

func (ActivityDefinition[Input, Output]) Handle 

func (a ActivityDefinition[Input, Output]) Handle(handler func(context.Context, Input) (Output, error)) HandlerRegistration

Handle binds the declaration's types and compatibility identities to a handler.

func (ActivityDefinition[Input, Output]) Result 

func (a ActivityDefinition[Input, Output]) Result(ctx context.Context, key string) (Output, error)

Result reads a completed earlier step in the current handler's workflow. The handler context carries execution identity; callers cannot read a future step or a result with a different activity type.

func (ActivityDefinition[Input, Output]) Step 

func (a ActivityDefinition[Input, Output]) Step(key string, input Input, options ...StepOption) (ActivitySpec, error)

Step binds input to a stable key within a workflow. Keys are independent of activity names, so one activity definition can appear more than once.

func (ActivityDefinition[Input, Output]) WithVersion 

func (a ActivityDefinition[Input, Output]) WithVersion(version, inputSchema, outputSchema string) ActivityDefinition[Input, Output]

WithVersion declares handler and payload compatibility explicitly. Keep the previous version registered until its forward and cleanup work has drained.

func (ActivityDefinition[Input, Output]) WorkflowResult 

func (a ActivityDefinition[Input, Output]) WorkflowResult(ctx context.Context, engine *Engine, workflowID string) (Output, error)

WorkflowResult reads the last step's output after successful completion.

type ActivityKind 

type ActivityKind string

ActivityKind distinguishes a forward step from its explicitly declared cleanup. 

const (
	ActivityForward      ActivityKind = "forward"
	ActivityCompensation ActivityKind = "compensation"
)

type ActivityMetadata 

type ActivityMetadata struct {
	Kind                  ActivityKind
	CompensatesActivityID *string
	WorkflowID            string
	ActivityID            string
	ActivityName          string
	ActivityVersion       string
	InputSchema           string
	OutputSchema          string
	StepKey               string
	WorkerID              string
	AttemptID             string
	AttemptNumber         int
	EffectKey             string
	AttemptDeadline       time.Time
}

ActivityMetadata identifies this delivery and its logical effect. EffectKey stays stable across replacement attempts; AttemptID and AttemptNumber change.

func Metadata 

func Metadata(ctx context.Context) (ActivityMetadata, bool)

type ActivitySpec 

type ActivitySpec struct {
	Name         string
	Version      string
	InputSchema  string
	OutputSchema string
	Key          string
	Input        json.RawMessage
	// contains filtered or unexported fields
}

ActivitySpec is a serialized step declaration persisted as part of a plan.

func NewActivity 

func NewActivity[Input any](name string, input Input, options ...StepOption) (ActivitySpec, error)

NewActivity constructs an untyped-output step. Prefer DefineActivity for typed handlers and results. Omitted keys receive stable positional keys at start.

type ActivityState 

type ActivityState string

ActivityState describes one persisted step's scheduling and execution state. 

const (
	ActivityBlocked   ActivityState = "blocked"
	ActivityReady     ActivityState = "ready"
	ActivityRunning   ActivityState = "running"
	ActivityWaiting   ActivityState = "retry_wait"
	ActivityCompleted ActivityState = "completed"
	ActivityFailed    ActivityState = "failed"
	ActivityTimedOut  ActivityState = "timed_out"
	ActivityCancelled ActivityState = "cancelled"
)

type Attempt 

type Attempt struct {
	ID            string
	ActivityID    string
	Number        int
	WorkerID      string
	State         AttemptState
	Error         string
	StartedAt     time.Time
	EndedAt       *time.Time
	RetryDecision string
	RetryDueAt    *time.Time
	DeadlineAt    time.Time
}

Attempt records a delivery's outcome, deadline, and retry decision.

type AttemptState 

type AttemptState string

AttemptState describes one delivery; every claim consumes the activity's budget. 

const (
	AttemptRunning   AttemptState = "running"
	AttemptCompleted AttemptState = "completed"
	AttemptFailed    AttemptState = "failed"
	AttemptAbandoned AttemptState = "abandoned"
	AttemptTimedOut  AttemptState = "timed_out"
	AttemptCancelled AttemptState = "cancelled"
)

type Claim 

type Claim struct {
	ActivityVersion       string
	InputSchema           string
	OutputSchema          string
	Kind                  ActivityKind
	CompensatesActivityID *string
	WorkflowID            string
	ActivityID            string
	ActivityName          string
	StepKey               string
	Position              int
	ActivityInput         json.RawMessage
	AttemptID             string
	AttemptNumber         int
	LeaseID               string
	LeaseToken            string
	WorkerID              string
	LeaseExpiresAt        time.Time
	ClaimedAt             time.Time
	AttemptDeadline       time.Time
}

Claim carries low-level delivery ownership. Keep LeaseToken private; it is a fencing capability, not a stable identity for deduplicating external effects.

type Clock 

type Clock interface {
	Now() time.Time
}

Clock controls runtime timers and injected test time. SystemClock uses PostgreSQL time for durable mutations; ManualClock controls exact boundaries.

type CompensationPlan 

type CompensationPlan struct {
	Name           string          `json:"name"`
	Version        string          `json:"version"`
	InputSchema    string          `json:"input_schema"`
	OutputSchema   string          `json:"output_schema"`
	Contract       string          `json:"contract"`
	Input          json.RawMessage `json:"input"`
	RetryPolicy    RetryPolicy     `json:"retry_policy"`
	AttemptTimeout time.Duration   `json:"attempt_timeout_ns"`
}

CompensationPlan is the immutable cleanup declaration stored on a forward activity. Cleanup handlers can also read committed forward step results.

type CompensationState 

type CompensationState string

CompensationState is independent of the original workflow failure or cancellation. 

const (
	CompensationNone      CompensationState = "none"
	CompensationRunning   CompensationState = "running"
	CompensationCompleted CompensationState = "completed"
	CompensationFailed    CompensationState = "failed"
)

type Definition 

type Definition[Input any] struct {
	Name        string
	Version     string
	InputSchema string

	Plan func(Input) ([]ActivitySpec, error)
	// contains filtered or unexported fields
}

Definition turns typed workflow input into a deterministic linear activity plan. A name and version must retain the same meaning while executions using them are in progress. Existing executions continue their stored plan.

func Define 

func Define[Input any](name string, plan func(Input) ([]ActivitySpec, error)) Definition[Input]

Define declares a workflow plan. The plan must be deterministic and free of side effects; duplicate keyed starts can evaluate it again before deduplication.

func (Definition[Input]) WithVersion 

func (d Definition[Input]) WithVersion(version, inputSchema string) Definition[Input]

WithVersion declares an immutable workflow version and input schema identity. Register old and new versions together while old executions drain.

type Engine 

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

Engine coordinates persisted executions. It is safe for concurrent operations.

func New 

func New(pool *pgxpool.Pool, registry *Registry, clock Clock, options ...EngineOption) (*Engine, error)

New uses an application-owned pool without doing I/O. Use SystemClock in production. The first operation checks schema compatibility.

func Open 

func Open(ctx context.Context, databaseURL string, registry *Registry, clock Clock, options ...EngineOption) (*Engine, error)

Open creates and owns a pool, verifies connectivity, and accepts a known schema prefix so the caller can apply pending migrations. It never migrates implicitly.

func (*Engine) Activities 

func (e *Engine) Activities(ctx context.Context, workflowID string) ([]Activity, error)

func (*Engine) Attempts 

func (e *Engine) Attempts(ctx context.Context, activityID string) ([]Attempt, error)

Attempts lists attempts for one activity in attempt-number order.

func (*Engine) Cancel 

func (e *Engine) Cancel(ctx context.Context, id, actor, reason string) (WorkflowExecution, error)

Cancel durably prevents further forward claims and fences active ownership. The scheduler settles the request without waiting for handler cooperation. Repeated and terminal requests preserve the original outcome and metadata.

func (*Engine) CheckSchema 

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

CheckSchema rejects missing, altered, and unknown migrations. Opening an engine allows a known older schema so its owner can call Migrate first.

func (*Engine) ClaimActivity 

func (e *Engine) ClaimActivity(ctx context.Context, workerID string, leaseDuration time.Duration) (Claim, error)

ClaimActivity is a low-level claim without handler capability filtering. Prefer Worker, which claims only compatible registered handlers.

func (*Engine) Close 

func (e *Engine) Close()

Close closes only a pool created by Open. Stop workers and the scheduler first.

func (*Engine) CompleteActivity 

func (e *Engine) CompleteActivity(ctx context.Context, claim Claim, output json.RawMessage) error

CompleteActivity atomically acknowledges current ownership, stores JSON output, and advances the stored plan. A duplicate or stale acknowledgement is fenced.

func (*Engine) FailActivity 

func (e *Engine) FailActivity(ctx context.Context, claim Claim, activityErr error) error

FailActivity records a non-nil task failure and its persisted retry decision.

func (*Engine) Heartbeat 

func (e *Engine) Heartbeat(ctx context.Context, claim Claim, extension time.Duration) (time.Time, error)

Heartbeat extends current ownership up to the attempt and phase deadlines. An expired or fenced token returns ErrLeaseLost, including at exact expiry.

func (*Engine) ListActivities 

func (e *Engine) ListActivities(ctx context.Context, workflowID string, request PageRequest) (Page[Activity], error)

ListActivities omits inputs, outputs, cleanup payloads, and diagnostic text by default. Contract identities and due times explain compatible waiting work.

func (*Engine) ListAttempts 

func (e *Engine) ListAttempts(ctx context.Context, activityID string, request PageRequest) (Page[Attempt], error)

func (*Engine) ListLeases 

func (e *Engine) ListLeases(ctx context.Context, activityID string, request PageRequest) (Page[Lease], error)

ListLeases never returns ownership tokens, even for sensitive inspection.

func (*Engine) ListTimeline 

func (e *Engine) ListTimeline(ctx context.Context, workflowID string, request PageRequest) (Page[TimelineEvent], error)

ListTimeline excludes event details unless sensitive inspection is requested.

func (*Engine) ListWorkflows 

func (e *Engine) ListWorkflows(ctx context.Context, filter WorkflowFilter, request PageRequest) (Page[WorkflowSummary], error)

ListWorkflows returns bounded live pages ordered by stable database ordinal. Start a new scan to observe new inserts and concurrent status changes.

func (*Engine) Migrate 

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

Migrate atomically applies pending embedded migrations under an exclusive lock.

func (*Engine) PurgeTerminal 

func (e *Engine) PurgeTerminal(ctx context.Context, policy RetentionPolicy) (RetentionResult, error)

PurgeTerminal deliberately deletes terminal execution details and history in bounded workflow batches. It is never called automatically by the scheduler. Keyed input moves to the deduplication record and remains retained indefinitely.

func (*Engine) RecoverAbandoned 

func (e *Engine) RecoverAbandoned(ctx context.Context, limit int) (int, error)

RecoverAbandoned reconciles at most limit workflows with expired ownership, deadlines, cancellation requests, or due retries. It returns reconciled work items, which need not equal the workflow count. Prefer Scheduler for a run loop.

func (*Engine) Timeline 

func (e *Engine) Timeline(ctx context.Context, workflowID string) ([]TimelineEvent, error)

func (*Engine) Workflow 

func (e *Engine) Workflow(ctx context.Context, workflowID string) (WorkflowExecution, error)

type EngineOption 

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

EngineOption configures optional engine diagnostics.

func WithObserver 

func WithObserver(observer Observer) EngineOption

WithObserver adds a non-nil diagnostic observer. Multiple observers are allowed.

type ExecutionExpiredError 

type ExecutionExpiredError struct{ ID string }

ExecutionExpiredError identifies a keyed execution whose details were purged. The retained start key still prevents another execution from using that key.

func (*ExecutionExpiredError) Error 

func (e *ExecutionExpiredError) Error() string

func (*ExecutionExpiredError) Unwrap 

func (e *ExecutionExpiredError) Unwrap() error

type Handler 

type Handler func(context.Context, json.RawMessage) (json.RawMessage, error)

Handler is the low-level JSON handler form. Handlers must tolerate redelivery.

type HandlerPanicError 

type HandlerPanicError struct{ Value any }

HandlerPanicError records an abnormal handler exit as a task failure. Panics in goroutines created by application handlers remain the application's concern.

func (*HandlerPanicError) Error 

func (e *HandlerPanicError) Error() string

func (*HandlerPanicError) Unwrap 

func (e *HandlerPanicError) Unwrap() error

type HandlerRegistration 

type HandlerRegistration struct {
	Name         string
	Version      string
	InputSchema  string
	OutputSchema string
	Handler      Handler
	// contains filtered or unexported fields
}

HandlerRegistration binds a handler to its durable compatibility identities.

func Handle 

func Handle[Input, Output any](name string, handler func(context.Context, Input) (Output, error)) HandlerRegistration

Handle adapts a Go handler without a typed declaration contract. Prefer an ActivityDefinition's Handle method for capability filtering and typed results.

type Lease 

type Lease struct {
	ID          string
	ActivityID  string
	AttemptID   string
	Token       string
	WorkerID    string
	State       LeaseState
	AcquiredAt  time.Time
	HeartbeatAt time.Time
	ExpiresAt   time.Time
	EndedAt     *time.Time
}

Lease records ownership timing. ListLeases never returns its Token.

type LeaseState 

type LeaseState string

LeaseState records ownership release or expiry. Only active, unexpired leases acknowledge work. 

const (
	LeaseActive   LeaseState = "active"
	LeaseReleased LeaseState = "released"
	LeaseExpired  LeaseState = "expired"
)

type ManualClock 

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

ManualClock controls durable timestamps and ticker wakeups in deterministic tests.

func NewManualClock 

func NewManualClock(now time.Time) *ManualClock

func (*ManualClock) Advance 

func (c *ManualClock) Advance(d time.Duration)

func (*ManualClock) NewTicker 

func (c *ManualClock) NewTicker(interval time.Duration) Ticker

NewTicker follows time.Ticker's buffered, coalescing behavior. Advancing time makes due ticks available without waiting for their consumers to run.

func (*ManualClock) Now 

func (c *ManualClock) Now() time.Time

type Observation 

type Observation struct {
	Type       string
	At         time.Time
	WorkflowID string
	ActivityID string
	AttemptID  string
	WorkerID   string
	StepKey    string
	Version    string
	Kind       ActivityKind
	Outcome    string
	Reason     string
	Duration   time.Duration
	ReadyAge   time.Duration
}

Observation contains diagnostics, never payloads, effect keys, ownership tokens, or arbitrary error text. Committed events are best effort notifications; the execution timeline remains the durable audit record.

type Observer 

type Observer interface {
	Observe(context.Context, Observation)
}

Observer must be safe for concurrent calls and return promptly without network waits. Transition callbacks run after commit, outside locks. Timing and runtime notifications are local observations. Panics cannot change committed outcomes.

func SlogObserver 

func SlogObserver(logger *slog.Logger) Observer

SlogObserver emits correlated structured diagnostics with no sensitive text.

type ObserverFunc 

type ObserverFunc func(context.Context, Observation)

ObserverFunc adapts a concurrent, prompt callback to Observer.

func (ObserverFunc) Observe 

func (f ObserverFunc) Observe(ctx context.Context, event Observation)

type Page 

type Page[T any] struct {
	Items      []T    `json:"items"`
	NextCursor string `json:"next_cursor,omitempty"`
}

Page is one bounded live read. Continue with NextCursor and identical filters; concurrent commits can change or remove rows between requests.

type PageRequest 

type PageRequest struct {
	Limit     int
	Cursor    string
	Sensitive bool
}

PageRequest bounds a live scan. Sensitive exposes diagnostic text and history details, never lease tokens. Application result APIs remain the payload path.

type PendingActivity 

type PendingActivity struct {
	Name        string
	Version     string
	State       ActivityState
	Kind        ActivityKind
	AvailableAt time.Time
}

PendingActivity identifies the first unfinished step, including version and retry due time.

type Registry 

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

Registry stores process-local definitions and compatible activity contracts.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry creates an empty concurrency-safe registry.

type RetentionPolicy 

type RetentionPolicy struct {
	CompletedBefore time.Time
	Limit           int
}

RetentionPolicy selects terminal executions strictly older than CompletedBefore. The cutoff is required and cannot be in the future. Limit defaults to 100, maximum 500.

type RetentionResult 

type RetentionResult struct {
	Deleted      int
	ExecutionIDs []string
}

RetentionResult identifies executions whose details were deleted atomically.

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts     int           `json:"max_attempts"`
	RetryErrors     bool          `json:"retry_errors"`
	InitialInterval time.Duration `json:"initial_interval"`
	MaximumInterval time.Duration `json:"maximum_interval"`
	JitterFraction  float64       `json:"jitter_fraction"`
}

RetryPolicy is persisted per step. MaxAttempts includes crashes and defaults to three. RetryErrors defaults to false; abandonment can still redeliver.

type RuntimeHealth 

type RuntimeHealth struct {
	Running             bool
	Healthy             bool
	ConsecutiveFailures int
	LastError           error
	LastSuccessAt       time.Time
}

RuntimeHealth describes the local Run loop, not fleet-wide readiness. LastError contains runtime errors, never a recorded handler failure.

type Scheduler 

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

Scheduler recovers expired ownership and drives persisted control transitions.

func NewScheduler 

func NewScheduler(engine *Engine, pollInterval time.Duration, batchSize int) (*Scheduler, error)

NewScheduler defaults to a one-second poll and batches of 100 workflows.

func (*Scheduler) Health 

func (s *Scheduler) Health() RuntimeHealth

Health describes this instance's Run loop, not fleet-wide scheduler availability.

func (*Scheduler) Run 

func (s *Scheduler) Run(ctx context.Context) error

Run polls until cancellation, a fatal error, or exhaustion of bounded reconnects.

func (*Scheduler) RunOnce 

func (s *Scheduler) RunOnce(ctx context.Context) (int, error)

RunOnce performs one bounded reconciliation with a five-second database timeout.

type StartOption 

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

StartOption configures start deduplication, phase deadlines, or reconciliation linkage.

func WithCompensationTimeout 

func WithCompensationTimeout(timeout time.Duration) StartOption

WithCompensationTimeout bounds cleanup from its activation, independently of the forward workflow deadline. The default is 24 hours.

func WithPredecessor 

func WithPredecessor(id string) StartOption

WithPredecessor links a new reconciliation execution to a terminal execution. The identifier remains recorded if retention later removes the predecessor.

func WithStartKey 

func WithStartKey(key string) StartOption

WithStartKey deduplicates starts within a workflow name/version and database. Reusing a key with JSON-equivalent input returns the original execution, including terminal executions. A conflicting input returns ErrStartConflict. After retention, matching input returns ExecutionExpiredError with the original ID. Retention never releases a start key for reuse.

func WithWorkflowTimeout 

func WithWorkflowTimeout(timeout time.Duration) StartOption

WithWorkflowTimeout bounds forward execution from its durable start. The default is 24 hours.

type StepOption 

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

StepOption configures persisted retry, timeout, or cleanup behavior.

func WithAttemptTimeout 

func WithAttemptTimeout(timeout time.Duration) StepOption

WithAttemptTimeout bounds each delivery, including heartbeats. The default is five minutes.

func WithCompensation 

func WithCompensation(cleanup ActivitySpec) StepOption

WithCompensation declares cleanup for this step if it completes before the workflow stops. Cleanup input and policy are persisted with the forward plan.

func WithRetry 

func WithRetry(policy RetryPolicy) StepOption

WithRetry sets the step's finite claim budget and optional handler-error retries.

type SystemClock 

type SystemClock struct{}

SystemClock uses host time for local diagnostics and PostgreSQL time for durable ownership.

func (SystemClock) Now 

func (SystemClock) Now() time.Time

type TaskError 

type TaskError struct {
	WorkflowID string
	ActivityID string
	AttemptID  string
	Cause      error
}

TaskError means a handler failure was successfully recorded. RunOnce exposes it to callers; Run keeps processing other executions. Unwrap preserves causes.

func (*TaskError) Error 

func (e *TaskError) Error() string

func (*TaskError) Unwrap 

func (e *TaskError) Unwrap() error

type Ticker 

type Ticker interface {
	Ticks() <-chan time.Time
	Stop()
}

Ticker supplies coalesced runtime wakeups and must be stopped by its owner.

type TickerClock 

type TickerClock interface {
	Clock
	NewTicker(time.Duration) Ticker
}

TickerClock also controls runtime waiting. Clocks that only supply Now keep using real tickers; ManualClock supplies both for deterministic loop tests.

type TimelineEvent 

type TimelineEvent struct {
	Sequence   int64
	WorkflowID string
	ActivityID *string
	AttemptID  *string
	Type       string
	Details    json.RawMessage
	RecordedAt time.Time
}

TimelineEvent records a committed transition in increasing sequence order.

type UnsupportedActivity 

type UnsupportedActivity struct {
	Name         string
	Version      string
	InputSchema  string
	OutputSchema string
	Contract     string
	ReadyCount   int64
}

type Worker 

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

Worker runs one compatible handler at a time. Use multiple instances for concurrency.

func NewWorker 

func NewWorker(engine *Engine, config WorkerConfig, registrations ...HandlerRegistration) (*Worker, error)

NewWorker requires a nonblank ID and at least one handler. Defaults are a 30 s lease, heartbeat at one third of it, 100 ms idle poll, and 5 s shutdown grace.

func (*Worker) Health 

func (w *Worker) Health() RuntimeHealth

Health describes this instance's Run loop. RunOnce does not maintain loop health.

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context) error

Run continues after recorded task failures and cooperative lease loss. Runtime failures use bounded reconnects. Cancellation requests handler shutdown within the grace period; an unresponsive handler prevents another invocation.

func (*Worker) RunOnce 

func (w *Worker) RunOnce(ctx context.Context) (bool, error)

RunOnce claims at most one activity and waits for its handler to finish. It heartbeats while the handler runs and fences completion if ownership is lost.

func (*Worker) UnsupportedActivities 

func (w *Worker) UnsupportedActivities(ctx context.Context) ([]UnsupportedActivity, error)

UnsupportedActivities reports up to 100 ready activity types this worker cannot claim. It does not classify blocked steps or claim global worker health.

type WorkerConfig 

type WorkerConfig struct {
	ID                  string
	LeaseDuration       time.Duration
	HeartbeatInterval   time.Duration
	PollInterval        time.Duration
	DatabaseTimeout     time.Duration
	ShutdownGracePeriod time.Duration
}

WorkerConfig controls local ownership and shutdown. Zero durations select defaults; DatabaseTimeout plus HeartbeatInterval must fit inside LeaseDuration.

type WorkflowExecution 

type WorkflowExecution struct {
	DefinitionVersion      string
	InputSchema            string
	ID                     string
	DefinitionName         string
	StartKey               *string
	Input                  json.RawMessage
	Output                 json.RawMessage
	State                  WorkflowState
	CreatedAt              time.Time
	UpdatedAt              time.Time
	CompletedAt            *time.Time
	TerminalReason         string
	LastError              string
	DeadlineAt             time.Time
	Timeout                time.Duration
	PreviousExecutionID    *string
	CancelActor            string
	CancelReason           string
	CancelRequestedAt      *time.Time
	CancelSettledAt        *time.Time
	OriginalState          WorkflowState
	CompensationState      CompensationState
	CompensationError      string
	CompensationTimeout    time.Duration
	CompensationDeadlineAt *time.Time
}

WorkflowExecution is a persisted execution snapshot, including application data. Use ListWorkflows for redacted operator summaries.

func Start 

func Start[Input any](ctx context.Context, engine *Engine, definition Definition[Input], input Input, options ...StartOption) (WorkflowExecution, error)

Start persists typed input, the full plan, and initial history atomically. The definition must be registered. Use WithStartKey to resolve ambiguous commits.

Example

This example needs PostgreSQL. The repository's cmd/example also runs worker and scheduler processes, reads typed results, and demonstrates deduplication. 

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	durable "github.com/KleitonBarone/durable-task-runner"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	double := durable.DefineActivity[int, int]("double")
	definition := durable.Define("calculate", func(input int) ([]durable.ActivitySpec, error) {
		step, err := double.Step("double-input", input)
		return []durable.ActivitySpec{step}, err
	})
	registry := durable.NewRegistry()
	if err := durable.Register(registry, definition); err != nil {
		log.Fatal(err)
	}
	engine, err := durable.Open(ctx, os.Getenv("DATABASE_URL"), registry, durable.SystemClock{})
	if err != nil {
		log.Fatal(err)
	}
	defer engine.Close()
	if err := engine.Migrate(ctx); err != nil {
		log.Fatal(err)
	}
	execution, err := durable.Start(ctx, engine, definition, 21,
		durable.WithStartKey("calculation/42"))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(execution.ID)
	// Application workers register double.Handle and run alongside one scheduler.
}

type WorkflowFilter 

type WorkflowFilter struct {
	ID                string
	State             WorkflowState
	DefinitionName    string
	DefinitionVersion string
	TerminalReason    string
	CompensationState CompensationState
	CreatedBefore     *time.Time
}

WorkflowFilter combines optional exact matches and an exclusive creation cutoff.

type WorkflowState 

type WorkflowState string

WorkflowState describes forward execution, cancellation, cleanup, or a terminal outcome. 

const (
	WorkflowRunning            WorkflowState = "running"
	WorkflowCompleted          WorkflowState = "completed"
	WorkflowFailed             WorkflowState = "failed"
	WorkflowTimedOut           WorkflowState = "timed_out"
	WorkflowCancelRequested    WorkflowState = "cancel_requested"
	WorkflowCancelled          WorkflowState = "cancelled"
	WorkflowCompensating       WorkflowState = "compensating"
	WorkflowCompensationFailed WorkflowState = "compensation_failed"
)

type WorkflowSummary 

type WorkflowSummary struct {
	ID                string
	DefinitionName    string
	DefinitionVersion string
	State             WorkflowState
	CreatedAt         time.Time
	UpdatedAt         time.Time
	CompletedAt       *time.Time
	Age               time.Duration
	DeadlineAt        time.Time
	TerminalReason    string
	HasError          bool   // Failure outcome or retained diagnostic text, even if redacted.
	LastError         string `json:",omitempty"`
	CancelActor       string `json:",omitempty"`
	CancelReason      string `json:",omitempty"`
	CancelRequestedAt *time.Time
	CancelSettledAt   *time.Time
	OriginalState     WorkflowState
	CompensationState CompensationState
	CompensationError string `json:",omitempty"`
	NextActivity      *PendingActivity
}

WorkflowSummary omits payloads and keys. Diagnostic text is opt-in through PageRequest.Sensitive. Age is measured using the engine's time authority.

Source Files

View all Source files

Directories

Path Synopsis
cmd
dev command
Command dev provides the repository's local development and verification loop.
 Command dev provides the repository's local development and verification loop.
durable command
Command durable migrates and inspects an application-owned engine database.
 Command durable migrates and inspects an application-owned engine database.
example command
Command example demonstrates application-owned startup, workers, and recovery.
 Command example demonstrates application-owned startup, workers, and recovery.
internal
testdb
Package testdb owns isolated PostgreSQL schemas for integration and process tests.
 Package testdb owns isolated PostgreSQL schemas for integration and process tests.
Package telemetry optionally adapts durable runtime observations to application-owned OpenTelemetry providers.
 Package telemetry optionally adapts durable runtime observations to application-owned OpenTelemetry providers.

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.