duro

package module
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 18 Imported by: 0

README

duro

Durable dataflow pipelines for Go.

CI Go Reference Go Report Card

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, Rescue does the same for error handling (best-effort segments, report-then-rethrow), and Status/Attach reconcile any persisted run ID from any process — with ListRuns/Steps/Cancel/Resume giving an admin tier the whole fleet, from an enqueue-only Client too.

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 (duro.Unbounded to lift the cap) ✅ 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; bound it with WithMaxIterations ✅ every iteration's verdict is checkpointed
duro.Rescue(name, pipeline, handler) except block: intercept an embedded pipeline's failure — swallow with a fallback or rethrow ✅ the handler is a checkpointed step
duro.Sub(name, pipeline) embed a pipeline as one named stage (whole-stream) ✅ its stages checkpoint as usual
duro.Via(name, pipeline) run an embedded pipeline for its effects — typically a fan-out — and pass the original item through ✅ 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 pipeline's single emitted value. A pipeline emitting nothing fails with ErrNoValue, and one emitting more than once fails with ErrMultipleValues rather than picking a value on your behalf: an unfolded Expand or FanOut would otherwise discard every result but one, silently.
  • duro.RunAll(ctx, input, pipeline) — returns every emitted value

Stage names must be unique within a pipeline — a name is how a stage is identified in the recorded step list, so duplicates would make ForkFromStage ambiguous. Arms of the same Branch or Switch may reuse names freely, since only one arm ever runs. Names beginning duro. are reserved for duro's own bookkeeping steps.

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"),
)

Error handling is a stage too. Pipelines are fail-fast by default; Rescue scopes that policy to a segment, covering the except-block shapes that used to force a hand-written workflow — best-effort segments, report-then-rethrow, and retry-then-swallow (stage retry options compose: the handler fires only after the embedded stages' retries are exhausted):

duro.Rescue("cover-art", coverArtPipeline, // best-effort: a failure must not kill the song
	func(_ context.Context, song Song, cause error) (Song, error) {
		slog.Warn("cover art failed", "song", song.ID, "cause", cause)
		return song, nil // swallow: pass the item through without art
	})

The handler runs as a checkpointed step, so effectful handlers replay consistently and a recovered run never flips a swallowed failure into a propagated one. On success the embedded emissions pass through unchanged; on failure partial emissions are discarded and only the handler's fallback is emitted. A top-level Pipe1(Rescue("run", pipeline, reportAndRethrow)) is the whole-pipeline except block.

And state can survive a fan-out. Via runs an embedded pipeline — typically a FanOut of child workflows — for its effects, then emits the item it was given, discarding the embedded emissions (Tap is to Step what Via is to Sub). The shape that used to force Run, RunAll-and-discard, Run in an imperative body is one registered pipeline:

duro.Pipe3(
	duro.Sub("research", researchPipeline),      // state flows in…
	duro.Via("process-units", unitsPipeline),    // fan-out runs; state flows on
	duro.Sub("finalize", finalizePipeline),      // …and continues afterwards
)

Embedded pipelines fold into the shape fingerprint, so editing an arm or body still fails fast on replay. Branch/Switch/Loop/Rescue/Via 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).

On the first child failure the stage fails with that child's error. What happens to the other children is a per-stage choice. The default is drain: siblings run to completion in the background — right when every completion has value on its own (cleanup and deletion fan-outs, where more work finishing on the failure path is strictly better). When surviving siblings are wasted spend once the batch's outcome is decided — cost-bearing generation, paid API calls — opt into cancellation:

duro.FanOut("units", Jobs, duro.Workflow(GenerateUnit), duro.WithCancelSiblings())

The first failed child promptly cancels every sibling not yet in a terminal state, running and never-dequeued alike; detection watches all children at once, so an early failure in a long batch is not discovered only after the children ahead of it finish. The stage still fails with the triggering child's error — cancelled siblings never mask it, live or on recovery replay, so a Rescue around the stage always sees the original failure — and a parent that crashes mid-cancellation re-issues it idempotently when it recovers. Cancellation also survives the parent's executor: each batch gets duro's cancellation watcher, an internal durable workflow (auto-registered by duro.New) that watches the same children from its own queue and cancels redundantly — any executor can dequeue or recover it, so a failure is acted on even if the process awaiting the batch dies. The watcher is a backstop polling at 5s by default (WithCancelWatchInterval tunes it); while the awaiting process lives, the stage itself detects failures within 250ms. Cancel mode awaits results as one assembled step named after the stage (instead of one DBOS.getResult step per child), so the option is part of the shape fingerprint: toggling it while runs are in flight trips the shape guard rather than misreading checkpoints. Cancellation does not cascade to workflows a cancelled child had itself started — see the WithCancelSiblings godoc for the nested-fan-out pattern.

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. max must be a positive bound or the explicit duro.Unbounded; a plain 0 is rejected at construction, so an unset config field can never silently mean "launch a step per item with no ceiling". It drains on failure by default too; duro.WithCancelSiblingSteps() is its cancellation opt-in — in-flight siblings get their step contexts cancelled (the function must honor cancellation, the same contract as WithTimeout) and unstarted items skip the function while still occupying their deterministic step slots, keeping recovery replay aligned.

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.

ListRuns is the fleet-wide view — every run of every registered pipeline, from any process attached to the system database — filtered, paged, and ordered by options; Steps inspects one run's checkpoints:

runs, err := duro.ListRuns(app,
	duro.WithNames("invoice", "refund"),
	duro.WithStates(duro.StateError, duro.StateRetriesExceeded),
	duro.WithCreatedAfter(time.Now().Add(-24*time.Hour)),
	duro.WithNewestFirst(), duro.WithLimit(50), duro.WithOffset(page*50))

steps, err := duro.Steps(app, runs[0].ID) // []StepStatus: ID, Name, Err, ChildID, StartedAt, CompletedAt

RunStatus carries the cheap per-run columns an admin view needs — QueueName, ParentID (set on FanOut children, so top-level runs are the ones without one), Attempts, StartedAt, ExecutorID — plus Input, the stored input as JSON text, loaded only when WithInput() asks for it. Everything else stays payload-free, and a failed run's Err is filled exactly as Status fills it. The options refuse to surprise: a membership filter given no values (WithIDs()) matches nothing rather than everything, and a non-positive WithLimit is an error rather than an empty page. Steps returns what the run has checkpointed — the shape checkpoint (duro.ShapeStepName) first, then one entry per durable stage execution — which is exactly what replays on recovery. duro's own durable plumbing is listed too: every cancel-enabled FanOut batch runs a watcher workflow named duro.CancelWatcherName on the duro.CancelWatchQueueName queue, so an admin view can show — or filter out — those rows by name, and a watcher stuck non-terminal is visible rather than hidden.

Cancelling and resuming

Cancel stops a live run at its next stage boundary (the stage in flight completes and is checkpointed); Resume revives a run under the same ID, replaying its checkpoints and continuing from the first stage that never finished:

err := duro.Cancel(app, runID)          // errors.Is(err, duro.ErrRunTerminal) when already finished
err := app.Resume(ctx, runID)           // bounded by ctx's deadline (or DefaultReadTimeout)
err := duro.ForkFromStage[R](app, ...)  // the re-run for everything Resume refuses

Resume takes exactly the runs no executor can still be executing: one that exceeded its recovery attempts, or one cancelled before it ever started. A run cancelled after it started is refused with ErrRunInFlight — cancellation lands at the next stage boundary, so the stage in flight keeps running on its executor, and nothing DBOS records says when it stops; resuming under the same ID could execute it twice. Re-run such a run with ForkFromStage: a new ID, the old run stays cancelled. Success and error runs are finished and return ErrRunTerminal (fork those too); pending, enqueued, and delayed runs return ErrRunActive — cancel first. Nothing silently no-ops. The state check is part of the transition itself: Resume is one guarded UPDATE of duro's own (which is why it is a method on App rather than a function of a context), so two concurrent resumes, or a resume racing a worker that already picked the run up, cannot both apply. A resumed run that still records its queue goes back on it; DBOS clears the queue when it cancels or dead-letters a run, so in practice a resumed run executes on DBOS's internal queue, outside its original queue's limits.

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
})

Worker-pool mode

DBOS recovers a crashed process's in-flight runs only when that same process restarts under the same executor ID. On ephemeral autoscaled infrastructure (Cloud Run, ECS, spot instances) a dead worker never returns under its old identity, so its PENDING runs strand. Worker-pool mode closes that gap without DBOS Conductor: every process heartbeats a lease, and a sweeper on each process re-enqueues the runs of any process whose lease has gone stale, where a live worker picks them up.

app, err := duro.New(ctx, duro.Config{
	Name:               "orders",
	DatabaseURL:        url,
	ApplicationVersion: gitSHA, // pin it — see below
}, duro.WithWorkerPool())

Enable it on every worker in the fleet. A killed worker's run resumes on a survivor within the stale threshold plus one sweep; a graceful Shutdown tombstones the lease so undrained runs are taken over on the next sweep with no wait. A run taken over goes back on the queue it came from, so it stays subject to that queue's concurrency and rate limits; only a run started directly (which has no queue) is re-enqueued on DBOS's internal queue. Takeover guarantees exactly-once workflow completion but, like all DBOS recovery, at-least-once step side effects — keep steps idempotent. Tune the cadence (defaults: 10s heartbeat / 60s stale / 30s sweep) with WithHeartbeatInterval, WithStaleThreshold, WithSweepInterval; New rejects a stale threshold under 2× the heartbeat interval, since a threshold that tight declares merely-slow workers dead and runs their live work twice. Leave real headroom for GC pauses — the defaults are 6×.

Shutdown's timeout bounds the drain; stopping maintenance and writing the tombstone are bounded separately and take milliseconds on a healthy database. Do not set the timeout to your whole SIGTERM grace period — the tombstone is written last, so it is what a SIGKILL takes away, and losing it costs the fast adoption of exactly the runs that could not drain.

Identity: application version and executor ID

Config.ApplicationVersion and Config.ExecutorID are the two knobs that scope recovery — previously reachable only through the DBOS__APPVERSION / DBOS__VMID environment variables, which still override them. Pin ApplicationVersion to a value you control (a git SHA, a release tag): its default is a hash of the binary, so every in-flight run strands the moment you deploy new code, recoverable only by rolling back to the exact previous binary. A run is only ever taken over by an executor on the same version — a new deploy never adopts (and replays on incompatible code) the previous version's runs. In worker-pool mode a process without an explicit ExecutorID is assigned a unique one automatically.

Enqueue-only client

A web tier that starts workflows but must never execute them uses a Client — no engine, no queue runners, but the same typed surface and the same status mapping as the workers:

// declared once, in a package both binaries import
var InvoiceJob = duro.NewJob[Batch, Invoice]("invoice")

// on the workers
duro.RegisterJob(app, InvoiceJob, invoicePipeline)

// in the web tier
c, err := duro.NewClient(ctx, duro.ClientConfig{DatabaseURL: url})
defer c.Shutdown(5 * time.Second)

handle, err := duro.Enqueue(c, Jobs, InvoiceJob, batch)
status, err := c.Status(handle.ID()) // duro.RunStatus, the same States as duro.Status
runs, err := c.ListRuns(duro.WithStates(duro.StateRetriesExceeded)) // the admin view, same options as duro.ListRuns

A Client carries the whole read and remediation surface — Status, StatusAll, ListRuns, Steps, Cancel, Resume — as thin calls into the same core the engine's functions use, so an admin site built on a Client sees exactly what the workers see and can never disagree with them about a run's state.

Those reads are bounded. DBOS retries a failed database read indefinitely until its context ends, which in a request handler means an outage parks a goroutine per call until the database returns; a Client instead runs every read on its own bounded context — ClientConfig.ReadTimeout, 30s by default — and WithContext adds the request's own deadline on top:

c, err := duro.NewClient(ctx, duro.ClientConfig{DatabaseURL: url, ReadTimeout: 10 * time.Second})

func list(w http.ResponseWriter, r *http.Request) {
	runs, err := c.WithContext(r.Context()).ListRuns(duro.WithNewestFirst(), duro.WithLimit(50))
	// ...
}

The bound covers a whole call, however many queries it issues, and the error says what ended it: a deadline — ReadTimeout or the request's — wraps context.DeadlineExceeded, a cancelled request wraps context.Canceled. Enqueue and Handle waits are not covered — dbos.Client offers no context for them. On the engine, derive a bounded context with dbos.WithTimeout(app.Context(), d) and pass it to duro.ListRuns and friends.

A Job is the pipeline's cross-process identity: its registered name and both of its types in one declaration, shared by the process that registers it and every process that enqueues it. That is what makes the enqueue checkable — enqueuing by bare string, a typo produces a run no worker can dispatch, which is inserted successfully, reported as a healthy enqueued, and waits forever, while a mismatched input or result type fails in another process long after the call site. With a Job all three are compile errors.

Enqueue stamps no application version by default, so any worker version runs the job (the web tier need not redeploy in lockstep with the workers); pin one with WithClientApplicationVersion.

Client is also how a worker hands a whole pipeline to the fleet rather than running it in-process: wf.Start always runs the pipeline on the calling process, and queues are for fan-out children (FanOut), so a worker that wants to enqueue a top-level run builds a Client alongside its engine.

Stale-run warnings and retention

Two more operational gaps DBOS open source leaves, both opt-in beside the worker pool and usable with or without it:

duro.New(ctx, cfg,
	duro.WithWorkerPool(),
	duro.WithStaleRunWarning(15*time.Minute), // surface non-terminal runs older than this
	duro.WithRetention(30*24*time.Hour),      // batch-delete terminal runs older than this
)

WithStaleRunWarning reports stranded runs — which DBOS otherwise surfaces nowhere — split into same-version (in limbo here) and other-version (recoverable only on their own version); pass a callback to also receive the counts. Set the age above the longest a healthy run legitimately takes: a run parked in a Delay stage stays pending for the whole pause. WithRetention bounds the otherwise-unbounded growth of terminal run history, deleting one batch per cycle under its own advisory lock — never the sweeper's, so housekeeping can never delay a takeover.

Both run on the sweep cadence, and both are database-wide: DBOS records no application name on a run, so they count and delete every matching run in the system database, including other applications' runs if they share it. Give each application its own database (or DBOS schema) before enabling retention.

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, a registered pipeline used directly as a FanOut child, and a strict batch whose first failure cancels the surviving fleet (WithCancelSiblings) under a Rescue that sees the original error.

  • examples/triage — durable control flow: Switch dispatch, a nested Branch, a polling Loop, shared segments with Sub, best-effort and report-then-rethrow error handling with Rescue, an archive fan-out threaded past with Via, 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/fleet — worker-pool mode: a two-process app — a duro.New+WithWorkerPool worker fleet and a duro.NewClient web tier that enqueues without an engine, and a -role=admin view on the same Client that lists the fleet's runs, dumps a run's checkpoints, and cancels or resumes one — plus a -crash flag that kills a worker mid-run so you can watch a survivor's sweeper take the run over and finish it, replaying the checkpointed steps and re-running only the in-flight one.

  • 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. On failure both drain surviving siblings by default; WithCancelSiblings (FanOut) and WithCancelSiblingSteps (Parallel) opt a stage into cancelling them instead, while still failing with the original error.

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, Rescue intercepts an embedded pipeline's failure with a checkpointed handler that swallows or rethrows it, Sub embeds a pipeline as one named stage, Via runs one for its effects and passes the original item through, 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. A pipeline another process enqueues is declared as a Job — its name and both types in one value — and registered with RegisterJob, so the name and types cannot drift between the binary that runs it and the binary that enqueues it. 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

Examples

Constants

View Source 
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
)
View Source 
const (
	// CancelWatcherName is the registered workflow name of duro's cancellation
	// watcher. Watcher runs are ordinary rows in the system database — one per
	// cancel-enabled FanOut batch — so they appear in ListRuns alongside
	// application runs; filter them in or out with WithNames(CancelWatcherName).
	// A watcher stuck non-terminal is worth an operator's attention. The name
	// is a durable identity: it never changes.
	CancelWatcherName = "duro.cancel-watcher"
	// CancelWatchQueueName is the duro-owned queue watcher runs execute on
	// (WithQueue(CancelWatchQueueName) lists them by queue). Also a durable
	// identity.
	CancelWatchQueueName = "duro.cancel-watch"
)

The cancellation watcher: a duro-internal durable workflow enqueued alongside every cancel-enabled fan-out batch. It watches the same children the parent's await step watches and cancels on failure, redundantly and idempotently — but because it is a queued workflow, any executor can dequeue or recover it. That is what keeps cancellation from depending on the parent's executor staying alive: if the parent dies right after a child fails, the watcher still cancels the survivors. duro.New registers the watcher workflow and its queue on every app; both names are durable identities and must never change.

View Source 
const DefaultReadTimeout = 30 * time.Second

DefaultReadTimeout bounds a Client's reads and remediations when ClientConfig.ReadTimeout is zero.

View Source 
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.

View Source 
const Unbounded = -1

Unbounded lifts Parallel's concurrency cap: every item's step is launched as soon as it arrives. Prefer a real bound — an unbounded Parallel over a large stream launches one in-process step per item with nothing holding it back. It exists so that removing the cap is something a caller states outright rather than something a zero-valued variable does by accident.

Variables

View Source 
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.

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

ErrMaxIterations is returned by a Loop stage configured with WithMaxIterations whose predicate did not report done within that many iterations. Without the option a Loop is unbounded; see WithMaxIterations.

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

ErrMultipleValues is returned by Run when the pipeline emits more than one value. Run yields a single result, so more than one emission is ambiguous rather than something Run could silently pick from: end the pipeline with Reduce or Collect to fold the stream, or call RunAll to receive every value.

View Source 
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).

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

ErrRunActive is returned by Resume when the run is still live — pending, enqueued, or delayed. Resuming a live run would re-enqueue it while an executor may still be running it, a double execution; Cancel it first.

View Source 
var ErrRunInFlight = errors.New("duro: run was cancelled mid-execution and its in-flight stage may still be running")

ErrRunInFlight is returned by Resume for a run that was cancelled after it had started executing. Cancellation lands at the next stage boundary, so the stage in flight keeps running on its executor until it finishes — and nothing DBOS records says when that is. Resuming under the same ID would re-enqueue the run while that stage may still be running, and would let the old executor carry on past it: a double execution. Re-run such a run under a new ID with ForkFromStage, which leaves this one cancelled.

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

ErrRunNotFound is returned by Status, Attach, Steps, Cancel, and Resume for an unknown workflow ID.

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

ErrRunTerminal is returned by Cancel and Resume when the run has already reached a final state the operation cannot change: any terminal state for Cancel; success or error for Resume, which only revives cancelled and retries-exceeded runs. Re-run a finished pipeline with ForkFromStage.

Functions

func Cancel added in v0.9.0 

func Cancel(ctx Context, workflowID string) error

Cancel stops a live run: an enqueued or delayed run never starts, and a pending one stops at its next stage boundary (the stage in flight finishes — keep stages idempotent). The run's queue is cleared and its state becomes StateCancelled. Runs already in a final state return ErrRunTerminal, so a cancel that changed nothing is never mistaken for one that did; unknown IDs return ErrRunNotFound.

Client.Cancel is the same operation from an enqueue-only process.

func RegisterQueues added in v0.2.0 

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 pipeline's single emitted value or the first stage error. Call it as the body of a registered DBOS workflow function.

Run is for pipelines that produce exactly one result. A pipeline that emits nothing fails with ErrNoValue, and one that emits more than once fails with ErrMultipleValues rather than having Run pick a value on the caller's behalf — a stage like Expand or FanOut left unfolded would otherwise discard every result but one, silently and with nothing to notice. Fold the stream with Reduce or Collect, or use RunAll to receive every value.

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.

Types

type App added in v0.2.0 

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 added in v0.2.0 

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

New initializes the application. Register pipelines and queues after New and before Launch. Options enable optional subsystems — see WithWorkerPool, WithRetention, and WithStaleRunWarning.

func (*App) Context added in v0.2.0 

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 added in v0.2.0 

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.

In worker-pool mode Launch lands the first heartbeat synchronously before starting queue runners (an executor must never be dequeuing while observably dead), then starts the heartbeat and sweeper goroutines.

func (*App) Resume added in v0.9.0 

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

Resume revives a run under its original ID: it is re-enqueued and picked up by an executor on its application version, where completed stages replay from their checkpoints and execution continues from the first stage that never finished. The recovery-attempt count starts over. A run that still records its queue returns to it; because DBOS clears a run's queue when it cancels or dead-letters it, in practice the resumed run executes on DBOS's internal queue, outside any concurrency or rate limit of the queue it was originally enqueued on.

Exactly two kinds of run resume, because they are the ones no executor can still be executing: a run that exceeded its recovery attempts, and a run cancelled before it ever started (while enqueued or delayed). A run cancelled after it started returns ErrRunInFlight — its in-flight stage may still be running and DBOS records nothing that says when it stops — and is re-run under a new ID with ForkFromStage instead. Success and error runs are finished and return ErrRunTerminal rather than a silent no-op; fork those too. Pending, enqueued, and delayed runs return ErrRunActive: Cancel first. Unknown IDs return ErrRunNotFound. The state check is part of the transition itself — one guarded UPDATE — so concurrent Resumes cannot both apply.

The call is bounded by ctx's deadline, or by DefaultReadTimeout when it has none, and ends early if ctx is cancelled. It is a method on App rather than a function of a Context because the transition is duro's own SQL statement, run on the App's database connection. Client.Resume is the same operation from an enqueue-only process.

func (*App) Shutdown added in v0.2.0 

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

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

In worker-pool mode the ordering matters: the sweeper stops first, the heartbeat keeps beating through DBOS's drain (so runs still executing here stay fresh and un-takeable however long the drain runs), and only once DBOS has stopped is the lease tombstoned — so any runs that could not drain are adopted by a survivor on its next sweep with no stale wait. Follow Shutdown promptly with process exit.

timeout bounds the drain. The two steps around it are bounded separately and take milliseconds against a healthy database: stopping maintenance cancels whatever query it has in flight rather than waiting it out, and the tombstone is a single primary-key UPDATE bounded by a few seconds of its own. Budget timeout plus a small constant for the whole call — never let timeout consume the entire SIGTERM grace period, or the tombstone is the part that is lost.

Calling Shutdown more than once is safe; the extra calls do nothing.

type Case added in v0.4.0 

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 added in v0.4.0 

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

When declares a Switch case.

type ChannelOption added in v0.2.0 

type ChannelOption func(*channelConfig)

ChannelOption configures a declared channel.

func Portable added in v0.2.0 

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 added in v0.2.0 

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 WithCancelSiblings added in v0.7.0 

func WithCancelSiblings() ChildOption

WithCancelSiblings makes the first failed child cancel every sibling that is not yet in a terminal state — running (PENDING) and never-dequeued (ENQUEUED) children alike — instead of letting them run to completion in the background. Failure detection is order-independent: the stage watches all children while awaiting, so an early failure in a long batch cancels promptly rather than after every child ahead of it finishes. The stage still fails with the triggering child's error; cancelled siblings surface as CANCELLED but never mask it, on the live run and on recovery replay alike, so a Rescue around the stage always sees the original failure.

Choose it when sibling work is worthless once the batch's outcome is decided (cost-bearing generation, paid API calls); leave the default when every completion has value on its own (cleanup or deletion fan-outs, where more finished children on the failure path is strictly better).

Semantics that differ from the default, beyond cancellation itself:

  • The stage's checkpoint layout changes: results are awaited and recorded as one step named after the stage, instead of one DBOS.getResult step per child. The option therefore participates in the pipeline's shape fingerprint — toggling it while runs are in flight trips the shape guard on recovery instead of misreading checkpoints. Treat toggling like any other pipeline edit.
  • On failure the fan-out fails as a unit: nothing is emitted downstream. Without the option, results before the first failure in input order are emitted before the stage fails.
  • Detection begins once the stage starts awaiting (all children enqueued). Enqueueing itself never blocks on child completion.

Cancellation does not cascade: DBOS cancels exactly the sibling children, and a cancelled child stops at the start of its next step. Workflows the cancelled child had itself started — grandchildren of this stage, including a nested FanOut's children — keep running to completion. To bound that cost, give the child's own fan-outs WithCancelSiblings (covers grandchild failures) and WithChildTimeout (bounds grandchild lifetime); cancelling a whole tree from outside remains dbos.CancelWorkflows over the grandchild IDs.

Cancellation is idempotent and re-issued on recovery: a parent that crashes mid-cancellation re-observes the failure when the stage resumes and cancels whatever is still live.

Nor does cancellation depend on this process surviving the await: alongside the batch, the stage enqueues duro's cancellation watcher — an internal durable workflow (registered by New as CancelWatcherName on the internal CancelWatchQueueName queue; both names are durable identities) that watches the same children and cancels redundantly. Any executor can dequeue or recover the watcher, so a failure is acted on even when the parent's executor dies mid-await. The stage requires the watcher to be registered — apps built with New always have it; a hand-rolled DBOS context without it fails the stage immediately with a clear error rather than silently weakening the guarantee.

func WithCancelWatchInterval added in v0.7.0 

func WithCancelWatchInterval(d time.Duration) ChildOption

WithCancelWatchInterval tunes how often the cancellation watcher polls child statuses (default 5s). The watcher is the backstop for when the process awaiting the fan-out dies — while that process lives, the stage's own await detects failures within 250ms regardless of this setting, and after cancellation is issued the watcher re-checks immediately rather than waiting out an interval. Lower it when orphaned spend must be cut faster after an executor loss; raise it to shave database load on very long-running batches. Recorded in the watcher's durable input, so a deploy that changes it affects new batches only.

It requires WithCancelSiblings and a positive duration — FanOut panics at construction time otherwise.

func WithChildAppVersion added in v0.2.0 

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 added in v0.2.0 

func WithChildAssumedRole(role string) ChildOption

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

func WithChildAuthenticatedRoles added in v0.2.0 

func WithChildAuthenticatedRoles(roles ...string) ChildOption

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

func WithChildAuthenticatedUser added in v0.2.0 

func WithChildAuthenticatedUser(user string) ChildOption

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

func WithChildDeduplicationID added in v0.2.0 

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 added in v0.2.0 

func WithChildDeduplicationPolicy(policy DeduplicationPolicy) ChildOption

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

func WithChildDelay added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 Client added in v0.8.0 

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

Client enqueues durable runs and reports their status without launching the engine — the abstraction for an enqueue-only process (a web tier) that starts workflows but must never execute them. It talks to the same system database as the workers, uses the same serialization, and reports status through the same mapping as the engine (Status/StatusAll), so a client and the workers can never disagree on what a run's state — or "terminal" — means. 

c, err := duro.NewClient(ctx, duro.ClientConfig{DatabaseURL: url})
defer c.Shutdown(5 * time.Second)
h, err := duro.Enqueue(c, Jobs, InvoiceJob, batch)
status, err := c.Status(h.ID())

func NewClient added in v0.8.0 

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

NewClient connects an enqueue-only client to the system database.

func (*Client) Cancel added in v0.9.0 

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

Cancel stops a live run, as the engine's Cancel does.

func (*Client) ListRuns added in v0.9.0 

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

ListRuns lists durable runs with the same options, mapping, and failed-run treatment as the engine's ListRuns — an admin tier's view of every run in the system, from a process that registers no workflows.

func (*Client) Resume added in v0.9.0 

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

Resume revives a cancelled or retries-exceeded run, as the engine's Resume does — the same guarded transition, on the client's own connection. The run executes on a worker; the client never runs it.

func (*Client) Shutdown added in v0.8.0 

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

Shutdown aborts any read still in flight and closes the client's system-database connections. WithContext views share those connections, so Shutdown on any of them closes the client for all. Calling it more than once is safe.

func (*Client) Status added in v0.8.0 

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

Status fetches a run's current status by workflow ID, using the same mapping as the engine's Status.

func (*Client) StatusAll added in v0.8.0 

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

StatusAll is the batch form of Status, sharing the engine's mapping core.

func (*Client) Steps added in v0.9.0 

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

Steps lists a run's executed steps, as the engine's Steps does.

func (*Client) WithContext added in v0.9.0 

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

WithContext returns a view of the client whose reads and remediations are bounded by ctx as well as by ReadTimeout — the seam for a request handler: 

runs, err := c.WithContext(r.Context()).ListRuns(duro.WithLimit(50))

A call cut short by ctx's deadline fails with an error wrapping context.DeadlineExceeded, one cut short by its cancellation with context.Canceled, so HTTP timeout classification sees the right cause.

The view shares the client's connections and its enqueue path; Enqueue and Handle waits are not bounded by ctx (dbos.Client offers no context for them). ctx must be non-nil.

type ClientConfig added in v0.8.0 

type ClientConfig struct {
	// DatabaseURL is the Postgres URL of the DBOS system database — the same one
	// the workers use.
	DatabaseURL string
	// Logger receives client and DBOS logs; slog.Default() when nil.
	Logger *slog.Logger
	// ReadTimeout bounds every read and remediation call — Status, StatusAll,
	// ListRuns, Steps, Cancel, Resume — at DefaultReadTimeout when zero. The
	// bound covers the whole call, however many queries it issues. DBOS
	// retries a failed database read indefinitely (backing off to 30s between
	// attempts) until its context ends, so without a bound an outage parks a
	// goroutine per call until the database returns; with one, each call
	// returns an error wrapping context.DeadlineExceeded at the deadline. Set
	// it below your request deadline and page ListRuns so no single call
	// needs longer. Negative is an error.
	//
	// It does not cover Enqueue, or waiting on a Handle: dbos.Client offers
	// no context for those.
	ReadTimeout time.Duration
}

ClientConfig configures a duro Client.

type Config added in v0.2.0 

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

	// ApplicationVersion pins the DBOS application version — the single most
	// operationally consequential DBOS knob. Recovery is scoped to it: a run is
	// only ever recovered (or, under worker-pool mode, taken over) by an
	// executor on the same version. Leaving it empty keeps DBOS's default (a
	// hash of the binary), which changes on every rebuild — so every in-flight
	// run strands the moment you deploy new code, recoverable only by rolling
	// back to the exact previous binary. Pin it to a value you control (a git
	// SHA, a release tag) so a redeploy of the same logical version resumes
	// in-flight work. The DBOS__APPVERSION environment variable, when set,
	// overrides this field.
	ApplicationVersion string
	// ExecutorID sets this process's executor identity. Empty keeps DBOS's
	// default ("local"). Worker-pool mode (WithWorkerPool) requires a distinct
	// ID per process and assigns a unique one when this is empty. The
	// DBOS__VMID environment variable, when set, overrides this field.
	ExecutorID string
}

Config configures a duro application.

type Context added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

type DeduplicationPolicy = dbos.DeduplicationPolicy

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

type EnqueueOption added in v0.8.0 

type EnqueueOption func(*enqueueConfig)

EnqueueOption configures a client Enqueue.

func WithClientApplicationVersion added in v0.8.0 

func WithClientApplicationVersion(version string) EnqueueOption

WithClientApplicationVersion pins the enqueued run to a specific application version, so only workers on that version dequeue it. By default a client stamps no version (NULL), meaning any worker version may run it — the right default for a web tier that should not need redeploying in lockstep with the workers.

func WithClientWorkflowID added in v0.8.0 

func WithClientWorkflowID(id string) EnqueueOption

WithClientWorkflowID assigns the run's workflow ID — the idempotency key. Enqueuing the same ID twice re-attaches to the first run instead of starting another.

type Event added in v0.2.0 

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 added in v0.2.0 

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

NewEvent declares a typed event key.

func (Event[V]) Get added in v0.2.0 

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 added in v0.2.0 

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

Key returns the event's key.

type Fork added in v0.2.0 

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 added in v0.2.0 

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 added in v0.4.0 

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; prefer AttachJob, which takes it from the job's declaration instead.

func AttachJob added in v0.8.0 

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

AttachJob is Attach with the result type taken from a Job rather than written out at the call site, so it cannot drift from the registration. Use it whenever the run belongs to a job you declared.

func Enqueue added in v0.8.0 

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

Enqueue places a run on the queue for a worker to execute, and returns a handle to it. The job supplies the registered workflow name and both types, so the call cannot name a pipeline that does not exist, or disagree with it about the input or result type — register the same job on the workers with RegisterJob. The run is serialized identically to an engine-side enqueue, so workers decode it transparently.

By default no application version is stamped (any worker version may run it); override with WithClientApplicationVersion.

func ForkFromStage added in v0.2.0 

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 added in v0.2.0 

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

ID returns the run's workflow ID.

func (Handle[R]) Result added in v0.2.0 

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

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

func (Handle[R]) Status added in v0.2.0 

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

Status returns the run's current status.

type Job added in v0.8.0 

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

Job is a registered pipeline's durable identity: its workflow name and both of its types in one declaration, shared by the process that registers the pipeline and every process that enqueues or awaits it.

It exists because a workflow name crossing a process boundary is otherwise three unchecked assumptions at once. Enqueuing by bare string, a typo produces a run no worker can ever dispatch — inserted successfully, reported as a healthy "enqueued", and waiting forever; a wrong input type fails to deserialize in another process long after the call site; a wrong result type surfaces when the caller reads the result. A Job makes all three the compiler's problem: declare it once beside the pipeline and both sides refer to the same value. 

// package jobs, imported by the worker and the web tier alike
var Invoice = duro.NewJob[Batch, Invoice]("invoice")

// worker
duro.RegisterJob(app, jobs.Invoice, invoicePipeline)

// web tier — name and both types come from the same declaration
handle, err := duro.Enqueue(client, Queue, jobs.Invoice, batch)

The zero Job is invalid; build one with NewJob.

func NewJob added in v0.8.0 

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

NewJob declares a pipeline's cross-process identity. Declaring is side-effect free — package level is the point — and registration happens separately through RegisterJob.

func (Job[P, R]) Name added in v0.8.0 

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

Name returns the workflow name the job is registered and enqueued under.

type ListOption added in v0.9.0 

type ListOption func(*listConfig)

ListOption narrows, pages, or enriches a ListRuns query. Every filter is conjunctive: a run is returned only if it matches all of them.

func WithCreatedAfter added in v0.9.0 

func WithCreatedAfter(t time.Time) ListOption

WithCreatedAfter restricts the listing to runs created at or after t. A zero time applies no bound.

func WithCreatedBefore added in v0.9.0 

func WithCreatedBefore(t time.Time) ListOption

WithCreatedBefore restricts the listing to runs created at or before t. A zero time applies no bound.

func WithIDs added in v0.9.0 

func WithIDs(ids ...string) ListOption

WithIDs restricts the listing to the given workflow IDs. Given no IDs, nothing matches. Unlike StatusAll, results follow the listing's sort order, not the order of the IDs.

func WithInput added in v0.9.0 

func WithInput() ListOption

WithInput loads each run's stored input into RunStatus.Input, as JSON text. It is off by default so the listing stays payload-free like StatusAll.

func WithLimit added in v0.9.0 

func WithLimit(n int) ListOption

WithLimit caps the number of runs returned; page with WithOffset. Without it the listing is unbounded. n must be positive: ListRuns reports a zero or negative limit as an error rather than silently returning nothing.

func WithNames added in v0.9.0 

func WithNames(names ...string) ListOption

WithNames restricts the listing to runs of the named pipelines — their registered workflow names, as in Register, RegisterJob, or Job.Name. Given no names, nothing matches.

func WithNewestFirst added in v0.9.0 

func WithNewestFirst() ListOption

WithNewestFirst orders the listing by creation time descending. The default is oldest first.

func WithOffset added in v0.9.0 

func WithOffset(n int) ListOption

WithOffset skips the first n runs of the listing, for paging with WithLimit. n must not be negative.

func WithQueue added in v0.9.0 

func WithQueue(name string) ListOption

WithQueue restricts the listing to runs currently recorded on the named queue (a Queue's Name). DBOS clears a run's queue when it is cancelled or exceeds its recovery attempts, so those runs match no queue.

func WithStates added in v0.9.0 

func WithStates(states ...State) ListOption

WithStates restricts the listing to runs in any of the given states. Given no states, nothing matches.

type Option added in v0.8.0 

type Option func(*appOptions)

Option configures a duro App; pass Options to New after the Config.

func WithRetention added in v0.8.0 

func WithRetention(d time.Duration) Option

WithRetention batch-deletes terminal runs that completed more than d ago, bounding the otherwise unbounded growth of workflow history (DBOS open source has no built-in retention). Each maintenance cycle deletes at most one batch, so a large backlog clears gradually over many cycles instead of in one long transaction; deletion holds its own advisory lock, never the sweeper's, so retention can never delay a takeover. Usable with or without WithWorkerPool.

Scope warning: DBOS records no application name on a run, so retention deletes every terminal run in the system database older than d — including runs belonging to other duro or DBOS applications that share it. Give each application its own database (or DBOS schema) before enabling this.

func WithStaleRunWarning added in v0.8.0 

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

WithStaleRunWarning surfaces runs older than age that are still waiting or running, split by whether this executor's application version can recover them (see StaleRunInfo) — so stranded runs, which DBOS otherwise reports nowhere, become visible. It logs a warning with the counts; pass an optional hook to also receive them programmatically (invoked only when there is something to report). Usable with or without WithWorkerPool.

Choose age above the longest a healthy run legitimately takes. Runs parked in a Delay stage stay PENDING for the whole pause, so an age below your longest Delay reports healthy runs as stale. (Debounced runs, which park in DELAYED, are already excluded.)

Scope warning: DBOS records no application name on a run, so the counts cover every run in the system database, including those of other applications sharing it.

func WithSweepInterval added in v0.8.0 

func WithSweepInterval(d time.Duration) Option

WithSweepInterval sets how often each process runs its maintenance cycle (default 30s): the scan for dead executors to take over, plus retention and the stale-run warning, which run on the same cadence. It is a plain Option, not a WorkerPoolOption, because it governs all three — an app using only WithRetention or WithStaleRunWarning can still tune it.

Fleet-wide the work is serialized by Postgres advisory locks, so this is per-process cadence, not global.

func WithWorkerPool added in v0.8.0 

func WithWorkerPool(opts ...WorkerPoolOption) Option

WithWorkerPool enables worker-pool mode: liveness heartbeats plus a sweeper that takes over the PENDING runs of dead executors (see the package-level worker-pool documentation). Every process in the fleet must enable it, and each needs a distinct executor identity — one is generated when neither Config.ExecutorID nor DBOS__VMID is set.

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 added in v0.2.0 

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 added in v0.2.0 

func Register[P, R any](ctx Context, name string, p Pipeline[P, R], opts ...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 RegisterJob added in v0.8.0 

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

RegisterJob registers a pipeline under a Job's name, tying the registration to the same declaration the enqueuing side uses. Prefer it over Register for any pipeline another process enqueues with Enqueue: the Job carries the name and both types, so a rename or a type change is a compile error on both sides instead of a run that strands on the queue. Otherwise identical to Register.

func RegisterScheduled added in v0.2.0 

func RegisterScheduled[R any](ctx Context, name, cronSchedule string, p Pipeline[time.Time, R], opts ...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 added in v0.2.0 

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 added in v0.2.0 

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

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

type Queue added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

func (q Queue) Name() string

Name returns the queue's name.

type QueueOption added in v0.2.0 

type QueueOption func(*queueConfig)

QueueOption configures a declared queue.

func WithConcurrency added in v0.2.0 

func WithConcurrency(n int) QueueOption

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

func WithPartitions added in v0.2.0 

func WithPartitions() QueueOption

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

func WithPriorities added in v0.2.0 

func WithPriorities() QueueOption

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

func WithRateLimit added in v0.2.0 

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

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

func WithWorkerConcurrency added in v0.2.0 

func WithWorkerConcurrency(n int) QueueOption

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

type RegisteredWorkflow added in v0.3.0 

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 added in v0.3.0 

func RegisterWorkflow[P, R any](ctx Context, name string, fn WorkflowFunc[P, R], opts ...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 added in v0.3.0 

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

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

type RunStatus added in v0.4.0 

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
	// StartedAt is when a queued run was dequeued to start. DBOS records no
	// start time for a run started directly on its executor, so it is zero
	// for those — and while a queued run is still enqueued or delayed.
	StartedAt          time.Time
	CompletedAt        time.Time // zero until terminal
	ApplicationVersion string
	ExecutorID         string // the executor that last ran (or is running) it
	// Attempts counts how many times execution has been started — 1 for a run
	// that completed without recovery, more after crashes or takeovers, 0
	// while still waiting on a queue.
	Attempts int
	// QueueName is the queue the run was enqueued on; "" for a run started
	// directly on its executor. DBOS clears it when a run is cancelled or
	// exceeds its recovery attempts.
	QueueName string
	// ParentID is the run that started this one — set for FanOut children and
	// other child workflows, "" for top-level runs.
	ParentID   string
	ForkedFrom string // original run's ID when this run was forked
	// Input is the run's input as JSON text — json.RawMessage(status.Input)
	// re-emits it. It is empty unless the run was listed with WithInput:
	// loading payloads is what the status path otherwise avoids. It is the
	// stored JSON, not a decoded value, so it is available from an
	// enqueue-only Client that registers no workflow types. (A string rather
	// than a byte slice keeps RunStatus comparable.)
	Input string
}

RunStatus is the cheap status view of a run: no input or output payloads are loaded or deserialized, making it safe for polling paths. The one exception is Input, which ListRuns fills only when asked with WithInput.

func ListRuns added in v0.9.0 

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

ListRuns lists durable runs — every registered pipeline and workflow, from any process attached to the system database — filtered, paged, and ordered by the options. With no options it returns every run, oldest first; page with WithLimit and WithOffset. Runs are reported through the same mapping as Status, with the same failed-run treatment: a failed run's recorded error is fetched in a second query scoped to the failed runs on the page, so RunStatus.Err is populated exactly as Status would populate it. No payloads are loaded unless WithInput asks for the input. 

runs, err := duro.ListRuns(app,
	duro.WithNames("invoice"),
	duro.WithStates(duro.StateError, duro.StateRetriesExceeded),
	duro.WithNewestFirst(), duro.WithLimit(50))

Client.ListRuns is the same query from an enqueue-only process.

func Status added in v0.4.0 

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 added in v0.4.0 

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 added in v0.4.0 

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 added in v0.4.0 

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 added in v0.2.0 

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 added in v0.2.0 

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. By default, children queued behind it are independent durable workflows and run to completion in the background — the right call when every completion has value on its own (cleanup, deletion). When surviving siblings are wasted spend once the batch has failed, opt into WithCancelSiblings, which cancels every non-terminal sibling promptly while still failing the stage with the original error.

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 added in v0.2.0 

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 added in v0.2.0 

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 added in v0.4.0 

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 added in v0.2.0 

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), and the pipeline fails with the first error in input order. By default every remaining step is drained to completion first; opt into WithCancelSiblingSteps to cancel in-flight siblings and skip unstarted items instead, while still failing with the first genuine error.

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 added in v0.2.0 

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 Rescue added in v0.5.0 

func Rescue[T, R any](name string, p Pipeline[T, R], handler func(ctx context.Context, in T, cause error) (R, error), opts ...StepOption) Stage[T, R]

Rescue is a durable except block: it runs the embedded pipeline for each item and intercepts its failure. On success the embedded pipeline's emissions flow downstream unchanged. On failure — after the embedded stages' own retry policies are exhausted — handler runs as a checkpointed step and decides the outcome: return a fallback R and nil error to swallow (the fallback is emitted downstream and the outer pipeline continues), or return an error to fail the outer pipeline with it (transform, or return cause unchanged to rethrow). handler receives the item that entered the embedded pipeline, so a pass-through swallow is `return in, nil` when the types match. opts apply to the handler step, like Branch's route opts.

A failed embedded run is treated as a unit: its partial emissions are discarded and only the handler's fallback is emitted. A successful embedded run that emits nothing drops the item, like Filter. Because the handler is a checkpointed step, effectful handlers (failure reports, warn logs) replay consistently, and the rescue decision is durable: recovery never flips a swallowed failure into a propagated one or vice versa. The cause is replay-stable by construction: the handler always receives a plain error carrying exactly the terminal failure's message — the same value a recovered run replays — so the handler cannot behave differently on recovery than it did live. Match causes by message; error identities (errors.Is/As) are deliberately not preserved, because they cannot survive recovery. A decision that needs the typed error belongs in the failing step itself or its WithRetryPredicate, which always see the live error.

Applied per item on multi-item streams. Only failures inside the embedded pipeline are rescued: upstream failures, and errors from the handler itself, propagate as usual. Rescue nests — the innermost enclosing Rescue wins — and Pipe1(Rescue(name, p, handler)) is the whole-pipeline except block.

func Send added in v0.2.0 

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 added in v0.2.0 

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 added in v0.4.0 

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 added in v0.4.0 

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 added in v0.2.0 

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.

func Via added in v0.6.0 

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

Via runs the embedded pipeline for each item — durably, to completion — then emits the ORIGINAL item downstream, discarding the embedded pipeline's emissions. Tap is to Step what Via is to Sub: the embedded pipeline exists for its effects (typically a FanOut of child workflows or a Parallel fleet), not its outputs, and the item that entered continues on afterwards. That is what lets a pipeline whose state must survive a fan-out stay a single registered pipeline instead of a hand-written workflow around several Run calls.

Via never drops items: a successful embedded run that emits nothing (a Filter dropped everything) still passes the original item through — embedded emissions are ignored entirely, zero included. This is the deliberate contrast with Sub, whose emissions ARE the stream. The re-emitted item is replay-stable because it came from upstream checkpoints; Via itself records no step.

An embedded failure fails the outer pipeline exactly like any stage failure — partial embedded effects before it remain checkpointed. Wrap in Rescue to swallow: Rescue(name, Pipe1(Via(...)), handler) is a best-effort fan-out that continues with the original item either way. Applied per item on multi-item streams, in stream order.

type StaleRunInfo added in v0.8.0 

type StaleRunInfo struct {
	// SameVersion counts non-terminal runs older than the warning age on this
	// executor's application version — in limbo, but recoverable by this fleet.
	SameVersion int
	// OtherVersion counts non-terminal runs older than the warning age on other
	// application versions — not recoverable here until an executor on their
	// version runs (see Config.ApplicationVersion).
	OtherVersion int
}

StaleRunInfo reports the counts a stale-run warning surfaces on each sweep (see WithStaleRunWarning). Stranded runs otherwise emit no signal at all.

type State added in v0.4.0 

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 added in v0.4.0 

func (s State) Failed() bool

Failed reports whether the run finished without succeeding.

func (State) Terminal added in v0.4.0 

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 added in v0.2.0 

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 WithCancelSiblingSteps added in v0.7.0 

func WithCancelSiblingSteps() StepOption

WithCancelSiblingSteps makes the first failed step of a Parallel stage cancel its sibling steps instead of letting the stage drain them: in-flight siblings have their step contexts cancelled (the step function must honor context cancellation for this to take effect, the same contract WithTimeout documents), and items whose steps have not started skip the function entirely. The stage still fails with the first genuinely-failed step's error, in input order — cancelled and skipped siblings never mask it, on the live run and on recovery replay alike, so a Rescue around the stage always sees the original failure.

Skipped and cancelled items still occupy their pre-assigned step slots, recording a marker error instead of running the function — the slot count stays deterministic, which is what keeps step IDs aligned on recovery (a recovered workflow that continues past the stage, e.g. through Rescue, would otherwise read misaligned checkpoints). Because the layout is unchanged, the option does not alter the pipeline's shape fingerprint. If the workflow is recovered and reaches the stage again, items with no recorded outcome re-execute, exactly as without the option.

A step's own retry policy runs before cancellation triggers: siblings are cancelled only when a step's error escapes its retries, matching when the stage would fail.

It applies only to Parallel; every other stage constructor rejects it at construction time. Sequential stages have nothing to cancel — an earlier failure already prevents later items from executing. For the child-workflow equivalent on FanOut, see WithCancelSiblings.

func WithMaxInterval added in v0.2.0 

func WithMaxInterval(d time.Duration) StepOption

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

func WithMaxIterations added in v0.8.0 

func WithMaxIterations(n int) StepOption

WithMaxIterations bounds a Loop: if the predicate has not reported done after n iterations, the stage fails with ErrMaxIterations instead of looping forever. A durable loop outlives the process running it, so an unbounded one whose predicate can never be satisfied keeps checkpointing across restarts until someone cancels the workflow by hand.

The bound is enforced from the checkpointed iteration count, so a recovered run resumes with the same budget it had rather than a fresh one.

It applies only to Loop; every other stage constructor rejects it at construction time.

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 added in v0.2.0 

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 added in v0.2.0 

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 StepStatus added in v0.9.0 

type StepStatus struct {
	ID   int    // position in the run's step sequence, from 0
	Name string // the stage name; ShapeStepName for the shape checkpoint that opens every pipeline run
	// Err is the error the step recorded, nil when it succeeded. A recorded
	// step error is final for that step: retries happen before recording.
	Err error
	// ChildID is the child run this step started — set for FanOut children
	// and other child workflows, "" for ordinary steps.
	ChildID     string
	StartedAt   time.Time
	CompletedAt time.Time
}

StepStatus is the record of one step a run has executed: its position, name, outcome, and timing. Step outputs are never decoded or returned.

func Steps added in v0.9.0 

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

Steps lists the steps a run has executed so far, in execution order — what a pipeline has checkpointed, which is exactly what replays on recovery. A pipeline run opens with the ShapeStepName checkpoint, then one entry per durable stage execution (per item for stages that ran per item; Pure stages record nothing). A run that has not started yet has no steps. Unknown IDs return ErrRunNotFound.

Step outputs are never decoded or returned, but DBOS's step query still reads the output column, so inspecting a run whose stages checkpoint large values costs that bandwidth — Steps is cheap in memory, not free on the wire. Prefer Status for polling; reserve Steps for inspection.

Client.Steps is the same query from an enqueue-only process.

type Stream added in v0.2.0 

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 added in v0.2.0 

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

NewStream declares a typed stream key.

func (Stream[V]) Key added in v0.2.0 

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

Key returns the stream's key.

func (Stream[V]) Read added in v0.2.0 

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 added in v0.2.0 

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 added in v0.2.0 

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

NewTopic declares a typed mailbox topic.

func (Topic[M]) Name added in v0.2.0 

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

Name returns the topic's name.

func (Topic[M]) Send added in v0.2.0 

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 WorkerPoolOption added in v0.8.0 

type WorkerPoolOption func(*appOptions)

WorkerPoolOption tunes worker-pool mode; pass WorkerPoolOptions to WithWorkerPool.

func WithHeartbeatInterval added in v0.8.0 

func WithHeartbeatInterval(d time.Duration) WorkerPoolOption

WithHeartbeatInterval sets how often each process refreshes its liveness lease (default 10s). Keep it well below the stale threshold.

func WithStaleThreshold added in v0.8.0 

func WithStaleThreshold(d time.Duration) WorkerPoolOption

WithStaleThreshold sets how long a lease may go unrefreshed before its executor is considered dead and its PENDING runs are taken over (default 60s). Must be well above WithHeartbeatInterval plus worst-case scheduling jitter and GC pauses — too low and a merely-slow process is treated as dead and its live run is run a second time elsewhere. Avoid sub-second values in production.

type WorkflowFunc added in v0.2.0 

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 WorkflowOption added in v0.6.0 

type WorkflowOption = dbos.WorkflowOption

WorkflowOption configures how a single run starts (DBOS's option type under a duro name, like Context): WithWorkflowID is the common one, and any dbos.WorkflowOption works unchanged. The alias lets consumer helpers name the type without importing dbos.

func WithWorkflowID added in v0.2.0 

func WithWorkflowID(id string) 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 WorkflowOption passes through Start unchanged.

type WorkflowRef added in v0.2.0 

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 added in v0.2.0 

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))

type WorkflowRegistrationOption added in v0.6.0 

type WorkflowRegistrationOption = dbos.WorkflowRegistrationOption

WorkflowRegistrationOption configures how a pipeline or workflow function is registered (DBOS's option type under a duro name); Register and RegisterWorkflow accept these. The alias lets consumer helpers name the type without importing dbos.

Source Files

View all Source files

Directories

Path Synopsis
examples
fleet command
Command fleet demonstrates duro's worker-pool mode: a fleet of interchangeable workers that recover each other's runs, an enqueue-only web tier that starts work without running an engine, and an admin view on that same client that lists, inspects, cancels, and resumes runs.
 Command fleet demonstrates duro's worker-pool mode: a fleet of interchangeable workers that recover each other's runs, an enqueue-only web tier that starts work without running an engine, and an admin view on that same client that lists, inspects, cancels, and resumes runs.
housekeeping command
Package main demonstrates duro's operational toolkit: cron pipelines (RegisterScheduled), burst-collapsing (RegisterDebounced), surgical replay of a finished run from a named stage (ForkFromStage), and the durable identity contract behind pipeline names (the stranded-run warning).
 Package main demonstrates duro's operational toolkit: cron pipelines (RegisterScheduled), burst-collapsing (RegisterDebounced), surgical replay of a finished run from a named stage (ForkFromStage), and the durable identity contract behind pipeline names (the stranded-run warning).
orders command
payments command
Package main demonstrates duro's signal and resilience toolkit on a payment flow: retries with a predicate, per-attempt timeouts, durable pauses, human-in-the-loop approval over a typed Topic, progress Events, and a receipt Stream drained by a second pipeline.
 Package main demonstrates duro's signal and resilience toolkit on a payment flow: retries with a predicate, per-attempt timeouts, durable pauses, human-in-the-loop approval over a typed Topic, progress Events, and a receipt Stream drained by a second pipeline.
thumbnails command
Package main demonstrates duro's parallelism toolkit on a thumbnail rendering fleet: declared queues with concurrency/rate/priority/partition controls, FanOut child options (idempotent IDs, deduplication, timeouts, delays, auth), a hand-written child workflow using duro.Context, in-process bounded Parallel, and a registered pipeline used directly as a FanOut child.
 Package main demonstrates duro's parallelism toolkit on a thumbnail rendering fleet: declared queues with concurrency/rate/priority/partition controls, FanOut child options (idempotent IDs, deduplication, timeouts, delays, auth), a hand-written child workflow using duro.Context, in-process bounded Parallel, and a registered pipeline used directly as a FanOut child.
triage command
Package main demonstrates duro's control-flow combinators on a support ticket triage pipeline: Switch dispatches by category, a nested Branch escalates urgent bugs, Loop durably polls an external system, Sub reuses a shared notification segment, Rescue scopes error handling to a best-effort segment (and to the whole pipeline), Via fans each resolution out to archive systems and passes the resolution through, and Collect folds the batch into a report — all inside one registered pipeline, no hand-written workflow function.
 Package main demonstrates duro's control-flow combinators on a support ticket triage pipeline: Switch dispatches by category, a nested Branch escalates urgent bugs, Loop durably polls an external system, Sub reuses a shared notification segment, Rescue scopes error handling to a best-effort segment (and to the whole pipeline), Via fans each resolution out to archive systems and passes the resolution through, and Collect folds the batch into a report — all inside one registered pipeline, no hand-written workflow function.

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.