website/src/content/posts/2026-02-25-queues-for-rivet-actors/page.mdx
Today we're releasing Queues for Rivet Actors: per-actor durable queues with a programmable run handler. Also known as the actor mailbox pattern.
run is a long-lived async function you control, not a callback. Selectively consume named queues, race messages against each other, cancel work mid-flight.Define typed queues, process them in a run handler, and send messages from a client.
Sending messages from a client:
import { createClient } from "rivetkit/client";
import type { registry } from "./actors";
const client = createClient<typeof registry>();
const handle = client.counter.getOrCreate(["main"]);
// Fire-and-forget
await handle.send("increment", { amount: 1 });
// Wait for a typed response
const result = await handle.send(
"increment",
{ amount: 5 },
{ wait: true, timeout: 5_000 },
);
if (result.status === "completed") {
console.log(result.response); // { value: 6 }
} else if (result.status === "timedOut") {
console.log("timed out");
}
The run handler is the heart of an actor. It's a long-lived async function that owns the actor's main processing. Instead of registering callbacks, you write it yourself: iterate queues, sleep between ticks, race signals against each other. You control exactly how and when messages are consumed.
Queues are a natural fit for AI agents. Use a prompt queue for incoming messages, a stop queue for cancellation, and SQLite for persistent chat history. The run handler processes messages durably, so the agent survives crashes and picks up where it left off.
<CodeGroup> ```ts Simple Agent import { actor, queue, setup } from "rivetkit"; import { db } from "rivetkit/db"; import { generateText, tool } from "ai"; import { openai } from "@ai-sdk/openai"; import { z } from "zod";const agent = actor({
// SQLite for persistent chat history
db: db({
onMigrate: async (db) => {
await db.execute( CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, role TEXT NOT NULL, content TEXT NOT NULL ) );
},
}),
queues: {
prompt: queue<{ content: string }>(),
},
run: async (c) => {
for await (const message of c.queue.iter()) {
// Save user message to SQLite
await c.db.execute(
"INSERT INTO messages (role, content) VALUES (?, ?)",
"user",
message.body.content,
);
// Load full chat history
const history = (await c.db.execute(
"SELECT role, content FROM messages ORDER BY id",
)) as { role: string; content: string }[];
// Generate with tool use
const result = await generateText({
model: openai("gpt-5"),
messages: history.map((row) => ({
role: row.role as "user" | "assistant",
content: row.content,
})),
tools: {
getWeather: tool({
description: "Get the weather for a location",
parameters: z.object({ location: z.string() }),
execute: async ({ location }) => `72F in ${location}`,
}),
},
maxSteps: 5,
});
// Save assistant response
await c.db.execute(
"INSERT INTO messages (role, content) VALUES (?, ?)",
"assistant",
result.text,
);
}
}, });
export const registry = setup({ use: { agent } });
```ts Cancellable Agent
import { actor, queue, setup } from "rivetkit";
import { db } from "rivetkit/db";
import { generateText, tool } from "ai";
import { openai } from "@ai-sdk/openai";
import { joinSignals } from "rivetkit/utils";
import { z } from "zod";
const agent = actor({
db: db({
onMigrate: async (db) => {
await db.execute(`
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
role TEXT NOT NULL,
content TEXT NOT NULL
)
`);
},
}),
queues: {
// Separate queues for prompts and stop signals
prompt: queue<{ content: string }, undefined>(),
stop: queue<{ reason?: string }>(),
},
run: async (c) => {
for await (const message of c.queue.iter({
names: ["prompt"],
completable: true,
})) {
const stopController = new AbortController();
const runSignal = joinSignals(c.abortSignal, stopController.signal);
// Cancel generation if a stop message arrives
c.queue
.next({ names: ["stop"], signal: runSignal })
.then((stopMessage) => {
if (stopMessage) stopController.abort();
})
.catch(() => {});
// Save user message to SQLite
await c.db.execute(
"INSERT INTO messages (role, content) VALUES (?, ?)",
"user",
message.body.content,
);
// Load full chat history
const history = (await c.db.execute(
"SELECT role, content FROM messages ORDER BY id",
)) as { role: string; content: string }[];
// Generate with tool use
const result = await generateText({
model: openai("gpt-5"),
messages: history.map((row) => ({
role: row.role as "user" | "assistant",
content: row.content,
})),
tools: {
getWeather: tool({
description: "Get the weather for a location",
parameters: z.object({ location: z.string() }),
execute: async ({ location }) => `72F in ${location}`,
}),
},
maxSteps: 5,
abortSignal: runSignal,
}).finally(() => {
stopController.abort();
});
// Save assistant response
await c.db.execute(
"INSERT INTO messages (role, content) VALUES (?, ?)",
"assistant",
result.text,
);
await message.complete();
}
},
});
export const registry = setup({ use: { agent } });
Three delivery modes depending on what the caller needs:
Feed queue messages into durable workflows. Each workflow step is checkpointed, so crashes pick up where they left off. Combine queues with sleep, join, race, rollback, and human-in-the-loop patterns.
<CodeSnippet file="examples/docs/blog-2026-02-25-queues-for-rivet-actors/workflow.ts" />Queues are part of the actor, not a separate service. The same actor has state, SQLite, events, and workflows, all built in. No external broker to provision, no connection strings, no infrastructure to manage.
Plus everything else that comes with Rivet Actors: scale to millions of instances, scale to zero, TypeScript-native, deploy on Cloudflare Workers, Vercel, Railway, or your own infra.
Queues are available today in RivetKit.
npm install rivetkit
import { queue } from "rivetkit";