General
Quickstart
Create your first durable workflow, then add replay-safe steps, queue-driven loops, setup, teardown, and reliable error handling.
Simple workflow
Use this when you need a short multi-step sequence.
import {
setup,
type WorkflowStepContextOf,
workflow,
} from "@rivet-dev/workflows";
export const invoiceActor = workflow({
state: {
invoiceId: null as string | null,
subtotal: 0,
tax: 0,
total: 0,
status: "idle" as "idle" | "complete",
},
run: async (ctx) => {
const subtotal = await ctx.step("load-subtotal", async (_ctx) =>
loadSubtotal(),
);
const tax = await ctx.step("calculate-tax", async (_ctx) =>
calculateTax(subtotal),
);
await ctx.step("save-invoice", async (step) =>
saveInvoice(step, subtotal, tax),
);
},
actions: {
getState: (c) => c.state,
},
});
async function loadSubtotal(): Promise<number> {
const response = await fetch("https://api.example.com/carts/main");
if (!response.ok) {
throw new Error(`load subtotal failed: ${response.status}`);
}
const cart = (await response.json()) as {
subtotal: number;
};
return cart.subtotal;
}
async function calculateTax(subtotal: number): Promise<number> {
const response = await fetch("https://api.example.com/tax/quote", {
method: "POST",
headers: {
"content-type": "application/json",
},
body: JSON.stringify({ subtotal }),
});
if (!response.ok) {
throw new Error(`tax quote failed: ${response.status}`);
}
const quote = (await response.json()) as {
tax: number;
};
return quote.tax;
}
async function saveInvoice(
ctx: WorkflowStepContextOf<typeof invoiceActor>,
subtotal: number,
tax: number,
): Promise<void> {
const total = subtotal + tax;
const response = await fetch("https://api.example.com/invoices", {
method: "POST",
headers: {
"content-type": "application/json",
},
body: JSON.stringify({ subtotal, tax, total }),
});
if (!response.ok) {
throw new Error(`save invoice failed: ${response.status}`);
}
const invoice = (await response.json()) as {
id: string;
};
ctx.state.invoiceId = invoice.id;
ctx.state.subtotal = subtotal;
ctx.state.tax = tax;
ctx.state.total = total;
ctx.state.status = "complete";
}
export const registry = setup({ use: { invoiceActor } });
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.
import {
queue,
setup,
type WorkflowStepContextOf,
workflow,
} from "@rivet-dev/workflows";
export const workflowCounter = workflow({
state: {
value: 0,
processed: 0,
lastOperationId: null as string | null,
},
queues: {
counter: queue<{
delta: number;
}>(),
},
run: async (ctx) => {
await ctx.loop("counter-loop", async (loopCtx) => {
const message = await loopCtx.queue.next("wait-counter-command");
await loopCtx.step("apply-counter-command", async (step) =>
applyCounterCommand(step, message.body.delta),
);
});
},
actions: {
getState: (c) => c.state,
},
});
async function applyCounterCommand(
ctx: WorkflowStepContextOf<typeof workflowCounter>,
delta: number,
): Promise<void> {
const response = await fetch("https://api.example.com/counter/apply", {
method: "POST",
headers: {
"content-type": "application/json",
},
body: JSON.stringify({ delta }),
});
if (!response.ok) {
throw new Error(`counter apply failed: ${response.status}`);
}
const result = (await response.json()) as {
nextValue: number;
operationId: string;
};
ctx.state.value = result.nextValue;
ctx.state.lastOperationId = result.operationId;
ctx.state.processed += 1;
}
export const registry = setup({ use: { workflowCounter } });
import { createClient } from "rivetkit/client";
import type { registry } from "./index";
const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.workflowCounter.getOrCreate(["main"]);
await handle.send("counter", { delta: 1 });
await handle.send("counter", { delta: 2 });
const state = await handle.getState();
console.log(state.value, state.processed);
Setup & teardown
Use this when the workflow should initialize resources, process queued commands, then clean up.
import {
Loop,
queue,
setup,
type WorkflowStepContextOf,
workflow,
} from "@rivet-dev/workflows";
type WorkMessage = {
amount: number;
};
type ControlMessage = {
type: "stop";
reason: string;
};
export const setupRunTeardownActor = workflow({
state: {
phase: "idle" as "idle" | "running" | "stopped",
total: 0,
processed: 0,
stopReason: null as string | null,
workerSessionId: null as string | null,
},
queues: {
work: queue<WorkMessage>(),
control: queue<ControlMessage>(),
},
run: async (ctx) => {
await ctx.step("setup", async (step) => setupWorkerSession(step));
const stopReason = await ctx.loop("worker-loop", async (loopCtx) => {
const message = await loopCtx.queue.next("wait-command", {
names: ["work", "control"],
});
if (message.name === "work") {
const work = message.body as WorkMessage;
await loopCtx.step("apply-work", async (step) =>
applyWorkerMessage(step, work),
);
return;
}
const control = message.body as ControlMessage;
if (control.type === "stop") {
return Loop.break(control.reason);
}
});
await ctx.step("teardown", async (step) =>
teardownWorkerSession(step, stopReason),
);
},
actions: {
getState: (c) => c.state,
},
});
async function setupWorkerSession(
ctx: WorkflowStepContextOf<typeof setupRunTeardownActor>,
): Promise<void> {
const response = await fetch("https://api.example.com/workers/session", {
method: "POST",
});
if (!response.ok) {
throw new Error(`worker setup failed: ${response.status}`);
}
const session = (await response.json()) as {
sessionId: string;
};
ctx.state.workerSessionId = session.sessionId;
ctx.state.phase = "running";
ctx.state.stopReason = null;
}
async function applyWorkerMessage(
ctx: WorkflowStepContextOf<typeof setupRunTeardownActor>,
work: WorkMessage,
): Promise<void> {
const response = await fetch("https://api.example.com/workers/process", {
method: "POST",
headers: {
"content-type": "application/json",
},
body: JSON.stringify({
sessionId: ctx.state.workerSessionId,
amount: work.amount,
}),
});
if (!response.ok) {
throw new Error(`worker process failed: ${response.status}`);
}
const result = (await response.json()) as {
appliedAmount: number;
};
ctx.state.total += result.appliedAmount;
ctx.state.processed += 1;
}
async function teardownWorkerSession(
ctx: WorkflowStepContextOf<typeof setupRunTeardownActor>,
stopReason: string,
): Promise<void> {
if (ctx.state.workerSessionId) {
const response = await fetch(
`https://api.example.com/workers/session/${ctx.state.workerSessionId}`,
{ method: "DELETE" },
);
if (!response.ok) {
throw new Error(`worker teardown failed: ${response.status}`);
}
}
ctx.state.phase = "stopped";
ctx.state.stopReason = stopReason;
}
export const registry = setup({ use: { setupRunTeardownActor } });
import { createClient } from "rivetkit/client";
import type { registry } from "./index";
const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.setupRunTeardownActor.getOrCreate(["main"]);
await handle.send("work", { amount: 5 });
await handle.send("work", { amount: 3 });
await handle.send("control", { type: "stop", reason: "maintenance" });
const state = await handle.getState();
console.log(state.phase, state.total, state.stopReason);
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: