engine

package
v0.10.5 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: Apache-2.0 Imports: 47 Imported by: 0

Documentation

Overview

Package engine is the dwarf workflow-orchestration engine.

The engine executes workflow graphs against a SQL database, scheduling and dispatching one task at a time per step, persisting state between steps, and driving fan-out/fan-in, retries, sleeps, subgraphs, and interrupts. It owns no transport of its own; a host wires it to the outside world through injected dependency interfaces and calls its operations.

Lifecycle

Build an engine with NewEngine and the Set* methods, register a Host (see SetHost), then Startup (opens the database, runs migrations, starts workers). Shutdown drains the workers and closes the database. In tests, NewEngineUnderTest wires isolated, auto-dropped SQLite databases keyed by a test name, so a test just configures the engine, calls Startup, and defers Shutdown (see NewEngineUnderTest). 

eng := engine.NewEngine()
eng.SetShard(ShardSpec{Index: 1, DSN: "postgres://user:pass@host:5432/dwarf"})
eng.SetHost(host)
err := eng.Startup(ctx)
if err != nil { ... }
defer eng.Shutdown(ctx)

Each Set* method returns an error. The live ones (SetMaxOpenConns - an expert override, SetTimeBudget, SetDefaultPriority) take effect immediately on a running engine; the construction-time-only ones (SetShard, SetWorkers, SetHost, SetLogger, SetMeterProvider, SetTracerProvider) return an error if called after Startup. Tuning derives from the facts the host declares plus what the engine observes: ShardSpec.VirtualCPUs drives each shard's connection budget and its placement weight; the budget is automatically split across the engine replicas sharing each database, counted from a shared peer registry the replicas heartbeat into (no replica count is ever declared, and nothing is sent between replicas to establish it); and the worker maximum is derived from the crash-recovery lease margin and the round-trip time measured at Startup, so it holds for any task duration (the pool grows into it only on demand, so short-task deployments never pay for the headroom). The SetWorkers/SetMaxOpenConns overrides exist for tests, benchmarks, and externally-constrained hosts.

Host

The engine reaches the outside world through a single injected Host interface (see SetHost):

  • LoadGraph fetches a workflow graph by name (called at Create; the graph JSON is then frozen on the flow), and on subgraph spawn.
  • ExecuteTask executes one task, given the Flow carrier with its state pre-populated.

That is the whole interface: the engine sends nothing between replicas, so a host wires no inter-replica transport for it. Replicas sharing a database coordinate by reading it.

The flow's opaque baggage (host identity/tenant/context, set in workflow.FlowOptions) rides on the dispatch context of every LoadGraph and ExecuteTask call; read it with workflow.BaggageFrom(ctx).

Operations

Create makes a flow and runs it; Await blocks until it stops; Run is Create+Await in one call. Snapshot/History/Step/List inspect; Resume continues a paused flow; Cancel/Continue manage lifecycle; Fork clones a terminal flow from a chosen step into a new flow for non-destructive recovery; Delete/Purge retain. See the repository's docs/ directory for guides.

Security model

Flow and step keys ("{shard}-{id}-{token}") are unguessable bearer capabilities, not authorization. Holding a flow key is by itself sufficient to act on that one flow — Resume, Cancel, Fork, Continue, Delete, and every introspection call — with no further check: the sole gate is the key (the numeric id plus its random flow_token). The engine performs no authentication, authorization, or rate limiting and has no notion of caller identity; its only vantage is the flow reference and the task URL, so ownership and tenancy are invisible to it. Authorizing an operation is therefore the host's responsibility: before calling the engine, verify the authenticated principal may act on the flow — typically from the baggage the host set at Create (see workflow.FlowOptions.Baggage), or the host's own record mapping a principal to the keys it was issued.

The token defends only against reference forgery and id enumeration: flow ids are sequential, so without the token a caller cannot fabricate a key for a flow it was never handed. That is defense in depth, not access control — a leaked, logged, or shared key grants its bearer full write access to that one flow, so treat a key like a password. The engine does not emit keys to traces or logs (telemetry carries only a token-free "{shard}-{id}" correlation id), and there is deliberately no operation that resolves a correlation id back to a key, which would be a capability-minting oracle.

List returns keys, tokens included — the only key-returning operation (Purge returns a count, not keys; free-text search is a field on workflow.Query consumed by List, not a separate operation). Exposing them to a principal is equivalent to granting the write capability for every flow they return, so a host must gate them by ownership and never surface them to less-than-fully-trusted callers. Key exposure is also transitive across an execution tree: Step and History navigation resolve and return the keys of neighboring steps, crossing flow boundaries into parent and subgraph-child flows, so a single step key reaches the whole tree (each neighbor key both discloses that step's state and can seed a Fork). Authorizing introspection by one flow's ownership is therefore insufficient when the caller holds any step key in that tree; treat the tree (its root) as the authorization unit, or restrict these surfaces to fully-trusted operators. Operations on an unknown or mismatched key return a uniform not-found (no existence oracle), but that is a hardening detail, not a substitute for host authorization.

Resource limits

The engine imposes no size or count limits — not on initial state, baggage, the frozen graph JSON, interrupt/resume payloads, forEach fan-out width (one step row per array element), or subgraph nesting depth. This is deliberate, the same division of labor as backpressure and time budgets: state size is workload-defined (a document-processing workflow may legitimately carry tens of megabytes per flow), and no single cap fits both that and a small control flow. Bounding resource use is therefore the host's job — it holds the caller identity and tenancy the engine cannot see. A host enforces quotas where it has that context: reject an over-large initial state or Baggage before Create, cap forEach input arrays in author space, and bound its own retention (Purge deletes at most 4096 roots per call, so a retention job loops). For a pass-through host that adds no policy of its own, this obligation flows through to the application using that host. (Deep subgraph nesting is bounded storage, not a crash vector: Fork clones the tree iteratively, so nesting depth costs no goroutine stack.)

Example

Wire an engine to a host, then create, start, and await a flow. 

package main

import (
	"context"
	"fmt"

	"github.com/microbus-io/dwarf/engine"
	"github.com/microbus-io/dwarf/workflow"
)

// exampleHost implements engine.Host. A real host loads graphs from a registry/file/database/RPC and
// dispatches tasks over a local call, RPC, or message bus; here an in-memory registry and a local
// function stand in. LoadGraph and ExecuteTask are required; the remaining Host methods (flow-stop
// notification and the cross-replica signals) are left as no-ops.
type exampleHost struct {
	graphs map[string]*workflow.Graph
}

func (h exampleHost) LoadGraph(ctx context.Context, name string) (*workflow.Graph, error) {
	return h.graphs[name], nil
}
func (h exampleHost) ExecuteTask(ctx context.Context, taskName string, f *workflow.Flow) error {
	f.SetString("greeting", "hello "+f.GetString("name"))
	return nil
}

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

	graphs := map[string]*workflow.Graph{}
	g := workflow.NewGraph("Greet")
	g.SetEndpoint("Hello", "Hello")
	g.AddTransition("Hello", workflow.END)
	graphs["greet"] = g

	eng := engine.NewEngine()
	eng.SetShard(engine.ShardSpec{Index: 1, DSN: "postgres://user:pass@localhost:5432/dwarf"})
	eng.SetHost(exampleHost{graphs: graphs})

	err := eng.Startup(ctx)
	if err != nil {
		panic(err)
	}
	defer eng.Shutdown(ctx)

	// Run is Create + Await in one call.
	_, out, err := eng.Run(ctx, "greet", map[string]any{"name": "ada"}, nil)
	if err != nil {
		panic(err)
	}
	fmt.Println(out.State.GetString("greeting"))
}

Index

Examples

Constants

View Source 
const (
	// Scoped by task name (the graph node name passed to ExecuteTask):
	FaultExecuteTask      = "executeTask"      // ExecuteTask returns a synthetic error (as if the task failed)
	FaultPanicExecuteTask = "panicExecuteTask" // ExecuteTask panics (exercises host-call panic isolation)

	// Scoped by workflow URL:
	FaultLoadGraph = "loadGraph" // LoadGraph returns a synthetic error

	// Scoped by task name; force the persistence/transition steps of a dispatch to fail:
	FaultTransitionCommit   = "transitionCommit"   // the post-completion transition transaction errors
	FaultCompleteFlowCommit = "completeFlowCommit" // the flow-completion transaction errors
	FaultContention         = "contention"         // a dispatch transaction returns a lock-contention error
	FaultLeaseStaleWrite    = "leaseStaleWrite"    // the completion write carries a stale lease_seq (zombie)
	FaultPersistErr         = "persistErr"         // the step-completion write returns a non-contention database error (consumed per attempt, so InjectN sets how many attempts fail)
	FaultSubgraphSpawnErr   = "subgraphSpawnErr"   // createSubgraphFlow errors after the caller step parked

	// Scoped by workflow URL of the subgraph child:
	FaultSubgraphReviveLost = "subgraphReviveLost" // completeSurgraphFlow skips reviving the parked caller

	// Process-wide, consumed per attempt (InjectN sets how many attempts fail):
	FaultCompleteSurgraphErr = "completeSurgraphErr" // completeSurgraphFlow returns a non-contention database error

	// Process-wide (no scope):
	FaultInterruptStaleWrite = "interruptStaleWrite" // handleInterrupt's in-tx leaf lease_seq read is forced to mismatch (zombie)
	FaultInterruptChainWrite = "interruptChainWrite" // handleInterrupt's combined chain UPDATE fails and applies nothing (deadlock victim)
	FaultDropSignalStop      = "dropSignalStop"      // signalStop delivers nothing (lost terminal wake)
	FaultDropDoorbell        = "dropDoorbell"        // the local work doorbell is dropped (the step waits for a refiller scan)
	FaultRecoveryResetErr    = "recoveryResetErr"    // the processStep recovery defer's own reset UPDATE errors
	FaultReapMidTree         = "reapMidTree"         // the reaper errors after deleting steps, before flows
	FaultReapSelectErr       = "reapSelectErr"       // the reaper's due-root SELECT errors
	FaultRefillScanErr       = piston.FaultScanErr   // the piston's priority-band scan errors (its name, so there is one catalogue)
	FaultPeerReadErr         = peers.FaultReadErr    // a shard's peer-registry reading fails: this replica goes BLIND on that shard (their names, so there is one catalogue)
	FaultPeerBeatErr         = peers.FaultBeatErr    // a shard's peer-registry beat writes nothing: this replica stops proving its liveness THERE, while running on
	FaultSlowPoolPush        = "slowPoolPush"        // recomputePools stalls between reading R and pushing the derived sizes
	FaultDeliverFailureErr   = "deliverFailureErr"   // deliverFlowFailureToParent drops the parked-caller re-dispatch (lost delivery); unscoped, or scoped by the parked caller's task name for per-level control
	FaultCancelCommit        = "cancelCommit"        // the Cancel transaction errors
	FaultResumeCommit        = "resumeCommit"        // the Resume transaction errors
	FaultForkCommit          = "forkCommit"          // the Fork clone transaction errors
)

--- Fault injection ---

View Source 
const (
	CheckpointResumeBeforeFlowWrite   = "resumeBeforeFlowWrite"   // resume(), just before its transaction's flow-status gate write
	CheckpointBeforeTransitionTx      = "beforeTransitionTx"      // processStep, after the step is marked completed, before the transition transaction
	CheckpointAfterCallerPark         = "afterCallerPark"         // processStep, after the subgraph caller step is parked, before createSubgraphFlow
	CheckpointBeforeRetryRewind       = "beforeRetryRewind"       // processStep, before the flow.Retry rewind transaction
	CheckpointBeforeCompleteFlowWrite = "beforeCompleteFlowWrite" // completeFlow(), just before its transaction's status-gate write
	CheckpointBeforeDeleteWrite       = "beforeDeleteWrite"       // deleteFlow(), just before its transaction's delete-stamp/interrupted-CAS write
	CheckpointBeforeReviveWrite       = "beforeReviveWrite"       // completeSurgraphFlow(), just before its transaction's caller-revive write
	CheckpointBeforeRecoveryReset     = "beforeRecoveryReset"     // processStep recovery defer, just before its fenced step-reset transaction

	// A COUNTING checkpoint (read with Visits, never a rendezvous), scoped by flow id so concurrent flows
	// count independently. No test arms Wait/Break on it, so Checkpoint just increments and returns; both the
	// count site (execution.go) and the read site (faninflowlock_test.go) consult e.seams directly.
	//
	// It exists for a single pin, and that pin guards a performance property no correctness test can see: a
	// NON-FINAL cohort arrival must issue ZERO flow-row statements. Grabbing the flow row for every arrival
	// serializes an entire cohort on one row (measured at fan-out width 64: 20 of 43 active backends queued on
	// that one statement), so the grab is deferred to the arrival that actually resolves the cohort. Nothing
	// about the flow's OUTCOME changes if someone reintroduces a per-arrival flow-row write - the fan-in still
	// fires and every existing fixture still passes - it just costs ~20% throughput silently. Hence a counter
	// rather than a state assertion.
	//
	// Counts are per flow and cumulative across the flow's whole life (Visits never resets), so a test reads
	// the delta it cares about. A Transact contention retry re-runs the closure and legitimately re-counts, so
	// a test asserting an exact count must be single-worker and contention-free.
	CheckpointFlowRowWrite = "flowRowWrite" // a dwarf_flows UPDATE issued by the transition transaction

	// Lifecycle rendezvous (fired at an event, used with Wait - not a freeze site): lets a test wait for
	// exact engine progress instead of polling status / sleeping. signalStop fires it BOTH unscoped and scoped
	// by (flowKey, status): the unscoped name catches whichever flow stops first, which is all a single-flow
	// test needs, while the scoped one lets a test wait for ONE flow to reach ONE status while other flows (a
	// subgraph child, a fan-out sibling, a peer replica's work) stop concurrently. Both are needed because a
	// scoped fire does not wake an unscoped waiter.
	//
	// The scoped form is meaningful only because signalStop runs POST-COMMIT: when it fires, the status is
	// durable, so a test reading the row immediately after sees it - the exact guarantee a status poll spun for.
	CheckpointFlowStopped = "flowStopped" // signalStop(), a flow just reached a stop (completed/failed/cancelled/interrupted)

	// Lifecycle rendezvous, fired BOTH unscoped and scoped by flow key when an Await has registered on the
	// latch board and is about to block. It is the mirror of CheckpointFlowStopped: that one says a flow
	// settled, this one says a caller was already waiting when it did.
	//
	// Registration order is not a correctness question - the board is POLLED, so a key that settles before
	// its caller arrives is reported by the next sweep, and `await` reads once before parking anyway. It is
	// a question about what a TEST exercised. Every test about a blocked caller being woken has to get its
	// Await onto the board before the thing it is waiting for happens, and the only alternative to this is
	// a sleep long enough to make that likely - which on a slow machine silently becomes its opposite: the
	// Await registers after the stop, its own first read answers it, every assertion still passes, and the
	// wake path under test never ran. A rendezvous makes the premise exact instead of probable.
	//
	// Fired through latch.Board's SetOnPark hook rather than from `await`, because registering and blocking
	// are deliberately one call there (the gap is the race Close is careful not to leave). Wired only when
	// the seams are enabled, so a production board carries no hook at all.
	CheckpointAwaitParked = "awaitParked" // Board.Latch(), an Await is on the board and about to block

	// Lifecycle rendezvous, fired at the END of a recovery pass - every detector in it has read the database
	// by then. recoveryLoop sweeps ON ENTRY (Startup), so that first pass runs CONCURRENTLY with the test body,
	// and a test that forges a wedge/orphan shape and then drives one detector itself is racing it: the sweep
	// sees the forged shape too and counts/logs it a second time. The pass is not instant - it is four scans on
	// one goroutine sharing the engine's connection pool, so against a loaded server it lands seconds after
	// Startup (measured on SQL Server: the sweep's own detectOrphanedFlows landed 27ms before a forge that
	// followed a Create+Await). Wait for this before forging; the wait is free once the sweep is behind (Visits).
	CheckpointRecoverySweepDone = "recoverySweepDone" // runRecoverySweep(), one full recovery pass is over

	// Fired per shard (scoped by the shard number, and once unscoped) when that shard's piston completes a
	// cycle that PUSHED - i.e. its cache partition now reflects the plan. Its name is the piston's, so there
	// is one catalogue, exactly like FaultRefillScanErr.
	//
	// What it exists for: `Offer` admits a step into an EMPTY partition unconditionally and stamps that
	// partition's band from the arrival, while `Cache.Pop` ranks partitions by that FROZEN band and never
	// consults the current global minimum - so a doorbell-admitted hint is indistinguishable from a
	// plan-selected one until the owning shard's next cycle reconciles it. Every shard reconciles on its own
	// cadence, so a test that needs the fleet's partitions to agree with the plan before it asserts dispatch
	// ORDER must wait for a cycle per shard; no amount of elapsed time substitutes, because one starved
	// piston is exactly the case that breaks it.
	CheckpointRefillCycleDone = piston.CheckpointCycleDone // a shard's piston reconciled its cache partition
	CheckpointRefillStole     = piston.CheckpointStole     // a shard's piston selected steps from OUTSIDE its residue class
)

--- Execution checkpoints ---

View Source 
const VariablePoolIdle = "poolIdle"

VariablePoolIdle is the idle-connection size the engine DERIVED for a shard, targeted by shard index (seamsJoin(VariablePoolIdle, strconv.Itoa(idx))). It is recorded wherever a derived size is pushed, so a test can read the half of shardPool's result that has no other witness: database/sql reports the configured max OPEN size through DBStats.MaxOpenConnections, but nothing reports the configured max idle - DBStats.Idle is the number of connections currently idle, which is traffic, not configuration. The value recorded is the engine's derivation, before the test-mode cap in internal/database clamps what the pool actually takes.

Variables

This section is empty.

Functions

This section is empty.

Types

type Engine 

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

Engine is the standalone workflow orchestration engine.

func NewEngine 

func NewEngine() *Engine

NewEngine creates a new workflow engine.

func NewEngineUnderTest added in v0.10.0 

func NewEngineUnderTest(testName string) *Engine

NewEngineUnderTest constructs an engine wired for testing: isolated, auto-dropped databases keyed by the given test name. It takes no testing.TB - a name is all it needs, so a host driving the engine under its own harness (one with no *testing.T to hand over) reaches it the same way a Go test does. The caller configures the engine (SetHost, SetShard, ...), drives Startup, and owns teardown: 

e := NewEngineUnderTest(t.Name())
defer e.Shutdown(ctx)
e.SetHost(h)
e.Startup(ctx)

The name is the isolation key: engines given the SAME name share one set of isolated databases (which is how a multi-replica test gives its peer engines shared state), and engines given DIFFERENT names are independent deployments. So a test standing up several INDEPENDENT engines gives each its own name (t.Name()+"#"+suffix), and a benchmark reused across warmup/measure passes gives each pass a fresh one.

Logging default: stderr at Error, so a CI failure surfaces the engine-level alarms (wedge sweeps / poll / refill faults) without the Info-level play-by-play noise (stderr, not t.Log, because a `go test` timeout panic drops buffered t.Log output but not stderr). DWARF_TEST_LOG_LEVEL overrides the level (e.g. "info" or "debug" for the flow-status play-by-play; "silent" or "off" for the discard logger). A benchmark or fuzz target silences the engine itself with SetLogger, since per-iteration logging would dominate the measurement / flood the fuzz output; SetLogger before Startup takes over entirely either way.

func (*Engine) Await 

func (e *Engine) Await(ctx context.Context, flowKey string) (*workflow.FlowOutcome, error)

Await blocks until a flow stops, then returns its outcome. Running out of time is an error; the flow is unaffected and keeps running, and the caller still holds the key to Await/Snapshot/Cancel it later.

Pass a ctx with a deadline. It is the only bound on the wait - a flow can run for as long as its work takes, and there is no notification the engine could time out on instead. A ctx without one is honored for a long fixed budget and then times out, which is a guard against blocking forever, not a wait to design around. A caller whose own deadline is shorter than the flow wants Poll, not Await.

func (*Engine) Cancel 

func (e *Engine) Cancel(ctx context.Context, flowKey string, reason string) error

Cancel aborts a flow.

func (*Engine) Continue 

func (e *Engine) Continue(ctx context.Context, threadKey string, additionalState any) (string, error)

Continue creates a new flow from the latest completed flow in a thread, inheriting that flow's policy (scheduling, baggage) - it does not take FlowOptions. For a turn with different policy, use Create with FlowOptions.ThreadKey.

func (*Engine) Create 

func (e *Engine) Create(ctx context.Context, workflowURL string, initialState any, opts *workflow.FlowOptions) (flowKey string, err error)

Create creates a new flow for a workflow and starts it, returning the running flow's key. opts carries the flow's policy (scheduling, DeleteOnCompletion, Baggage, ThreadKey); nil uses defaults. For a flow that must wait for an external trigger, have the entry task call flow.Interrupt and resume it with Resume (which, unlike a separate start, also delivers a payload).

Example

Create makes and runs a flow, and accepts FlowOptions for scheduling, notifications, thread membership, and the opaque host baggage carried with the flow. 

package main

import (
	"context"
	"fmt"

	"github.com/microbus-io/dwarf/engine"
	"github.com/microbus-io/dwarf/workflow"
)

func main() {
	var eng *engine.Engine // obtained from NewEngine().…Startup(ctx)
	ctx := context.Background()

	flowKey, err := eng.Create(ctx, "greet", map[string]any{"name": "ada"},
		&workflow.FlowOptions{
			Priority:    10,                             // lower runs first
			FairnessKey: "tenant-42",                    // fair scheduling bucket
			Baggage:     map[string]any{"actor": "ada"}, // read with workflow.BaggageFrom(ctx)
		})
	if err != nil {
		panic(err)
	}

	// Create runs the flow immediately; Await blocks until it stops.
	out, _ := eng.Await(ctx, flowKey)
	fmt.Println(out.Status)
}

func (*Engine) DB added in v0.10.0 

func (e *Engine) DB() *database.ShardSet

DB exposes the engine's shard set so a black-box test in another package can inspect flow/step rows directly. Test-only: it panics outside a test binary (!testing.Testing()), so production code never reaches the shards through it. The guard is testing.Testing() rather than "was this built by NewEngineUnderTest" so it also serves a test that builds its engine with the raw NewEngine (e.g. restart_test, which needs a real file DSN and no test-DB harness) - such an engine is not in test mode but is still a legitimate test engine.

func (*Engine) Delete 

func (e *Engine) Delete(ctx context.Context, flowKey string) error

Delete removes a flow and its steps.

func (*Engine) Fingerprint 

func (e *Engine) Fingerprint(ctx context.Context, flowKey string) (fingerprint string, status string, err error)

Fingerprint returns a fingerprint and status for change detection.

func (*Engine) FlowsStarted added in v0.10.0 

func (e *Engine) FlowsStarted() int64

FlowsStarted returns the count of flows this engine has started (Create/Continue/Fork/subgraph), a cheap atomic read that needs no configured meter. FlowsTerminated is its completion counterpart; started - terminated approximates in-flight work for backpressure.

func (*Engine) FlowsTerminated added in v0.10.0 

func (e *Engine) FlowsTerminated() int64

FlowsTerminated returns the count of flows this engine has completed/failed/cancelled. See FlowsStarted.

func (*Engine) Fork added in v0.8.0 

func (e *Engine) Fork(ctx context.Context, stepKey string, stateOverrides any) (string, error)

Fork clones a terminal flow's prefix up to the given step into a new, self-contained running flow and re-executes from that step with optional stateOverrides applied to it. The original flow is never modified. The fork inherits the original's scheduling and baggage (it does not take FlowOptions). Returns the new flow's key.

func (*Engine) History 

func (e *Engine) History(ctx context.Context, flowKey string) ([]workflow.FlowStep, error)

History returns the step-by-step execution history of a flow.

func (*Engine) HistoryMermaid 

func (e *Engine) HistoryMermaid(ctx context.Context, flowKey string, w io.StringWriter) error

HistoryMermaid writes the execution DAG of a flow as a Mermaid diagram.

func (*Engine) List 

func (e *Engine) List(ctx context.Context, query workflow.Query) ([]workflow.FlowSummary, string, error)

List queries flows by status, workflow name, or thread key, newest first, with cursor pagination (Query.Limit, default 100; the returned cursor fetches the next page). Query.Limit is a per-shard cap divided across shards, not a hard ceiling on the total: a multi-shard page can hold up to shards*ceil(Limit/shards) summaries (see Query.Limit). Pass Query.Shard, or truncate the page, for a strict count.

"Newest first" is per shard, not global. On a MULTI-SHARD fleet each shard contributes its own newest flows and the results are grouped by shard, so the concatenation is not in one descending time order - shard 2's newest flow follows shard 1's oldest returned one. There is no cross-shard order to give: the flow ids are per-shard sequences (a shard with fewer flows has lower ids, so they do not compare), and created_at would compare different database servers' clocks. A single-shard engine - the default - is globally newest-first. A caller that needs one ordered view across shards sorts the page itself, choosing what to trust; a UI that must not show interleaving artifacts can page one shard at a time with Query.Shard.

func (*Engine) Poll added in v0.9.3 

func (e *Engine) Poll(ctx context.Context, flowKey string) (*workflow.FlowOutcome, error)

Poll returns a flow's current outcome, blocking up to the ctx deadline for it to stop. Unlike Await, a ctx timeout is not an error: it returns the current non-terminal outcome, whose Stopped() reports false, so a caller bridging an open-ended flow to a bounded request (e.g. an HTTP poll) can answer within its budget and re-poll. A real failure still returns an error.

The ctx deadline IS the budget, so pass one; without it Poll blocks on the same long fallback budget as Await before answering, which is rarely what a polling caller wants.

func (*Engine) Purge 

func (e *Engine) Purge(ctx context.Context, query workflow.Query) (int, error)

Purge marks flows matching a query (and their subgraph subtrees) for deletion; a background reaper removes them shortly after. Marked flows are excluded from List/History immediately. Returns the count of roots marked - no more than 4096 per call; iterate to mark more. Query.Limit is divided per shard exactly as in List (up to ceil(Limit/shards) roots per shard), so a multi-shard call can mark more than Limit roots. Running flows are skipped.

func (*Engine) Resume 

func (e *Engine) Resume(ctx context.Context, flowKey string, resumeData any) error

Resume continues a flow paused by flow.Interrupt.

func (*Engine) Run 

func (e *Engine) Run(ctx context.Context, workflowURL string, initialState any, opts *workflow.FlowOptions) (flowKey string, outcome *workflow.FlowOutcome, err error)

Run creates, starts, and awaits a flow in one call, returning the new flow's key alongside its outcome (the key is the flow's identity, not part of the outcome). opts carries scheduling and the opaque host Baggage; nil opts uses defaults.

Error semantics differ by phase. A create failure returns flowKey "" and a nil outcome - no flow exists. An await failure - most commonly the caller's ctx expiring before the flow stops - leaves the flow running (it is durable and not bound to this call) and returns its flowKey with a nil outcome and the error, so the caller keeps a handle to Await/Snapshot/Cancel it later. Run never cancels the flow on the caller's behalf; a caller that wants the flow torn down on timeout calls Cancel explicitly.

func (*Engine) Seams added in v0.10.0 

func (e *Engine) Seams() *seamster.Seamster

Seams exposes the test instrumentation seams (fault injection and execution checkpoints) so a black-box test in another package can drive the engine's internal race windows deterministically. Test-only: it panics outside a test binary (!testing.Testing()); the seams are inert unless armed, and arming is itself only possible under testing.Testing(), so this exposes no production behaviour.

func (*Engine) SetDebugLogger added in v0.4.0 

func (e *Engine) SetDebugLogger() error

SetDebugLogger is a convenience that wires a human-readable text logger to stderr at debug level - shorthand for SetLogger(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))). It is meant for development and test runs where you want to see the engine's (and its sequel DB layer's) internal logging without standing up an OTEL pipeline. Output goes to stderr, not stdout, so it never mixes with a program's data stream - the standard convention for diagnostic logs. Because it routes through SetLogger, it counts as an explicitly-set logger, so it also reaches sequel via the engine's existing SetLogger wiring (sequel's migration logs appear too). Construction-time only.

func (*Engine) SetDefaultPriority added in v0.4.0 

func (e *Engine) SetDefaultPriority(p int) error

SetDefaultPriority sets the default priority for new flows: an integer >= 1, lower runs first. Live: read fresh on each Create, so it takes effect on a running engine immediately.

The lower bound is not cosmetic. The refiller selects the strict-minimum band with a `priority=(SELECT MIN(priority) ...)` subquery, but a step is only a candidate at all through predicates the scan applies to a positive band; a flow stamped with a non-positive priority would sit `pending` forever, invisible to selection, while the doorbell re-rang it in a loop. The upper bound guards the column's int32 width - an int that overflows it (3_000_000_000, say) would wrap NEGATIVE and produce exactly that hang. `FlowOptions.Priority` is separately validated at Create, where 0 means "unset, take this default".

func (*Engine) SetEngineID added in v0.10.0 

func (e *Engine) SetEngineID(id int64) error

SetEngineID pins this replica's identity, overriding the random identifier minted per instance. The identity is what a replica registers in the shared peer registry to be counted for the connection-pool split across replicas. The default is random and fresh on every restart, which is correct for the common case (including several engines in one process, which must count as distinct replicas) but leaves a stale registry entry behind when a replica crashes: the entry lingers until it ages out, transiently over-counting replicas and shrinking every live replica's pool share in the meantime. Pinning a value that is STABLE across a replica's restarts (for example, derived from the deployment's own per-instance identity) lets a restarted replica reuse its entry instead, so a crash-restart never inflates the count.

The id must be positive (0 is reserved) and UNIQUE among replicas concurrently sharing the same databases: two live replicas with the same id register as one, under-counting replicas and over-sizing pools. Leave it unset (random) unless a stable, unique value is genuinely available - a wrong stable id is worse than the random default. Construction-time only.

func (*Engine) SetHost added in v0.4.0 

func (e *Engine) SetHost(h Host) error

SetHost registers the host the engine reaches the outside world through: it loads graphs, executes tasks, and (optionally) receives flow-stop notifications and carries cross-replica coordination signals. A host must implement LoadGraph and ExecuteTask; the remaining Host methods may be no-ops. Construction-time only.

func (*Engine) SetLogger added in v0.4.0 

func (e *Engine) SetLogger(l *slog.Logger) error

SetLogger sets the structured logger. The engine logs through the *Context variants (DebugContext/InfoContext/WarnContext/ErrorContext) so a handler that reads the context - e.g. the otelslog bridge - can correlate each record with the active step span. A host routes logs to OTEL by passing a logger whose handler bridges there. Defaults to a discard logger: until a logger is injected the engine (and its sequel DB layer) stay silent rather than writing to the application-owned slog.Default(). A nil logger resets to that silent default. Construction-time only - the engine resolves the logger once at Startup.

func (*Engine) SetMaxOpenConns added in v0.4.0 

func (e *Engine) SetMaxOpenConns(n int) error

SetMaxOpenConns is an expert override that pins every shard's connection pool to exactly n open (and idle) connections, replacing the per-shard budget the engine derives from ShardSpec.VirtualCPUs. Operators normally never call this - provide VirtualCPUs instead and let the engine size the pool at the measured knee (12x the database's CPU count at 32 vCPUs or more, 6x below). The override exists for benchmarking (pool-size sweeps) and for deployments whose connection budget is constrained by something the engine cannot see (e.g. a shared database or an external pooler). Live: pushes to every open shard immediately.

func (*Engine) SetMeterProvider added in v0.4.0 

func (e *Engine) SetMeterProvider(mp metric.MeterProvider) error

SetMeterProvider sets the OpenTelemetry MeterProvider the engine builds its dwarf_* instruments from. Defaults to the global otel.GetMeterProvider() (the no-op provider unless the host configures the OTEL SDK). The engine creates instruments under the "github.com/microbus-io/dwarf" scope; the provider's Resource carries the host service's identity. Construction-time only - the engine resolves the meter once at Startup.

func (*Engine) SetRefillInterval added in v0.10.0 

func (e *Engine) SetRefillInterval(d time.Duration) error

SetRefillInterval is an expert override that PINS every piston's cycle period to d, replacing the value the engine derives from capacity/vCPUs/R (deriveRefillInterval). d <= 0 restores derivation. Operators normally never call this - the derived period tracks the cache sizing it depends on automatically.

The override exists for benchmarking (scan-rate sweeps): the period is measured, not tuned, so finding its optimum needs to hold it at a series of fixed values - INCLUDING values below the 20ms minimum gap, since the unlimited-scanning arm is one of the measured reference points (it costs 8% candidate churn and a 77% worse p99 while buying no throughput, and that number has to stay reproducible). So a pinned interval also lowers the pipeline's MinGap to match when it is the tighter of the two; restoring derivation restores the default gap. Without that the fuse would silently clamp every sub-20ms arm of a sweep to 20ms and quietly flatten the interesting end of the curve.

Live: applies on the next recomputeRefillIntervals, and this triggers one immediately on a running engine.

func (*Engine) SetShard added in v0.10.0 

func (e *Engine) SetShard(spec ShardSpec) error

SetShard registers a database shard. Call once per shard; see ShardSpec for field semantics. When no shard is registered, Startup opens a single default shard 1 (in test mode, an isolated in-memory database).

Construction-time only: shards are opened and migrated at Startup and the set is immutable for the engine's life, so a call on a running engine is rejected. Changing the shard set requires a coordinated restart of every replica (a maintenance window): a flow created on a shard unknown to a peer replica is unroutable (404) there.

func (*Engine) SetTimeBudget added in v0.4.0 

func (e *Engine) SetTimeBudget(d time.Duration) error

SetTimeBudget sets the default duration for a single task execution, used by any flow that does not override it via FlowOptions.TimeBudget. Live: read fresh on each Create (existing flows keep the budget frozen at their own Create). It must be positive.

The lower bound is not cosmetic. The budget is the ExecuteTask call's context deadline and, via the step row, the size of the crash-recovery lease - so a non-positive default hands every new flow a deadline that has already passed, failing every task instantly, engine-wide and silently. Unlike Priority and FairnessWeight, 0 cannot mean "unset" here: this setter IS the default. The floor is a millisecond rather than a nanosecond because the budget is persisted in MILLISECONDS: a positive duration below 1ms truncates to zero and produces exactly the expired-deadline case.

func (*Engine) SetTracerProvider added in v0.4.0 

func (e *Engine) SetTracerProvider(tp trace.TracerProvider) error

SetTracerProvider sets the OpenTelemetry TracerProvider the engine builds its spans from. Defaults to the global otel.GetTracerProvider() (the no-op provider unless the host configures the OTEL SDK). The engine emits one span per flow and one span per task, nested to mirror the call structure, under the "github.com/microbus-io/dwarf" scope; the provider's Resource carries the host's identity. The host injects only the provider - it writes no span or context code. Construction-time only - the engine resolves the tracer once at Startup.

func (*Engine) SetWorkers added in v0.4.0 

func (e *Engine) SetWorkers(n int) error

SetWorkers is an expert override that pins the maximum number of worker goroutines, replacing the derived default. The default is the lease-margin ceiling: the largest pool that keeps a synchronized completion storm (every in-flight task released at once by a recovering downstream) draining inside the crash-recovery lease margin, derived at Startup from each shard's connection budget and its measured round-trip time. It contains no assumption about task duration, so it is correct for any workload; the pool grows into it only on demand, so a short-task deployment never pays for the headroom.

Operators normally never call this. Reasons to: memory (the ceiling can be tens of thousands of workers, and each in-flight step also holds its state map - a bound the engine cannot see); a deliberately smaller global concurrency cap; deterministic tests (SetWorkers(1) serializes dispatch); and benchmark sweeps. Setting it ABOVE the ceiling is allowed - an operator may consciously trade the risk of duplicate task execution in a storm for long-task throughput - and is logged as a warning at Startup. SetWorkers(0) is a valid shape: a replica that creates, awaits, and serves reads but never executes tasks. Construction-time only: the pool bound is fixed at Startup, so a call on a running engine is rejected.

func (*Engine) ShardInfo 

func (e *Engine) ShardInfo(ctx context.Context) ([]ShardSummary, error)

ShardInfo returns health and size summaries for all shards.

func (*Engine) Shutdown 

func (e *Engine) Shutdown(ctx context.Context) error

Shutdown stops all worker goroutines and closes database connections. Idempotent: a call on an engine that is not running (never started, or already shut down) is a no-op, so it is safe to defer and to call more than once.

func (*Engine) Snapshot 

func (e *Engine) Snapshot(ctx context.Context, flowKey string) (*workflow.FlowOutcome, error)

Snapshot returns the current state and status of a flow.

func (*Engine) Startup 

func (e *Engine) Startup(ctx context.Context) error

Startup initializes the engine: opens database connections, runs migrations, and starts worker goroutines.

func (*Engine) Step 

func (e *Engine) Step(ctx context.Context, stepKey string) (*workflow.FlowStep, error)

Step returns details of a single step.

type Host added in v0.4.0 

type Host interface {
	// LoadGraph fetches a workflow graph definition by its URL (the addressable resolve key passed to
	// Create). The flow's opaque baggage rides on ctx; read it with workflow.BaggageFrom(ctx) if loading
	// is identity-dependent (authz, per-actor graphs). The engine validates the returned graph
	// (graph.Validate) at Create and at subgraph spawn, so the host need not: returning (nil, nil) yields a
	// 404 and a structurally invalid graph a 400, rather than a later dispatch-time failure.
	LoadGraph(ctx context.Context, workflowURL string) (*workflow.Graph, error)

	// ExecuteTask executes a single task within a workflow. taskURL is the task's dispatch URL (the real
	// downstream address), not the graph node name. The flow carrier has its state pre-populated; the
	// executor should call the task and let it write changes to the flow. The flow's opaque baggage rides
	// on ctx - read it with workflow.BaggageFrom(ctx) (e.g. to mint a token).
	//
	// Execution is at-least-once and may be concurrent: a task can run more than once, and if a worker's
	// lease is lost while it is still running (a task that overruns its ctx deadline, or a forward DB-clock
	// step past the lease) a second worker re-runs it in parallel. The engine guarantees the flow's
	// persisted state reflects exactly one execution, but exactly-once side effects are the task's
	// responsibility - tasks must be idempotent. Honor the ctx deadline (it bounds the step's time budget);
	// a task that ignores it can only be recovered by lease expiry, not cancelled.
	ExecuteTask(ctx context.Context, taskURL string, flow *workflow.Flow) error
}

Host is the contract between the dwarf engine and the surrounding host application. The engine owns no transport of its own; it reaches workflow graphs and tasks exclusively through the host. Register it once via Engine.SetHost.

THE ENGINE SENDS NOTHING TO ITS PEERS, and a host therefore wires no inter-replica transport for it. Replicas sharing a database coordinate entirely by reading it: pending work, flow outcomes and fleet membership are all discovered by polling, on cadences the engine sets, so a fleet converges with no message passing between replicas at any volume - not per step, not per flow, not per deployment event. What a host must supply is exactly what only it can: the graphs and the task dispatch below.

type ShardSpec added in v0.10.0 

type ShardSpec struct {
	// Index identifies the shard. Indices must be >= 1 and unique, but need not be contiguous (shards
	// 1 and 99 are fine). The index is encoded into every flow key created on the shard and drives
	// routing, so the index-to-DSN mapping must be identical across all replicas and stable across
	// restarts.
	Index int
	// DSN is the connection string of the shard's database (dialect auto-detected), used EXACTLY as
	// given - the engine never formats or rewrites it, so a percent-encoded credential (a password
	// "p@ss" written "p%40ss") survives intact. Each shard is declared with its own DSN; there is no
	// template. An empty DSN is only valid in test mode, where the DSN IS a template (a "%d" is replaced
	// with the shard index, which is what gives each shard its own isolated test database).
	DSN string
	// VirtualCPUs is the CPU count of the shard's database server - a fact off the instance's spec
	// sheet. It drives the shard's connection budget (the pool is capped at the measured knee - 12x the
	// CPU count on a server of 32 vCPUs or more, 6x below that, beyond which connections only queue - and
	// on smaller servers actively destabilize or collapse throughput) and its placement weight (new flows are distributed across shards in proportion to
	// measured capacity). Left at 0, the engine assumes 2 - the smallest machine any major cloud sells
	// as a current-generation instance, so the assumed pool stays safe even if the real machine is
	// smaller. Declare it: a large database sized as if it were a 2-CPU one runs at a fraction of its
	// capacity.
	VirtualCPUs int
	// Cordoned excludes the shard from new-flow placement. Everything already resident proceeds
	// normally: existing flows keep executing, and subgraph children, thread continuations (Continue),
	// and forks - all shard-pinned - are still created on it. Use for retiring or overloaded shards.
	Cordoned bool
}

ShardSpec declares one database shard: the facts about it the operator can readily provide, from which the engine derives its tuning (connection budget, placement weight).

type ShardSummary 

type ShardSummary struct {
	Shard     int    `json:"shard,omitzero"`
	Error     string `json:"error,omitzero"`
	LatencyMs int    `json:"latencyMs,omitzero"`
	Steps     int    `json:"steps,omitzero"`
	Flows     int    `json:"flows,omitzero"`
}

ShardSummary is the health/size summary of a single database shard.

type TaskHandler 

type TaskHandler func(ctx context.Context, flow *workflow.Flow) error

TaskHandler is the signature for a test task handler. Read the flow's baggage, if any, with workflow.BaggageFrom(ctx).

type TestProxy 

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

TestProxy routes graph fetches and task dispatches to registered handlers. It implements the Host interface for use with Engine.SetHost.

A multi-replica test needs nothing else from it: replicas coordinate through the database they share, so several engines pointed at one test database ARE a fleet, with no relay between them to stand up.

func NewTestProxy 

func NewTestProxy() *TestProxy

NewTestProxy creates a new test proxy with empty handler registries.

func (*TestProxy) ExecuteTask 

func (p *TestProxy) ExecuteTask(ctx context.Context, taskURL string, flow *workflow.Flow) error

ExecuteTask implements Host.

func (*TestProxy) HandleGraph 

func (p *TestProxy) HandleGraph(name string, graph *workflow.Graph)

HandleGraph registers a workflow graph under the given name. The name should match the workflow URL passed to Engine.Create or Engine.Run.

func (*TestProxy) HandleTask 

func (p *TestProxy) HandleTask(name string, handler TaskHandler)

HandleTask registers a task handler under the given name. The name should match the task URL registered via graph.SetEndpoint.

func (*TestProxy) LoadGraph 

func (p *TestProxy) LoadGraph(ctx context.Context, workflowURL string) (*workflow.Graph, error)

LoadGraph implements Host.

Source Files

View all Source files

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.