Workflows

A workflow is a set of tasks with dependencies. Each task is one job on a queue that you bound to a worker. VarMQ queues a task when every task it depends on has ended. Use a workflow when one job needs the output of another, or when finished work must be undone after a later failure.

Workflows need Go 1.27. Add and Undo are generic methods, and Go supports those from 1.27.

A first workflow

You need three things: a store, a workflow and its tasks.

1. Create a store with NewWorkflowStore. It keeps the state of each workflow in memory.
2. Create a Workflow with NewWorkflow.
3. Add the tasks, then call Submit.
4. Call Wait or WaitContext to block until the workflow ends.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	// Each step has its own worker, queue and data type.
	upload := varmq.NewResultWorker(func(j varmq.Job[string]) (string, error) {
		return "s3://bucket/" + j.Data(), nil
	}).BindQueue()

	// A result worker's return value is the task's result.
	thumb := varmq.NewResultWorker(func(j varmq.Job[int]) (string, error) {
		var src string
		if err := varmq.TaskOf(j).Result("upload", &src); err != nil {
			return "", err
		}
		return fmt.Sprintf("%s@%dpx", src, j.Data()), nil
	}, 2).BindQueue()

	email := varmq.NewErrWorker(func(j varmq.Job[string]) error {
		t := varmq.TaskOf(j)
		for _, dep := range []string{"thumb-s", "thumb-l"} {
			var url string
			if err := t.Result(dep, &url); err != nil {
				return err
			}
			fmt.Println("email to", j.Data(), "with", url)
		}
		return nil
	}).BindQueue()

	wf := varmq.NewWorkflow(varmq.NewWorkflowStore(), varmq.WithWorkflowID("upload-1"))
	up := wf.Add("upload", upload, "photo.png")
	small := wf.Add("thumb-s", thumb, 64).After(up)
	large := wf.Add("thumb-l", thumb, 512).After(up)
	wf.Add("email", email, "ann@example.com").After(small, large)

	if err := wf.Submit(context.Background()); err != nil {
		panic(err)
	}

	// Wait with a deadline, so a stuck workflow cannot block forever.
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	rec, err := wf.WaitContext(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println("workflow:", rec.State)
	for _, t := range rec.Tasks {
		fmt.Printf("  %-8s %-10s result: %d bytes\n", t.Name, t.State, t.ResultSize)
	}
}

upload runs first. The two thumbnail tasks start after it, on a worker with concurrency 2, so they can run at the same time. email starts only after both have completed. Each thumbnail task reads the result of upload, and email reads both thumbnail results. The email task has no result, because it runs on an error worker.

Tasks and dependencies

wf.Add(name, queue, data) adds a task and returns its *Task. The task is one job with data on queue. The compiler checks that data has the job type of the queue.

• A task name uses only A-Z, a-z, 0-9, _, . and -, with 1 to 64 characters. It must be unique in the workflow.
• The task’s job ID is wf:<workflow ID>:<task name>. You see it as j.ID() in the worker function.
• Every Task method returns the same *Task, so you can chain the calls.

A dependency must be a task of the same workflow, added before the task that waits for it. So the order of your Add calls is a valid run order, and a workflow cannot have a cycle. A task with no dependencies starts at Submit.

When a dependency fails

By default, a task is cancelled when one of its dependencies fails or is cancelled. Its own dependents are then cancelled too, and so on. A task with Always runs anyway. In its worker, Failed(dep) tells you which dependency failed.

A failed task does not fail the workflow when it has dependents and all of them use Always. Those tasks handle the failure. Any other failed task makes the workflow end as failed.

Without other options, tasks on other branches still run after a failure. WithFailFast() changes that. At the first failure that makes the workflow fail, every waiting task is cancelled. Queued tasks that have not started are cancelled too. The failed task’s own Always dependents still run.

Which queues a task can use

A task can use a queue handle that BindQueue, BindPriorityQueue, WithQueue or WithPriorityQueue returns. These handles implement TaskQueue, whose WorkflowItem method turns task data into a job. You do not call it yourself. Each handle must also be in the scope of the workflow’s store: see workflows across processes. Queues covers how to bind a queue.

The deprecated binding methods for shared storage return handles that a task cannot use. Passing one to Add does not compile. Bind that storage with WithQueue or WithPriorityQueue instead. A topic subscription cannot be a task’s queue either.

A built-in in-memory queue takes a task’s job even when the queue is full. A custom queue bound with WithQueue can refuse it. The task then fails with the reason queue refused promotion.

Passing results between tasks

Only a task on a result worker leaves a result. VarMQ stores the value the worker function returns as JSON. A task on a plain or error worker leaves no result.

In the worker function, TaskOf(j) returns the *TaskContext of the job. It is nil when the job is not a workflow task. It has the workflow ID, the task name and these methods:

• Result(dep, &v) reads the result of the dependency dep into v.
• Failed(dep) reports whether the dependency dep failed or was cancelled.

Result returns ErrDepFailed when dep failed or was cancelled. It returns ErrNoResult when dep is not a dependency of this task, or left no result.

A result is limited to 64 KiB. A bigger result fails the task with ErrResultTooLarge. To change the limit, pass WithResultLimit to NewWorkflowStore. For a large output, store the data somewhere else and return its key.

A result that cannot be encoded as JSON fails the task. Task data that cannot be encoded as JSON makes Submit fail.

Submitting and waiting

Submit(ctx) checks the whole workflow, stores it and queues the tasks that have no dependencies. It returns every mistake that you made while you added tasks, joined in one error. For example, it reports a duplicate task name or a dependency from another workflow. A Submit that fails leaves the workflow unsubmitted, so you can call it again, for example after a store error. A mistake in the tasks stays in the workflow, so build a new workflow to fix it.

Each workflow has an ID. WithWorkflowID sets it, and ID() returns it. Without the option, VarMQ makes a random one. A second Submit with the same ID and the same tasks returns nil and changes nothing. With the same ID and different tasks, it returns ErrWorkflowExists.

After a successful Submit, you cannot change the workflow. A later Add or Task method does nothing, and the next Submit returns ErrWorkflowSubmitted.

To follow a workflow, use one of these:

• Wait() blocks until the workflow is in a final state.
• WaitContext(ctx) does the same, but returns ctx.Err() when ctx ends.
• Status(ctx) returns the current state and does not block.

All three return a WorkflowRecord. It has the workflow’s State and one TaskRecord per task, in Add order. A task record has the task’s State, the Reason it failed, its ResultSize and the record of its Undo.

Workflow states

The state is a WorkflowState. Its Go constants are WorkflowRunning, WorkflowCompleted, WorkflowFailed, WorkflowCompensating, WorkflowCompensated, WorkflowCompensationFailed and WorkflowCancelled.

A task’s State is waiting while a dependency has not ended, and pending once its job is queued. It ends as completed, failed or cancelled. The in-memory store shows a running task as pending.

Undo

Undo(queue, data) gives a task a second job that reverses its work, such as a refund for a payment. The undo job runs only if its task completed and the workflow then fails. It also runs after a cancel, unless you turned that off: see cancelling a workflow.

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	steps := varmq.NewErrWorker(func(j varmq.Job[string]) error {
		if j.Data() == "charge card" {
			fmt.Println("run: charge card - declined")
			return errors.New("card declined")
		}
		fmt.Println("run:", j.Data())
		return nil
	}).BindQueue()

	undos := varmq.NewErrWorker(func(j varmq.Job[string]) error {
		fmt.Println("undo:", j.Data(), "- for task", varmq.TaskOf(j).Name)
		return nil
	}).BindQueue()

	wf := varmq.NewWorkflow(varmq.NewWorkflowStore(), varmq.WithWorkflowID("order-1"))
	reserve := wf.Add("reserve", steps, "reserve stock").Undo(undos, "release stock")
	label := wf.Add("label", steps, "print label").After(reserve) // no undo
	wf.Add("charge", steps, "charge card").After(label)

	if err := wf.Submit(context.Background()); err != nil {
		panic(err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	rec, err := wf.WaitContext(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println("workflow:", rec.State)
	for _, t := range rec.Tasks {
		line := fmt.Sprintf("  %-8s %s", t.Name, t.State)
		if t.Reason != "" {
			line += " - " + t.Reason
		}
		if t.Undo != nil {
			line += fmt.Sprintf(" (undo: %s)", t.Undo.State)
		}
		fmt.Println(line)
	}
}

charge fails, so the workflow fails. reserve completed and has an undo, so VarMQ queues release stock. label completed too, but it has no undo, so nothing undoes it. The workflow ends as compensated.

How undo jobs run:

• They run in reverse order. An undo job waits until the undo jobs of all completed tasks that depend on its task have ended. Tasks without an undo are skipped.
• A failed undo job does not stop the others. The workflow then ends as compensation_failed.
• A task can have one undo job. A second Undo call makes Submit fail.
• The undo job’s ID is the task’s job ID with :undo added. On a priority queue, it gets priority 0.
• In the undo job’s worker, TaskOf(j).Undo is true. Result can read the result of the task itself, for example a payment ID to refund.
• An undo job’s own result is not kept.

The task record shows an undo as waiting until an undo phase reaches it. In a completed workflow, the undo stays waiting, because it never runs.

Cancelling a workflow

Cancel(ctx) stops a workflow:

if err := wf.Cancel(ctx); err != nil {
	return err
}
rec, err := wf.Wait() // returns when the workflow has ended

What happens to each task:

• A waiting task is cancelled at once.
• A queued task stays in its queue. When a worker takes it, the task ends as cancelled with the reason workflow cancelled, and its function does not run.
• A running task finishes.

The workflow ends only when no task is queued or running. A paused or stopped worker keeps its queued tasks, so Wait blocks until that worker takes them.

By default, the undo jobs of the completed tasks then run, and the workflow ends as compensated or compensation_failed. When no completed task has an undo, it ends as cancelled. Pass WithUndoOnCancel(false) to NewWorkflow to skip the undo jobs on a cancel. A failure still runs them.

The store keeps the WithUndoOnCancel value with the workflow. So a cancel from any handle follows it. Cancel on a workflow that has ended does nothing and returns nil.

Cancelling one task’s job is not a workflow cancel: see cancelling a job. The task ends as cancelled, and the workflow handles it like a failed task. On shared storage, a cancel does not stop a task that still waits: cancel the workflow instead. A Purge of a queue has the same effect on the tasks in it.

The REST API lists workflows, shows one and cancels one, with ?undo=false to skip the undo jobs: see workflows.

Workflows across processes

OpenWorkflow(store, id) returns a handle to a workflow that was submitted before. Use it where you do not have the *Workflow that NewWorkflow returned, for example in an HTTP handler.

h := varmq.OpenWorkflow(store, "order-1")
rec, err := h.Status(ctx)

On this handle, Cancel, Status, Wait and WaitContext work. Submit returns ErrWorkflowSubmitted. If the store does not have the ID, Cancel, Status, Wait and WaitContext return ErrWorkflowNotFound.

The store from NewWorkflowStore is in-memory only. It lives in your process, and it loses every workflow when the process exits. Each call to NewWorkflowStore makes a separate store. So OpenWorkflow finds a workflow only in the same process, with the same store. The store keeps an ended workflow for 7 days. WithWorkflowRetention changes that.

A workflow’s queues must all be in the scope of its store. The scope of the in-memory store is the in-memory queues of your process. A task on a queue in another scope makes Submit return an error that wraps ErrScopeMismatch.

For a workflow that continues after a restart, or that spans processes, the storage keeps the workflow state in a WorkflowStore. Each adapter has its own store constructor: redisq.NewWorkflowStore and sqliteq.NewWorkflowStore. Pass that store to NewWorkflow and OpenWorkflow. Bind its queues with WithQueue or WithPriorityQueue. A producer handle from NewQueue or NewPriorityQueue can be a task’s queue too.

Every process that runs tasks must create the store. The store registers itself by its scope, and a worker finds it by the scope of the queue it takes a task from. A worker in a process without that store does not run the task. It reports varmq: no workflow store for scope …, at most once a minute for each scope, on Errs() and to the error handler, and leaves the job for another try.

Workflows on Redis

redisq v1.6.0 keeps workflows in Redis. Every redisq queue can carry workflow tasks. redisq.NewWorkflowStore makes the store for the client’s Redis server, database and namespace.

qs := redisq.New("redis://localhost:6375")
defer qs.Close()

store, err := redisq.NewWorkflowStore(qs)
if err != nil {
	return err
}
defer store.Close()

upload, thumb := qs.NewQueue("upload"), qs.NewPriorityQueue("thumb")

wf := varmq.NewWorkflow(store, varmq.WithWorkflowID("upload-1"))
up := wf.Add("upload", varmq.NewResultQueue[string, string](upload), "photo.png")
wf.Add("thumb", varmq.NewResultPriorityQueue[int, string](thumb), 64).After(up)
if err := wf.Submit(ctx); err != nil {
	return err
}
rec, err := wf.WaitContext(ctx)

The process that submits needs no worker. Workers in any process run the tasks. Each of those processes creates the store too, and binds the same redisq queue:

store, err := redisq.NewWorkflowStore(qs) // registers the store for this Redis scope
if err != nil {
	return err
}
defer store.Close()

w := varmq.NewResultWorker(func(j varmq.Job[string]) (string, error) {
	return "s3://bucket/" + j.Data(), nil
})
w.WithQueue(qs.NewQueue("upload"))

• The workflow survives a restart. Redis keeps its state. So OpenWorkflow(store, "upload-1") works in any process that uses the same Redis server, database and namespace.
• A task ends in one step. One script records the end of a task and queues the tasks that wait for it. So a crash never leaves a workflow between the two.
• Store options go to NewWorkflowStore. WithWorkflowResultLimit sets the largest task result, 64 KiB by default. WithWorkflowRetention sets how long an ended workflow is kept, 7 days by default.
• Closing. store.Close unregisters the store and leaves the client open. Closing the client also closes its store.
• One Redis credential per scope in a process, with the same ACL user and TLS settings. The scope is the address, database and namespace. The registered store ends the tasks of every client of that scope over its own connection.
• Every process that runs a task must use redisq v1.6.0 or newer.
• Do not call SetExpiration on a queue that carries tasks. Submit refuses such a queue.
• Redis Cluster is not supported for workflows.
• Protect write access to Redis. Anyone who can add jobs to a task’s queue can add a copy of a task.

The redisq repository has examples/workflow, which runs one workflow across three processes.

Workflows on SQLite

sqliteq v0.4.0 keeps workflows in SQLite. sqliteq.NewWorkflowStore makes the store for the client’s root.

qs, err := sqliteq.New("data")
if err != nil {
	return err
}
defer qs.Close()

store, err := sqliteq.NewWorkflowStore(qs)
if err != nil {
	return err
}
defer store.Close()

reserve, _ := qs.NewQueue("reserve")
charge, _ := qs.NewQueue("charge")

wf := varmq.NewWorkflow(store)
r := wf.Add("reserve", varmq.NewResultQueue[string, string](reserve), "sku-1")
wf.Add("charge", varmq.NewErrQueue[int](charge), 100).After(r)
err = wf.Submit(ctx)

• Where the state lives. By default, the store keeps workflows in their own file, __sqliteq_workflows.db, next to the queue files. With WithSingleFile, it uses tables in that one file.
• How a task ends. With one file per queue, a task’s end and the tasks it releases cannot share one transaction. The store writes the released tasks to an outbox in the same transaction as the end. Then it pushes each released task at once. A loop pushes any rows that are left, when the store opens and every second after that. A task can be pushed twice, and the pre-run check skips a copy whose task already ended. With WithSingleFile, the end, the pushes and the acknowledgement are one transaction.
• After a crash. A task that the loop pushes starts at the worker’s next poll, because the loop cannot wake workers.
• Options: WithWorkflowResultLimit and WithWorkflowRetention, with the same defaults as redisq.
• Closing. Close the store before the client. Closing the client does not close the store.
• One process. sqliteq queues serve one process, so a workflow on SQLite runs in one process too. It survives a restart of that process.
• Protect the files. Anyone who can write the queue files or __sqliteq_workflows.db can add or end tasks, and anyone who can read them sees task data and results. Give the folder to the service user only.

For the errors that workflow calls return, see errors.

Next: Context.