duro

README

duro

Durable dataflow pipelines for Go.

duro lets you write DBOS durable workflows as typed dataflow pipelines. Every stage is checkpointed to Postgres: if your process dies mid-pipeline, the workflow resumes exactly where it left off — completed stages replay from their checkpoints instead of re-running. And the API is designed so that code which would corrupt recovery either doesn't compile or fails fast with a clear error.

var OrderPipeline = duro.Pipe4(
	duro.Step("validate", validateOrder),
	duro.Step("reserve", reserveInventory),
	duro.Step("charge", chargePayment, duro.WithMaxRetries(3)),
	duro.Step("notify", sendConfirmation),
)

// at startup:
orders := duro.Register(app, "orders", OrderPipeline)
// anywhere:
handle, err := orders.Start(app, order)
confirmation, err := handle.Result()

Kill the process right after reserve completes. On restart, validate and reserve replay from Postgres in microseconds — without executing your functions — and the workflow continues at charge. Durable orchestration without writing a single line of recovery code.

Why duro

• Durable by construction — each stage runs inside dbos.RunAsStep; DBOS checkpoints the result and recovers crashed workflows automatically.
• Typed end to end — Pipe4(Step[Order→Validated], Step[Validated→Reservation], …) type-checks the whole chain; mismatched stages don't compile.
• Streams, not just sequences — Expand, Filter, and Reduce process collections item by item, with a checkpoint per stage execution.
• Safety as a feature — three guard layers (compiler, construction-time, execution-time) make the classic durable-workflow footguns hard to fire; see Built-in safety.
• The whole DBOS toolkit, typed — messaging, events, and streams (both directions), scheduled and debounced pipelines, stage-level forking, and per-child queue controls — each as a small, composable surface over the DBOS primitive it wraps.
• Durable control flow — Branch, Switch, and Loop make routing and polling checkpointed stages instead of hand-written workflow code, and Status/Attach reconcile any persisted run ID from any process.

Installation

go get github.com/lemonberrylabs/duro

Requires Go 1.26+ and PostgreSQL (DBOS's system database).

Quickstart

A complete durable application — note the single import:

package main

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

	"github.com/lemonberrylabs/duro"
)

type Order struct {
	ID          string
	AmountCents int
}

type Receipt struct {
	OrderID   string
	PaymentID string
}

var ChargePipeline = duro.Pipe2(
	duro.Step("charge", func(_ context.Context, o Order) (Receipt, error) {
		return Receipt{OrderID: o.ID, PaymentID: "pay-" + o.ID}, nil // call your payment provider here
	}, duro.WithMaxRetries(3)),
	duro.Tap("audit", func(_ context.Context, r Receipt) error {
		fmt.Printf("audited %s\n", r.OrderID)
		return nil
	}),
)

func main() {
	app, err := duro.New(context.Background(), duro.Config{
		Name:        "quickstart",
		DatabaseURL: os.Getenv("DBOS_SYSTEM_DATABASE_URL"),
	})
	if err != nil {
		panic(err)
	}

	charge := duro.Register(app, "charge-order", ChargePipeline)

	if err := app.Launch(); err != nil {
		panic(err)
	}
	defer app.Shutdown(5 * time.Second)

	handle, err := charge.Start(app, Order{ID: "42", AmountCents: 1999})
	if err != nil {
		panic(err)
	}
	receipt, err := handle.Result()
	if err != nil {
		panic(err)
	}
	fmt.Printf("charged: %+v\n", receipt)
}

createdb quickstart
DBOS_SYSTEM_DATABASE_URL=postgres://$USER@localhost:5432/quickstart go run .

The primitives

Stage	What it does	Durable?
duro.Step(name, fn)	transform T → R	✅ checkpointed step
duro.Tap(name, fn)	side effect, passes T through	✅ checkpointed step
duro.Filter(name, pred)	drop items failing the predicate	✅ predicate is a step
duro.Expand(name, fn)	one item → many (T → []R), emitted in order	✅ checkpointed step
duro.Reduce(name, fn, seed)	fold the stream; emits the final accumulator	✅ one step per accumulation
duro.FanOut(name, queue, wf)	parallel map: each item runs as a child workflow on a declared Queue	✅ every child is a durable workflow
duro.Parallel(name, max, fn)	parallel map: concurrent steps in-process, at most max at a time	✅ pre-assigned step per item
duro.Delay(name, d)	durable pause per item	✅ recovery resumes the remaining time
duro.Send(name, topic, fn)	message another workflow's mailbox on a typed Topic	✅ no re-send on replay
duro.Recv(name, topic, timeout)	pause until a Topic message arrives; emits it	✅ no double-consume on replay
duro.SetEvent(name, event, fn)	publish per-item progress on a typed Event	✅ checkpointed
duro.GetEvent(name, event, fn, timeout)	read another workflow's Event; emits it	✅ replay returns the observed value
duro.ToStream(name, stream)	append items to a typed durable Stream, closed at completion	✅ checkpointed
duro.FromStream(name, stream, fn)	drain another workflow's Stream; emits its values	✅ the whole read is one checkpoint
duro.Branch(name, pred, then, els)	route each item through one of two pipelines	✅ the verdict is a checkpointed step
duro.Switch(name, route, When(k, p)...)	multi-way dispatch to case pipelines	✅ the route key is a checkpointed step
duro.Loop(name, body, until)	repeat a pipeline until done; durable polling with Delay	✅ every iteration's verdict is checkpointed
duro.Sub(name, pipeline)	embed a pipeline as one named stage (whole-stream)	✅ its stages checkpoint as usual
duro.Collect(name)	fold the stream into a slice of every item	✅ checkpointed; empty stream → empty slice
duro.Pure(name, fn)	cheap reshaping between stages	❌ re-executes on replay — must be deterministic and side-effect free
duro.UnsafeOperator(name, op)	escape hatch for raw ro operators	⚠️ you're on your own (runtime guards still apply)

Stages compose with duro.Pipe1…Pipe8 into a Pipeline[P, R]. Register one as a workflow (duro.Register, below) or run it inside a hand-written workflow body:

• duro.Run(ctx, input, pipeline) — returns the last emitted value
• duro.RunAll(ctx, input, pipeline) — returns every emitted value

For the rare logic no combinator fits (mostly interop with existing DBOS code), hand-written workflows are declared against duro.Context — an alias for DBOS's context type — and registered with duro.RegisterWorkflow:

func InvoiceWorkflow(ctx duro.Context, b Batch) (Invoice, error) {
	return duro.Run(ctx, b, InvoicePipeline)
}

invoice := duro.RegisterWorkflow(app, "invoice", InvoiceWorkflow) // before app.Launch
handle, err := invoice.Start(app, batch)

Stage options

Every step-backed stage takes functional options:

• Retries — WithMaxRetries(n), WithBaseInterval(d), WithMaxInterval(d), WithBackoffFactor(f) for the exponential-backoff envelope, and WithRetryPredicate(pred) to spend retries on transient errors only:

  duro.Step("charge", chargePayment,
      duro.WithMaxRetries(5),
      duro.WithBaseInterval(200*time.Millisecond),
      duro.WithMaxInterval(10*time.Second),
      duro.WithRetryPredicate(func(err error) bool { return !errors.Is(err, ErrCardDeclined) }),
  )
  

• Timeouts — WithTimeout(d) cancels each attempt's step context after d (per attempt; the stage function must honor cancellation). For a durable deadline on a whole pipeline run, start its workflow from a context derived with dbos.WithTimeout; for child workflows, see WithChildTimeout below.

• Serialization — declare a channel with duro.Portable() (duro.NewTopic[M](name, duro.Portable())) to write its payloads in DBOS's cross-language format so Python/TypeScript DBOS apps can consume them. Readers need nothing special — DBOS decodes by each value's recorded serialization.

Multi-item pipelines

Durability per item, not just per workflow — each stage execution below is its own checkpoint, so a crash mid-batch resumes at the exact item and stage where it stopped:

func InvoiceWorkflow(ctx duro.Context, b Batch) (Invoice, error) {
	return duro.Run(ctx, b, duro.Pipe5(
		duro.Expand("explode", func(_ context.Context, b Batch) ([]LineItem, error) {
			return b.Items, nil
		}),
		duro.Filter("in-stock", func(_ context.Context, li LineItem) (bool, error) {
			return li.InStock, nil
		}),
		duro.Step("price", priceItem),
		duro.Tap("audit", auditItem),
		duro.Reduce("total", func(_ context.Context, acc Invoice, p PricedItem) (Invoice, error) {
			acc.TotalCents += p.TotalCents
			acc.ItemCount++
			return acc, nil
		}, Invoice{BatchID: b.ID}),
	))
}

Durable control flow

Branching, dispatch, and loops are stages, not hand-written code — each decision runs as a checkpointed step (the same mechanism Filter uses), so a recovered workflow re-reads its recorded decisions and walks the same stage sequence:

duro.Pipe3(
	duro.Expand("explode", explodeTickets),
	duro.Switch("dispatch", func(_ context.Context, t Ticket) (string, error) { return t.Category, nil },
		duro.When("billing", billingPipeline),
		duro.When("bug", duro.Pipe1(
			duro.Branch("urgency", isUrgent,
				escalatePipeline, // contains a duro.Loop polling until fixed
				filePipeline,
			),
		)),
	),
	duro.Collect[Resolution]("report"),
)

Embedded pipelines fold into the shape fingerprint, so editing an arm or body still fails fast on replay. Branch/Switch/Loop apply per item; Sub(name, pipeline) embeds a segment over the whole stream (an inner Reduce folds everything). See examples/triage.

Parallel fan-out

Need "20 jobs, at most 4 at a time, merge the results"? FanOut runs each item as a child workflow on a DBOS queue, so parallelism, rate limits, and distribution across processes are all governed by the queue — and every child is independently durable:

var Jobs = duro.NewQueue("jobs", duro.WithConcurrency(4)) // declared once, referenced by value

var ProcessAll = duro.Pipe3(
	duro.Expand("explode", func(_ context.Context, js []Job) ([]Job, error) { return js, nil }),
	duro.FanOut("process", Jobs, duro.Workflow(ProcessJob)), // or pass a Registered/PipelineWorkflow directly
	duro.Reduce("merge", mergeResults, Merged{}),
)

Registering a pipeline with Register automatically registers every queue it references — no separate registration call, no name to keep in sync. (For pipelines run with Run inside hand-written workflows, call duro.RegisterQueues(app, Jobs) once at startup.) Queue knobs are duro options: WithConcurrency, WithWorkerConcurrency, WithRateLimit, WithPriorities, WithPartitions. Declaring the same queue name twice with different configurations fails loudly at registration.

Children are enqueued in stream order with IDs derived from the parent's step counter, so a recovered parent re-attaches to its children instead of spawning duplicates; results are awaited and emitted in input order, each checkpointed in the parent. Determinism is preserved because DBOS checkpoints both the spawns and the awaits.

Child workflows are configured with ChildOptions. Identity options derive a per-child value from the item; policy options apply to every child:

duro.FanOut("process", Jobs, duro.Workflow(ProcessJob),
	duro.WithChildID(func(j Job) string { return "job-" + j.ID }), // idempotency across runs
	duro.WithChildDeduplicationID(func(j Job) string { return j.CustomerID }),
	duro.WithChildDeduplicationPolicy(duro.DeduplicationReturnExisting),
	duro.WithChildPartitionKey(func(j Job) string { return j.Region }),
	duro.WithChildPriority(2),                    // queue must enable priorities
	duro.WithChildDelay(time.Minute),             // start children DELAYED
	duro.WithChildTimeout(10*time.Minute),        // durable per-child deadline
	duro.WithChildAuthenticatedUser("billing-svc"),
	duro.WithChildAppVersion(version),            // pin recovery to a code version
	duro.WithPortableChildren(),                  // cross-language serialization
)

Item-derived options are typed by the stage's item; a mismatch panics at construction time, like every other stage validation. A registered pipeline is itself a valid FanOut child — pass it directly: duro.FanOut("sub", Jobs, childPipeline).

When you don't need queue-level distribution or per-child durability, duro.Parallel(name, max, fn) is the lightweight sibling: concurrent steps inside the workflow process (built on dbos.Go, which pre-assigns each step's ID deterministically), bounded by max, with results in input order — no queue, no polling latency.

Signals, events, and streams

Messaging goes through typed channels: declare a Topic, Event, or Stream once, and both sides reference the value. The key lives in one place and the payload type is compiler-checked — sending an Approval where the receiver expects an Invoice doesn't compile:

var (
	Approvals = duro.NewTopic[Approval]("approvals")
	Receipts  = duro.NewStream[Receipt]("receipts")
)

duro.Pipe5(
	duro.Step("prepare", prepare),
	duro.Delay[Prepared]("cool-off", 24*time.Hour),        // durable sleep — survives restarts
	duro.Recv[Prepared]("await-approval", Approvals, 72*time.Hour),
	duro.Step("execute", execute),                          // human-in-the-loop, durably
	duro.ToStream("publish", Receipts),                     // readers consume incrementally
)

• Delay checkpoints its wake-up deadline: a workflow recovered mid-sleep sleeps only the remaining time; a replayed sleep is instant.
• Recv parks the pipeline until a message arrives on the workflow's mailbox; receipt is checkpointed so recovery never consumes a second message. Send is the in-pipeline counterpart; from a plain client, Approvals.Send(app, workflowID, approval).
• SetEvent publishes per-item progress readable while the pipeline runs; ToStream appends each item to a durable stream, closed when the pipeline completes. Clients read with event.Get(app, id, timeout) and stream.Read(app, id).
• Both have read-side stages: GetEvent durably observes another workflow's event, and FromStream drains another workflow's stream into the pipeline — one checkpoint for the whole read, so replay never re-reads a stream that has since changed. Bound the wait with duro.WithTimeout.

Pipelines as workflows

Register makes a pipeline a first-class DBOS workflow — no wrapper function to write — and two variants cover scheduling and burst-collapsing:

wf := duro.Register(app, "invoice", invoicePipeline) // after duro.New, before app.Launch
handle, err := wf.Start(app, batch)                  // → duro.Handle[Invoice]
result, err := handle.Result()

// Cron pipelines: input is the tick time, enforced at compile time.
duro.RegisterScheduled(app, "nightly-report", "0 0 2 * * *",
	reportPipeline) // Pipeline[time.Time, Report]

// Debounced pipelines: bursts collapse into one run with the last input.
deb := duro.RegisterDebounced(app, "reindex", reindexPipeline)
deb.Debounce(app, userID, 30*time.Second, req) // each call pushes the start back

The registered name is the pipeline's durable identity: in-flight runs are recovered by looking it up, so register the same name on every process start. app.Launch() checks for runs recorded under names that are no longer registered and warns about each one — a renamed pipeline is a startup warning, not a silent recovery failure.

Checking on runs

Persist a run's workflow ID and reconcile it later from any process — the check-on-read pattern — with duro-owned types, no Handle required:

status, err := duro.Status(app, runID) // errors.Is(err, duro.ErrRunNotFound) for unknown IDs
switch {
case status.State.Failed():   // error | cancelled | retries_exceeded
	markFailed(record, status.Err) // the recorded failure reason, always non-nil when Failed
case status.State.Terminal(): // success
	markDone(record)
default:                      // pending | enqueued | delayed
	// still working
}

Status never loads run payloads (failure reasons for failed runs are fetched in a scoped second query), so it is safe to poll. StatusAll(app, ids...) is the batch form; duro.Attach[R](app, runID) reconnects a restarted process to a live handle so it can await the result, not just poll. Handle.Status() returns the same RunStatus.

Forking from a stage

ForkFromStage restarts an existing run from a named stage: earlier stages replay from checkpoints, the named stage and everything after re-execute — optionally on a different application version (the recovery tool for re-running a workflow on fixed code after a bad deploy):

handle, err := duro.ForkFromStage[Confirmation](app, duro.Fork{
	WorkflowID:         failedRunID,
	Stage:              "charge",
	ApplicationVersion: fixedVersion, // optional; inherits when empty
})

Built-in safety

DBOS replay determinism requires that the Nth step call on recovery is the same logical operation as in the original run. Reactive operators make that easy to violate — concurrency reorders steps between runs, timers change emission counts — and the worst failure is silent: two identically-named steps swapping positions hand one step the other's checkpointed output. duro guards this in three layers:

1. Compile time. PipeN only accepts Stage values, which only duro's constructors can create (nominal struct typing). Raw ro operators don't type-check, and Run owns the source — there is no way to plug in a channel-fed or timer-based stream.
2. Construction time. Run checkpoints the pipeline's shape (ordered stage kinds + names) as a hidden first step, duro.shape. A replay that builds a different pipeline — non-deterministic construction, changed code — fails immediately with a shape-mismatch error instead of reading misaligned checkpoints.
3. Execution time. Every stage asserts it runs on the goroutine the pipeline was subscribed on, catching smuggled concurrency before it can race DBOS's step counter. After any stage failure, a shared abort flag stops items behind the failure from executing further stages — fail-fast, like sequential code.

What remains yours, as in every durable-workflow system: Pure functions and pipeline construction must be deterministic.

How it works

duro.Context (DBOS's context type) implements context.Context, and samber/ro — the reactive engine under duro's hood — propagates the subscription context unchanged through its operators. duro.Run subscribes the composed pipeline with the workflow's context; each stage recovers it and executes its function via dbos.RunAsStep, one durable checkpoint per stage execution. Synchronous emission on the workflow goroutine keeps step order deterministic — exactly what DBOS replay requires.

Example apps

Each example is a runnable app with its own README; together they cover the whole feature set:

• examples/payments — signals & resilience: retry options and predicates, per-attempt timeouts, durable pauses, human-in-the-loop approval over a typed Topic, progress Events, and a portable receipt Stream drained by a second pipeline.

• examples/thumbnails — fan-out fleets: queue declarations (concurrency, rate limits, priorities, partitions), every child option (idempotent IDs, deduplication, timeouts, delays, auth), a hand-written child on duro.Context with RunAll and Parallel, and a registered pipeline used directly as a FanOut child.

• examples/triage — durable control flow: Switch dispatch, a nested Branch, a polling Loop, shared segments with Sub, and Collect — plus the replay proof that recorded decisions drive the same path.

• examples/housekeeping — operations: cron pipelines, debounced bursts, forking a finished run from a named stage after a fix, and a three-phase walkthrough of the durable-identity contract and the stranded-run warning.

• examples/orders — the fundamentals: the same order workflow written both as plain sequential DBOS steps and as a duro pipeline (their recorded checkpoints are identical), plus a crash-recovery demo:

  createdb duro_demo
  cd examples/orders
  go run .                                      # all variants + step-sequence dumps
  go run . -variant=duro -crash-after=reserve   # die mid-workflow…
  go run .                                      # …recover: watch replayed steps keep their old timestamps
  

Status

Experimental. The durability semantics are covered by a test suite that runs against a real Postgres — including parity with handwritten DBOS code, re-runs with zero re-execution, mid-pipeline fork/replay, and one test per safety guard. Both underlying dependencies (dbos-transact-golang, samber/ro) are pre-1.0, so expect pinned versions and occasional churn until they stabilize.

Contributions welcome — see CONTRIBUTING.md.

Acknowledgements

• Hat tip to @samber — duro exists because ro's operator model (and its meticulous context propagation) turned out to compose beautifully with durable execution. If you like lo, go look at ro.
• The DBOS team, whose Postgres-backed durable workflow engine does all the heavy lifting here.

License

MIT © Lemonberry Labs

Documentation

Overview

Package duro is a durable dataflow DSL: reactive pipelines whose every stage runs as a checkpointed DBOS step. Pipelines are powered by samber/ro internally, but the public API only accepts duro stages, making it a compile error to insert a raw ro operator that could break durability.

A workflow body is a pipe of typed stages:

func OrderWorkflow(ctx duro.Context, o Order) (Confirmation, error) {
	return duro.Run(ctx, o, duro.Pipe4(
		duro.Step("validate", validateOrder),
		duro.Step("reserve", reserveInventory),
		duro.Step("charge", chargePayment, duro.WithMaxRetries(3)),
		duro.Step("notify", sendConfirmation),
	))
}

Durability relies on DBOS replay determinism: on recovery the workflow function re-executes and the Nth step call must be the same logical operation as in the original run. duro enforces this with three layers:

• Compile time: PipeN only accepts Stage values, which can only be built by this package's constructors. Concurrent or time-based ro operators cannot be expressed. Sources cannot be swapped either — Run feeds the workflow input through ro.Of internally.
• Construction time: Run checkpoints the pipeline's shape (ordered stage kinds and names) as a hidden first step named "duro.shape". If a replay constructs a different shape — non-deterministic pipeline construction, changed code — Run fails immediately instead of letting stages read misaligned checkpoints.
• Execution time: every stage asserts it runs on the goroutine the pipeline was subscribed on, failing fast if an operator smuggled in concurrency; and once any stage fails, a shared abort flag prevents items behind the failure from executing further stages (fail-fast, like sequential workflow code).

Escape hatches: Pure wraps a deterministic, side-effect-free transform that is NOT checkpointed (it re-executes on every replay — it must be pure), and UnsafeOperator admits an arbitrary ro operator with no safety guarantees beyond the runtime guards. Both participate in the shape fingerprint.

For parallelism, use FanOut (child workflows on a DBOS queue — bounded, distributed, per-child durability) or Parallel (concurrent steps in-process via dbos.Go — lightweight, no queue). Both preserve replay determinism: work is spawned and awaited in stream order on the workflow goroutine.

The rest of DBOS's workflow toolkit is available as stages: Delay (durable sleep), Send/Recv (durable mailbox messaging — external signals and human-in-the-loop pauses), SetEvent/GetEvent (progress events published and read durably), and ToStream/FromStream (durable streams written and drained durably). Messaging goes through typed channels — Topic, Event, Stream — declared once and referenced by both sides, so keys and payload types cannot drift; declare a channel with Portable() to serialize its payloads in DBOS's cross-language format.

Control flow is durable too: Branch and Switch route each item through embedded pipelines by a checkpointed decision, Loop repeats a pipeline until a checkpointed verdict says done, Sub embeds a pipeline as one named stage, and Collect folds the stream into a slice. Embedded pipelines are part of the shape fingerprint.

Pipelines are also registrable as first-class workflows: Register names a pipeline as a DBOS workflow, RegisterScheduled runs one on a cron schedule (typed Pipeline[time.Time, R]), RegisterDebounced collapses bursts of triggers into a single run, and RegisterWorkflow covers hand-written workflow functions. Runs are tracked by workflow ID from any process: Status/StatusAll reconcile a persisted ID against the engine, Attach reconnects to a live handle, and ForkFromStage restarts a completed or failed run from a named stage — optionally onto a different application version.

Example

Example shows a DBOS workflow written as a durable pipeline: each stage runs as a checkpointed DBOS step, so a crashed workflow resumes after the last completed stage. Register the workflow with dbos.RegisterWorkflow and start it with dbos.RunWorkflow as usual.

package main

import (
	"context"

	"github.com/dbos-inc/dbos-transact-golang/dbos"

	"github.com/lemonberrylabs/duro"
)

func main() {
	type Order struct {
		ID          string
		AmountCents int
	}
	type Receipt struct {
		OrderID   string
		PaymentID string
	}

	chargeOrder := func(ctx dbos.DBOSContext, o Order) (Receipt, error) {
		return duro.Run(ctx, o, duro.Pipe2(
			duro.Step("charge", func(_ context.Context, o Order) (string, error) {
				return "pay-" + o.ID, nil // call your payment provider here
			}, duro.WithMaxRetries(3)),
			duro.Step("receipt", func(_ context.Context, paymentID string) (Receipt, error) {
				return Receipt{OrderID: o.ID, PaymentID: paymentID}, nil
			}),
		))
	}
	_ = chargeOrder
}

Index

• Constants
• Variables
• func RegisterQueues(ctx Context, queues ...Queue) error
• func Run[P, R any](ctx Context, in P, p Pipeline[P, R]) (R, error)
• func RunAll[P, R any](ctx Context, in P, p Pipeline[P, R]) ([]R, error)
• func WithWorkflowID(id string) dbos.WorkflowOption
• type App
  • 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 Case
  • func When[T, R any](key string, p Pipeline[T, R]) Case[T, R]
• type ChannelOption
  • func Portable() ChannelOption
• type ChildOption
  • 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
• type Context
• type Debouncer
  • 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
• type Event
  • 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
• type Handle
  • func Attach[R any](ctx Context, workflowID string) (Handle[R], error)
  • 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() (RunStatus, error)
• type Pipeline
  • 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 PipelineWorkflow
  • func Register[P, R any](ctx Context, name string, p Pipeline[P, R], ...) *PipelineWorkflow[P, R]
  • func RegisterScheduled[R any](ctx Context, name, cronSchedule string, p Pipeline[time.Time, R], ...) *PipelineWorkflow[time.Time, 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
  • func NewQueue(name string, opts ...QueueOption) Queue
  • func (q Queue) Name() string
• type QueueOption
  • 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 RegisteredWorkflow
  • 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)
• type RunStatus
  • 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 Delay[T any](name string, d time.Duration) Stage[T, T]
  • func Expand[T, R any](name string, fn func(ctx context.Context, in T) ([]R, error), ...) Stage[T, R]
  • func FanOut[T, R any](name string, queue Queue, wf WorkflowRef[T, R], opts ...ChildOption) Stage[T, R]
  • func Filter[T any](name string, pred func(ctx context.Context, in T) (bool, error), ...) Stage[T, T]
  • 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 Loop[T any](name string, body Pipeline[T, T], ...) Stage[T, T]
  • func Parallel[T, R any](name string, maxConcurrent int, fn func(ctx context.Context, in T) (R, error), ...) Stage[T, R]
  • func Pure[T, R any](name string, fn func(in T) R) Stage[T, R]
  • func Recv[T, M any](name string, topic Topic[M], timeout time.Duration) Stage[T, M]
  • func Reduce[T, A any](name string, fn func(ctx context.Context, acc A, in T) (A, error), seed A, ...) Stage[T, A]
  • 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 Step[T, R any](name string, fn func(ctx context.Context, in T) (R, error), opts ...StepOption) Stage[T, R]
  • 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]
  • func Tap[T any](name string, fn func(ctx context.Context, in T) error, opts ...StepOption) Stage[T, T]
  • func ToStream[T any](name string, stream Stream[T]) Stage[T, T]
  • func UnsafeOperator[T, R any](name string, op func(ro.Observable[T]) ro.Observable[R]) Stage[T, R]
• type State
  • func (s State) Failed() bool
  • func (s State) Terminal() bool
• type StepOption
  • func WithBackoffFactor(factor float64) StepOption
  • func WithBaseInterval(d time.Duration) StepOption
  • func WithMaxInterval(d time.Duration) StepOption
  • func WithMaxRetries(n int) StepOption
  • func WithRetryPredicate(pred func(error) bool) StepOption
  • func WithTimeout(d time.Duration) StepOption
• type Stream
  • 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
  • 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
• type WorkflowRef
  • func Workflow[T, R any](fn WorkflowFunc[T, R]) WorkflowRef[T, R]

Examples

• Package

Constants

const (
	// DeduplicationReject (the default) fails the enqueue of a child whose
	// deduplication ID is already held by an active child.
	DeduplicationReject = dbos.DeduplicationPolicyReject
	// DeduplicationReturnExisting returns the existing child's handle
	// instead, so both items observe the first child's result.
	DeduplicationReturnExisting = dbos.DeduplicationPolicyReturnExisting
)

const ShapeStepName = "duro.shape"

ShapeStepName is the name of the hidden bookkeeping step Run records as the pipeline's first checkpoint. It holds the pipeline's shape fingerprint, which Run compares on replay to fail fast on non-deterministic pipeline construction.

Variables

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

ErrAborted marks stage executions skipped because an earlier stage already failed. It never surfaces from Run: the first failure is the pipeline's error, and ErrAborted only travels through already-terminated downstream observers.

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

ErrNoValue is returned by Run when the pipeline completes without emitting any value (for example, when a Filter stage drops every item).

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

ErrRunNotFound is returned by Status and Attach for an unknown workflow ID.

Functions

func RegisterQueues

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

RegisterQueues registers declared queues with DBOS. Register does this automatically for every queue its pipeline references; call RegisterQueues yourself only for pipelines run directly with Run/RunAll inside hand-written workflows.

func Run

func Run[P, R any](ctx Context, in P, p Pipeline[P, R]) (R, error)

Run executes the pipeline durably inside a DBOS workflow, feeding it the input value and blocking until completion. It returns the last emitted value, the first stage error, or ErrNoValue if the pipeline emits nothing. Call it as the body of a registered DBOS workflow function.

func RunAll

func RunAll[P, R any](ctx Context, in P, p Pipeline[P, R]) ([]R, error)

RunAll is Run for pipelines whose final stage legitimately emits multiple items: it returns every emitted value.

func WithWorkflowID

func WithWorkflowID(id string) dbos.WorkflowOption

WithWorkflowID assigns a run's workflow ID — the standard idempotency key: starting the same ID twice re-attaches to the first run instead of running again. Every other dbos.WorkflowOption passes through Start unchanged.

Types

type App

type App struct {
	dbos.DBOSContext
	// contains filtered or unexported fields
}

App owns the DBOS lifecycle so applications never touch it directly:

app, err := duro.New(ctx, duro.Config{Name: "orders", DatabaseURL: url})
wf := duro.Register(app, "invoice", invoicePipeline) // register everything...
err = app.Launch()                                   // ...then launch
defer app.Shutdown(5 * time.Second)
handle, err := wf.Start(app, batch)

Launch also checks for stranded runs: in-flight workflows recorded under names no longer registered (a renamed pipeline) are reported as warnings instead of silently never recovering.

*App satisfies Context, so it can be passed wherever duro expects one. Calling raw dbos package functions directly is different: several inspect the concrete context type, so hand them Context() rather than the App itself.

func New

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

New initializes the application. Register pipelines and queues after New and before Launch.

func (*App) Context

func (a *App) Context() Context

Context returns the underlying DBOS context — for calling raw dbos package functions directly. Everything in duro accepts the App itself.

func (*App) Launch

func (a *App) Launch() error

Launch starts DBOS: workflow recovery, queue runners, and schedulers. Call it after all registrations. It then warns about stranded runs — see App.

func (*App) Shutdown

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

Shutdown stops DBOS, waiting up to timeout for in-flight work to settle.

type Case

type Case[T, R any] struct {
	// contains filtered or unexported fields
}

Case pairs a route key with the pipeline that handles it; see Switch.

func When

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

When declares a Switch case.

type ChannelOption

type ChannelOption func(*channelConfig)

ChannelOption configures a declared channel.

func Portable

func Portable() ChannelOption

Portable makes every payload written through the channel serialize in DBOS's cross-language portable JSON format, so non-Go DBOS applications (Python, TypeScript) can consume it. Readers need nothing special — DBOS decodes by each value's recorded serialization.

type ChildOption

type ChildOption func(*childConfig)

ChildOption configures the child workflows a FanOut stage enqueues. Policy options (priority, delay, timeout, auth, serialization, version) apply uniformly to every child; identity options (workflow ID, deduplication ID, partition key) derive a per-child value from the stream item. Item-derived options are typed by the stage's item type — FanOut panics at construction time if they were built for a different type.

func WithChildAppVersion

func WithChildAppVersion(version string) ChildOption

WithChildAppVersion pins children to a specific application version, overriding the parent's. This affects which executors recover them.

func WithChildAssumedRole

func WithChildAssumedRole(role string) ChildOption

WithChildAssumedRole records the assumed role on every child workflow's status.

func WithChildAuthenticatedRoles

func WithChildAuthenticatedRoles(roles ...string) ChildOption

WithChildAuthenticatedRoles records the authenticated roles on every child workflow's status.

func WithChildAuthenticatedUser

func WithChildAuthenticatedUser(user string) ChildOption

WithChildAuthenticatedUser records the authenticated user on every child workflow's status.

func WithChildDeduplicationID

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

WithChildDeduplicationID derives a queue deduplication ID from each item. While a child holding the ID is active on the queue, enqueueing another with the same ID is rejected — or returns the existing child's handle under dbos.DeduplicationPolicyReturnExisting (see WithChildDeduplicationPolicy).

func WithChildDeduplicationPolicy

func WithChildDeduplicationPolicy(policy DeduplicationPolicy) ChildOption

WithChildDeduplicationPolicy sets how a colliding deduplication ID is handled (default DeduplicationReject).

func WithChildDelay

func WithChildDelay(d time.Duration) ChildOption

WithChildDelay delays each child's dequeue by d: children start in the DELAYED status and become runnable once the delay expires.

func WithChildID

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

WithChildID derives each child's workflow ID from its item, making child runs idempotent under an application-level key (e.g. an order ID): starting the same pipeline twice re-attaches to the same children instead of spawning duplicates. Without it, child IDs derive from the parent's step counter, which is idempotent per parent run but not across runs.

func WithChildPartitionKey

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

WithChildPartitionKey derives each child's queue partition key from its item. The queue must be registered with dbos.WithPartitionQueue; each partition then gets its own concurrency limits.

func WithChildPriority

func WithChildPriority(priority uint) ChildOption

WithChildPriority sets every child's queue priority (lower runs first). The queue must be registered with dbos.WithPriorityEnabled.

func WithChildTimeout

func WithChildTimeout(d time.Duration) ChildOption

WithChildTimeout gives every child a durable workflow deadline of d from its enqueue time: the deadline is stored with the child's status, survives recovery, and cancels the child when it expires. A timed-out child fails the pipeline when its result is awaited.

func WithPortableChildren

func WithPortableChildren() ChildOption

WithPortableChildren stores each child's inputs, step outputs, events, messages, and streams in DBOS's cross-language portable JSON format, so non-Go DBOS applications can read them.

type Config

type Config struct {
	// Name identifies the application in the system database.
	Name string
	// DatabaseURL is the Postgres URL of the DBOS system database.
	DatabaseURL string
	// Logger receives duro and DBOS logs; slog.Default() when nil.
	Logger *slog.Logger
}

Config configures a duro application.

type Context

type Context = dbos.DBOSContext

Context is the durable execution context every workflow runs under: it carries the checkpoint state duro's stages record to. It is DBOS's context type under a duro name (a type alias), so it satisfies context.Context and remains directly usable with any dbos API — but declaring and running workflows never requires importing dbos:

func Process(ctx duro.Context, job Job) (Result, error)

type Debouncer

type Debouncer[P, R any] struct {
	// contains filtered or unexported fields
}

Debouncer collapses bursts of pipeline starts into a single run; see RegisterDebounced.

func RegisterDebounced

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

RegisterDebounced registers the pipeline as a workflow and returns its debouncer. Cap the total postponement with dbos.WithDebouncerTimeout. Like Register, call it after New and before Launch.

func (*Debouncer[P, R]) Debounce

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

Debounce postpones the pipeline's start by delay. Every further call with the same key pushes the start back and replaces the input; when the delay lapses, the pipeline runs once with the last input. Different keys debounce independently. Every call returns a handle to the same eventual run.

type DeduplicationPolicy

type DeduplicationPolicy = dbos.DeduplicationPolicy

DeduplicationPolicy controls how a colliding child deduplication ID is handled; see WithChildDeduplicationPolicy.

type Event

type Event[V any] struct {
	// contains filtered or unexported fields
}

Event is a typed key-value event channel: a value of type V published on a workflow under one key. Write with a SetEvent stage; read with a GetEvent stage or Event.Get.

var Progress = duro.NewEvent[int]("last-item")

func NewEvent

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

NewEvent declares a typed event key.

func (Event[V]) Get

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

Get reads the event's value from the given workflow, blocking until it is set or the timeout elapses. Inside a workflow the read is checkpointed; from a plain client it is a direct read. Pipelines should prefer the GetEvent stage.

func (Event[V]) Key

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

Key returns the event's key.

type Fork

type Fork struct {
	WorkflowID string // the pipeline run to fork
	Stage      string // the stage to restart from (its first execution, for stages that ran per item)

	// ForkedID names the forked run; auto-generated when empty.
	ForkedID string
	// ApplicationVersion pins the forked run to a different code version —
	// the recovery tool for rerunning a workflow on fixed code after a bad
	// deploy.
	ApplicationVersion string
	// Queue enqueues the forked run on the named queue instead of running it
	// on the internal one; QueuePartitionKey partitions it there.
	Queue             string
	QueuePartitionKey string
}

Fork describes where to restart an existing pipeline run. WorkflowID and Stage are required; every other field is optional and its zero value means "inherit from the original run".

type Handle

type Handle[R any] struct {
	// contains filtered or unexported fields
}

Handle tracks one durable pipeline run. It is returned by Start and ForkFromStage; Result blocks until the run completes.

func Attach

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

Attach reconnects to an existing run by workflow ID and returns its handle — how a restarted process awaits a result instead of just polling Status. R must match the workflow's result type.

func ForkFromStage

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

ForkFromStage restarts an existing pipeline run from a named stage: stages before it replay from the original run's checkpoints, the named stage and everything after re-execute. The pipeline's shape guard replays too, so a fork onto changed pipeline code fails fast instead of misreading checkpoints.

func (Handle[R]) ID

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

ID returns the run's workflow ID.

func (Handle[R]) Result

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

Result blocks until the run completes and returns its result or error.

func (Handle[R]) Status

func (h Handle[R]) Status() (RunStatus, error)

Status returns the run's current status.

type Pipeline

type Pipeline[P, R any] struct {
	// contains filtered or unexported fields
}

Pipeline is a composed chain of stages from input P to result R. Pipelines are immutable and stateless: build them once (package level is fine) and run them from any workflow with Run or RunAll.

func Pipe1

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

Pipe1 composes a pipeline from 1 stage.

func Pipe2

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

Pipe2 composes a pipeline from 2 stages.

func Pipe3

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

Pipe3 composes a pipeline from 3 stages.

func Pipe4

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]

Pipe4 composes a pipeline from 4 stages.

func Pipe5

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]

Pipe5 composes a pipeline from 5 stages.

func Pipe6

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], s6 Stage[F, G]) Pipeline[A, G]

Pipe6 composes a pipeline from 6 stages.

func Pipe7

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], s6 Stage[F, G], s7 Stage[G, H]) Pipeline[A, H]

Pipe7 composes a pipeline from 7 stages.

func Pipe8

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], s6 Stage[F, G], s7 Stage[G, H], s8 Stage[H, I]) Pipeline[A, I]

Pipe8 composes a pipeline from 8 stages.

type PipelineWorkflow

type PipelineWorkflow[P, R any] struct {
	// contains filtered or unexported fields
}

PipelineWorkflow is a pipeline registered as a DBOS workflow. Every registration shares one generic runner method, so DBOS's configured instance mechanism keys each registration by the pipeline's name — that is what ConfigName returns, and why runs must go through Start (which selects this instance) rather than a bare dbos.RunWorkflow.

func Register

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

Register turns a pipeline into a registered DBOS workflow under the given name, and registers every queue the pipeline references. Call it after New and before Launch; run the result with Start. Workflow-level registration options (recovery attempts via dbos.WithMaxRetries, ...) pass through opts.

The name is the pipeline's durable identity: in-flight runs are recovered by looking it up, so it must be registered on every process start. Launch warns about runs whose name is no longer registered.

func RegisterScheduled

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

RegisterScheduled registers the pipeline as a scheduled (cron) workflow: every tick starts a durable run whose input is the scheduled time. The schedule uses cron syntax with seconds precision ("*/30 * * * * *" = every 30 seconds). Requiring Pipeline[time.Time, R] makes the DBOS rule that scheduled workflows take a time.Time input a compile-time guarantee.

func (*PipelineWorkflow[P, R]) ConfigName

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

ConfigName implements dbos.ConfiguredInstance: the workflow name uniquely keys this pipeline's registration.

func (*PipelineWorkflow[P, R]) Start

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

Start runs (or, with dbos.WithQueue, enqueues) the pipeline as a durable workflow and returns its handle. It accepts any dbos.WorkflowOption — workflow ID, queue, priority, deduplication, auth. For a durable deadline, pass a context derived with dbos.WithTimeout.

type Queue

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

Queue is a declared DBOS workflow queue. Declare it once (package level is fine) and reference the value everywhere it is used — the queue's name lives in exactly one place, so writer and reader can never drift:

var Jobs = duro.NewQueue("jobs", duro.WithConcurrency(4))
...
duro.FanOut("process", Jobs, duro.Workflow(ProcessJob))

Queues referenced by a pipeline are registered automatically when the pipeline is registered with Register; pipelines run directly with Run/RunAll (no Register) need RegisterQueues before workflows start enqueueing.

func NewQueue

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

NewQueue declares a queue. Declaring is side-effect free; registration happens through Register (automatic for the pipeline's queues) or RegisterQueues.

func (Queue) Name

func (q Queue) Name() string

Name returns the queue's name.

type QueueOption

type QueueOption func(*queueConfig)

QueueOption configures a declared queue.

func WithConcurrency

func WithConcurrency(n int) QueueOption

WithConcurrency caps how many workflows from the queue run concurrently across all executors.

func WithPartitions

func WithPartitions() QueueOption

WithPartitions makes the queue partitioned: children enqueued with WithChildPartitionKey get per-partition concurrency limits.

func WithPriorities

func WithPriorities() QueueOption

WithPriorities enables priority scheduling: children enqueued with WithChildPriority run lowest-number-first.

func WithRateLimit

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

WithRateLimit caps how many workflows may start within each period — backpressure for external services.

func WithWorkerConcurrency

func WithWorkerConcurrency(n int) QueueOption

WithWorkerConcurrency caps how many workflows from the queue a single executor runs concurrently.

type RegisteredWorkflow

type RegisteredWorkflow[P, R any] struct {
	// contains filtered or unexported fields
}

RegisteredWorkflow is a hand-written workflow function registered under a durable name; see RegisterWorkflow. It is a WorkflowRef, so it passes directly as a FanOut child.

func RegisterWorkflow

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

RegisterWorkflow registers a hand-written workflow function under the given name. Reach for it only when a workflow needs imperative control flow around its pipelines — branching between them, looping, post-processing a RunAll — since Register covers pipelines themselves. Call it before Launch; the name is the workflow's durable identity, so register the same function under the same name on every process start. Workflow-level registration options pass through opts.

func (*RegisteredWorkflow[P, R]) Start

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

Start runs (or, with dbos.WithQueue, enqueues) the workflow and returns its handle. It accepts any dbos.WorkflowOption, like PipelineWorkflow's Start.

type RunStatus

type RunStatus struct {
	ID    string
	Name  string // registered workflow (pipeline) name
	State State
	// Err is the recorded failure for failed runs — nil otherwise. Cancelled
	// and retries-exceeded runs that recorded no error get a synthesized one,
	// so Err is always non-nil when State.Failed().
	Err                error
	CreatedAt          time.Time
	UpdatedAt          time.Time
	CompletedAt        time.Time // zero until terminal
	ApplicationVersion string
	ForkedFrom         string // original run's ID when this run was forked
}

RunStatus is the cheap status view of a run: no input or output payloads are loaded or deserialized, making it safe for polling paths.

func Status

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

Status fetches a run's current status by workflow ID — the reconcile primitive for consumers that persist run IDs and check on them later. It works from any process attached to the same system database; no Handle needed.

func StatusAll

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

StatusAll is the batch form of Status: it returns the status of every listed run that exists, in the requested order, silently omitting unknown IDs (compare lengths to detect them).

The common case is one payload-free query. DBOS stores a run's failure message alongside its output, so when the batch contains failed runs their recorded errors are fetched in a second query scoped to just those runs — healthy polling stays cheap, failure reasons still surface.

type Stage

type Stage[T, R any] struct {
	// contains filtered or unexported fields
}

Stage is one typed pipeline segment. Stages are nominal: only this package's constructors can build them, which is what keeps arbitrary ro operators out of durable pipelines at compile time.

func Branch

func Branch[T, R any](name string, pred func(ctx context.Context, in T) (bool, error), then, els Pipeline[T, R], opts ...StepOption) Stage[T, R]

Branch is durable two-way dispatch: the predicate runs as a checkpointed step and each item flows through then or els accordingly. Both arms must produce the same output type — the compiler holds routing honest.

func Collect

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

Collect folds the stream into a slice of every item, in order — the standard final stage for a registered pipeline that should return all values rather than the last one. An empty stream yields an empty slice.

func Delay

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

Delay is a durable pause: each item passing through sleeps for d via dbos.Sleep, which checkpoints the wake-up deadline — a workflow recovered mid-sleep resumes sleeping only for the remaining time, and a replayed sleep completes instantly. Use it for pacing between durable stages; remember it applies per item on multi-item streams.

func Expand

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

Expand is a durable one-to-many transform (a flattening FlatMap): fn runs as a checkpointed DBOS step and each element of its result is emitted downstream in order.

func FanOut

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

FanOut is a durable parallel map: each item starts the referenced workflow as a child on the queue, and once the stream completes, results are awaited and emitted downstream in input order. Parallelism, rate limits, and distribution across processes are governed entirely by the queue's declaration:

var Jobs = duro.NewQueue("jobs", duro.WithConcurrency(4))
...
duro.Pipe3(
	duro.Expand("explode", split),
	duro.FanOut("process", Jobs, duro.Workflow(ProcessJob)),
	duro.Reduce("merge", merge, seed),
)

The child can be a hand-written DBOS workflow (wrap it with Workflow) or a registered pipeline (pass the *PipelineWorkflow directly). Child workflows are configured with ChildOptions — identity (WithChildID, WithChildDeduplicationID, WithChildPartitionKey), scheduling (WithChildPriority, WithChildDelay, WithChildTimeout), and metadata (WithChildAuthenticatedUser, WithChildAppVersion, WithPortableChildren).

FanOut is the sanctioned form of concurrency inside a duro pipeline: it is deterministic because children are enqueued in stream order (child workflow IDs derive from the parent's step counter, so a recovered parent re-attaches to its children instead of spawning duplicates) and awaited in that same order (each result is checkpointed in the parent). Every child is itself a durable workflow.

On the first child failure, FanOut fails the pipeline with that child's error. Children queued behind it are independent durable workflows and run to completion in the background; cancel them with dbos.CancelWorkflows if that is not what you want.

func Filter

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

Filter is a durable filter: the predicate runs as a checkpointed DBOS step, so effectful or non-deterministic predicates still replay consistently on recovery. Items for which the predicate returns false are dropped.

func FromStream

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

FromStream durably drains the stream written by another workflow: fn derives the source workflow ID from the item, the stream is read to its close (blocking while the writer is still active), and each value is emitted downstream in order — the item itself is discarded, like Recv. The whole read runs as one checkpointed step, so a recovered workflow replays the values it already collected instead of re-reading a stream that may have changed. Pair it with WithTimeout to bound how long the stage waits for the writer to finish.

func GetEvent

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

GetEvent durably reads the event published by another workflow: fn derives the source workflow ID from the item, and the event's value is emitted downstream in place of the item (reshape beforehand if you need both). The read blocks until the event is set or the timeout elapses, and is checkpointed — a recovered workflow replays the value it already observed instead of re-reading.

func Loop

func Loop[T any](name string, body Pipeline[T, T], until func(ctx context.Context, in T) (bool, error), opts ...StepOption) Stage[T, T]

Loop durably repeats the body pipeline until the until predicate — a checkpointed step — reports done, then emits the final value. Each iteration feeds the body's last emitted value back in (a body that emits nothing drops the item, like Filter). On replay the recorded predicate verdicts reproduce the exact iteration count. Pair the body with Delay for durable polling. Applied per item on multi-item streams.

Iterations are unbounded, and a durable loop is more durable than a bug deserves: a predicate that can never report done keeps checkpointing and resumes across restarts. Give the loop a natural bound (track attempts in T and fail past a limit), or stop a runaway run with dbos.CancelWorkflow.

func Parallel

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

Parallel is a durable parallel Step: items execute fn concurrently as DBOS steps within the workflow process, at most maxConcurrent at a time (unbounded if maxConcurrent <= 0), and results are emitted downstream in input order once the stream completes.

Parallel is the in-process, lightweight sibling of FanOut: no queue and no child workflows, just concurrent steps inside the current workflow. Use FanOut when work should distribute across processes, survive independently, or obey queue-level rate limits; use Parallel when a bounded burst of concurrent steps in this process is enough.

Determinism is preserved because each step's ID is assigned on the workflow goroutine at launch time, in stream order (dbos.Go exists precisely for this), and outcomes are collected in that same order. On recovery, completed steps replay from their checkpoints without re-running fn.

If any step fails, results for items before the failure are still emitted downstream (matching sequential fail-fast semantics), all remaining steps are drained, and the pipeline fails with the first error in input order.

func Pure

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

Pure is a non-durable transform: fn is NOT checkpointed and re-executes on every replay, so it must be deterministic and side-effect free. Use it for cheap reshaping between durable stages; anything effectful or fallible belongs in Step.

func Recv

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

Recv durably waits for the next message on the topic and emits it downstream, consuming one message per upstream item (the item itself is discarded — reshape beforehand if you need it). Receipt is checkpointed, so a recovered workflow does not consume a second message. A zero or negative timeout means dbos.Recv's no-wait behavior; if no message arrives in time, the stage fails the pipeline.

Recv is how a pipeline pauses for an external signal — a payment confirmation, a human approval — sent to this workflow's ID with a Send stage or Topic.Send from anywhere.

func Reduce

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

Reduce is a durable fold: each accumulation runs as a checkpointed DBOS step, and the final accumulator is emitted when the source completes.

func Send

func Send[T, M any](name string, topic Topic[M], fn func(in T) (destinationID string, message M, err error)) Stage[T, T]

Send durably sends one message per item on the topic (dbos.Send is checkpointed, so a recovered workflow does not re-send). fn derives the destination workflow ID and the message from the item, which passes through unchanged. The message type is the topic's — a mismatch with the receiving side is a compile error.

func SetEvent

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

SetEvent durably publishes the event on the workflow for each item and passes the item through unchanged. Read it with a GetEvent stage or Event.Get — the classic use is exposing pipeline progress to the outside world while the workflow runs.

func Step

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

Step is a durable Map: it transforms each item by running fn as a checkpointed DBOS step. On recovery, completed executions are replayed from the database instead of re-running fn.

func Sub

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

Sub embeds a pipeline as a single named stage — reuse a pipeline segment across pipelines without a wrapper workflow. Unlike Branch/Switch/Loop, Sub applies to the whole stream, not per item: an embedded Reduce folds everything flowing through it.

func Switch

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

Switch is durable multi-way dispatch: route runs as a checkpointed step, and each item flows through the case pipeline matching the returned key — on replay the recorded key routes the item down the same arm. A key with no matching case fails the pipeline. Applied per item on multi-item streams; every arm's outputs are emitted downstream in stream order.

func Tap

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

Tap is a durable side effect: fn runs as a checkpointed DBOS step and the item passes through unchanged.

func ToStream

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

ToStream durably appends each item to the workflow's stream and passes it through unchanged; the stream is closed when the pipeline completes. Readers drain it with a FromStream stage or Stream.Read — the way to expose a pipeline's per-item output while it is still running, instead of waiting for the final result. The stream's type is the pipeline's item type, checked at compile time.

func UnsafeOperator

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

UnsafeOperator admits an arbitrary ro operator into a durable pipeline. duro cannot guarantee replay determinism for it: the operator must be synchronous, order-preserving, and deterministic, or recovery will fail — loudly if step names misalign or execution changes goroutines, silently if identically-named steps swap positions. Prefer the safe constructors.

type State

type State string

State is a run's lifecycle state.

const (
	StatePending         State = "pending"  // running or ready to run
	StateEnqueued        State = "enqueued" // waiting on a queue
	StateDelayed         State = "delayed"  // waiting for its start delay
	StateSuccess         State = "success"
	StateError           State = "error" // completed with an error
	StateCancelled       State = "cancelled"
	StateRetriesExceeded State = "retries_exceeded" // exceeded max recovery attempts
)

func (State) Failed

func (s State) Failed() bool

Failed reports whether the run finished without succeeding.

func (State) Terminal

func (s State) Terminal() bool

Terminal reports whether the run has reached a final state.

type StepOption

type StepOption func(*stepConfig)

StepOption configures how a durable stage executes as a DBOS step.

func WithBackoffFactor

func WithBackoffFactor(factor float64) StepOption

WithBackoffFactor sets the exponential multiplier applied to the retry delay after each attempt (default 2.0).

func WithBaseInterval

func WithBaseInterval(d time.Duration) StepOption

WithBaseInterval sets the initial delay between retries (default 100ms).

func WithMaxInterval

func WithMaxInterval(d time.Duration) StepOption

WithMaxInterval caps the delay between retries (default 5s).

func WithMaxRetries

func WithMaxRetries(n int) StepOption

WithMaxRetries sets the maximum number of automatic retries for the stage when its function returns an error. Zero (the default) means no retries. This is the step-level retry limit (DBOS's WithStepMaxRetries), distinct from workflow recovery attempts, which are configured at registration.

func WithRetryPredicate

func WithRetryPredicate(pred func(error) bool) StepOption

WithRetryPredicate restricts which errors are retried: when the stage function returns an error for which pred is false, the stage stops immediately with that error even if retries remain. Use it to spend retries on transient failures only.

func WithTimeout

func WithTimeout(d time.Duration) StepOption

WithTimeout bounds each execution attempt of the stage function: the step context is cancelled after d and the attempt fails with the context's error. Retries get a fresh deadline. DBOS has no native step timeout, so the deadline is enforced in-process per attempt — the stage function must honor context cancellation for the timeout to take effect. For a durable deadline on a whole pipeline, start its workflow from a context derived with dbos.WithTimeout; for child workflows, see WithChildTimeout.

type Stream

type Stream[V any] struct {
	// contains filtered or unexported fields
}

Stream is a typed durable stream channel: values of type V appended by one workflow and drained by readers. Write with a ToStream stage; read with a FromStream stage or Stream.Read.

var Receipts = duro.NewStream[Receipt]("receipts")

func NewStream

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

NewStream declares a typed stream key.

func (Stream[V]) Key

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

Key returns the stream's key.

func (Stream[V]) Read

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

Read drains the stream written by the given workflow, blocking until the writer closes it or becomes inactive; closed reports whether the stream was cleanly closed. Read is for clients — inside a pipeline use the FromStream stage, which checkpoints the read so replay never observes a changed stream.

type Topic

type Topic[M any] struct {
	// contains filtered or unexported fields
}

Topic is a typed mailbox channel: messages of type M sent to a workflow's mailbox under one topic name. Write with a Send stage or Topic.Send; read with a Recv stage.

var Approvals = duro.NewTopic[Approval]("approvals")

func NewTopic

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

NewTopic declares a typed mailbox topic.

func (Topic[M]) Name

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

Name returns the topic's name.

func (Topic[M]) Send

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

Send delivers a message to the destination workflow's mailbox on this topic. It works from anywhere: inside a workflow it is checkpointed (no re-send on replay); from a plain client it is a direct send. Pipelines should prefer the Send stage.

type WorkflowFunc

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

WorkflowFunc is a hand-written durable workflow function, the kind RegisterWorkflow registers and Workflow adapts into a FanOut child. Registered pipelines (Register) never touch this type.

type WorkflowRef

type WorkflowRef[T, R any] interface {
	// contains filtered or unexported methods
}

WorkflowRef identifies a workflow FanOut can start: a registered pipeline (*PipelineWorkflow, which carries its own dispatch metadata) or a hand-written DBOS workflow wrapped with Workflow.

func Workflow

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

Workflow adapts a hand-written, dbos-registered workflow function into a WorkflowRef:

duro.FanOut("process", Jobs, duro.Workflow(ProcessJob))