Queue Messages

Send durable queue messages to Rivet Actors and consume them from run loops.

Rivet Actors include a pull-based queue for durable background processing.

Send Messages (Client)

Use handle.send(name, body) for fire-and-forget:

const handle = client.worker.getOrCreate(["main"]);

await handle.send("jobs", { id: "job-1" });

Use wait: true for request/response:

const result = await handle.send(
  "jobs",
  { id: "job-1" },
  { wait: true, timeout: 30_000 },
);

if (result.status === "completed") {
  console.log(result.response);
} else {
  console.log("timed out");
}

Queue Schema

Define queue message types under queues. Use complete when a queue supports manual completion responses.

import { actor, queue } from "rivetkit";

const worker = actor({
  state: {},
  queues: {
    jobs: queue<{ id: string }, { ok: true }>(),
    logs: queue<{ line: string }>(),
  },
  actions: {},
});

Receive Messages (Actor)

next

next returns an array and can block until messages are available.

const messages = await c.queue.next({
  names: ["jobs"],
  count: 10,
  timeout: 1000,
  signal: abortController.signal,
});

If no messages arrive before timeout, next returns [].

tryNext

tryNext is non-blocking and immediately returns [] when empty.

const messages = await c.queue.tryNext({ names: ["jobs"], count: 10 });

iter

iter returns an async iterator yielding one message at a time.

for await (const message of c.queue.iter({
  names: ["jobs"],
  signal: abortController.signal,
})) {
  // process message
}

Iterate All Queue Names

Use iter() without names to consume across all queue names.

for await (const message of c.queue.iter()) {
  // process message
}

Completable Messages

Use completable: true to receive messages that expose message.complete(...).

for await (const message of c.queue.iter({ names: ["jobs"], completable: true })) {
  await message.complete({ ok: true });
}

The message stays in the queue until message.complete(...) is called.

Abort Behavior

Use c.aborted for loop exit conditions when needed.

Never wrap c.queue.next(...) in try/catch for normal shutdown handling. Queue receive calls throw special abort errors during actor shutdown so the run handler can stop cleanly.