engine

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, interrupts, backpressure, and circuit breakers. 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 With* builder methods, inject at least a GraphLoader and a TaskExecutor, then Startup (opens the database, runs migrations, starts workers). Shutdown drains the workers and closes the database. In tests, RunInTest replaces Startup/Shutdown with per-test SQLite databases and t.Cleanup.

eng := engine.NewEngine().
	WithDSN("postgres://user:pass@host:5432/dwarf").
	WithGraphLoader(loadGraph).
	WithTaskExecutor(runTask)
if err := eng.Startup(ctx); err != nil { ... }
defer eng.Shutdown(ctx)

The With* configuration methods are atomic and may be called after Startup for hot reconfiguration.

Dependency interfaces ¶

The engine reaches the outside world through four injection points:

• GraphLoader fetches a workflow graph by name (called once at Create; the graph JSON is then frozen on the flow).
• TaskExecutor executes one task, given the Flow carrier with its state pre-populated.
• FlowStoppedCallback (optional) is fired when a flow stops, for flows started via StartNotify.
• PeerNotifier (optional) carries cross-replica coordination signals; nil in single-replica mode.

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

Operations ¶

Create or CreateTask makes a flow; Start (or StartNotify) runs it; Await blocks until it stops; Run is Create+Start+Await in one call. Snapshot/History/Step/List inspect; Resume/ResumeBreak continue a paused flow; Cancel/Restart/RestartFrom/Continue manage lifecycle; Delete/Purge retain. See the repository's docs/ directory for guides.

package main

import (
	"context"
	"fmt"

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

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

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

	eng := engine.NewEngine().
		WithDSN("postgres://user:pass@localhost:5432/dwarf").
		WithGraphLoader(func(ctx context.Context, name string) (*workflow.Graph, error) {
			return graphs[name], nil
		}).
		WithTaskExecutor(func(ctx context.Context, taskName string, f *workflow.Flow) error {
			f.SetString("greeting", "hello "+f.GetString("name"))
			return nil
		})

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

	// Run is Create + Start + 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["greeting"])
}

Output: