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

selvedge

package module
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 28 Imported by: 0

README

Selvedge

Selvedge is an embedded durable-execution engine for Go. It runs inside your application and stores workflow state in a database.

Workflows coordinate activities. Activities perform I/O and record their results for replay. Signal and timer waits suspend execution until work can resume.

Derived from Romancy

Selvedge is derived from Yasushi Itoh's Romancy v0.6.2, licensed under the MIT License. Selvedge is a separate project that will diverge from Romancy.

Removed from Romancy: LLM and MCP integrations, public channels and groups, Saga compensation, transactional outbox, CloudEvents and HTTP handlers, recur, MySQL, CLI tools, examples, and website.

Engine core

  • App manages workflow execution, storage, and background tasks.
  • DefineWorkflow and RegisterWorkflow define and register typed workflows.
  • DefineActivity creates activities with retry and timeout options. RegisterActivity makes them available to recovery workers.
  • WorkflowContext provides execution state and activity identifiers.
  • The replay engine reads stored history and returns recorded results.
  • Receive, Sleep, and SleepUntil provide signal and timer waits.
  • CancelWorkflow requests cancellation of an instance. GetWorkflowResult reads its status and output.
  • Workflow hooks and hooks/otel provide observability.

hooks/otel emits a completed span for each lifecycle callback. Start, retry, and cache-hit callbacks are markers; completion spans use the reported duration. Correlate callbacks through instance and command ID attributes. Spans inherit the callback context and do not remain open across suspension or worker handoff.

Signals and cancellation

App.Signal(ctx, instanceID, channel, payload) stores a JSON payload for one workflow instance. The inbox accepts signals immediately after instance creation.

Receive[T](ctx, channel) consumes the oldest pending signal from that channel. Each channel has its own sequence. Replay reads the recorded payload from history.

// Outside the workflow:
err := app.Signal(ctx, instanceID, "confirmation", true)

// Inside the workflow:
confirmed, err := selvedge.Receive[bool](workflowCtx, "confirmation")
if err != nil {
    return false, err
}
return confirmed, nil

Channel names pass through unchanged and contain 1–255 bytes. Every signal wakes its instance, including during an active execution. Unknown instances return ErrInstanceNotFound. Closed instances return *InstanceClosedError, which also matches ErrInstanceClosed through errors.Is.

Cancellation

CancelWorkflow stores a durable request and wakes the instance. Attached scopes observe this request at blocking waits: Sleep, Receive, Await, Timer.Get, Future.Get, and ReceiveWithTimeout. A stored outcome takes precedence over cancellation. At a new Receive, cancellation takes precedence over a pending signal.

WithCancel(ctx) returns a child context and a cancel function. Detached(ctx) returns a parentless context that ignores the durable request and ancestor cancellation. These functions issue no commands and write no history. WithContext retains the same scope.

The first cancel call inserts one cancel_scope history row, even for an empty scope. One fenced transaction cancels unfinished activities and timers in that scope and its attached descendants. It also cancels their queued or running tasks and deletes their timer subscriptions. Repeated calls, including after ancestor cancellation, write nothing and consume no command position. Replay repeats the same cancel calls without new writes.

New scope and child cancellation records include their targets so replay rejects changes to which commands are cancelled. Older records remain readable, but lack the target metadata needed for that check.

A scope cancellation matches context.Canceled and is not a WorkflowCancelledError. Commands issued in a locally cancelled scope record cancellation without creating tasks or timer subscriptions. A nonpositive ReceiveWithTimeout still records an immediate timeout. Await evaluates its predicate before cancellation. IsReady, Pending, ReceiveAsync, and Now ignore cancellation.

Future.Get and Timer.Get use their issuing scope, regardless of the context passed to Get. Newly cancelled outcomes become ready at the cancel command's position. A stored outcome remains unchanged, including its delivery position.

Detached cleanup can run after cancellation:

_, cancelErr := selvedge.Receive[string](ctx, "work")
if errors.Is(cancelErr, context.Canceled) {
    if _, err := cleanup.Execute(selvedge.Detached(ctx), input); err != nil {
        return output, err
    }
}
return output, cancelErr

WithCancel(Detached(ctx)) supports its own local cancellation. Detached scopes retain lease and shutdown cancellation. WorkflowContext.Done and Err report those Go context signals, never scope cancellation. WorkflowContext.Cancel was removed. The cancel function must run on the workflow coroutine, outside Await predicates. Calls during coroutine closure write nothing.

A durable request alone does not stop activity tasks. Future.Get cancels its unfinished task when it observes the request. Scope cancellation stops its unfinished tasks immediately. Running attempts lose renewal, receive context cancellation, and cannot overwrite the recorded cancellation.

An uncaught cancellation error closes the workflow as cancelled only if a durable request exists. Without that request, a scope cancellation error closes it as failed. A workflow that catches cancellation and returns successfully completes normally. Closure cancels every unfinished task, including tasks from detached scopes.

Signals have no idempotency key.

ReceiveWithTimeout[T](ctx, channel, timeout) returns a signal or an error that matches context.DeadlineExceeded. A nonpositive timeout records an immediate timeout without consuming a pending signal. Positive waits prefer cancellation, then a pending signal, then the timer fire. A signal or cancellation retires the timer in the same transaction as its observation. A later fire cannot wake the instance or change that observation.

NewTimer(ctx, duration, opts...) returns a *Timer without suspending the workflow. Timer.Key() returns its history key. Timer.IsReady() reads the delivered view, like activity futures. Timer.Get(ctx) returns the recorded expiry time or cancellation and caches that outcome for this execution. A nonpositive duration expires immediately. Sleep and NewTimer followed by Get share the same command and support replay with either form. Timer Get consumes no command position and delivers no batch.

Await(ctx, cond) waits until a predicate is true in the delivered view. future.IsReady() reports whether the future has been delivered or a Get has already returned its outcome in this execution. Pending(ctx, channel) counts that channel's delivered, unconsumed signals. ReceiveAsync[T](ctx, channel) records either the next such signal or that none was available, without waiting or observing cancellation.

Arrival and delivery are separate. Activity workers, timer fires, and signal senders record arrivals. A completing Await or Receive, or Sleep's cancellation path, commits a delivery batch under the workflow lease. Await delivers exactly the arrivals its predicate evaluated. Other batching waits deliver all arrivals at commit time. Get never delivers a batch; passing a recorded timer fire also leaves arrivals undelivered until a later batching wait. Replay reconstructs the view from durable delivery and consumption positions, so later arrivals cannot change earlier readiness or pending counts.

Await predicates must be deterministic and side-effect free. They may read IsReady, Pending, and local values. Workflow commands inside a predicate fail the run, and a predicate panic fails the run. An immediately true pre-check returns without issuing a command or writing history. Replay runs that same pre-check, then uses the recorded observation instead of reevaluating a recorded Await's candidate batch. A never-true predicate parks the instance and costs one replay per input. Nil predicates and invalid Pending channels fail the run.

Use IsReady and Pending only on the workflow coroutine, including inside Await predicates. Off-coroutine calls to these pure reads are unsupported and are not detected.

Starting, signalling and cancelling from other processes

A client App can start with the same database configuration as a worker and register no workflows. Use StartWorkflowByName(ctx, client, name, input) to start by name, client.Signal(ctx, id, channel, payload) to send JSON payloads, and CancelWorkflow(ctx, client, id, reason) to request cancellation. These operations also accept child instance IDs.

WaitWorkflow[Output](ctx, client, id) waits for a terminal result. Workflow failures appear in result.Error; cancelling the wait context stops the wait. GetWorkflowResult reads the current result without waiting. Application error messages and typed cancellation errors survive storage and replay.

Call App.Start before using the client APIs. A start rejected before the insert matches ErrWorkflowStartRejected. An unstarted App also matches ErrAppNotStarted. Duplicate IDs, including closed and child IDs, match ErrWorkflowAlreadyStarted and cannot be reused. Storage errors can leave the start outcome uncertain: retrying an ID whose insert committed returns ErrWorkflowAlreadyStarted.

Unknown IDs return *WorkflowNotFoundError, which matches ErrInstanceNotFound. Signals to closed instances and cancellation of completed or failed instances return *InstanceClosedError with the terminal status. Cancelling an already cancelled instance succeeds. Signal encoding failures match ErrInvalidSignalPayload and do not write a signal.

Apps in one process share notifications for the same database. PostgreSQL also notifies other processes of runnable and closed instances. SQLite processes use polling. WithMaxPollInterval(d) caps workflow and activity polling delays (default one minute). WithWaitPollInterval(d) sets completion waits' fallback polling interval (default one second). Nonpositive values leave the current setting unchanged. SQLite memory databases do not share the local bus.

There are no separate run IDs: the instance ID identifies the run. RunTimeout, Memo, SearchAttributes, and start RetryPolicy are not supported.

Install

Use Go 1.27.3 or later.

go get github.com/CaliLuke/selvedge

Run a workflow

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/CaliLuke/selvedge"
)

func run() error {
    ctx := context.Background()
    app := selvedge.NewApp(selvedge.WithDatabase("file:workflow.db"))

    greet := selvedge.DefineActivity("greet",
        func(ctx context.Context, name string) (string, error) {
            return "Hello, " + name, nil
        },
    )
    workflow := selvedge.DefineWorkflow("greeting",
        func(ctx *selvedge.WorkflowContext, name string) (string, error) {
            return greet.Execute(ctx, name)
        },
    )
    if err := selvedge.RegisterActivity(app, greet); err != nil {
        return err
    }
    selvedge.RegisterWorkflow(app, workflow)

    if err := app.Start(ctx); err != nil {
        return err
    }
    defer func() { _ = app.Shutdown(ctx) }()

    id, err := selvedge.StartWorkflow(ctx, app, workflow, "Ada")
    if err != nil {
        return err
    }
    for {
        result, err := selvedge.GetWorkflowResult[string](ctx, app, id)
        if err != nil {
            return err
        }
        switch result.Status {
        case "completed":
            fmt.Println(result.Output)
            return nil
        case "failed":
            return result.Error
        }
        time.Sleep(20 * time.Millisecond)
    }
}

func main() {
    if err := run(); err != nil {
        panic(err)
    }
}

Activities start from durable task rows and run without a database transaction around user code. Each attempt claims its task before it runs. Sync Execute tries to claim and run its own task inline. If another worker holds the task, the workflow yields until the stored result is available. Execute reads results and failures from history, including after an inline attempt.

ExecuteAsync(ctx, input) returns a *Future[O] after reserving the activity. Issue several futures serially to let activity workers run them concurrently, then call future.Get(ctx) for each result. future.Key() is fixed at issue time. Get consumes no command position and repeated calls return the same value or error. It can run a claimable task inline; otherwise it yields for redispatch. Execute and ExecuteAsync use the same history command, so switching between them preserves replay compatibility. Futures belong to their issuing execution and must be used serially from workflow code. Completing without awaiting a future cancels its unfinished task.

GetActivityInfo(ctx) provides InstanceID, WorkflowName, ActivityID, ActivityName, and Attempt inside an activity. Attempt counts task claims, not retry iterations. Activities no longer receive a WorkflowContext.

Every App runs activity workers. Register each activity with RegisterActivity(app, activity) before App.Start so workers can recover pending tasks after a restart. Execute also registers its definition with the issuing App. A different definition with the same name returns ErrActivityNameConflict. Reusing the same definition pointer is allowed. Tasks remain queued if no worker registers their activity name.

WithMaxConcurrentActivities limits background attempts (default 10). WithActivityPollInterval sets the base polling interval (default one second), with adaptive backoff during idle periods. New tasks wake local workers and PostgreSQL listeners. Activity claims use the workflow lease configuration for expiry and renewal. A lost claim cancels the attempt context. Shutdown releases interrupted claims for another worker. Activities must honor context cancellation for prompt shutdown.

Activity retries, timeouts and heartbeats

Each task records its resolved ActivityOptions when it is reserved. Use WithActivityDefaults on the definition and WithActivityOptions on an invocation. Non-zero invocation fields override the defaults field by field. WithRetryPolicy and WithTimeout remain shorthands. Existing tasks keep their recorded options after code changes, including replay and another worker's retries. Definition-level NonRetryableErrors are matched by the worker running the attempt and are not recorded; per-call NonRetryableErrors are rejected.

Claim acquisition commits only after its full result has been read. Interrupted acquisition rolls back ownership and does not consume an attempt or retry budget. Once acquisition succeeds, shutdown releases the claim through its activity runner.

A claim runs one attempt. A retryable failure returns the task to the durable queue with a failure count and a not-before time. Backoff survives a restart and does not hold a workflow coroutine or worker slot. A lapsed claim counts as a failure and can be reclaimed immediately, without backoff. Success, non-retryable failure, or retry exhaustion records the single final history value. Cancellation and workflow closure stop tasks even during backoff.

GetActivityInfo(ctx) includes a copy of the recorded options. Its Attempt counts claims, starting at one. Graceful release consumes an attempt number but not the retry budget, so the number can exceed MaxAttempts. Retry and failure hooks report that same claim number and fire only after their durable write. ListActivityTasks(ctx, app, instanceID) returns tasks in issue order, including recorded options, claim and failure counts, and the last retry error.

WithActivityQueues(queues...) restricts both background workers and inline Get attempts to those queues. Omitting it serves every queue; an empty list serves none. An empty task queue name denotes the default queue. Workflow and child task queues are not supported.

StartToCloseTimeout sets the attempt's context deadline and caps its durable claim, including every renewal. HeartbeatTimeout requires the activity to call RecordHeartbeat(ctx) to retain its claim. The call marks progress in memory without database I/O; a renewal tick extends the claim only after new progress. With no heartbeat timeout, claims renew automatically. Heartbeat calls return the context cause after cancellation or claim loss, and ErrNotActivityContext outside an attempt.

ScheduleToStartTimeout limits each wait for a worker, measured from reservation, the retry's due time, graceful release, or claim lapse. Deadlines expire strictly after the stored deadline. Schedule timeout is final; start-to-close and heartbeat lapses consume retry budget. The leader records exhausted lapsed claims and schedule timeouts. ActivityTimeoutError identifies the timeout kind and matches context.DeadlineExceeded, including when nested in RetryExhaustedError.

Activities should honor their context. An activity that ignores cancellation can continue running after its claim expires while another attempt starts. Its late result cannot replace the durable outcome. Inline attempts use the same deadline and heartbeat rules; losing a claim yields the workflow and releases its worker. Pre-S15 databases must be recreated because the activity task schema changed.

Workflow code runs on one coroutine per execution. A pending Sleep or Receive yields to the dispatcher without returning. The dispatcher closes that coroutine, runs its defers, and releases the lease. A later dispatch replays from the beginning.

Workflow defers must not block: the dispatcher waits for them without a time limit. During coroutine closure, the runtime rejects commands from defers without hooks or writes. Workflow panics and unexpected runtime.Goexit calls fail the instance with *WorkflowPanicError, including the value and stack. An activity function panic also fails the instance with *WorkflowPanicError and is not retried.

Workflow code must issue workflow commands serially. Commands inside activity functions and overlapping commands from user goroutines fail the run with nondeterminism. The runtime also rejects commands while the workflow coroutine is not running. If a command reaches a blocking yield on a foreign goroutine, Yield returns a sticky nondeterminism error without a dispatcher handoff.

The runtime cannot detect non-overlapping foreign commands that finish without a blocking yield while the workflow coroutine runs pure user code. Such use remains unsupported. Activities can use GetActivityInfo to read metadata.

S7 removes the public SuspendSignal family and the unused OnEvent* hooks.

Child workflows

StartChild[O](ctx, workflowName, input, opts...) starts a child by name and returns a *Child[O]. Only the worker that runs the child needs to register its workflow. The parent command and child instance commit in one transaction.

child.ID() returns the instance ID. The default is <parentID>/<historyKey>. WithChildID(id) supplies an explicit ID, and child.Key() returns the history key. A duplicate ID returns ErrWorkflowAlreadyStarted. The recorded error also returns on replay. The parent can catch this error and continue.

child.Get(ctx) returns the stored result or yields until an outcome exists. It caches its outcome for the execution and consumes no command position or delivery batch. Child failures preserve their error envelopes. The parent can catch child panic and nondeterminism errors. Invalid result data fails the parent.

child.IsReady() reads the delivered view, like activity futures. It becomes true after a delivery batch or after Get returns. It is safe inside Await predicates. The child methods belong to their issuing execution and require serial use on the workflow coroutine.

child.Cancel(ctx) records one command and requests cancellation of a running child. It resolves an unfinished handle immediately, without waiting for the child to close. Further calls on that handle consume no command position. A stored result takes precedence over cancellation.

When the parent completes, fails, or cancels, its close transaction requests cancellation of every running child. Each child observes the request at its next blocking yield. Closing that child applies the same policy to its own children. A parent cancellation observed by a blocked Get records the outcome and requests child cancellation together.

Child RunTimeout, TaskQueue, and RetryPolicy are not supported. Scope integration and detached child handles belong to S12b.

Workflow time

Use Now(ctx) for time-based decisions in workflow code. It returns a (time.Time, error). Each live call writes one fenced history observation; values are not cached or batched. Replay returns the recorded value, even if the deadline has passed since the previous execution.

Values use UTC and contain no monotonic clock reading. Live and replay values compare identically with both == and Equal. Each execution tracks the largest time returned so far, including replayed values and timer expiries observed by Sleep or Timer.Get. Contexts made with WithContext share this floor, so a backward clock step cannot lower a later live value. On error, Now returns the floor and a sticky error that fails the execution even if workflow code ignores it.

WithClock(func() time.Time) configures the wall clock used by Now and timer expiry; the default (also for a nil clock) is time.Now. Sleep and NewTimer set expiry to the larger of that clock and the workflow time floor, plus the duration. Nonpositive durations still expire immediately. SleepUntil computes its duration against that same base. Replay keeps the stored expiry. Observing a fire advances the floor, so a later live Now cannot precede it. Issuing a timer or checking IsReady does not advance the floor. Lease clocks and the timer poller's store clock are unchanged.

Like other workflow commands, Now must be called serially from workflow code. Calls while the coroutine is closing return an error and the floor, without writing history.

Queries and status

SetQueryHandler(ctx, name, handler) registers a zero-argument handler for the current execution. A registration replaces the previous handler with that name. A nil handler removes it. Names contain 1–255 bytes. Registration consumes no command position and writes no history. Live executions store handlers without calling them. Register handlers on the workflow coroutine, outside Await predicates.

// Inside the workflow:
err := selvedge.SetQueryHandler(ctx, "pending", func() (int, error) {
    return selvedge.Pending(ctx, "results"), nil
})

// Outside the workflow:
count, err := selvedge.Query[int](ctx, app, instanceID, "pending")

Each Query takes one consistent database snapshot and replays the workflow from its beginning. It stops before the first unrecorded command, blocking wait, or other storage action that requires live state. The selected handler runs before that replay closes. Each query therefore costs one full replay. Answers are not recorded. A later query can reach a later position and return a different answer.

In-memory SQLite databases (":memory:" with shared cache) do not use WAL. A snapshot on these databases blocks writers for the duration of its read transaction.

Handlers can read local values, Pending, and activity, timer, or child IsReady. These reads use the delivered view at the replay position. Arrived events stay invisible until a recorded delivery makes them visible. A recorded Get can return without delivering other events.

Queries take no lease, fire no hooks, and write no workflow state. Commands and guarded calls in handlers return ErrQueryCommand and fail only the query. Handler errors and panics return QueryFailedError, which preserves wrapped errors. Results always pass through JSON encoding and decoding into the requested type. Encoding and decoding errors also return QueryFailedError.

Queries work on running and closed instances. The querying App must register the workflow code. Otherwise, Query returns WorkflowNotRegisteredError. A handler absent at the replay position returns QueryNotFoundError. An unknown instance returns WorkflowNotFoundError. Replay mismatches return the original replay error. Caller cancellation is checked at command boundaries. Workflow code and handlers must not perform external I/O or block outside workflow commands.

Describe(ctx, app, instanceID) returns WorkflowDescription with the workflow name, status, cancellation request and reason, parent links, and timestamps. WorkflowStatus has four values: StatusRunning, StatusCompleted, StatusFailed, and StatusCancelled. A newly created instance reports StatusRunning before execution. Describe returns WorkflowNotFoundError for an unknown instance.

Replay checks

Use ExportHistory(ctx, app, instanceID) to capture one consistent snapshot of an instance. The App must have initialized storage. Marshal the returned RecordedHistory with json.Marshal, and load it with ParseHistory. The selvedge.history/v1 format contains command history, workflow metadata, and signal delivery positions. It excludes ownership, leases, and signal payloads. Missing event values remain distinct from recorded JSON null results. ParseHistory rejects unknown fields and unsupported formats or versions.

Register the workflow on an App, then call ReplayHistory(ctx, app, history) to check its code against the snapshot. This App need not be started. Replay takes no lease, writes no state, fires no hooks, and neither registers nor executes activities. Workflow code must still avoid external I/O and blocking outside workflow commands. Query and replay checks use the same read-only execution path.

A running history can stop at its recorded frontier. A closed history must reach the recorded close: completed outputs must contain the same JSON value, failed outcomes must preserve the error kind and message, and cancelled outcomes must preserve cancellation. JSON object order and equivalent number spellings do not change an output. Command changes and leftover commands produce NondeterminismError; changed outcomes produce ReplayOutcomeError. ReplayError wraps the divergence and the workflow's original Go error, so errors.Is and errors.As can inspect either cause.

ReplayInstance combines export and verification. For example, check one page of running instances using registered workflow code:

func checkRunning(ctx context.Context, app *selvedge.App) error {
    instances, err := app.FindInstancesWithOptions(ctx, selvedge.FindInstancesOptions{
        StatusFilter: selvedge.StatusRunning,
        Limit: 100,
    })
    if err != nil {
        return err
    }
    for _, instance := range instances {
        if err := selvedge.ReplayInstance(ctx, app, instance.InstanceID); err != nil {
            return err
        }
    }
    return nil
}

Check exported histories before deployment, and drain running instances before changing their workflow code. A successful check covers only the recorded history; it does not predict commands on paths the instance has not reached.

Storage and migrations

SQLite and PostgreSQL are the supported storage backends. PostgreSQL also supports LISTEN/NOTIFY for workflow wakeups.

WithDatabase accepts a SQLite path, a file: URL, a sqlite:// URL, or a PostgreSQL connection URL.

App.Start applies embedded migrations by default. WithAutoMigrate(false) disables this behavior. WithMigrationsFS supplies an alternative migration filesystem.

internal/migrations/sql/ is the schema source of truth. The module embeds the migration files for each dialect, and tests apply those same files. Released migrations are immutable. Add schema changes as new timestamped migrations for both SQLite and PostgreSQL.

Version 0.7.0 is the first Selvedge release and establishes its schema baseline. Automatic upgrades from Romancy or earlier development schemas are not supported; use a fresh database for this release. Those schemas can share the initial migration version while lacking required tables, columns, or constraints, so App.Start cannot upgrade them by reapplying the initial migration.

Error envelopes use snake_case JSON fields, including panic and nondeterminism details. Older development envelope formats have no compatibility shim.

Schema application is transactional: SQLite uses BEGIN IMMEDIATE; PostgreSQL takes a transaction-scoped advisory lock before checking schema_migrations. SQL errors abort and roll back the schema application.

Migration lock waits use the context passed to App.Start. Without a deadline, startup can wait indefinitely unless that context is canceled. This applies to SQLite writer locks and PostgreSQL advisory locks.

Development

go build ./...
go vet ./...
go test -race ./...
make lint

For PostgreSQL storage tests, start the development database and run:

docker compose up -d postgres
SELVEDGE_TEST_POSTGRES_URL='postgres://selvedge:selvedge@localhost:5433/selvedge_test?sslmode=disable' \
    go test -tags=postgres,pglocal ./internal/storage

Tests with the integration build tag use PostgreSQL testcontainers by default and require Docker. Set SELVEDGE_TEST_POSTGRES_URL to reuse a test server instead; each fixture creates and removes its own database. That test account must have permission to create databases.

go test -race -tags=integration -count=1 -timeout=25m ./...

CI runs the race detector, both PostgreSQL storage suites, and the integration suite against a PostgreSQL service with separate databases for each fixture.

Dispatch filters use registered workflow names. The storage FindDispatchable variadic names filter treats omitted names or a nil slice as all workflows, and a non-nil empty slice as none.

Workers sharing a database must use distinct workflow names for different implementations. Dispatch does not scope names by owner_service; workers that register the same name can execute each other's instances. Use separate databases when services need independent namespaces.

WorkflowContext.Storage() exposes read-only history lookup for diagnostics. It does not expose database connections or storage mutations.

The development Compose file also includes Jaeger for OpenTelemetry traces.

License

This project is licensed under the MIT License - see the LICENSE file for details.

Documentation

Overview

Package selvedge provides a durable execution framework for Go.

Index

Examples

Constants

View Source 
const (
	StatusRunning   = storage.StatusRunning
	StatusCompleted = storage.StatusCompleted
	StatusFailed    = storage.StatusFailed
	StatusCancelled = storage.StatusCancelled
)

Variables

View Source 
var ErrActivityNameConflict = errors.New("activity name already registered with another definition")

ErrActivityNameConflict rejects different definitions registered under one name.

View Source 
var ErrAppNotStarted = errors.New("app not started")

ErrAppNotStarted means the App has no initialized storage or has shut down.

View Source 
var ErrDeterminismViolation = replay.ErrDeterminismViolation

ErrDeterminismViolation indicates non-deterministic behavior during replay.

View Source 
var ErrFenced = storage.ErrFenced

ErrFenced means another execution owns the instance.

View Source 
var ErrHistoryKeyConflict = storage.ErrHistoryKeyConflict

ErrHistoryKeyConflict indicates duplicate claims or conflicting stored commands.

View Source 
var ErrInstanceClosed = storage.ErrInstanceClosed
View Source 
var ErrInstanceNotFound = storage.ErrInstanceNotFound
View Source 
var ErrInvalidSignalChannel = storage.ErrInvalidSignalChannel
View Source 
var ErrInvalidSignalPayload = errors.New("invalid signal payload")

ErrInvalidSignalPayload means a signal payload could not be encoded as JSON.

View Source 
var ErrInvalidWorkflowState = errors.New("invalid workflow state")

ErrInvalidWorkflowState indicates an invalid workflow state transition.

View Source 
var ErrLeaseLost = replay.ErrLeaseLost

ErrLeaseLost means the execution abandoned its expired local lease belief.

View Source 
var ErrNotActivityContext = errors.New("not an activity context")

ErrNotActivityContext rejects heartbeats outside an activity attempt.

View Source 
var ErrQueryCommand = replay.ErrQueryCommand

ErrQueryCommand matches an attempted workflow command in a query handler.

View Source 
var ErrUnsupportedHistoryFormat = errors.New("unsupported history format")

ErrUnsupportedHistoryFormat rejects unknown history formats or versions.

View Source 
var ErrWorkflowAlreadyCancelled = storage.ErrWorkflowAlreadyCancelled

ErrWorkflowAlreadyCancelled indicates an operation on a cancelled workflow.

View Source 
var ErrWorkflowAlreadyCompleted = storage.ErrWorkflowAlreadyCompleted

ErrWorkflowAlreadyCompleted indicates an operation on a completed workflow.

View Source 
var ErrWorkflowAlreadyFailed = storage.ErrWorkflowAlreadyFailed

ErrWorkflowAlreadyFailed indicates an operation on a failed workflow.

View Source 
var ErrWorkflowAlreadyStarted = storage.ErrWorkflowAlreadyStarted

ErrWorkflowAlreadyStarted indicates that an instance ID already exists.

View Source 
var ErrWorkflowStartRejected = errors.New("workflow start rejected")

ErrWorkflowStartRejected means no instance insert was attempted. Duplicate IDs and uncertain storage failures do not match this error.

Functions

func Await 

func Await(ctx *WorkflowContext, cond func() bool) error

Await waits until cond is true in the delivered view. Predicates must be deterministic and side-effect free; only IsReady, Pending and local reads are allowed. A never-true predicate parks at one replay per input. Replay runs the pre-check but never reevaluates a recorded command's candidate batch.

func CancelWorkflow 

func CancelWorkflow(ctx context.Context, app *App, instanceID, reason string) error

CancelWorkflow durably requests cancellation and wakes a running instance. Attached scopes observe it at blocking waits. Detached scopes ignore it. A request alone does not stop activity tasks. Future.Get cancels its unfinished task when it observes the request. A stored outcome wins over cancellation. WithCancel provides local cancellation. Detached supports cleanup after either form of cancellation. Closure cancels all unfinished tasks, including detached tasks. A workflow that catches cancellation can complete normally.

func EmbeddedMigrationsFS 

func EmbeddedMigrationsFS() fs.FS

EmbeddedMigrationsFS returns the embedded schemas rooted at sqlite/ and postgresql/. internal/migrations/sql/ is the only schema source of truth.

func IsTerminalError 

func IsTerminalError(err error) bool

IsTerminalError returns true if the error is or wraps a TerminalError.

func Now 

func Now(ctx *WorkflowContext) (value time.Time, err error)

Now returns recorded workflow time in UTC, without a monotonic reading. Each live call writes one fenced history observation; replay returns that observation. Live time never falls below earlier Now values or timer expiries observed by Sleep or Timer.Get in this execution. On error, Now returns the last observed time and the run fails even if the caller ignores the error.

func Pending 

func Pending(ctx *WorkflowContext, channel string) int

Pending counts delivered, unconsumed signals at the current command position. It performs no I/O and is safe inside Await predicates. Like IsReady, it must run serially on the workflow coroutine; off-coroutine use is not detected.

func Query 

func Query[R any](ctx context.Context, app *App, instanceID, name string) (R, error)

Query answers from a consistent snapshot replay in the calling App. Each call costs one full replay, takes no lease, and writes no workflow state. The result always passes through JSON into R, including when R is any. In-memory SQLite snapshots (":memory:" with shared cache, without WAL) block writers until the snapshot read transaction ends.

func Receive 

func Receive[T any](ctx *WorkflowContext, channel string) (T, error)

Receive returns the next signal from this instance's channel in arrival order. Replay returns the recorded payload. Cancellation is observed before consuming a pending signal. An empty inbox yields to the dispatcher without returning.

func ReceiveAsync 

func ReceiveAsync[T any](ctx *WorkflowContext, channel string) (value T, received bool, err error)

ReceiveAsync records the next delivered signal, or records that none was available. It never yields, delivers new events, or observes cancellation.

func ReceiveWithTimeout 

func ReceiveWithTimeout[T any](ctx *WorkflowContext, channel string, timeout time.Duration) (T, error)

ReceiveWithTimeout receives the oldest signal on channel or returns an error matching context.DeadlineExceeded. A nonpositive timeout records an immediate timeout without consuming signals. Positive waits prefer cancellation, then a signal, then the timer fire, and retire a losing timer atomically.

func RecordHeartbeat 

func RecordHeartbeat(ctx context.Context) error

RecordHeartbeat records progress for the current activity attempt without I/O. It returns the context's cancellation cause after the attempt has stopped.

func RegisterActivity 

func RegisterActivity[I, O any](app *App, a *Activity[I, O]) error

RegisterActivity makes an activity available for recovered tasks on this App. Register the same definition pointer again freely; a different definition is a conflict.

func RegisterWorkflow 

func RegisterWorkflow[I, O any](app *App, workflow Workflow[I, O])

RegisterWorkflow registers a workflow with the application. The workflow can later be started by name or by type.

func ReplayHistory 

func ReplayHistory(ctx context.Context, app *App, history *RecordedHistory) error

ReplayHistory checks registered code against a history without live storage or activities.

func ReplayInstance 

func ReplayInstance(ctx context.Context, app *App, id string) error

ReplayInstance exports one instance and checks it against the App's registered workflow.

func SetQueryHandler 

func SetQueryHandler[R any](ctx *WorkflowContext, name string, handler func() (R, error)) error

SetQueryHandler replaces an execution-local handler without issuing a command. A nil handler removes the name. Call only on the workflow coroutine.

func Sleep 

func Sleep(ctx *WorkflowContext, duration time.Duration, opts ...SleepOption) error

Sleep suspends the workflow for a specified duration. The workflow will resume after the duration has elapsed.

During replay, if the timer already fired, this returns immediately. Otherwise, it reserves and arms the timer under the execution fence, then yields to the dispatcher. The engine releases the lease and preserves pending input.

When the timer expires, the workflow will be resumed.

func SleepUntil 

func SleepUntil(ctx *WorkflowContext, t time.Time, opts ...SleepOption) error

SleepUntil suspends the workflow until a specific time.

func StartWorkflow 

func StartWorkflow[I, O any](
	ctx context.Context,
	app *App,
	workflow Workflow[I, O],
	input I,
	opts ...StartOption,
) (string, error)

StartWorkflow starts a new workflow instance. Returns the instance ID.

func StartWorkflowByName 

func StartWorkflowByName(ctx context.Context, app *App, workflow string, input any, opts ...StartOption) (string, error)

StartWorkflowByName starts a workflow without requiring local registration. The returned instance ID also identifies the run; IDs cannot be reused.

Types

type Activity 

type Activity[I, O any] struct {
	// contains filtered or unexported fields
}

Activity represents a single unit of work within a workflow. Activities are the only way to perform I/O or side effects in workflows. I is the input type, O is the output type.

func DefineActivity 

func DefineActivity[I, O any](
	name string,
	fn func(ctx context.Context, input I) (O, error),
	opts ...ActivityOption[I, O],
) *Activity[I, O]

DefineActivity creates a new activity.

func (*Activity[I, O]) Execute 

func (a *Activity[I, O]) Execute(ctx *WorkflowContext, input I, opts ...ExecuteOption) (O, error)

Execute issues the activity and waits for its durable outcome.

func (*Activity[I, O]) ExecuteAsync 

func (a *Activity[I, O]) ExecuteAsync(ctx *WorkflowContext, input I, opts ...ExecuteOption) (*Future[O], error)

ExecuteAsync reserves an activity in issue order without waiting for its result. Get reads the durable outcome. Un-awaited tasks are cancelled when the workflow closes.

func (*Activity[I, O]) Name 

func (a *Activity[I, O]) Name() string

Name returns the activity name.

type ActivityInfo 

type ActivityInfo struct {
	InstanceID, WorkflowName, ActivityID, ActivityName string
	Attempt                                            int
	Options                                            ActivityOptions
}

ActivityInfo is immutable metadata for the current durable activity claim.

func GetActivityInfo 

func GetActivityInfo(ctx context.Context) *ActivityInfo

GetActivityInfo returns a copy of claim metadata, or nil outside an activity.

type ActivityOption 

type ActivityOption[I, O any] func(*Activity[I, O])

ActivityOption configures an activity.

func WithActivityDefaults 

func WithActivityDefaults[I, O any](options ActivityOptions) ActivityOption[I, O]

WithActivityDefaults sets the definition's activity execution options.

func WithRetryPolicy 

func WithRetryPolicy[I, O any](policy *retry.Policy) ActivityOption[I, O]

WithRetryPolicy sets the retry policy for the activity.

func WithTimeout 

func WithTimeout[I, O any](d time.Duration) ActivityOption[I, O]

WithTimeout sets the timeout for each activity execution attempt.

type ActivityOptions 

type ActivityOptions struct {
	Queue                                                         string
	StartToCloseTimeout, ScheduleToStartTimeout, HeartbeatTimeout time.Duration
	RetryPolicy                                                   *retry.Policy
}

ActivityOptions configures the durable execution policy recorded at reservation. Non-zero per-call fields override definition defaults. A nil policy inherits.

type ActivityTaskInfo 

type ActivityTaskInfo struct {
	Key, Name, Queue, Status string
	Seq                      int64
	Options                  ActivityOptions
	Attempts, Failures       int
	LastError                error
}

ActivityTaskInfo describes the recorded policy and execution state of an activity.

func ListActivityTasks 

func ListActivityTasks(ctx context.Context, app *App, id string) ([]ActivityTaskInfo, error)

ListActivityTasks returns the instance's activities in reservation order.

type ActivityTimeoutError 

type ActivityTimeoutError struct {
	Activity string
	Kind     TimeoutKind
	Attempt  int
}

ActivityTimeoutError describes a recorded activity deadline failure.

func (*ActivityTimeoutError) Error 

func (e *ActivityTimeoutError) Error() string

func (*ActivityTimeoutError) Unwrap 

func (e *ActivityTimeoutError) Unwrap() error

Unwrap identifies a timeout as context.DeadlineExceeded.

type App 

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

App is the main entry point for Selvedge. It manages workflow execution, storage, and background tasks.

func NewApp 

func NewApp(opts ...Option) *App

NewApp creates a new Selvedge application.

func (*App) FindInstances 

func (a *App) FindInstances(ctx context.Context, inputFilters map[string]any) ([]*storage.WorkflowInstance, error)

FindInstances searches for workflow instances with input filters. This is a convenience method for searching workflows by their input data.

The inputFilters parameter maps JSON paths to expected values:

  • "customer.id" -> "12345" matches input like {"customer": {"id": "12345"}}
  • "status" -> "active" matches input like {"status": "active"}

Values are compared with exact match. Supported value types:

  • string: Exact string match
  • int/float64: Numeric comparison
  • bool: Boolean comparison
  • nil: Matches null values

Example: 

instances, err := app.FindInstances(ctx, map[string]any{
    "order.customer_id": "cust_123",
    "order.status": "pending",
})

func (*App) FindInstancesWithOptions 

func (a *App) FindInstancesWithOptions(ctx context.Context, opts storage.ListInstancesOptions) ([]*storage.WorkflowInstance, error)

FindInstancesWithOptions searches for workflow instances with full options. Use this when you need pagination, status filters, or other advanced options in addition to input filters.

Example

This example compiles using only the public package. Register workflows and start app before calling it with stored instances. 

package main

import (
	"context"

	"github.com/CaliLuke/selvedge"
)

func main() {
	ctx := context.Background()
	app := selvedge.NewApp()
	instances, err := app.FindInstancesWithOptions(ctx, selvedge.FindInstancesOptions{
		StatusFilter: selvedge.StatusRunning,
		Limit:        100,
	})
	if err != nil {
		return
	}
	for _, instance := range instances {
		if err := selvedge.ReplayInstance(ctx, app, instance.InstanceID); err != nil {
			return
		}
	}
}

func (*App) GetInstance 

func (a *App) GetInstance(ctx context.Context, instanceID string) (*storage.WorkflowInstance, error)

GetInstance retrieves a workflow instance by ID.

func (*App) Shutdown 

func (a *App) Shutdown(ctx context.Context) error

Shutdown gracefully shuts down the application. Cleanup continues if the caller times out; subsequent calls wait for the same cleanup operation.

func (*App) Signal 

func (a *App) Signal(ctx context.Context, instanceID, channel string, payload any) error

Signal durably appends a JSON payload to one instance's named inbox and wakes it. Signals are accepted from instance creation until closure, including while an execution holds the lease. Channel names pass through unchanged (1–255 bytes).

func (*App) Start 

func (a *App) Start(ctx context.Context) error

Start initializes the application and starts background tasks.

func (*App) Storage 

func (a *App) Storage() storage.Storage

Storage returns the storage instance for advanced usage.

func (*App) WorkerID 

func (a *App) WorkerID() string

WorkerID returns the worker ID.

type ApplicationError 

type ApplicationError struct {
	Type         string
	Message      string
	NonRetryable bool
	Details      any
	Cause        error
}

ApplicationError carries a portable activity or workflow failure. Details must be JSON-serializable and is returned as JSON-decoded data.

func NewApplicationError 

func NewApplicationError(message, errType string, details any, cause error) *ApplicationError

NewApplicationError constructs a retryable application failure.

func NewNonRetryableApplicationError 

func NewNonRetryableApplicationError(message, errType string, details any, cause error) *ApplicationError

NewNonRetryableApplicationError constructs a failure that must not be retried.

func (*ApplicationError) Error 

func (e *ApplicationError) Error() string

Error returns the message, followed by the cause when one is set.

func (*ApplicationError) Unwrap 

func (e *ApplicationError) Unwrap() error

Unwrap returns the cause, so errors.Is and errors.As see through the error.

type Child 

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

Child is a child workflow outcome belonging to its issuing execution. Use its methods serially from the workflow coroutine.

func StartChild 

func StartChild[O any](ctx *WorkflowContext, workflow string, input any, opts ...ChildOption) (*Child[O], error)

StartChild starts a workflow by name. Registration is required only on a worker that executes the child. The parent command and child creation commit together. A locally cancelled scope records cancellation without creating an instance.

func (*Child[O]) Cancel 

func (c *Child[O]) Cancel(ctx *WorkflowContext) error

Cancel records one child_cancel command per handle, resolves an unfinished handle immediately, and requests cancellation of its running child.

func (*Child[O]) Get 

func (c *Child[O]) Get(ctx *WorkflowContext) (O, error)

Get returns the stored outcome and consumes no command position or delivery batch. Cancellation follows the issuing scope, regardless of ctx's scope. A stored outcome wins over cancellation.

func (*Child[O]) ID 

func (c *Child[O]) ID() string

ID returns the child ID, even when local cancellation prevented its creation.

func (*Child[O]) IsReady 

func (c *Child[O]) IsReady() bool

IsReady reads only the delivered view or an outcome observed by Get. It is safe in Await predicates; off-coroutine calls are unsupported and not detected.

func (*Child[O]) Key 

func (c *Child[O]) Key() string

Key returns the parent's child command key.

type ChildOption 

type ChildOption func(*childOptions)

ChildOption configures a child workflow start.

func WithChildID 

func WithChildID(id string) ChildOption

WithChildID supplies a stable instance ID instead of parentID/historyKey.

type ExecuteOption 

type ExecuteOption func(*executeOptions)

ExecuteOption configures activity execution.

func WithActivityID 

func WithActivityID(id string) ExecuteOption

WithActivityID specifies a manual activity ID. Commands must be issued serially from workflow code. IDs starting with $ or # are reserved. Execute rejects these IDs with ErrHistoryKeyConflict before running the activity.

func WithActivityOptions 

func WithActivityOptions(options ActivityOptions) ExecuteOption

WithActivityOptions overrides non-zero fields for this activity reservation.

type FindInstancesOptions 

type FindInstancesOptions = storage.ListInstancesOptions

FindInstancesOptions filters and pages instances for App.FindInstancesWithOptions.

type Future 

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

Future is an activity outcome tied to the workflow execution that issued it. Use it serially from that workflow's coroutine. Get may be called repeatedly.

func (*Future[O]) Get 

func (f *Future[O]) Get(ctx *WorkflowContext) (O, error)

Get returns the stored outcome, or yields until it is available. It consumes no command position. A stored outcome takes precedence over cancellation.

func (*Future[O]) IsReady 

func (f *Future[O]) IsReady() bool

IsReady reports a delivered or previously observed outcome. It consumes no position and performs no I/O. Use only on the workflow coroutine, including Await predicates; off-coroutine calls are unsupported and not detected.

func (*Future[O]) Key 

func (f *Future[O]) Key() string

Key returns the history key assigned when the activity was issued.

type InstanceClosedError 

type InstanceClosedError = storage.InstanceClosedError

InstanceClosedError reports the terminal status that rejected a signal.

type LockAcquisitionError 

type LockAcquisitionError struct {
	InstanceID string
	WorkerID   string
	Reason     string
}

LockAcquisitionError indicates failure to acquire a workflow lock.

func (*LockAcquisitionError) Error 

func (e *LockAcquisitionError) Error() string

type NondeterminismError 

type NondeterminismError = replay.NondeterminismError

NondeterminismError reports a replay command divergence.

type Option 

type Option func(*appConfig)

Option configures an App.

func WithActivityPollInterval 

func WithActivityPollInterval(d time.Duration) Option

WithActivityPollInterval sets the base activity poll interval before adaptive backoff.

func WithActivityQueues 

func WithActivityQueues(queues ...string) Option

WithActivityQueues restricts workers and inline attempts to these queues. Without this option an App serves every queue; an empty list serves none.

func WithAutoMigrate 

func WithAutoMigrate(enabled bool) Option

WithAutoMigrate enables or disables automatic database migrations on startup.

When enabled (default), selvedge will automatically apply pending dbmate-compatible migrations from the embedded internal/migrations/sql/ directory during App.Start().

This is compatible with the dbmate CLI tool and uses the same schema_migrations table for tracking applied migrations.

Set to false if you prefer to manage migrations manually using dbmate CLI: 

dbmate -d internal/migrations/sql/sqlite up

func WithClock 

func WithClock(clock func() time.Time) Option

WithClock sets the wall clock used by Now and timer expiry. A nil clock uses time.Now. It does not change lease clocks or the timer poller.

func WithDatabase 

func WithDatabase(url string) Option

WithDatabase sets the database connection URL. Supported formats:

  • SQLite: "file:path/to/db.db" or "sqlite://path/to/db.db"
  • PostgreSQL: "postgres://user:pass@host:port/dbname"

func WithHooks 

func WithHooks(h hooks.WorkflowHooks) Option

WithHooks sets the workflow lifecycle hooks. A nil value installs NoOpHooks, replacing any hooks configured by earlier options.

func WithLeaderHeartbeatInterval 

func WithLeaderHeartbeatInterval(d time.Duration) Option

WithLeaderHeartbeatInterval sets the interval for leader election heartbeat. The leader will renew its lease at this interval. Default: 15 seconds.

func WithLeaderLeaseDuration 

func WithLeaderLeaseDuration(d time.Duration) Option

WithLeaderLeaseDuration sets the duration for which a leader holds its lease. Should be at least 3x the heartbeat interval to allow for missed heartbeats. Default: 45 seconds.

func WithLease 

func WithLease(ttl, renewInterval, margin, maxSkew time.Duration) Option

WithLease configures execution leases. Start rejects unsafe configurations.

func WithListenNotify 

func WithListenNotify(enabled *bool) Option

WithListenNotify configures PostgreSQL LISTEN/NOTIFY usage. - nil (default): auto-detect based on database URL (enabled for PostgreSQL) - true: force enable (fails if not PostgreSQL) - false: force disable (use polling only)

func WithMaxConcurrentActivities 

func WithMaxConcurrentActivities(n int) Option

WithMaxConcurrentActivities bounds the number of background activity attempts.

func WithMaxConcurrentResumptions 

func WithMaxConcurrentResumptions(n int) Option

WithMaxConcurrentResumptions sets the maximum number of concurrent workflow resumptions. This limits goroutine spawning in background tasks to prevent resource exhaustion. Default: 10.

func WithMaxConcurrentTimers 

func WithMaxConcurrentTimers(n int) Option

WithMaxConcurrentTimers sets the maximum number of concurrent timer handlers. Default: 10.

func WithMaxPollInterval 

func WithMaxPollInterval(d time.Duration) Option

WithMaxPollInterval caps workflow and activity polling delays, including jitter. The default is one minute. Nonpositive values leave the configured cap unchanged.

func WithMaxTimersPerBatch 

func WithMaxTimersPerBatch(n int) Option

WithMaxTimersPerBatch sets the maximum number of timers to process per polling cycle. Default: 100.

func WithMaxWorkflowsPerBatch 

func WithMaxWorkflowsPerBatch(n int) Option

WithMaxWorkflowsPerBatch sets the maximum number of workflows to process per polling cycle. Default: 100.

func WithMigrationsFS 

func WithMigrationsFS(migrationsFS fs.FS) Option

WithMigrationsFS sets a custom filesystem for database migrations.

By default, selvedge uses embedded migrations from internal/migrations/sql/. Use this option to provide custom migrations from a different source.

The filesystem should contain subdirectories for each database type:

  • sqlite/
  • postgresql/

Each subdirectory should contain .sql files in dbmate format with -- migrate:up and -- migrate:down sections.

func WithNotifyReconnectDelay 

func WithNotifyReconnectDelay(d time.Duration) Option

WithNotifyReconnectDelay sets the delay before reconnecting after a LISTEN/NOTIFY connection failure. Default: 60 seconds.

func WithServiceName 

func WithServiceName(name string) Option

WithServiceName sets the service name for identification.

func WithShutdownTimeout 

func WithShutdownTimeout(d time.Duration) Option

WithShutdownTimeout sets the timeout for graceful shutdown.

func WithTimerCheckInterval 

func WithTimerCheckInterval(d time.Duration) Option

WithTimerCheckInterval sets the interval for checking expired timers.

func WithWaitPollInterval 

func WithWaitPollInterval(d time.Duration) Option

WithWaitPollInterval sets the completion wait fallback interval (default one second). Nonpositive values leave the configured interval unchanged.

func WithWorkerID 

func WithWorkerID(id string) Option

WithWorkerID sets a custom worker ID. If not set, a UUID will be generated.

func WithWorkflowResumptionInterval 

func WithWorkflowResumptionInterval(d time.Duration) Option

WithWorkflowResumptionInterval sets the interval for the workflow resumption task. This task finds runnable workflows and expired leases, then dispatches them. This is essential for load balancing in multi-worker environments. Default: 1 second.

type QueryFailedError 

type QueryFailedError struct {
	Name string
	Err  error
}

QueryFailedError wraps a handler error, panic, or JSON conversion failure.

func (*QueryFailedError) Error 

func (e *QueryFailedError) Error() string

func (*QueryFailedError) Unwrap 

func (e *QueryFailedError) Unwrap() error

type QueryNotFoundError 

type QueryNotFoundError struct{ Name string }

QueryNotFoundError means no handler of this name exists at the replay stop.

func (*QueryNotFoundError) Error 

func (e *QueryNotFoundError) Error() string

type RecordedEvent 

type RecordedEvent struct {
	Seq          int64           `json:"seq"`
	Key          string          `json:"key"`
	Command      string          `json:"command"`
	EventType    string          `json:"event_type"`
	Data         json.RawMessage `json:"data,omitempty"`
	DataType     string          `json:"data_type"`
	Binary       []byte          `json:"binary"`
	Reservation  json.RawMessage `json:"reservation,omitempty"`
	DeliveredSeq int64           `json:"delivered_seq"`
}

RecordedEvent contains one command or workflow-close history row.

type RecordedHistory 

type RecordedHistory struct {
	Format   string           `json:"format"`
	Instance RecordedInstance `json:"instance"`
	Events   []RecordedEvent  `json:"events"`
	Signals  []RecordedSignal `json:"signals"`
}

RecordedHistory is a portable snapshot of the inputs needed to replay one instance.

func ExportHistory 

func ExportHistory(ctx context.Context, app *App, id string) (*RecordedHistory, error)

ExportHistory reads one consistent snapshot from an initialized App.

func ParseHistory 

func ParseHistory(data []byte) (*RecordedHistory, error)

ParseHistory strictly decodes the portable Selvedge history format.

type RecordedInstance 

type RecordedInstance struct {
	InstanceID       string          `json:"instance_id"`
	WorkflowName     string          `json:"workflow_name"`
	Status           WorkflowStatus  `json:"status"`
	Input            json.RawMessage `json:"input,omitempty"`
	Output           json.RawMessage `json:"output,omitempty"`
	CancelRequested  bool            `json:"cancel_requested"`
	CancelReason     string          `json:"cancel_reason"`
	ParentInstanceID string          `json:"parent_instance_id"`
	ParentKey        string          `json:"parent_key"`
}

RecordedInstance contains durable workflow metadata without execution ownership.

type RecordedSignal 

type RecordedSignal struct {
	ID           int64  `json:"id"`
	Channel      string `json:"channel"`
	DeliveredSeq int64  `json:"delivered_seq"`
	ConsumedSeq  int64  `json:"consumed_seq"`
}

RecordedSignal preserves delivery and consumption positions without its payload.

type ReplayDecodeError 

type ReplayDecodeError struct {
	InstanceID string
	Key        string
	GoType     string
	Err        error
}

ReplayDecodeError reports a recorded value that cannot be decoded into the Go output type. It is a workflow failure, not a retryable activity failure.

func (*ReplayDecodeError) Error 

func (e *ReplayDecodeError) Error() string

func (*ReplayDecodeError) Unwrap 

func (e *ReplayDecodeError) Unwrap() error

type ReplayError 

type ReplayError struct {
	InstanceID  string
	Divergence  error
	WorkflowErr error
}

ReplayError preserves both the replay divergence and the live workflow error.

func (*ReplayError) Error 

func (e *ReplayError) Error() string

func (*ReplayError) Unwrap 

func (e *ReplayError) Unwrap() []error

Unwrap exposes the divergence and original Go error for errors.Is and errors.As.

type ReplayOutcomeError 

type ReplayOutcomeError struct {
	InstanceID                     string
	RecordedStatus, ReplayedStatus WorkflowStatus
	Recorded, Replayed             json.RawMessage
}

ReplayOutcomeError reports a workflow close that differs from its recorded outcome.

func (*ReplayOutcomeError) Error 

func (e *ReplayOutcomeError) Error() string

type RetryExhaustedError 

type RetryExhaustedError struct {
	ActivityName string
	Attempts     int
	LastErr      error
}

RetryExhaustedError indicates that all retry attempts have been exhausted.

func (*RetryExhaustedError) Error 

func (e *RetryExhaustedError) Error() string

func (*RetryExhaustedError) Unwrap 

func (e *RetryExhaustedError) Unwrap() error

type SleepOption 

type SleepOption func(*sleepOptions)

SleepOption configures sleep behavior.

func WithSleepID 

func WithSleepID(id string) SleepOption

WithSleepID sets a custom timer ID for the sleep. This is useful for identifying timers in logs and for deterministic replay. The same prefixes as WithActivityID are reserved; Sleep rejects them with ErrHistoryKeyConflict.

type StartOption 

type StartOption func(*startOptions)

StartOption configures workflow start options.

func WithInstanceID 

func WithInstanceID(id string) StartOption

WithInstanceID specifies a custom instance ID. If not provided, a UUID will be generated.

type TerminalError 

type TerminalError struct {
	Err error
}

TerminalError wraps an error to indicate it should not be retried. When an activity returns a TerminalError, the workflow will fail without attempting any retries.

func NewTerminalError 

func NewTerminalError(err error) *TerminalError

NewTerminalError creates a new TerminalError wrapping the given error.

func NewTerminalErrorf 

func NewTerminalErrorf(format string, args ...any) *TerminalError

NewTerminalErrorf creates a new TerminalError with a formatted message.

func (*TerminalError) Error 

func (e *TerminalError) Error() string

func (*TerminalError) Unwrap 

func (e *TerminalError) Unwrap() error

type TimeoutKind 

type TimeoutKind string

TimeoutKind identifies which activity deadline elapsed. 

const (
	// TimeoutScheduleToStart means the scheduled task was not claimed in time.
	TimeoutScheduleToStart TimeoutKind = "schedule_to_start"
	// TimeoutStartToClose means the attempt exceeded its execution deadline.
	TimeoutStartToClose TimeoutKind = "start_to_close"
	// TimeoutHeartbeat means the activity claim expired without renewed progress.
	TimeoutHeartbeat TimeoutKind = "heartbeat"
)

type Timer 

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

Timer is a durable timer tied to its issuing workflow execution. Use it serially on the workflow coroutine. Get may be called repeatedly.

func NewTimer 

func NewTimer(ctx *WorkflowContext, d time.Duration, opts ...SleepOption) (*Timer, error)

NewTimer issues a sleep command without yielding. Nonpositive durations expire immediately; positive durations keep their first recorded deadline on replay.

func (*Timer) Get 

func (t *Timer) Get(ctx *WorkflowContext) (time.Time, error)

Get returns the recorded expiry or cancellation without consuming a command position or delivering a batch. A stored fire takes precedence over cancellation.

func (*Timer) IsReady 

func (t *Timer) IsReady() bool

IsReady reads only the delivered view or an outcome returned by Get. It is safe in Await predicates. Off-coroutine calls are unsupported and not detected.

func (*Timer) Key 

func (t *Timer) Key() string

Key returns the history key assigned at issue time.

type Workflow 

type Workflow[I, O any] interface {
	// Name returns the unique name of the workflow.
	Name() string

	// Execute runs the workflow logic.
	Execute(ctx *WorkflowContext, input I) (O, error)
}

Workflow defines the interface for a durable workflow. I is the input type, O is the output type.

func GetWorkflow 

func GetWorkflow[I, O any](name string) Workflow[I, O]

GetWorkflow retrieves a registered workflow by name. Returns nil if not found.

type WorkflowCancelledError 

type WorkflowCancelledError struct {
	InstanceID string
	Reason     string
}

WorkflowCancelledError indicates that a workflow has been cancelled.

func (*WorkflowCancelledError) Error 

func (e *WorkflowCancelledError) Error() string

func (*WorkflowCancelledError) Unwrap 

func (e *WorkflowCancelledError) Unwrap() error

Unwrap makes workflow cancellation match context.Canceled.

type WorkflowContext 

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

WorkflowContext provides context for workflow execution. It provides command positions, replay state, and recorded history.

func Detached 

func Detached(ctx *WorkflowContext) *WorkflowContext

Detached creates a parentless scope that ignores durable workflow cancellation. It retains the execution's lease and shutdown context. Use it for cleanup.

func NewWorkflowContext 

func NewWorkflowContext(
	ctx context.Context,
	instanceID, workerID, workflowName string,
) *WorkflowContext

NewWorkflowContext creates a standalone context without recorded history. Commands require an engine-managed workflow coroutine.

func NewWorkflowContextFromExecution 

func NewWorkflowContextFromExecution(execCtx *replay.ExecutionContext) *WorkflowContext

NewWorkflowContextFromExecution creates a WorkflowContext from a replay ExecutionContext.

func WithCancel 

func WithCancel(ctx *WorkflowContext) (scoped *WorkflowContext, cancel func())

WithCancel creates a child cancellation scope without issuing a command. Its cancel function must run on the workflow coroutine, outside predicates. The first call records one cancel_scope command, even for an empty scope.

func (*WorkflowContext) App 

func (c *WorkflowContext) App() *App

App returns the App reference for this workflow context. Returns nil if not set.

func (*WorkflowContext) Context 

func (c *WorkflowContext) Context() context.Context

Context returns the underlying context.Context.

func (*WorkflowContext) Done 

func (c *WorkflowContext) Done() <-chan struct{}

Done signals lease or shutdown cancellation, never scope cancellation.

func (*WorkflowContext) Err 

func (c *WorkflowContext) Err() error

Err reports lease or shutdown cancellation, never scope cancellation.

func (*WorkflowContext) Hooks 

func (c *WorkflowContext) Hooks() hooks.WorkflowHooks

Hooks returns the workflow hooks for observability. Returns nil if hooks are not available (e.g., not using replay engine).

func (*WorkflowContext) InstanceID 

func (c *WorkflowContext) InstanceID() string

InstanceID returns the workflow instance ID.

func (*WorkflowContext) IsReplaying 

func (c *WorkflowContext) IsReplaying() bool

IsReplaying returns true if the workflow is being replayed.

func (*WorkflowContext) SetApp 

func (c *WorkflowContext) SetApp(app *App)

SetApp sets the App reference for this workflow context. This is typically called internally when creating the context.

func (*WorkflowContext) Storage 

func (c *WorkflowContext) Storage() WorkflowStorage

Storage returns read-only history access, or nil outside the replay engine.

func (*WorkflowContext) WithContext 

func (c *WorkflowContext) WithContext(ctx context.Context) *WorkflowContext

WithContext returns a copy of the WorkflowContext with a new underlying context.

func (*WorkflowContext) WorkerID 

func (c *WorkflowContext) WorkerID() string

WorkerID returns the worker ID that is executing this workflow.

func (*WorkflowContext) WorkflowName 

func (c *WorkflowContext) WorkflowName() string

WorkflowName returns the workflow name.

type WorkflowDescription 

type WorkflowDescription struct {
	InstanceID, WorkflowName                  string
	Status                                    WorkflowStatus
	CancelRequested                           bool
	CancelReason, ParentInstanceID, ParentKey string
	StartedAt, UpdatedAt                      time.Time
}

WorkflowDescription exposes durable status and parent metadata.

func Describe 

func Describe(ctx context.Context, app *App, instanceID string) (*WorkflowDescription, error)

Describe reads status and cancellation metadata from one consistent snapshot.

type WorkflowFunc 

type WorkflowFunc[I, O any] struct {
	// contains filtered or unexported fields
}

WorkflowFunc is a convenience type for workflows defined as functions.

func DefineWorkflow 

func DefineWorkflow[I, O any](name string, fn func(ctx *WorkflowContext, input I) (O, error)) *WorkflowFunc[I, O]

DefineWorkflow creates a new workflow from a function.

func (*WorkflowFunc[I, O]) Execute 

func (w *WorkflowFunc[I, O]) Execute(ctx *WorkflowContext, input I) (O, error)

Execute runs the workflow function.

func (*WorkflowFunc[I, O]) Name 

func (w *WorkflowFunc[I, O]) Name() string

Name returns the workflow name.

type WorkflowNotFoundError 

type WorkflowNotFoundError struct {
	InstanceID string
}

WorkflowNotFoundError indicates that a workflow instance was not found.

func (*WorkflowNotFoundError) Error 

func (e *WorkflowNotFoundError) Error() string

func (*WorkflowNotFoundError) Unwrap 

func (e *WorkflowNotFoundError) Unwrap() error

Unwrap makes a missing workflow match ErrInstanceNotFound.

type WorkflowNotRegisteredError 

type WorkflowNotRegisteredError struct{ WorkflowName string }

WorkflowNotRegisteredError means the querying App lacks the workflow runner.

func (*WorkflowNotRegisteredError) Error 

func (e *WorkflowNotRegisteredError) Error() string

type WorkflowPanicError 

type WorkflowPanicError = replay.WorkflowPanicError

WorkflowPanicError preserves a recovered workflow panic and its stack.

type WorkflowResult 

type WorkflowResult[O any] struct {
	InstanceID string
	Status     string
	Output     O
	Error      error
}

WorkflowResult represents the result of a workflow execution.

func GetWorkflowResult 

func GetWorkflowResult[O any](ctx context.Context, app *App, instanceID string) (*WorkflowResult[O], error)

GetWorkflowResult retrieves the result of a workflow instance.

func WaitWorkflow 

func WaitWorkflow[O any](ctx context.Context, app *App, instanceID string) (*WorkflowResult[O], error)

WaitWorkflow waits for a terminal result. Workflow failures are in Result.Error. Cancellation of ctx stops only the wait. GetWorkflowResult is the nonblocking form.

type WorkflowStatus 

type WorkflowStatus = storage.WorkflowStatus

WorkflowStatus is the durable state of a workflow instance.

type WorkflowStorage 

type WorkflowStorage interface {
	GetHistoryEvent(context.Context, string, string) (*storage.HistoryEvent, error)
}

WorkflowStorage exposes recorded history for diagnostics. Workflow commands must use the workflow APIs rather than direct database writes.

Source Files

View all Source files

Directories

Path Synopsis
Package hooks provides lifecycle hooks for workflow observability.
 Package hooks provides lifecycle hooks for workflow observability.
otel
Package otel exports Selvedge lifecycle callbacks as completed OpenTelemetry spans.
 Package otel exports Selvedge lifecycle callbacks as completed OpenTelemetry spans.
internal
migrations
Package migrations provides automatic dbmate-compatible migration support.
 Package migrations provides automatic dbmate-compatible migration support.
notify
Package notify provides PostgreSQL LISTEN/NOTIFY functionality for real-time notifications.
 Package notify provides PostgreSQL LISTEN/NOTIFY functionality for real-time notifications.
replay
storage
Package storage provides the storage layer for Selvedge.
 Package storage provides the storage layer for Selvedge.
storage/faultstore
Package faultstore provides storage fault injection for engine tests.
 Package faultstore provides storage fault injection for engine tests.
testkit
Package testkit provides deterministic execution clocks for tests.
 Package testkit provides deterministic execution clocks for tests.
Package retry provides retry policies for activities.
 Package retry provides retry policies for activities.

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.