dispatch

Dispatch: durable background jobs, workflows and cron for Go

Dispatch is an open-source durable execution library for Go. You import it, hand it a store, and register background jobs, multi-step workflows and cron schedules as plain Go functions. Retries with backoff, a dead letter queue, distributed workers with leader-elected cron, checkpointed workflows that resume after a crash, and a history-based durable workflow runtime with signals, queries and child workflows all run inside your own process, against PostgreSQL, SQLite, MongoDB, Redis or memory. There's no separate server to deploy. It runs standalone or as a Forge extension with a REST admin API.

Documentation · Quick start · Features · Jobs · Workflows · Cron · Stores · Forge · Examples · FAQ

Quick start

You need Go 1.26 or later. This uses the in-memory store, so there's nothing else to install:

mkdir hello-dispatch
cd hello-dispatch
go mod init example.com/hello-dispatch
go get github.com/xraph/dispatch@latest

Save this as main.go:

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/xraph/dispatch"
	"github.com/xraph/dispatch/engine"
	"github.com/xraph/dispatch/job"
	"github.com/xraph/dispatch/store/memory"
	"github.com/xraph/dispatch/workflow"
)

type Signup struct {
	Email string `json:"email"`
}

var done = make(chan struct{})

var SendWelcome = job.NewDefinition("send_welcome",
	func(ctx context.Context, in Signup) error {
		fmt.Println("job: welcome email to", in.Email)
		return nil
	},
)

var Onboard = workflow.NewWorkflow("onboard",
	func(wf *workflow.Workflow, in Signup) error {
		if err := wf.Step("create-account", func(ctx context.Context) error {
			fmt.Println("workflow: account created for", in.Email)
			return nil
		}); err != nil {
			return err
		}

		return wf.Step("notify", func(ctx context.Context) error {
			fmt.Println("workflow: team notified")
			close(done)
			return nil
		})
	},
)

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

	d, err := dispatch.New(
		dispatch.WithStore(memory.New()),
		dispatch.WithConcurrency(4),
	)
	if err != nil {
		log.Fatal(err)
	}

	eng, err := engine.Build(d)
	if err != nil {
		log.Fatal(err)
	}
	engine.Register(eng, SendWelcome)
	engine.RegisterWorkflow(eng, Onboard)

	if err := d.Start(ctx); err != nil {
		log.Fatal(err)
	}
	defer d.Stop(ctx)

	in := Signup{Email: "ada@example.com"}
	if _, err := engine.Enqueue(ctx, eng, "send_welcome", in, job.WithMaxRetries(5)); err != nil {
		log.Fatal(err)
	}
	if _, err := engine.StartWorkflow(ctx, eng, "onboard", in); err != nil {
		log.Fatal(err)
	}

	select {
	case <-done:
		time.Sleep(200 * time.Millisecond) // give the job worker a moment
	case <-time.After(10 * time.Second):
		log.Fatal("timed out")
	}
}

Run it:

go run .
# workflow: account created for ada@example.com
# workflow: team notified
# job: welcome email to ada@example.com

The workflow prints first because StartWorkflow runs it in the calling goroutine, checkpointing each step as it goes, while the job waits in the queue for one of the four workers to poll it. Payloads are JSON, so whatever you pass to Enqueue has to marshal cleanly.

The memory store is for development and tests. Swap in postgres.New(groveDB) from store/postgres (see Stores) and the same jobs, workflow checkpoints and cron entries survive a restart and spread across every process that runs the engine.

Why Dispatch?

Most job queues give you jobs, and you build workflows, schedules and failure handling on top. Dispatch gives you all of it as a library inside your Go service, so a job handler is a typed function you can unit test and the queue is a table in the database you already run.

The architecture guide shows how the dispatcher, the engine, the worker pool and the stores fit together.

Background jobs

A job definition is a name, a typed handler and options. Register it with the engine, then enqueue it by name:

var ResizeImage = job.NewDefinition("resize_image",
	func(ctx context.Context, in ResizeInput) error {
		if in.Width <= 0 {
			return fmt.Errorf("bad width %d: %w", in.Width, dispatch.ErrPermanent)
		}
		return resize(ctx, in)
	},
)

engine.Register(eng, ResizeImage)

_, err := engine.Enqueue(ctx, eng, "resize_image", ResizeInput{Width: 640},
	job.WithQueue("media"),
	job.WithMaxRetries(5),
	job.WithTimeout(2*time.Minute),
	job.WithPriority(10),
	job.WithRunAt(time.Now().Add(time.Hour)),
)

Pass the queue, retries, timeout and priority to Enqueue. job.NewDefinition accepts the same options, but Enqueue doesn't read them from the definition today, so set them per call. Unless you say otherwise, your job gets 3 retries, the default queue and a 5 minute timeout. When it runs out of retries, or returns an error wrapping dispatch.ErrPermanent, it moves to the dead letter queue. Workers only poll the queues you give dispatch.WithQueues, and engine.WithQueueConfig adds per-queue rate limits and concurrency.

Workflows

Dispatch has two workflow models. You can run both in one engine, and which one you pick mostly comes down to how long your process lives and whether something outside it has to talk to it while it runs.

Checkpointed workflows (workflow package) are ordinary Go functions made of named steps. Each completed step is recorded, so if your process dies partway through, the next engine start resumes the run and skips the steps that already finished. You also get Parallel, Sleep, WaitForEvent, WaitForAll, WaitForAny, StepWithCompensation for sagas, child workflows and versioned definitions with NewWorkflowV. Failed runs can be replayed from a chosen step.

Durable workflows (durable and durable/runtime) replay a deterministic handler against recorded history, in the style of Temporal. You schedule activities, timers and signal receives as futures, and the runtime gives you signals, queries, selectors, async activity completion, child workflows, cancellation, execution timeouts, continue-as-new and whole-workflow retry policies. Turn it on with engine.WithDurableWorkflows(...). It needs a store that implements durable.Store, which today means store/memory or store/postgres. Workflow code has to stay deterministic, so keep network calls, clocks and randomness in activities. The examples/ directory has a runnable program for each feature.

Cron

err := engine.RegisterCron(ctx, eng, &cron.Definition[Report]{
	Name:     "nightly-report",
	Schedule: "0 2 * * *",
	JobName:  "build_report",
	Payload:  Report{Kind: "daily"},
})

Each tick enqueues an ordinary job, so your cron runs get the same retries, DLQ and middleware as everything else. One worker holds cron leadership at a time. Registering the same name again changes nothing, so you can call this on every start without worrying about duplicate schedules piling up in the store.

Stores

store.Store combines the job, workflow, cron, DLQ, event, cluster and artifact stores, and every backend implements all of them.

You hand the SQL and MongoDB stores a Grove database, and the Redis store a Grove KV store. The persistent stores have a Migrate method that creates their tables or collections. store/storetest holds the conformance suites if you want to write your own. See the store docs for each backend.

Use it with Forge

The github.com/xraph/dispatch/extension package plugs Dispatch into a Forge app. It depends on the Grove extension, picks up the *grove.DB it provides (Postgres, SQLite or MongoDB), runs the store migrations on start, builds the engine and provides *engine.Engine through DI. It also mounts the REST admin API under /dispatch/v1.

import (
	"github.com/xraph/dispatch/cron"
	"github.com/xraph/dispatch/engine"
	dispatchext "github.com/xraph/dispatch/extension"
	"github.com/xraph/forge"
	_ "github.com/xraph/grove/drivers/pgdriver"
	groveext "github.com/xraph/grove/extension"
)

jobs := dispatchext.New(dispatchext.WithConcurrency(8))

app := forge.NewApp(forge.AppConfig{
	Name:        "accounts",
	HTTPAddress: ":8080",
	Extensions: []forge.Extension{
		groveext.New(groveext.WithDSN("postgres", "postgres://localhost:5432/accounts?sslmode=disable")),
		jobs,
	},
})

err := app.RegisterHook(forge.PhaseAfterRegister, func(ctx context.Context, _ forge.App) error {
	eng := jobs.Engine()
	engine.Register(eng, SendWelcome)

	return engine.RegisterCron(ctx, eng, &cron.Definition[Signup]{
		Name:     "nightly-digest",
		Schedule: "0 2 * * *",
		JobName:  "send_welcome",
		Payload:  Signup{Email: "team@example.com"},
	})
}, forge.LifecycleHookOptions{Name: "register-jobs"})

err = app.Router().POST("/signup", func(ctx forge.Context) error {
	eng, err := forge.Inject[*engine.Engine](app.Container())
	if err != nil {
		return err
	}

	j, err := engine.Enqueue(ctx.Context(), eng, "send_welcome", Signup{Email: "ada@example.com"})
	if err != nil {
		return err
	}

	return ctx.JSON(http.StatusAccepted, map[string]string{"job_id": j.ID.String()})
})

jobs.Engine() is nil until the extension has registered, which is why the job registration sits in a PhaseAfterRegister hook. The admin routes (/dispatch/v1/jobs, /dispatch/v1/workflows/runs, /dispatch/v1/dlq, /dispatch/v1/crons, /dispatch/v1/stats and their actions) authenticate through the Forge auth extension and authorize with Warden by default. Pass WithRemoteSecurity to bring your own authenticator, or WithDisableRoutes() to mount nothing.

Configuration

The extension reads extensions.dispatch, then falls back to a top-level dispatch key.

extensions:
  dispatch:
    base_path: /dispatch
    grove_database: primary
    resources:
      enabled: true
      cpu_overcommit: 1.0
      memory_fraction: 0.8

Worker tuning is code-only: WithConcurrency, WithQueues, WithPollInterval, WithHeartbeatInterval, WithStaleJobThreshold and the WithCron* options. A dispatch: block nested under the extension key parses, but nothing reads it, so set those from Go. WithMiddleware, WithBackoff and WithExtension add job middleware, a retry strategy and lifecycle hooks. The Forge extension guide has more.

Examples

Every directory under examples/ is a runnable program for one feature of the durable runtime, built with the rest of the module:

Run one with GOWORK=off go run ./examples/durable-signals. A larger resource-aware example lives in _examples/resources, which the Go tool skips in ./....

Documentation

The exported Go API is on pkg.go.dev. The docs source is in docs/, with design notes in docs/architecture/.

FAQ

How do I run background jobs in Go without Redis or a separate server?

Use Dispatch with the PostgreSQL or SQLite store. Your queue lives in the database you already back up, the workers run inside your own process, and there's nothing new to deploy. Run more copies of your service and they share the work.

Does Dispatch retry failed jobs?

Yes. Each job retries up to its MaxRetries (3 by default) with exponential backoff and jitter, then lands in the dead letter queue, where you can replay it. Return an error that wraps dispatch.ErrPermanent when a retry can't help.

How is Dispatch different from Temporal?

Temporal is a cluster you run and talk to over the network. Dispatch is a Go library that keeps its state in your database. Its durable runtime borrows the same ideas (deterministic replay, activities, signals, queries, child workflows), and its checkpointed workflows are a simpler option for step-by-step processes.

Do cron jobs run once when I have several instances?

Yes. Cron entries live in the store, and only the instance holding cron leadership enqueues a tick. If it stops, another one takes over.

Can I use Dispatch without Forge?

Yes. The quick start above doesn't touch Forge. You only need it for the extension package.

Contributing

Issues and focused pull requests are welcome on GitHub. Use GOWORK=off on every Go command so you see the module graph CI sees:

git clone https://github.com/xraph/dispatch.git
cd dispatch
GOWORK=off go build ./...
make test-race
make lint
GOWORK=off go test -race -tags integration ./store/postgres/... ./store/redis/... ./extension/contract/...

The integration tests start containers and fail without Docker, and CI never runs them, so run them yourself when you touch a store. Some PostgreSQL tests also need a throwaway server in DISPATCH_READ_TEST_DSN or DISPATCH_LIFECYCLE_TEST_DSN and skip without one. Skip make build, make run and make dev: they point at a cmd/dispatch that doesn't exist. AGENTS.md explains the qualification/ modules, every test tier and the release workflow. Commits follow Conventional Commits.

License

MIT. See LICENSE.

Built by Rex Raphael and contributors, as part of the Forge forgery stack.