duro

package module

Go to main page

Versions in this module

v0

v0.9.0

Aug 26, 2026

Changes in this version

+ const CancelWatchQueueName

+ const CancelWatcherName

+ const DefaultReadTimeout

+ var ErrRunActive = errors.New("duro: run is still active")

+ var ErrRunInFlight = errors.New(...)

+ var ErrRunTerminal = errors.New("duro: run is already in a final state")

+ func Cancel(ctx Context, workflowID string) error

type App

+ func (a *App) Resume(ctx context.Context, workflowID string) error

type Client

+ func (c *Client) Cancel(workflowID string) error

+ func (c *Client) ListRuns(opts ...ListOption) ([]RunStatus, error)

+ func (c *Client) Resume(workflowID string) error

+ func (c *Client) Steps(workflowID string) ([]StepStatus, error)

+ func (c *Client) WithContext(ctx context.Context) *Client

type ClientConfig

+ ReadTimeout time.Duration

+ type ListOption func(*listConfig)

+ func WithCreatedAfter(t time.Time) ListOption

+ func WithCreatedBefore(t time.Time) ListOption

+ func WithIDs(ids ...string) ListOption

+ func WithInput() ListOption

+ func WithLimit(n int) ListOption

+ func WithNames(names ...string) ListOption

+ func WithNewestFirst() ListOption

+ func WithOffset(n int) ListOption

+ func WithQueue(name string) ListOption

+ func WithStates(states ...State) ListOption

type RunStatus

+ Attempts int

+ ExecutorID string

+ Input string

+ ParentID string

+ QueueName string

+ StartedAt time.Time

+ func ListRuns(ctx Context, opts ...ListOption) ([]RunStatus, error)

+ type StepStatus struct

+ ChildID string

+ CompletedAt time.Time

+ Err error

+ ID int

+ Name string

+ StartedAt time.Time

+ func Steps(ctx Context, workflowID string) ([]StepStatus, error)

v0.8.1

Jul 27, 2026

v0.8.0

Jul 27, 2026

Changes in this version

+ const Unbounded

+ var ErrMaxIterations = errors.New("duro: loop exceeded its maximum iterations")

+ var ErrMultipleValues = errors.New("duro: pipeline emitted more than one value")

+ type Client struct

+ func NewClient(ctx context.Context, cfg ClientConfig) (*Client, error)

+ func (c *Client) Shutdown(timeout time.Duration)

+ func (c *Client) Status(workflowID string) (RunStatus, error)

+ func (c *Client) StatusAll(workflowIDs ...string) ([]RunStatus, error)

+ type ClientConfig struct

+ DatabaseURL string

+ Logger *slog.Logger

type Config

+ ApplicationVersion string

+ ExecutorID string

+ type EnqueueOption func(*enqueueConfig)

+ func WithClientApplicationVersion(version string) EnqueueOption

+ func WithClientWorkflowID(id string) EnqueueOption

type Handle

+ func AttachJob[P, R any](ctx Context, job Job[P, R], workflowID string) (Handle[R], error)

+ func Enqueue[P, R any](c *Client, queue Queue, job Job[P, R], input P, opts ...EnqueueOption) (Handle[R], error)

+ type Job struct

+ func NewJob[P, R any](name string) Job[P, R]

+ func (j Job[P, R]) Name() string

+ type Option func(*appOptions)

+ func WithRetention(d time.Duration) Option

+ func WithStaleRunWarning(age time.Duration, hook ...func(StaleRunInfo)) Option

+ func WithSweepInterval(d time.Duration) Option

+ func WithWorkerPool(opts ...WorkerPoolOption) Option

type PipelineWorkflow

+ func RegisterJob[P, R any](ctx Context, job Job[P, R], p Pipeline[P, R], ...) *PipelineWorkflow[P, R]

+ type StaleRunInfo struct

+ OtherVersion int

+ SameVersion int

type StepOption

+ func WithMaxIterations(n int) StepOption

+ type WorkerPoolOption func(*appOptions)

+ func WithHeartbeatInterval(d time.Duration) WorkerPoolOption

+ func WithStaleThreshold(d time.Duration) WorkerPoolOption

v0.7.0

Jul 24, 2026

Changes in this version

type ChildOption

+ func WithCancelSiblings() ChildOption

+ func WithCancelWatchInterval(d time.Duration) ChildOption

type StepOption

+ func WithCancelSiblingSteps() StepOption

v0.6.0

Jul 22, 2026

Changes in this version

type Stage

+ func Via[T, R any](name string, p Pipeline[T, R]) Stage[T, T]

+ type WorkflowOption = dbos.WorkflowOption

+ type WorkflowRegistrationOption = dbos.WorkflowRegistrationOption

v0.5.0

Jul 22, 2026

Changes in this version

type Stage

+ func Rescue[T, R any](name string, p Pipeline[T, R], ...) Stage[T, R]

v0.4.0

Jul 21, 2026

Changes in this version

+ var ErrRunNotFound = errors.New("duro: run not found")

+ type Case struct

+ func When[T, R any](key string, p Pipeline[T, R]) Case[T, R]

type Handle

+ func Attach[R any](ctx Context, workflowID string) (Handle[R], error)

+ type RunStatus struct

+ ApplicationVersion string

+ CompletedAt time.Time

+ CreatedAt time.Time

+ Err error

+ ForkedFrom string

+ ID string

+ Name string

+ State State

+ UpdatedAt time.Time

+ func Status(ctx Context, workflowID string) (RunStatus, error)

+ func StatusAll(ctx Context, workflowIDs ...string) ([]RunStatus, error)

type Stage

+ func Branch[T, R any](name string, pred func(ctx context.Context, in T) (bool, error), ...) Stage[T, R]

+ func Collect[T any](name string, opts ...StepOption) Stage[T, []T]

+ func Loop[T any](name string, body Pipeline[T, T], ...) Stage[T, T]

+ func Sub[T, R any](name string, p Pipeline[T, R]) Stage[T, R]

+ func Switch[T, R any](name string, route func(ctx context.Context, in T) (string, error), ...) Stage[T, R]

+ type State string

+ const StateCancelled

+ const StateDelayed

+ const StateEnqueued

+ const StateError

+ const StatePending

+ const StateRetriesExceeded

+ const StateSuccess

+ func (s State) Failed() bool

+ func (s State) Terminal() bool

v0.3.0

Jul 21, 2026

Changes in this version

+ type RegisteredWorkflow struct

+ func RegisterWorkflow[P, R any](ctx Context, name string, fn WorkflowFunc[P, R], ...) *RegisteredWorkflow[P, R]

+ func (w *RegisteredWorkflow[P, R]) Start(ctx Context, in P, opts ...dbos.WorkflowOption) (Handle[R], error)

v0.2.1

Jul 21, 2026

v0.2.0

Jul 21, 2026

Changes in this version

+ const DeduplicationReject

+ const DeduplicationReturnExisting

+ func RegisterQueues(ctx Context, queues ...Queue) error

+ func WithWorkflowID(id string) dbos.WorkflowOption

+ type App struct

+ func New(ctx context.Context, cfg Config) (*App, error)

+ func (a *App) Context() Context

+ func (a *App) Launch() error

+ func (a *App) Shutdown(timeout time.Duration)

+ type ChannelOption func(*channelConfig)

+ func Portable() ChannelOption

+ type ChildOption func(*childConfig)

+ func WithChildAppVersion(version string) ChildOption

+ func WithChildAssumedRole(role string) ChildOption

+ func WithChildAuthenticatedRoles(roles ...string) ChildOption

+ func WithChildAuthenticatedUser(user string) ChildOption

+ func WithChildDeduplicationID[T any](fn func(in T) string) ChildOption

+ func WithChildDeduplicationPolicy(policy DeduplicationPolicy) ChildOption

+ func WithChildDelay(d time.Duration) ChildOption

+ func WithChildID[T any](fn func(in T) string) ChildOption

+ func WithChildPartitionKey[T any](fn func(in T) string) ChildOption

+ func WithChildPriority(priority uint) ChildOption

+ func WithChildTimeout(d time.Duration) ChildOption

+ func WithPortableChildren() ChildOption

+ type Config struct

+ DatabaseURL string

+ Logger *slog.Logger

+ Name string

+ type Context = dbos.DBOSContext

+ type Debouncer struct

+ func RegisterDebounced[P, R any](ctx Context, name string, p Pipeline[P, R], opts ...dbos.DebouncerOption) *Debouncer[P, R]

+ func (d *Debouncer[P, R]) Debounce(ctx Context, key string, delay time.Duration, input P) (Handle[R], error)

+ type DeduplicationPolicy = dbos.DeduplicationPolicy

+ type Event struct

+ func NewEvent[V any](key string, opts ...ChannelOption) Event[V]

+ func (e Event[V]) Get(ctx Context, workflowID string, timeout time.Duration) (V, error)

+ func (e Event[V]) Key() string

+ type Fork struct

+ ApplicationVersion string

+ ForkedID string

+ Queue string

+ QueuePartitionKey string

+ Stage string

+ WorkflowID string

+ type Handle struct

+ func ForkFromStage[R any](ctx Context, f Fork) (Handle[R], error)

+ func (h Handle[R]) ID() string

+ func (h Handle[R]) Result() (R, error)

+ func (h Handle[R]) Status() (dbos.WorkflowStatus, error)

+ type PipelineWorkflow struct

+ func RegisterScheduled[R any](ctx Context, name, cronSchedule string, p Pipeline[time.Time, R], ...) *PipelineWorkflow[time.Time, R]

+ func Register[P, R any](ctx Context, name string, p Pipeline[P, R], ...) *PipelineWorkflow[P, R]

+ func (w *PipelineWorkflow[P, R]) ConfigName() string

+ func (w *PipelineWorkflow[P, R]) Start(ctx Context, in P, opts ...dbos.WorkflowOption) (Handle[R], error)

+ type Queue struct

+ func NewQueue(name string, opts ...QueueOption) Queue

+ func (q Queue) Name() string

+ type QueueOption func(*queueConfig)

+ func WithConcurrency(n int) QueueOption

+ func WithPartitions() QueueOption

+ func WithPriorities() QueueOption

+ func WithRateLimit(limit int, period time.Duration) QueueOption

+ func WithWorkerConcurrency(n int) QueueOption

type Stage

+ func Delay[T any](name string, d time.Duration) Stage[T, T]

+ func FanOut[T, R any](name string, queue Queue, wf WorkflowRef[T, R], opts ...ChildOption) Stage[T, R]

+ func FromStream[T, V any](name string, stream Stream[V], fn func(in T) (workflowID string), ...) Stage[T, V]

+ func GetEvent[T, V any](name string, event Event[V], fn func(in T) (workflowID string), ...) Stage[T, V]

+ func Parallel[T, R any](name string, maxConcurrent int, fn func(ctx context.Context, in T) (R, error), ...) Stage[T, R]

+ func Recv[T, M any](name string, topic Topic[M], timeout time.Duration) Stage[T, M]

+ func Send[T, M any](name string, topic Topic[M], ...) Stage[T, T]

+ func SetEvent[T, V any](name string, event Event[V], fn func(in T) V) Stage[T, T]

+ func ToStream[T any](name string, stream Stream[T]) Stage[T, T]

type StepOption

+ func WithBackoffFactor(factor float64) StepOption

+ func WithMaxInterval(d time.Duration) StepOption

+ func WithRetryPredicate(pred func(error) bool) StepOption

+ func WithTimeout(d time.Duration) StepOption

+ type Stream struct

+ func NewStream[V any](key string, opts ...ChannelOption) Stream[V]

+ func (s Stream[V]) Key() string

+ func (s Stream[V]) Read(ctx Context, workflowID string) (values []V, closed bool, err error)

+ type Topic struct

+ func NewTopic[M any](name string, opts ...ChannelOption) Topic[M]

+ func (t Topic[M]) Name() string

+ func (t Topic[M]) Send(ctx Context, destinationID string, message M) error

+ type WorkflowFunc = dbos.Workflow[P, R]

+ type WorkflowRef interface

+ func Workflow[T, R any](fn WorkflowFunc[T, R]) WorkflowRef[T, R]

v0.1.0

Jul 6, 2026

Changes in this version

+ const ShapeStepName

+ var ErrAborted = errors.New("duro: pipeline aborted by an earlier stage failure")

+ var ErrNoValue = errors.New("duro: pipeline completed without emitting a value")

+ func RunAll[P, R any](ctx dbos.DBOSContext, in P, p Pipeline[P, R]) ([]R, error)

+ func Run[P, R any](ctx dbos.DBOSContext, in P, p Pipeline[P, R]) (R, error)

+ type Pipeline struct

+ func Pipe1[A, B any](s1 Stage[A, B]) Pipeline[A, B]

+ func Pipe2[A, B, C any](s1 Stage[A, B], s2 Stage[B, C]) Pipeline[A, C]

+ func Pipe3[A, B, C, D any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D]) Pipeline[A, D]

+ func Pipe4[A, B, C, D, E any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D], s4 Stage[D, E]) Pipeline[A, E]

+ func Pipe5[A, B, C, D, E, F any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D], s4 Stage[D, E], s5 Stage[E, F]) Pipeline[A, F]

+ func Pipe6[A, B, C, D, E, F, G any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D], s4 Stage[D, E], s5 Stage[E, F], ...) Pipeline[A, G]

+ func Pipe7[A, B, C, D, E, F, G, H any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D], s4 Stage[D, E], s5 Stage[E, F], ...) Pipeline[A, H]

+ func Pipe8[A, B, C, D, E, F, G, H, I any](s1 Stage[A, B], s2 Stage[B, C], s3 Stage[C, D], s4 Stage[D, E], s5 Stage[E, F], ...) Pipeline[A, I]

+ type Stage struct

+ func Expand[T, R any](name string, fn func(ctx context.Context, in T) ([]R, error), ...) Stage[T, R]

+ func Filter[T any](name string, pred func(ctx context.Context, in T) (bool, error), ...) Stage[T, T]

+ func Pure[T, R any](name string, fn func(in T) R) Stage[T, R]

+ func Reduce[T, A any](name string, fn func(ctx context.Context, acc A, in T) (A, error), seed A, ...) Stage[T, A]

+ func Step[T, R any](name string, fn func(ctx context.Context, in T) (R, error), opts ...StepOption) Stage[T, R]

+ func Tap[T any](name string, fn func(ctx context.Context, in T) error, opts ...StepOption) Stage[T, T]

+ func UnsafeOperator[T, R any](name string, op func(ro.Observable[T]) ro.Observable[R]) Stage[T, R]

+ type StepOption func(*stepConfig)

+ func WithBaseInterval(d time.Duration) StepOption

+ func WithMaxRetries(n int) StepOption

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL