runnerq

package module
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Jun 15, 2026 License: MIT Imports: 15 Imported by: 0

README

RunnerQ

Durable workflows and queues for Go, backed by Postgres.

Add crash-proof background jobs and multi-step workflows to your Go app in a few lines — no separate orchestrator to run, no new infrastructure. RunnerQ is a library: import it, point it at the Postgres you already have, and write workflows as ordinary Go functions. When a process crashes mid-workflow, it resumes from where it left off without redoing completed work.

func (h *Checkout) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
    // Each step is checkpointed in Postgres. Crash after the charge and
    // restart — the charge is NOT repeated; the workflow resumes at shipping.
    receipt, err := ctx.Run("charge-card", func() (json.RawMessage, error) {
        return chargeCard(payload)   // runs exactly once
    })
    if err != nil {
        return nil, err
    }

    ship, err := ctx.ActivityExecutor.Activity("ship").Step("ship").Payload(payload).Execute(ctx.Ctx)
    if err != nil {
        return nil, err
    }
    return ship.GetResult(ctx.Ctx)   // parent parks here — frees the worker, survives deploys
}

go get github.com/alob-mtc/runnerq-go

See it for real: examples/02-crash-and-resume charges an order, lets you Ctrl-C it, and resumes on restart without double-charging. That demo is the whole pitch in 30 seconds.

Why RunnerQ

Background work that touches the real world — charging cards, sending email, calling APIs, orchestrating multi-step jobs — has to survive crashes, deploys, and retries without doing things twice. The usual options are a heavy workflow server (a cluster to operate) or a plain task queue (no durability — you hand-roll idempotency and recovery yourself).

RunnerQ gives you durable execution as a library:

  • A workflow is just a Go function. Orchestration is normal control flow — loops, conditionals, error handling — not a DSL or a DAG.
  • Steps are checkpointed. Completed work is skipped on replay, so retries and restarts are cheap and side effects don't repeat.
  • Workflows pause for free. Sleep for days or wait for a webhook while holding zero workers — paused workflows are just rows in Postgres.
  • Only Postgres. No orchestrator cluster, no broker, no control-plane bill. Scale by adding stateless worker processes.
When to use it
  • Use RunnerQ when you want durable, crash-safe workflows or queues with minimal new infrastructure, and you're already running Postgres.
  • Reach for a dedicated workflow server (Temporal, Cadence) if you need many SDK languages or want workflow orchestration decoupled from your app and your database.
  • A plain queue (River, Asynq, SQS) is enough if you only need fire-and-forget tasks and don't need workflows, steps, signals, or durable timers.

Features

Durable workflows that survive crashes

A handler that calls ctx.Run steps is a durable workflow. Each step's result is checkpointed; if the process dies, the handler replays and completed steps return their stored results instead of re-executing.

receipt, err := ctx.Run("charge-card", func() (json.RawMessage, error) {
    return chargeCard(payload)   // at most once, even across retries and restarts
})
Orchestration that fast-forwards on retry

Spawn child activities with .Step(name). A retried parent reattaches to the children it already launched instead of duplicating them, and parents that await children park in the database — no goroutine, no lease, no retry burned while they wait.

fut, _ := ctx.ActivityExecutor.Activity("reserve").Step("reserve").Payload(p).Execute(ctx.Ctx)
reserved, _ := fut.GetResult(ctx.Ctx)   // memoized on replay
Durable timers

Sleep inside a workflow for seconds or weeks. The deadline is persisted — a restart resumes the remainder — and long sleeps hold no worker.

ctx.Sleep("cooling-off", 24 * time.Hour)   // survives restarts; frees the worker
Signals — pause for the outside world

Wait for an approval, a webhook, or a payment confirmation. The workflow parks until a signal arrives, delivered from anywhere — even another process.

decision, err := ctx.WaitForSignal("approval", 48*time.Hour)   // parks, holds no worker
// from an HTTP handler, a Slack action, a Kafka consumer:
engine.Signal(ctx, activityID, "approval", payload)
Exactly-once enqueue

Idempotency keys make enqueuing exactly-once — dedupe webhooks and event handlers with one option.

executor.Activity("process_event").
    Payload(p).
    IdempotencyKeyOption(eventID, runnerq.ReturnExisting).   // duplicate deliveries collapse to one
    Execute(ctx)
A real queue underneath

Priorities, exponential-backoff retries, a dead-letter queue, scheduling, and worker-level type filtering so slow jobs can't starve latency-sensitive ones.

executor.Activity("send_email").
    Payload(p).
    Priority(runnerq.PriorityHigh).
    MaxRetries(5).
    Timeout(2 * time.Minute).
    Delay(30 * time.Second).
    Execute(ctx)
A built-in console

An embedded web dashboard (no npm, no build step) with live queue stats, activity browsing, results, and lifecycle timelines over SSE.

mux.Handle("/console/", http.StripPrefix("/console", ui.RunnerQUI(inspector)))

How it compares

RunnerQ Temporal / Cadence Plain task queue (River, Asynq, …)
Durable workflows ✅ ✅ ❌
Deploy model a Go library a server cluster to operate a library
Infrastructure Postgres you already run server + its own datastore Postgres or Redis
Workflow code plain Go, step-memoized plain code, full replay-determinism n/a
Durable timers / signals ✅ ✅ ❌
Scale add stateless worker processes scale the cluster add workers

The economics: RunnerQ adds one dependency you almost certainly already have — Postgres. There's no orchestrator cluster to size, secure, and pay for, and no separate control plane. Workers are stateless; scale out by running more of your own binary. The trade-off versus a dedicated server is deliberate: durable execution that fits inside your app and your database, rather than a system you operate alongside it.

Quick start

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"

    "github.com/alob-mtc/runnerq-go"
    "github.com/alob-mtc/runnerq-go/storage/postgres"
)

type Greeting struct{ runnerq.DefaultDeadLetterHandler }

func (h *Greeting) ActivityType() string { return "greeting" }

func (h *Greeting) Handle(ctx runnerq.ActivityContext, payload json.RawMessage) (json.RawMessage, error) {
    greeting, err := ctx.Run("compose", func() (json.RawMessage, error) {
        return json.Marshal("hello, " + string(payload))
    })
    return greeting, err
}

func main() {
    ctx := context.Background()

    backend, err := postgres.New(ctx, "postgres://postgres:runnerq@localhost:5432/runnerq", "my_app")
    if err != nil {
        log.Fatal(err)
    }
    defer backend.Close()

    engine, err := runnerq.Builder().Backend(backend).MaxWorkers(8).Build()
    if err != nil {
        log.Fatal(err)
    }
    engine.RegisterActivity("greeting", &Greeting{})
    go engine.Start(ctx)

    future, _ := engine.GetActivityExecutor().
        Activity("greeting").
        Payload(json.RawMessage(`"world"`)).
        Execute(ctx)

    result, _ := future.GetResult(ctx)
    fmt.Println(string(result))   // "hello, world"
}

Examples

Runnable, realistic, copy-pasteable — each in examples/ with its own README. docker compose up -d once, then go run . in any of them.

# Example Shows
01 hello-workflow a durable two-step workflow
02 crash-and-resume kill it mid-flight; it resumes without redoing work
03 steps-and-retries child activities memoized across a retry
04 checkpoint-side-effects charge once, even when the handler retries
05 durable-sleep timers that survive restarts
06 human-approval pause for an approval, delivered over HTTP
07 fan-out spawn many children, run them in parallel
08 exactly-once-webhook idempotency keys collapse duplicate deliveries
09 cross-process-futures enqueue in a web tier, await by ID anywhere
10 retries-and-dead-letter backoff, the dead-letter queue, OnDeadLetter
11 retention TTL cleanup of completed workflows
12 workload-isolation many worker fleets sharing one queue
13 console the built-in observability dashboard

Documentation

Full reference in docs/:

Status

RunnerQ is pre-1.0. The durable-execution core (steps, checkpoints, signals, durable sleep, crash recovery) is built and integration-tested against Postgres, but APIs may still change before 1.0. See RELEASING.md for the stability policy.

License

MIT — see LICENSE.

Documentation

Overview

Package runnerq provides a durable activity queue and worker system for Go.

Features

  • Priority-based activity processing (Critical, High, Normal, Low)
  • Activity scheduling with precise timestamp-based scheduling
  • Intelligent retry mechanism with exponential backoff
  • Dead letter queue handling for activities exceeding retry limits
  • Concurrent activity processing with configurable worker pools
  • Graceful shutdown with proper cleanup
  • Activity orchestration enabling activities to execute other activities
  • Comprehensive error handling with retryable and non-retryable types
  • Pluggable storage backends (PostgreSQL built-in)
  • Worker-level activity type filtering
  • Queue statistics and monitoring
  • Web-based observability console

Quick Start 

backend, _ := postgres.New(ctx, "postgres://localhost/mydb", "my_app")
engine, _ := runnerq.Builder().
    Backend(backend).
    QueueName("my_app").
    MaxWorkers(8).
    Build()

engine.RegisterActivity("send_email", &SendEmailHandler{})
engine.Start(ctx)

Index

Constants

View Source 
const DefaultMaxActivityDepth uint16 = 32

DefaultMaxActivityDepth is the default cap when MaxActivityDepth is unset.

Variables

This section is empty.

Functions

func IsActivityNotFound added in v0.3.0 

func IsActivityNotFound(err error) bool

IsActivityNotFound reports whether err is a signal-to-a-missing-activity error (the target was never enqueued, or has completed and been swept). External deliverers (webhooks, reconciliation) usually treat this as "already settled" rather than a failure.

func IsSignalTimeout added in v0.2.0 

func IsSignalTimeout(err error) bool

IsSignalTimeout reports whether err is a WaitForSignal timeout.

func SignalActivity added in v0.2.0 

func SignalActivity(ctx context.Context, backend storage.Storage, activityID uuid.UUID, name string, payload json.RawMessage) error

SignalActivity delivers an external signal to an activity from ANY process that shares the database — no engine required (e.g. a webhook receiver or approval service holding only a backend handle). The payload is persisted (buffered) even if the activity hasn't reached its WaitForSignal yet, and a parked waiter is woken immediately. payload may be nil for a pure notification. Repeated signals with the same name overwrite the payload.

Returns a not-found error when no activity with that ID exists in the backend's queue.

func SignalActivityByKey added in v0.3.0 

func SignalActivityByKey(ctx context.Context, backend storage.Storage, activityType string, idempotencyKey string, name string, payload json.RawMessage) error

SignalActivityByKey delivers a signal to whichever activity currently owns (activityType, idempotencyKey) in the backend's queue, instead of addressing it by internal activity ID. This lets an external deliverer (a webhook, a reconciliation job) wake a durable workflow using the same business key the workflow was enqueued with — no client-side reference→ID bookkeeping required.

activityType and idempotencyKey together must match what the workflow was enqueued with via IdempotencyKeyOption: keys are namespaced per activity type internally, so the type is required to resolve the right instance. The signal name is unchanged from SignalActivity: the key pair selects the workflow instance, the name selects which of its waits to satisfy. A key owns at most one activity (idempotency primary key), so there is no fan-out ambiguity. When the key is unclaimed (never enqueued, or completed and retention-swept) it returns an ErrActivityNotFoundW worker error (check with IsActivityNotFound) — typically treated as "already settled / nothing to wake".

Resolution and delivery are two steps rather than one transaction; this only matters under key-reuse behaviors (AllowReuse/AllowReuseOnFailure) that repoint a key mid-flight, in which case the signal targets the owner at delivery time. For the common ReturnExisting workflow case the owner is stable, and the delivery itself (store + wake) remains atomic.

func WaitAll added in v0.2.0 

func WaitAll(ctx context.Context, futures ...*ActivityFuture) ([]json.RawMessage, error)

WaitAll awaits every future and returns their results in order. Inside a handler, sequential awaiting is already efficient under replay semantics: the first pending child parks the parent, and on each wake every completed child fast-forwards — so no parallel-wait machinery is needed. Propagate the error unchanged, as with GetResult.

Types

type ActivityBuilder 

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

ActivityBuilder builds and executes activities with fluent configuration.

func (*ActivityBuilder) AsRoot 

func (b *ActivityBuilder) AsRoot() *ActivityBuilder

AsRoot detaches this spawn from its parent's lineage. The resulting activity becomes a root (no parent, root is itself, depth 0) regardless of the surrounding handler context. Use for fire-and-forget side jobs whose lifecycle is logically independent of the parent (audit logs, async telemetry, etc.).

func (*ActivityBuilder) Delay 

func (b *ActivityBuilder) Delay(d time.Duration) *ActivityBuilder

Delay sets the delay before execution.

func (*ActivityBuilder) Execute 

func (b *ActivityBuilder) Execute(ctx context.Context) (*ActivityFuture, error)

Execute enqueues the activity and returns an ActivityFuture.

func (*ActivityBuilder) IdempotencyKeyOption 

func (b *ActivityBuilder) IdempotencyKeyOption(key string, behavior OnDuplicate) *ActivityBuilder

IdempotencyKeyOption sets the idempotency key and behavior for duplicate detection.

func (*ActivityBuilder) MaxRetries 

func (b *ActivityBuilder) MaxRetries(retries uint32) *ActivityBuilder

MaxRetries sets the maximum number of retry attempts (0 for unlimited).

func (*ActivityBuilder) MaxRetryDelay 

func (b *ActivityBuilder) MaxRetryDelay(d time.Duration) *ActivityBuilder

MaxRetryDelay caps the exponential backoff delay between retries. Defaults to 1 hour if not set. Non-positive durations are ignored. Values below 1 second are rounded up to 1 second.

func (*ActivityBuilder) Metadata 

func (b *ActivityBuilder) Metadata(key, value string) *ActivityBuilder

Metadata sets a single metadata key/value pair on the activity.

func (*ActivityBuilder) MetadataMap 

func (b *ActivityBuilder) MetadataMap(metadata map[string]string) *ActivityBuilder

MetadataMap sets multiple metadata fields on the activity. Values are merged with existing metadata and overwrite duplicate keys.

func (*ActivityBuilder) Payload 

func (b *ActivityBuilder) Payload(payload json.RawMessage) *ActivityBuilder

Payload sets the JSON payload for the activity.

func (*ActivityBuilder) Priority 

func (b *ActivityBuilder) Priority(p ActivityPriority) *ActivityBuilder

Priority sets the priority level.

func (*ActivityBuilder) Step added in v0.2.0 

func (b *ActivityBuilder) Step(name string) *ActivityBuilder

Step makes this spawn a named, memoized step of the calling handler. The idempotency key is derived automatically from (root activity, parent activity, name) with ReturnExisting behavior, so when a retried parent re-issues the same spawn it gets the SAME child back instead of creating a duplicate — and GetResult returns the child's stored result instantly if it already completed. This is what lets a retried orchestrator fast-forward through work it finished on a previous attempt.

Contract: step names must be stable across retries and unique among this parent's spawns (two spawns sharing a name resolve to one child). Only valid when called from inside an activity handler; incompatible with AsRoot and IdempotencyKeyOption.

func (*ActivityBuilder) Timeout 

func (b *ActivityBuilder) Timeout(d time.Duration) *ActivityBuilder

Timeout sets the maximum execution time.

type ActivityContext 

type ActivityContext struct {
	// ActivityID is the unique identifier for this activity instance.
	ActivityID uuid.UUID

	// ActivityType is the type of activity being executed.
	ActivityType string

	// RetryCount is the current retry attempt number (0 for first execution).
	RetryCount uint32

	// Metadata is custom metadata associated with the activity.
	Metadata map[string]string

	// Ctx is the Go context for cancellation and deadline propagation.
	Ctx context.Context

	// ActivityExecutor allows executing other activities from within a handler.
	// Spawns made through this executor are tagged as children of this activity.
	ActivityExecutor ActivityExecutor

	// ParentActivityID is the direct parent of this activity, if any.
	ParentActivityID *uuid.UUID

	// RootActivityID is the root of the lineage tree this activity belongs to.
	// For root activities, RootActivityID equals ActivityID.
	RootActivityID uuid.UUID

	// Depth is the lineage depth of this activity (0 for roots).
	Depth uint16
	// contains filtered or unexported fields
}

ActivityContext is provided to activity handlers during execution.

func (ActivityContext) Run added in v0.2.0 

func (c ActivityContext) Run(name string, fn func() (json.RawMessage, error)) (json.RawMessage, error)

Run executes fn as a named, checkpointed step: its successful result (or permanent failure) is persisted, and when a retried handler reaches the same step it returns the stored outcome WITHOUT re-running fn. Use it for local side effects that must not repeat across retries — payments, emails, non-idempotent API calls.

Semantics:

  • success → result stored; later attempts return it instantly.
  • NonRetryError → failure stored; later attempts return the same error without re-running fn.
  • retryable error → nothing stored; fn runs again on the next attempt.
  • crash AFTER fn but BEFORE the result commits → fn runs again. Run is at-least-once with at-most-once-per-recorded-success; fn should be as idempotent as the external system allows (e.g. pass an idempotency key to the payment provider).

Step names must be stable across retries and unique within the handler.

func (ActivityContext) Sleep added in v0.2.0 

func (c ActivityContext) Sleep(name string, d time.Duration) error

Sleep is a durable timer: the wake deadline is persisted under the step name on first execution, so a handler that crashes or is redeployed mid-sleep resumes with only the REMAINDER of the wait — a 24h sleep does not restart from zero, and an already-elapsed sleep returns immediately on replay.

When the remaining wait fits inside the activity's timeout budget, Sleep waits in-process. When it doesn't, Sleep YIELDS: it returns a sentinel error that the caller MUST propagate unchanged (`if err != nil { return nil, err }`); the engine intercepts it, parks the activity as scheduled until the wake time without consuming a retry, and re-invokes the handler afterwards — earlier Run/Step checkpoints fast-forward and this Sleep returns nil.

Step names must be stable across retries and unique within the handler.

func (ActivityContext) WaitForSignal added in v0.2.0 

func (c ActivityContext) WaitForSignal(name string, timeout time.Duration) (json.RawMessage, error)

WaitForSignal blocks until an external signal named name is delivered to THIS activity (via WorkerEngine.Signal or runnerq.SignalActivity from any process sharing the database) and returns its payload. timeout bounds the wait, measured from the FIRST attempt that reached this call — replays share the persisted deadline, they don't restart it; 0 means wait forever. On timeout it returns a non-retryable error (check with IsSignalTimeout).

Signals are buffered: one delivered before the handler reaches this call — or before the activity even started — is returned immediately, including on replay. Repeated signals with the same name overwrite the payload (last write wins).

Like Sleep, a wait that doesn't fit the handler's timeout budget YIELDS: the sentinel error MUST be propagated unchanged; the engine parks the activity (no retry consumed, no worker held) until delivery wakes it or the wait deadline passes. Signal names must be stable across retries and unique within the handler.

type ActivityError 

type ActivityError struct {
	Retryable bool
	Message   string
}

ActivityError represents an error from activity handler execution. Retryable indicates whether the activity should be retried.

func NewNonRetryError 

func NewNonRetryError(msg string) *ActivityError

NewNonRetryError creates an ActivityError that should not be retried.

func NewRetryError 

func NewRetryError(msg string) *ActivityError

NewRetryError creates an ActivityError that triggers a retry.

func (*ActivityError) Error 

func (e *ActivityError) Error() string

func (*ActivityError) IsRetryable 

func (e *ActivityError) IsRetryable() bool

IsRetryable implements the RetryableError interface.

type ActivityExecutor 

type ActivityExecutor interface {
	Activity(activityType string) *ActivityBuilder
}

ActivityExecutor allows executing activities, enabling orchestration.

type ActivityFuture 

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

ActivityFuture represents a pending activity result that can be awaited.

func FutureFor added in v0.2.0 

func FutureFor(backend storage.Storage, activityID uuid.UUID) *ActivityFuture

FutureFor reconstructs an awaitable future from an activity ID — futures are rehydratable across processes and restarts: any process holding a backend for the same queue can await any activity's result.

func (*ActivityFuture) ActivityID added in v0.2.0 

func (f *ActivityFuture) ActivityID() uuid.UUID

ActivityID returns the awaited activity's ID — the externally-shareable handle for a future. Hand it to another process and reconstruct the future there with FutureFor.

func (*ActivityFuture) GetResult 

func (f *ActivityFuture) GetResult(ctx context.Context) (json.RawMessage, error)

GetResult waits for and returns the completed activity result. The wait is notification-driven when the backend supports it (the Postgres backend does), with a slow table re-check as fallback — including when the caller is a different process from the worker producing the result.

Called from inside an activity handler, GetResult waits in-process for a short grace and then YIELDS, parking the parent activity — no goroutine held, no lease, no retry consumed — until the child's completion wakes it. The handler replays, earlier Step/Run checkpoints fast-forward, and this call returns the now-available result. The sentinel error MUST be propagated unchanged, same as Sleep and WaitForSignal. A parent workflow can therefore outlive any number of handler invocations and deploys.

Called from outside a handler (a server process holding a future), it blocks until the result exists or ctx is done.

type ActivityHandler 

type ActivityHandler interface {
	// ActivityType returns the activity type string this handler processes.
	ActivityType() string

	// Handle processes the activity with the given payload and context.
	// Returns:
	//   (result, nil)       - completed successfully, result may be nil
	//   (nil, RetryError)   - failed but should be retried
	//   (nil, NonRetryError)- failed permanently
	Handle(ctx ActivityContext, payload json.RawMessage) (json.RawMessage, error)

	// OnDeadLetter is called when an activity enters the dead letter state.
	// The default behavior (if not overridden by embedding DefaultDeadLetterHandler) is a no-op.
	OnDeadLetter(ctx ActivityContext, payload json.RawMessage, errorMsg string)
}

ActivityHandler is the interface that all activity handlers must implement. Implementations should be safe for concurrent use.

type ActivityOption 

type ActivityOption struct {
	Priority             *ActivityPriority
	MaxRetries           uint32
	TimeoutSeconds       uint64
	MaxRetryDelaySeconds *uint64
	DelaySeconds         *uint64
	IdempotencyKey       *IdempotencyConfig
	Metadata             map[string]string
}

ActivityOption configures an activity's execution parameters.

type ActivityPriority 

type ActivityPriority int

ActivityPriority determines the order of execution. Higher priority activities are processed first. 

const (
	PriorityLow      ActivityPriority = 1
	PriorityNormal   ActivityPriority = 2
	PriorityHigh     ActivityPriority = 3
	PriorityCritical ActivityPriority = 4
)

func (ActivityPriority) MarshalJSON 

func (p ActivityPriority) MarshalJSON() ([]byte, error)

func (ActivityPriority) String 

func (p ActivityPriority) String() string

func (*ActivityPriority) UnmarshalJSON 

func (p *ActivityPriority) UnmarshalJSON(data []byte) error

type ActivityStatus 

type ActivityStatus string

ActivityStatus tracks the activity lifecycle. 

const (
	StatusPending    ActivityStatus = "Pending"
	StatusRunning    ActivityStatus = "Running"
	StatusCompleted  ActivityStatus = "Completed"
	StatusFailed     ActivityStatus = "Failed"
	StatusRetrying   ActivityStatus = "Retrying"
	StatusDeadLetter ActivityStatus = "DeadLetter"
)

type DefaultDeadLetterHandler 

type DefaultDeadLetterHandler struct{}

DefaultDeadLetterHandler provides a no-op OnDeadLetter implementation. Embed this in your handler struct if you don't need dead letter handling.

func (DefaultDeadLetterHandler) OnDeadLetter 

func (DefaultDeadLetterHandler) OnDeadLetter(_ ActivityContext, _ json.RawMessage, _ string)

OnDeadLetter is a no-op implementation.

type IdempotencyConfig 

type IdempotencyConfig struct {
	Key      string
	Behavior OnDuplicate
}

IdempotencyConfig holds idempotency key and its behavior.

type MetricsSink 

type MetricsSink interface {
	IncCounter(name string, value uint64)
	ObserveDuration(name string, dur time.Duration)
}

MetricsSink allows collecting metrics about activity processing. Implement this interface to integrate with your preferred metrics system.

type NoopMetrics 

type NoopMetrics struct{}

NoopMetrics is the default metrics sink that discards all metrics.

func (NoopMetrics) IncCounter 

func (NoopMetrics) IncCounter(_ string, _ uint64)

func (NoopMetrics) ObserveDuration 

func (NoopMetrics) ObserveDuration(_ string, _ time.Duration)

type OnDuplicate 

type OnDuplicate int

OnDuplicate defines behavior when an activity with the same idempotency key exists. 

const (
	// AllowReuse always creates a new activity, updating the idempotency record.
	AllowReuse OnDuplicate = iota
	// ReturnExisting returns the existing ActivityFuture if key exists.
	ReturnExisting
	// AllowReuseOnFailure allows reuse only if the previous activity failed.
	AllowReuseOnFailure
	// NoReuse returns an error if the key exists.
	NoReuse
)

type ResultState 

type ResultState int

ResultState indicates whether an activity result is success or failure. 

const (
	ResultOk ResultState = iota
	ResultErr
)

type RetentionConfig added in v0.2.0 

type RetentionConfig struct {
	// Completed is how long trees whose root completed successfully are kept.
	// Zero keeps them forever.
	Completed time.Duration `json:"completed,omitempty"`
	// Failed is how long trees whose root is failed or dead_letter are kept —
	// a separate clock so failures can be held longer for inspection.
	// Zero keeps them forever.
	Failed time.Duration `json:"failed,omitempty"`
	// Interval is the sweep cadence. Defaults to 10 minutes.
	Interval time.Duration `json:"interval,omitempty"`
	// BatchSize is the max root trees deleted per sweep transaction.
	// Defaults to 100.
	BatchSize int `json:"batch_size,omitempty"`
}

RetentionConfig opts the engine into deleting old terminal workflow trees. Without it, activities, events, results, and idempotency keys are kept forever. The deletion unit is a whole tree (terminal root with no non-terminal descendants), so retries can never find their children's results missing. One engine per queue sweeps at a time (advisory-lock leadership in the backend); running it on every engine is safe.

type RetryableError 

type RetryableError interface {
	IsRetryable() bool
}

RetryableError is an interface for errors that know if they are retryable. Handlers can return any error implementing this interface to control retry behavior.

type WorkerConfig 

type WorkerConfig struct {
	// QueueName is used as a prefix to avoid conflicts between different applications.
	QueueName string `json:"queue_name"`

	// MaxConcurrentActivities is the maximum number of activities processed concurrently.
	MaxConcurrentActivities int `json:"max_concurrent_activities"`

	// SchedulePollIntervalSeconds is the interval for polling scheduled activities.
	// When nil, defaults to 5 seconds.
	// Only effective for backends that don't handle scheduling natively in Dequeue().
	SchedulePollIntervalSeconds *uint64 `json:"schedule_poll_interval_seconds,omitempty"`

	// LeaseMS is the lease duration in milliseconds for claimed activities.
	// Defaults to 60000 ms (60s).
	LeaseMS *uint64 `json:"lease_ms,omitempty"`

	// ReaperIntervalSeconds is how often the reaper scans for expired leases.
	// Defaults to 5 seconds.
	ReaperIntervalSeconds *uint64 `json:"reaper_interval_seconds,omitempty"`

	// ReaperBatchSize is the max number of expired items to requeue per reaper tick.
	// Defaults to 100.
	ReaperBatchSize *int `json:"reaper_batch_size,omitempty"`

	// ActivityTypes restricts this engine to only dequeue specific activity types.
	// When nil, workers dequeue all activity types.
	ActivityTypes []string `json:"activity_types,omitempty"`

	// MaxActivityDepth caps how deep the parent/child activity tree can grow.
	// A handler attempting to spawn a child beyond this depth will receive ErrDepthExceeded.
	// When zero, defaults to 32.
	MaxActivityDepth uint16 `json:"max_activity_depth,omitempty"`

	// Retention enables deletion of old terminal workflow trees. Nil (the
	// default) keeps everything forever.
	Retention *RetentionConfig `json:"retention,omitempty"`

	// ShutdownGraceSeconds bounds the entire shutdown drain — worker loops,
	// in-flight activity goroutines, and worker-pool deregistration all run
	// in parallel under this single
	// budget. When the budget expires, Start() returns even if some
	// goroutines are still in flight (those are then orphaned for the
	// remaining process lifetime, which is fine on a SIGTERM). Default 30s.
	ShutdownGraceSeconds *uint64 `json:"shutdown_grace_seconds,omitempty"`
}

WorkerConfig controls queue behavior and resource usage.

func DefaultWorkerConfig 

func DefaultWorkerConfig() WorkerConfig

DefaultWorkerConfig returns a WorkerConfig with sensible defaults.

type WorkerEngine 

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

WorkerEngine is the main activity processing engine.

func NewWorkerEngineWithBackend 

func NewWorkerEngineWithBackend(backend storage.Storage, config WorkerConfig) *WorkerEngine

NewWorkerEngineWithBackend creates a WorkerEngine from a custom backend.

func (*WorkerEngine) Backend 

func (e *WorkerEngine) Backend() storage.Storage

Inspector returns a QueueInspector for observability operations. Import the observability package and use NewQueueInspector(backend) instead for decoupled usage.

func (*WorkerEngine) GetActivityExecutor 

func (e *WorkerEngine) GetActivityExecutor() ActivityExecutor

GetActivityExecutor returns an ActivityExecutor for orchestrating activities. Spawns made through the returned executor are roots (no parent lineage).

func (*WorkerEngine) MaxConcurrentActivities 

func (e *WorkerEngine) MaxConcurrentActivities() int

MaxConcurrentActivities returns the max workers config for inspector use.

func (*WorkerEngine) RegisterActivity 

func (e *WorkerEngine) RegisterActivity(activityType string, handler ActivityHandler)

RegisterActivity registers an activity handler for a given activity type.

func (*WorkerEngine) SetMetrics 

func (e *WorkerEngine) SetMetrics(sink MetricsSink)

SetMetrics sets the metrics sink.

func (*WorkerEngine) Signal added in v0.2.0 

func (e *WorkerEngine) Signal(ctx context.Context, activityID uuid.UUID, name string, payload json.RawMessage) error

Signal delivers an external signal to an activity — see SignalActivity.

func (*WorkerEngine) SignalByKey added in v0.3.0 

func (e *WorkerEngine) SignalByKey(ctx context.Context, activityType string, idempotencyKey string, name string, payload json.RawMessage) error

SignalByKey delivers a signal addressed by (activityType, idempotencyKey) — see SignalActivityByKey.

func (*WorkerEngine) Start 

func (e *WorkerEngine) Start(ctx context.Context) error

Start starts the worker engine and blocks until shutdown or error.

func (*WorkerEngine) Stop 

func (e *WorkerEngine) Stop()

Stop initiates a graceful shutdown of the worker engine.

type WorkerEngineBuilder 

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

WorkerEngineBuilder provides fluent configuration for WorkerEngine.

func Builder 

func Builder() *WorkerEngineBuilder

Builder creates a new WorkerEngineBuilder.

func (*WorkerEngineBuilder) ActivityTypes 

func (b *WorkerEngineBuilder) ActivityTypes(types []string) *WorkerEngineBuilder

ActivityTypes restricts this engine to only dequeue the specified types.

func (*WorkerEngineBuilder) Backend 

func (b *WorkerEngineBuilder) Backend(backend storage.Storage) *WorkerEngineBuilder

Backend sets the storage backend. Required.

func (*WorkerEngineBuilder) Build 

func (b *WorkerEngineBuilder) Build() (*WorkerEngine, error)

Build creates the WorkerEngine with configured settings.

func (*WorkerEngineBuilder) MaxWorkers 

func (b *WorkerEngineBuilder) MaxWorkers(max int) *WorkerEngineBuilder

MaxWorkers sets the maximum concurrent workers.

func (*WorkerEngineBuilder) Metrics 

func (b *WorkerEngineBuilder) Metrics(sink MetricsSink) *WorkerEngineBuilder

Metrics sets the metrics sink.

func (*WorkerEngineBuilder) QueueName 

func (b *WorkerEngineBuilder) QueueName(name string) *WorkerEngineBuilder

QueueName sets the queue name.

func (*WorkerEngineBuilder) Retention added in v0.2.0 

func (b *WorkerEngineBuilder) Retention(cfg RetentionConfig) *WorkerEngineBuilder

Retention opts the engine into deleting old terminal workflow trees — see RetentionConfig. Safe to set on every engine in a cluster; the backend elects one sweeper per queue.

func (*WorkerEngineBuilder) SchedulePollInterval 

func (b *WorkerEngineBuilder) SchedulePollInterval(interval time.Duration) *WorkerEngineBuilder

SchedulePollInterval sets the interval for polling scheduled activities.

func (*WorkerEngineBuilder) ShutdownGrace 

func (b *WorkerEngineBuilder) ShutdownGrace(d time.Duration) *WorkerEngineBuilder

type WorkerEngineWrapper 

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

WorkerEngineWrapper provides activity execution capabilities. It implements ActivityExecutor.

func (*WorkerEngineWrapper) Activity 

func (w *WorkerEngineWrapper) Activity(activityType string) *ActivityBuilder

Activity creates a fluent activity builder.

type WorkerError 

type WorkerError struct {
	Kind    WorkerErrorKind
	Message string
	Cause   error
}

WorkerError represents an error from the worker engine.

func IsWorkerError 

func IsWorkerError(err error) (*WorkerError, bool)

IsWorkerError extracts a *WorkerError from err (if any).

func WorkerErrorFromStorage 

func WorkerErrorFromStorage(err error) *WorkerError

WorkerErrorFromStorage converts a StorageError to a WorkerError.

func (*WorkerError) Error 

func (e *WorkerError) Error() string

func (*WorkerError) IsRetryable 

func (e *WorkerError) IsRetryable() bool

IsRetryable returns true if this error may be resolved by retrying.

func (*WorkerError) Unwrap 

func (e *WorkerError) Unwrap() error

type WorkerErrorKind 

type WorkerErrorKind int

WorkerErrorKind classifies worker engine errors. 

const (
	ErrCustom WorkerErrorKind = iota
	ErrQueue
	ErrSerializationW
	ErrTimeoutW
	ErrExecution
	ErrHandlerNotFound
	ErrBackend
	ErrDatabase
	ErrConfiguration
	ErrShutdown
	ErrAlreadyRunning
	ErrScheduling
	ErrDuplicateActivityW
	ErrIdempotencyConflictW
	ErrDepthExceeded
	ErrUnknown
	// ErrSignalTimeoutW means WaitForSignal's timeout elapsed before the
	// signal was delivered. Not retryable: retrying replays the persisted
	// wait deadline and times out again immediately — the handler must
	// decide what a missing signal means.
	ErrSignalTimeoutW
	// ErrActivityNotFoundW means a signal was addressed to an activity (by ID
	// or by idempotency key) that does not exist in the queue — never
	// enqueued, or already completed and retention-swept. Callers delivering
	// an external event typically treat this as "already settled / nothing to
	// wake" rather than a failure.
	ErrActivityNotFoundW
)

Source Files

View all Source files

Directories

Path Synopsis
examples
01-hello-workflow command
01 — Hello, Workflow
 01 — Hello, Workflow
02-crash-and-resume command
02 — Crash & Resume (the flagship demo)
 02 — Crash & Resume (the flagship demo)
03-steps-and-retries command
03 — Steps & Retries
 03 — Steps & Retries
04-checkpoint-side-effects command
04 — Checkpoint Side Effects with ctx.Run
 04 — Checkpoint Side Effects with ctx.Run
05-durable-sleep command
05 — Durable Sleep
 05 — Durable Sleep
06-human-approval command
06 — Human-in-the-Loop with Signals
 06 — Human-in-the-Loop with Signals
07-fan-out command
07 — Fan-Out and WaitAll
 07 — Fan-Out and WaitAll
08-exactly-once-webhook command
08 — Exactly-Once Webhook Ingestion
 08 — Exactly-Once Webhook Ingestion
09-cross-process-futures command
09 — Cross-Process Futures
 09 — Cross-Process Futures
10-retries-and-dead-letter command
10 — Retries and the Dead-Letter Queue
 10 — Retries and the Dead-Letter Queue
11-retention command
11 — Retention
 11 — Retention
12-workload-isolation command
12 — Workload Isolation
 12 — Workload Isolation
13-console command
13 — The Observability Console
 13 — The Observability Console
14-console-durable-steps command
14 — Console: durable step history & "blocked on"
 14 — Console: durable step history & "blocked on"
ui
postgres

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.