Skip to main content
Sign In
Concepts

Queues & Messaging

Coordinate workflows with queue messages, request and response patterns, drains, signals, and durable loops that resume after restarts.

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 + 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 } });
Edit this page