Skip to main content
Sign In

Workflows

Build durable, replayable workflows on Rivet Actors: steps, queue waits, timers, joins, races, rollback, versioning, and recovery patterns that survive restarts.

Automate agent work that outlives a process. Steps record their results, wait for people or systems, and resume after crashes and deploys. Every workflow is a Rivet Actor, so each run gets the Actor’s identity, state, queue, and scheduling for free.

A workflow is a durable, replayable Actor definition. It supports the full Actor configuration, including actions and lifecycle hooks, while its run function uses replay-safe workflow primitives.

  • Survives restarts: workflow progress is saved automatically.
  • Re-runs safely: replay follows the same recorded steps.
  • Event-driven: workflows can pause for queue messages, then continue.
  • Waiting costs nothing: a run blocked on a queue wait or timer sleeps until the message or deadline arrives.
  • Inspectable: the run inspector in the Rivet dashboard traces every step’s status, timing, attempts, inputs, outputs, and errors, and can replay eligible steps after a fix.

Quickstart

Simple workflow

Use this when you need a short multi-step sequence.

Loops

This is the recommended workflow shape for most Actor workloads.

  • Use a queue wait inside the loop to receive the next unit of work.
  • Keep Actor state changes in a single workflow loop.
  • This gives you one durable workflow that manages all Actor progress.

Setup and teardown

Use this when the workflow should initialize resources, process queued commands, then clean up.

Steps

Step-only access to Actor APIs

state, vars, db, client(), and connection/event APIs are only valid inside ctx.step(...) callbacks.

Use non-step workflow code for control flow only: queue waits, sleeps, loops, joins, races, and rollback boundaries. Keep Actor-local side effects in steps.

Handling terminal failures as data

Use tryStep when a step failure should produce data instead of failing the whole workflow.

import { setup, workflow } from "@rivet-dev/workflows";
export const paymentActor = workflow({
	state: {
		status: "pending" as "pending" | "manual-review" | "paid",
		reason: null as string | null,
	},
	run: async (ctx) => {
		const charge = await ctx.tryStep({
			name: "charge-card",
			maxRetries: 3,
			run: async (_ctx) => await chargeCard("order-123"),
		});
		await ctx.step("store-charge-result", async (step) => {
			if (!charge.ok) {
				step.state.status = "manual-review";
				step.state.reason = charge.failure.error.message;
				return;
			}
			step.state.status = "paid";
			step.state.reason = null;
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function chargeCard(orderId: string): Promise<string> {
	return `charge-${orderId}`;
}
export const registry = setup({ use: { paymentActor } });

Use try when you want to recover from terminal step, join, or race failures inside a named block.

async function runPaymentFlow(ctx: any) {
  return await ctx.try("payment-flow", async (blockCtx: any) => {
    const auth = await blockCtx.step("authorize", async (blockCtx) =>
      authorizeOrder("order-123"),
    );
    const capture = await blockCtx.step("capture", async (blockCtx) =>
      captureOrder("order-123"),
    );
    return { auth, capture };
  });
}

async function authorizeOrder(orderId: string): Promise<string> {
  return `auth-${orderId}`;
}

async function captureOrder(orderId: string): Promise<string> {
  return `capture-${orderId}`;
}
  • tryStep and try only catch terminal failures. Retry backoff, sleeps, queue waits, eviction, and history divergence still rethrow.
  • Catching a failure does not undo it. state and vars mutations made before the failure remain visible after tryStep or try returns, so use explicit compensating steps when a caught failure needs cleanup.
  • RollbackError is not caught by default. Pass catch: ["rollback"] when you want rollback failures returned as data.

Checkpoint-friendly loop design

Use this when you need reliable replay and resume semantics across crashes and restarts.

import { setup, workflow } from "@rivet-dev/workflows";

type PaymentMessage = {
	id: string;
	amount: number;
};
export const checkpointFriendlyActor = workflow({
	state: {
		appliedCount: 0,
		totalAmount: 0,
		lastPaymentId: null as string | null,
	},
	run: async (ctx) => {
		await ctx.loop("payment-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-payment", {
				timeout: 30000,
			});
			if (!message) return;
			const payment = message.body as PaymentMessage;
			await loopCtx.rollbackCheckpoint("apply-payment-checkpoint");
			const plan = (await loopCtx.step("build-plan", async (_loopCtx) =>
				buildPaymentPlan(payment),
			)) as {
				paymentId: string;
				amount: number;
			};
			await loopCtx.step("apply-side-effects", async (step) => {
				step.state.appliedCount += 1;
				step.state.totalAmount += plan.amount;
				step.state.lastPaymentId = plan.paymentId;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
function buildPaymentPlan(payment: PaymentMessage): {
	paymentId: string;
	amount: number;
} {
	return {
		paymentId: payment.id,
		amount: payment.amount,
	};
}
export const registry = setup({ use: { checkpointFriendlyActor } });

Queues and messaging

Queue

Use this for fire-and-forget commands where the client does not need a reply.

Use the Loops example as the baseline pattern.

Request/response (using queue)

Use this when the caller needs a response from queued processing.

Queue-driven worker

Use this when external systems enqueue work and the Actor should process each item durably.

import { setup, workflow } from "@rivet-dev/workflows";

type Job = {
	id: string;
	amount: number;
};
export const queueWorkerActor = workflow({
	state: {
		processed: 0,
		totalAmount: 0,
	},
	run: async (ctx) => {
		await ctx.loop("worker-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-job", {
				timeout: 30000,
			});
			if (!message) return;
			const job = message.body as Job;
			await loopCtx.step("process-job", async (step) => {
				step.state.processed += 1;
				step.state.totalAmount += job.amount;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { queueWorkerActor } });

Request/response over queue (async RPC)

Use this when you want decoupled Actor-to-Actor communication with durable waits and explicit completion.

Batch drainer

Use this when throughput matters and handling one message at a time is too expensive.

import {
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type MetricMessage = {
	value: number;
};
export const batchDrainerActor = workflow({
	state: {
		pending: [] as number[],
		flushedBatches: 0,
		lastBatchTotal: 0,
	},
	run: async (ctx) => {
		await ctx.loop("drain-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-metric", {
				timeout: 5000,
			});
			const pendingCount = await loopCtx.step(
				"buffer-message",
				async (step) => {
					if (message) {
						step.state.pending.push((message.body as MetricMessage).value);
					}
					return step.state.pending.length;
				},
			);
			if (pendingCount < 5) return;
			await loopCtx.step("flush-batch", async (step) => flushBatch(step));
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
function flushBatch(
	ctx: WorkflowStepContextOf<typeof batchDrainerActor>,
): void {
	const total = ctx.state.pending.reduce(
		(sum: number, value: number) => sum + value,
		0,
	);
	ctx.state.lastBatchTotal = total;
	ctx.state.flushedBatches += 1;
	ctx.state.pending = [];
}
export const registry = setup({ use: { batchDrainerActor } });

Bounded drain and concurrency cap

Use this when inbound work can spike and you need predictable per-iteration limits.

import {
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type WorkMessage = {
	id: string;
	value: number;
};
const MAX_PER_ITERATION = 10;
const CONCURRENCY_LIMIT = 3;
async function processWork(value: number): Promise<number> {
	return value * 2;
}
async function runWithLimit<T>(
	limit: number,
	items: T[],
	fn: (item: T) => Promise<void>,
): Promise<void> {
	let nextIndex = 0;
	const workers = Array.from({ length: limit }, async () => {
		while (nextIndex < items.length) {
			const current = items[nextIndex];
			nextIndex += 1;
			await fn(current);
		}
	});
	await Promise.all(workers);
}
export const boundedDrainActor = workflow({
	state: {
		processed: 0,
		lastWindowSize: 0,
		lastWindowTotal: 0,
	},
	run: async (ctx) => {
		await ctx.loop("bounded-drain-loop", async (loopCtx) => {
			const window: WorkMessage[] = [];
			for (let i = 0; i < MAX_PER_ITERATION; i += 1) {
				const [message] = await loopCtx.queue.nextBatch("wait-work", {
					timeout: i === 0 ? 30000 : 10,
				});
				if (!message) break;
				window.push(message.body as WorkMessage);
			}
			if (window.length === 0) return;
			await loopCtx.step("process-window", async (step) =>
				processWindow(step, window),
			);
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function processWindow(
	ctx: WorkflowStepContextOf<typeof boundedDrainActor>,
	window: WorkMessage[],
): Promise<void> {
	let windowTotal = 0;
	await runWithLimit(CONCURRENCY_LIMIT, window, async (work) => {
		const result = await processWork(work.value);
		windowTotal += result;
	});
	ctx.state.processed += window.length;
	ctx.state.lastWindowSize = window.length;
	ctx.state.lastWindowTotal = windowTotal;
}
export const registry = setup({ use: { boundedDrainActor } });

Signal-driven control loop

Use this when workflow progress should be triggered by commands/events instead of fixed polling intervals.

import {
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type ControlSignal = {
	kind: "pause" | "resume" | "stop";
};
export const controlLoopActor = workflow({
	state: {
		mode: "running" as "running" | "paused" | "stopped",
		handledSignals: 0,
	},
	run: async (ctx) => {
		await ctx.loop("control-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-signal", {
				timeout: 30000,
			});
			if (!message) return;
			const signal = message.body as ControlSignal;
			await loopCtx.step("apply-signal", async (step) =>
				applyControlSignal(step, signal.kind),
			);
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
function applyControlSignal(
	ctx: WorkflowStepContextOf<typeof controlLoopActor>,
	kind: ControlSignal["kind"],
): void {
	ctx.state.handledSignals += 1;
	if (kind === "pause") ctx.state.mode = "paused";
	if (kind === "resume") ctx.state.mode = "running";
	if (kind === "stop") ctx.state.mode = "stopped";
}
export const registry = setup({ use: { controlLoopActor } });

Human approval gate

Use this when an operation must pause for a user or system decision before continuing.

import { queue, setup, workflow } from "@rivet-dev/workflows";
export const approvalGateActor = workflow({
	state: { status: "pending" as string },
	queues: {
		approval: queue<{
			approved: boolean;
		}>(),
	},
	run: async (ctx) => {
		await ctx.step("validate-order", async (step) => {
			await validateOrder("order-123");
			step.state.status = "awaiting_approval";
		});
		const decision = await ctx.queue.next("wait-approval");
		if (decision.body.approved) {
			await ctx.step("fulfill-order", async (step) => {
				await fulfillOrder("order-123");
				step.state.status = "fulfilled";
			});
		} else {
			await ctx.step("cancel-order", async (step) => {
				await cancelOrder("order-123");
				step.state.status = "cancelled";
			});
		}
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function validateOrder(orderId: string): Promise<void> {
	const res = await fetch(
		`https://api.example.com/orders/${orderId}/validate`,
		{ method: "POST" },
	);
	if (!res.ok) throw new Error("Order validation failed");
}
async function fulfillOrder(orderId: string): Promise<void> {
	await fetch(`https://api.example.com/orders/${orderId}/fulfill`, {
		method: "POST",
	});
}
async function cancelOrder(orderId: string): Promise<void> {
	await fetch(`https://api.example.com/orders/${orderId}/cancel`, {
		method: "POST",
	});
}
export const registry = setup({ use: { approvalGateActor } });

Timers and concurrency

Timers

Use queue messages as the trigger source, then sleep durably inside the workflow.

Timeouts

Use step timeouts and retries for slow or flaky dependencies.

Step timeouts are critical by default and fail immediately. Set retryOnTimeout: true if a timeout should retry like any other error using maxRetries.

Workflows use roll-forward semantics everywhere. When a step throws, any state or vars mutations it made before failing are never rolled back, whether the step retries or the failure is caught by tryStep or try. The next attempt observes whatever the failed attempt already wrote, so write steps idempotently: check before you increment, or move the mutation after the fallible work.

import { queue, setup, workflow } from "@rivet-dev/workflows";

async function chargeCard(orderId: string): Promise<string> {
	return `charge-${orderId}`;
}
export const timeoutActor = workflow({
	state: {
		lastChargeId: null as string | null,
	},
	queues: {
		charge: queue<{
			orderId: string;
		}>(),
	},
	run: async (ctx) => {
		await ctx.loop("charge-loop", async (loopCtx) => {
			const message = await loopCtx.queue.next("wait-charge");
			const chargeId = await loopCtx.step<string>({
				name: "charge-card",
				timeout: 5000,
				retryOnTimeout: true,
				maxRetries: 5,
				retryBackoffBase: 200,
				retryBackoffMax: 2000,
				run: async (_loopCtx) => await chargeCard(message.body.orderId),
			});
			await loopCtx.step("save-charge", async (step) => {
				step.state.lastChargeId = chargeId;
			});
		});
	},
});
export const registry = setup({ use: { timeoutActor } });

Join

Use join when several independent tasks can run in parallel.

Race

Use race when you need first-winner behavior.

Fan-out / fan-in (join)

Use this when independent work items can run in parallel and you need a single merged result.

import { setup, workflow } from "@rivet-dev/workflows";
export const fanInOutActor = workflow({
	state: {
		total: 0,
	},
	run: async (ctx) => {
		await ctx.loop("join-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-refresh", {
				timeout: 30000,
			});
			if (!message) return;
			const joined = await loopCtx.join("parallel-work", {
				users: {
					run: async (branchCtx) =>
						await branchCtx.step("fetch-users", (_branchCtx) =>
							fetchCount("/users"),
						),
				},
				orders: {
					run: async (branchCtx) =>
						await branchCtx.step("fetch-orders", (_branchCtx) =>
							fetchCount("/orders"),
						),
				},
				invoices: {
					run: async (branchCtx) =>
						await branchCtx.step("fetch-invoices", (_branchCtx) =>
							fetchCount("/invoices"),
						),
				},
			});
			await loopCtx.step("merge-results", async (step) => {
				step.state.total = joined.users + joined.orders + joined.invoices;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function fetchCount(path: string): Promise<number> {
	const res = await fetch(`https://api.example.com${path}`);
	if (!res.ok) throw new Error(`fetch ${path} failed: ${res.status}`);
	return (
		(await res.json()) as {
			count: number;
		}
	).count;
}
export const registry = setup({ use: { fanInOutActor } });

Scatter-gather across Actors

Use this when multiple Actors can process independent parts of a request in parallel, then return a merged response.

import {
	actor,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type ScatterMessage = {
	input: number;
};
export const shardActor = actor({
	actions: {
		compute: async (_c, input: number) => input * 10,
	},
});
export const scatterGatherActor = workflow({
	state: {
		lastSum: 0,
	},
	run: async (ctx) => {
		await ctx.loop("scatter-gather-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-scatter", {
				timeout: 30000,
			});
			if (!message) return;
			const scatter = message.body as ScatterMessage;
			const gathered = await loopCtx.join("gather", {
				shardA: {
					run: async (joinCtx) =>
						await joinCtx.step("call-shard-a", async (step) =>
							callShard(step, "a", scatter.input),
						),
				},
				shardB: {
					run: async (joinCtx) =>
						await joinCtx.step("call-shard-b", async (step) =>
							callShard(step, "b", scatter.input),
						),
				},
				shardC: {
					run: async (joinCtx) =>
						await joinCtx.step("call-shard-c", async (step) =>
							callShard(step, "c", scatter.input),
						),
				},
			});
			await loopCtx.step("aggregate", async (step) => {
				step.state.lastSum =
					gathered.shardA + gathered.shardB + gathered.shardC;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function callShard(
	ctx: WorkflowStepContextOf<typeof scatterGatherActor>,
	shardId: "a" | "b" | "c",
	input: number,
): Promise<number> {
	const client = ctx.client();
	const handle = client.shardActor.getOrCreate([shardId]);
	return await handle.compute(input);
}
export const registry = setup({ use: { scatterGatherActor, shardActor } });

Poll and backoff loop

Use this when an external dependency has variable availability and retries should slow down after failures.

import { setup, workflow } from "@rivet-dev/workflows";

async function pollExternal(attempt: number): Promise<boolean> {
	return attempt % 3 === 0;
}
export const pollBackoffActor = workflow({
	state: {
		attempts: 0,
		backoffMs: 100,
		status: "unknown" as "unknown" | "healthy" | "retrying",
	},
	run: async (ctx) => {
		await ctx.loop("poll-loop", async (loopCtx) => {
			const success = await loopCtx.step("poll-target", async (step) => {
				step.state.attempts += 1;
				return pollExternal(step.state.attempts);
			});
			if (success) {
				await loopCtx.step("reset-backoff", async (step) => {
					step.state.status = "healthy";
					step.state.backoffMs = 100;
				});
				await loopCtx.sleep("healthy-interval", 1000);
				return;
			}
			const retryDelay = await loopCtx.step("grow-backoff", async (ctx) => {
				ctx.state.status = "retrying";
				ctx.state.backoffMs = Math.min(ctx.state.backoffMs * 2, 5000);
				return ctx.state.backoffMs;
			});
			await loopCtx.sleep("retry-delay", retryDelay);
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { pollBackoffActor } });

Coordinator to worker RPC

Use this when one Actor coordinates work by calling actions on other Actors.

import {
	actor,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type TaskMessage = {
	taskId: string;
	workerId: string;
	value: number;
};
export const workerActor = actor({
	actions: {
		runTask: async (_c, value: number) => value * 2,
	},
});
export const coordinatorActor = workflow({
	state: {
		lastTaskId: null as string | null,
		lastResult: 0,
	},
	run: async (ctx) => {
		await ctx.loop("orchestrator-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-task", {
				timeout: 30000,
			});
			if (!message) return;
			const task = message.body as TaskMessage;
			const result = await loopCtx.step("dispatch-rpc", async (step) =>
				dispatchTask(step, task),
			);
			await loopCtx.step("record-result", async (step) => {
				step.state.lastTaskId = task.taskId;
				step.state.lastResult = result as number;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function dispatchTask(
	ctx: WorkflowStepContextOf<typeof coordinatorActor>,
	task: TaskMessage,
): Promise<number> {
	const client = ctx.client();
	const worker = client.workerActor.getOrCreate([task.workerId]);
	return await worker.runTask(task.value);
}
export const registry = setup({ use: { coordinatorActor, workerActor } });

Child workers

Use this when one workflow coordinates many child workers (Actors or worker workflows) and manages their lifecycle.

import {
	actor,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type BatchMessage = {
	payload: number;
};
export const childWorkerActor = actor({
	actions: {
		process: async (_c, payload: number) => payload * 3,
	},
});
export const orchestratorActor = workflow({
	state: {
		lastTotal: 0,
	},
	run: async (ctx) => {
		await ctx.step("start-children", async (step) => startChildren(step));
		await ctx.loop("orchestrate-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-batch", {
				timeout: 30000,
			});
			if (!message) return;
			const batch = message.body as BatchMessage;
			const results = await loopCtx.join("collect-updates", {
				a: {
					run: async (joinCtx) =>
						await joinCtx.step("run-child-a", async (step) =>
							runChildWorker(step, "child-a", batch.payload),
						),
				},
				b: {
					run: async (joinCtx) =>
						await joinCtx.step("run-child-b", async (step) =>
							runChildWorker(step, "child-b", batch.payload),
						),
				},
				c: {
					run: async (joinCtx) =>
						await joinCtx.step("run-child-c", async (step) =>
							runChildWorker(step, "child-c", batch.payload),
						),
				},
			});
			await loopCtx.step("reconcile", async (step) => {
				step.state.lastTotal = results.a + results.b + results.c;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function startChildren(
	ctx: WorkflowStepContextOf<typeof orchestratorActor>,
): Promise<void> {
	const client = ctx.client();
	await client.childWorkerActor.getOrCreate(["child-a"]).process(0);
	await client.childWorkerActor.getOrCreate(["child-b"]).process(0);
	await client.childWorkerActor.getOrCreate(["child-c"]).process(0);
}
async function runChildWorker(
	ctx: WorkflowStepContextOf<typeof orchestratorActor>,
	workerId: "child-a" | "child-b" | "child-c",
	payload: number,
): Promise<number> {
	const client = ctx.client();
	return await client.childWorkerActor.getOrCreate([workerId]).process(payload);
}
export const registry = setup({ use: { orchestratorActor, childWorkerActor } });

Timeout and fallback Actor

Use this when a primary Actor call might be slow or unavailable and you need a deterministic fallback path.

import {
	actor,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";
export const primaryServiceActor = actor({
	actions: {
		fetchValue: async () => {
			await new Promise((resolve) => setTimeout(resolve, 500));
			return "primary";
		},
	},
});
export const fallbackServiceActor = actor({
	actions: {
		fetchValue: async () => "fallback",
	},
});
export const timeoutFallbackActor = workflow({
	state: {
		lastSource: "none" as "none" | "primary" | "fallback",
		lastValue: "",
	},
	run: async (ctx) => {
		await ctx.loop("timeout-loop", async (loopCtx) => {
			await loopCtx.queue.nextBatch("wait-request", {
				timeout: 30000,
			});
			const winner = await loopCtx.race("primary-vs-timeout", [
				{
					name: "primary",
					run: async (raceCtx) =>
						await raceCtx.step("call-primary", async (step) =>
							callPrimaryValue(step),
						),
				},
				{
					name: "timeout",
					run: async (raceCtx) => {
						await raceCtx.sleep("primary-timeout", 200);
						return "timeout";
					},
				},
			]);
			let value = winner.value as string;
			let source: "primary" | "fallback" = "primary";
			if (winner.winner === "timeout") {
				value = (await loopCtx.step("fallback-call", async (step) =>
					callFallbackValue(step),
				)) as string;
				source = "fallback";
			}
			await loopCtx.step("record-choice", async (step) => {
				step.state.lastSource = source;
				step.state.lastValue = value;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function callPrimaryValue(
	ctx: WorkflowStepContextOf<typeof timeoutFallbackActor>,
): Promise<string> {
	const client = ctx.client();
	const primary = client.primaryServiceActor.getOrCreate(["main"]);
	return await primary.fetchValue();
}
async function callFallbackValue(
	ctx: WorkflowStepContextOf<typeof timeoutFallbackActor>,
): Promise<string> {
	const client = ctx.client();
	const fallback = client.fallbackServiceActor.getOrCreate(["main"]);
	return await fallback.fetchValue();
}
export const registry = setup({
	use: { timeoutFallbackActor, primaryServiceActor, fallbackServiceActor },
});

Failure and recovery

Error hooks

Use onError when you want a best-effort notification for workflow failures.

  • Step failures include the attempt number, retry counts, whether the step will retry, and the next retry delay.
  • Workflow failures also include terminal errors outside steps, such as rollback failures or code/history mismatches.
  • The hook is observational. It is not part of workflow replay, so use it for logging, metrics, or updating non-critical Actor state.
  • This is also a good place to forward workflow failures to Sentry or another error reporting pipeline.
import {
	event,
	setup,
	type WorkflowErrorEvent,
	workflow,
} from "@rivet-dev/workflows";
export const errorHookActor = workflow(
	{
		state: {
			lastError: null as WorkflowErrorEvent | null,
		},
		events: {
			workflowError: event<[WorkflowErrorEvent]>(),
		},
		run: async (ctx) => {
			await ctx.step({
				name: "sync-ledger",
				maxRetries: 3,
				retryBackoffBase: 250,
				retryBackoffMax: 1000,
				run: async (_ctx) => {
					throw new Error("ledger unavailable");
				},
			});
		},
		actions: {
			getState: (c) => c.state,
		},
	},
	{
		onError: (c, event) => {
			c.state.lastError = event;
			c.broadcast("workflowError", event);
		},
	},
);
export const registry = setup({ use: { errorHookActor } });

Rollback

Use rollback checkpoints before steps that have compensating actions.

import { queue, setup, workflow } from "@rivet-dev/workflows";
export const checkoutActor = workflow({
	state: { status: "pending" as string },
	queues: {
		orders: queue<{
			orderId: string;
		}>(),
	},
	run: async (ctx) => {
		await ctx.loop("checkout-loop", async (loopCtx) => {
			const message = await loopCtx.queue.next("wait-order");
			await loopCtx.rollbackCheckpoint("checkout-checkpoint");
			await loopCtx.step<string>({
				name: "reserve-inventory",
				run: (_loopCtx) => reserveInventory(message.body.orderId),
				rollback: async (_rollbackCtx, id) => {
					await releaseInventory(id as string);
				},
			});
			await loopCtx.step<string>({
				name: "charge-card",
				run: (_loopCtx) => chargeCard(message.body.orderId),
				rollback: async (_rollbackCtx, chargeId) => {
					await refundCharge(chargeId as string);
				},
			});
			await loopCtx.step("confirm", async (step) => {
				step.state.status = "confirmed";
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function reserveInventory(orderId: string): Promise<string> {
	const res = await fetch("https://api.example.com/inventory/reserve", {
		method: "POST",
		body: JSON.stringify({ orderId }),
	});
	return (
		(await res.json()) as {
			reservationId: string;
		}
	).reservationId;
}
async function releaseInventory(reservationId: string): Promise<void> {
	await fetch(`https://api.example.com/inventory/${reservationId}/release`, {
		method: "POST",
	});
}
async function chargeCard(orderId: string): Promise<string> {
	const res = await fetch("https://api.stripe.com/v1/charges", {
		method: "POST",
		headers: { Authorization: `Bearer ${process.env.STRIPE_KEY}` },
		body: JSON.stringify({ orderId }),
	});
	return (
		(await res.json()) as {
			id: string;
		}
	).id;
}
async function refundCharge(chargeId: string): Promise<void> {
	await fetch("https://api.stripe.com/v1/refunds", {
		method: "POST",
		headers: { Authorization: `Bearer ${process.env.STRIPE_KEY}` },
		body: JSON.stringify({ charge: chargeId }),
	});
}
export const registry = setup({ use: { checkoutActor } });

Cross-Actor saga (compensating actions)

Use this when a workflow spans multiple Actors and each side effect may need compensation.

import {
	actor,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type CheckoutMessage = {
	orderId: string;
	amount: number;
};
export const inventoryActor = actor({
	actions: {
		reserve: async (_c, orderId: string) => `reserve-${orderId}`,
		release: async (_c, reservationId: string) => reservationId,
	},
});
export const billingActor = actor({
	actions: {
		charge: async (_c, amount: number) => `charge-${amount}`,
		refund: async (_c, chargeId: string) => chargeId,
	},
});
export const checkoutSagaActor = workflow({
	state: {
		completedOrders: 0,
	},
	run: async (ctx) => {
		await ctx.loop("checkout-loop", async (loopCtx) => {
			const [message] = await loopCtx.queue.nextBatch("wait-order", {
				timeout: 30000,
			});
			if (!message) return;
			const checkout = message.body as CheckoutMessage;
			await loopCtx.rollbackCheckpoint("checkout-saga");
			await loopCtx.step({
				name: "reserve-inventory",
				run: async (ctx) => reserveInventoryForCheckout(ctx, checkout.orderId),
				// Rollback callbacks only receive a rollback context, not actor
				// APIs like client(). Compensate with direct external calls.
				rollback: async (_rollbackCtx, output) => {
					await releaseInventoryForCheckout(output as string);
				},
			});
			await loopCtx.step({
				name: "charge-card",
				run: async (ctx) => chargeCheckout(ctx, checkout.amount),
				rollback: async (_rollbackCtx, output) => {
					await refundCheckout(output as string);
				},
			});
			await loopCtx.step("mark-complete", async (step) =>
				markOrderComplete(step),
			);
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
async function reserveInventoryForCheckout(
	ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
	orderId: string,
): Promise<string> {
	const client = ctx.client();
	const inventory = client.inventoryActor.getOrCreate(["main"]);
	return await inventory.reserve(orderId);
}
async function releaseInventoryForCheckout(
	reservationId: string,
): Promise<void> {
	await fetch("https://api.example.com/inventory/release", {
		method: "POST",
		body: JSON.stringify({ reservationId }),
	});
}
async function chargeCheckout(
	ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
	amount: number,
): Promise<string> {
	const client = ctx.client();
	const billing = client.billingActor.getOrCreate(["main"]);
	return await billing.charge(amount);
}
async function refundCheckout(chargeId: string): Promise<void> {
	await fetch("https://api.example.com/billing/refund", {
		method: "POST",
		body: JSON.stringify({ chargeId }),
	});
}
function markOrderComplete(
	ctx: WorkflowStepContextOf<typeof checkoutSagaActor>,
): void {
	ctx.state.completedOrders += 1;
}
export const registry = setup({
	use: { checkoutSagaActor, inventoryActor, billingActor },
});

Versioning and migrations

Versioned workflow evolution

Use this when workflow structure changes across deployments and old histories must still replay.

import { setup, workflow } from "@rivet-dev/workflows";
export const versionedWorkflowActor = workflow({
	state: {
		runs: 0,
	},
	run: async (ctx) => {
		await ctx.step("validate-v2", async (step) => {
			step.state.runs += 1;
		});
		await ctx.removed("validate-v1", "step");
		await ctx.loop("main-loop-v2", async (loopCtx) => {
			await loopCtx.sleep("idle", 500);
			await loopCtx.step("heartbeat-v2", async (step) => {
				step.state.runs += 1;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { versionedWorkflowActor } });

Version gates with getVersion

Use ctx.getVersion(name, latest) to branch behavior when you change a workflow’s logic while old instances are still in flight. It returns the version this instance is pinned to at that point:

  • A fresh instance resolves to latest.
  • An instance that already executed past this point under older code resolves to 1 (the implicit floor).

The resolved version is recorded in history, so replays are deterministic and each instance stays on the branch it started on. Inside a loop, every iteration resolves independently, so in-flight iterations finish on the old branch while new iterations pick up latest.

import { setup, workflow } from "@rivet-dev/workflows";
export const versionGateActor = workflow({
	state: {
		processed: 0,
	},
	run: async (ctx) => {
		await ctx.loop("process-loop", async (loopCtx) => {
			// Gate the changed code path. Each loop iteration resolves its version
			// independently: an iteration that already ran under the old code
			// (in-flight across the deploy) resolves to version 1, while a fresh
			// iteration resolves to `latest` (2 here). The resolved value is pinned
			// in history, so replays stay deterministic.
			const version = await loopCtx.getVersion("process-message", 2);
			if (version === 1) {
				// Preserve the original behavior for in-flight iterations.
				await loopCtx.step("process-v1", async (step) => {
					step.state.processed += 1;
				});
			} else {
				// New behavior for iterations that begin after this deploy.
				await loopCtx.step("process-v2", async (step) => {
					step.state.processed += 1;
				});
			}
			await loopCtx.sleep("idle", 1000);
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { versionGateActor } });

latest must be an integer >= 1, and the gate name must be unique within its scope like any other entry. Once every old instance has drained, retire the gate by replacing the call with ctx.removed(name, "version_check").

Migrations

  • Keep workflow entry names stable once deployed.
  • If an old entry was removed or renamed, call ctx.removed(name, originalType).
  • To change behavior at a point while old instances are still running, gate it with ctx.getVersion(name, latest) (see Version gates).
  • This keeps replay compatible across deployments.

Patterns

These are common workflow shapes used in production systems.

Store workflow progress in state and broadcast

Store progress in state so replay and recovery always restore it. Broadcast state changes so clients can render progress in realtime.

Cron (queue-driven)

Rivet scheduling triggers actions. For cron-like workflows, use a small scheduled action as a bridge that enqueues work, then process that work in the workflow loop.

import {
	queue,
	type ScheduledFireInfo,
	setup,
	workflow,
} from "@rivet-dev/workflows";
export const cronActor = workflow({
	state: {
		runs: 0,
		lastRunAt: null as number | null,
	},
	queues: {
		"cron-tick": queue<{
			scheduledAt: number;
		}>(),
	},
	onCreate: async (c) => {
		await c.cron.every({
			name: "workflow-tick",
			interval: 60000,
			action: "enqueueCronTick",
			args: [],
			maxHistory: 100,
		});
	},
	actions: {
		enqueueCronTick: async (c, fire: ScheduledFireInfo) => {
			await c.queue.send("cron-tick", { scheduledAt: fire.scheduledAt });
		},
		getState: (c) => c.state,
	},
	run: async (ctx) => {
		await ctx.loop("cron-loop", async (loopCtx) => {
			const message = await loopCtx.queue.next("wait-cron-tick");
			await loopCtx.step("run-cron-job", async (step) => {
				step.state.runs += 1;
				step.state.lastRunAt = message.body.scheduledAt;
			});
		});
	},
});
export const registry = setup({ use: { cronActor } });

Setup and teardown around a loop

Use this when you need one-time initialization before a long-lived loop, plus cleanup when the Actor stops sleeping or is destroyed.

import { setup, workflow } from "@rivet-dev/workflows";

function openResource(): string {
	return "connected";
}
function closeResource(_resource: string): void {}
export const setupRunTeardownActor = workflow({
	vars: {
		resource: null as string | null,
	},
	state: {
		initialized: false,
		ticks: 0,
	},
	onWake: (c) => {
		c.vars.resource = openResource();
	},
	onSleep: (c) => {
		if (!c.vars.resource) return;
		closeResource(c.vars.resource);
		c.vars.resource = null;
	},
	run: async (ctx) => {
		await ctx.step("setup", async (step) => {
			if (!step.vars.resource) step.vars.resource = openResource();
			step.state.initialized = true;
		});
		await ctx.loop("main-loop", async (loopCtx) => {
			await loopCtx.sleep("tick", 1000);
			await loopCtx.step("tick-step", async (step) => {
				step.state.ticks += 1;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { setupRunTeardownActor } });

Reference

Debugging

  • GET /inspector/workflow-history returns workflow history status for an Actor.
  • Response includes isWorkflowEnabled and history.
  • In non-dev mode, inspector endpoints require authorization.

Recommendations

  • Prefer queue-driven loops for long-lived workflows.
  • Structure long-lived workflows with setup and teardown around the main loop.
  • Keep Actor state changes and side effects inside steps.
  • Store workflow progress in state and broadcast updates as progress changes.
  • Use timeouts and rollback for external side effects.
  • Write step bodies idempotently. state and vars mutations from a failed attempt are never rolled back, whether the step retries or tryStep/try catches the failure.

Deploy

In development, Rivet runs a local control plane for you and keeps Actor data on your machine.

To scale Rivet in production, pick how much of it you want to run yourself:

Fully managed

Rivet runs the control plane and your nodes. Nothing to operate.

Bring your own compute

Rivet runs the control plane. Your nodes run on your own infrastructure.

Full self-hosting

You run the control plane and your nodes. No dependency on Rivet Cloud.

If you are running your own nodes, follow the guide for your hosting provider:

Docker Compose

Kubernetes

Virtual Machine

Hetzner

AWS ECS

Railway

Render

Custom Platform

Edit this page Last updated November 10, 2026