Deprecated

Queues

Migrate actors off the deprecated queue APIs to actions and SQLite.

Queues were deprecated in 3.0 and will be removed in a future version. They still work today, so you can migrate at your own pace.

What is deprecated

Queues delivered durable messages to an actor’s run loop. The following are deprecated:

The queues actor option, the queue() schema helper, and canPublish.
c.queue and all of its methods: send, next, nextBatch, tryNext, tryNextBatch, iter, waitForNames, waitForAvailable, complete and enqueueAndWait.
handle.send() and conn.send() on the client.
ctx.queue.send, ctx.queue.next and ctx.queue.nextBatch in workflows.

The Rust SDK no longer has a queue API. Use actions, and store pending work in SQLite with ctx.sql().

import { actor, queue } from "rivetkit";

// An actor that still uses the deprecated `queues` option and `c.queue`
const counter = actor({
  queues: {
    increment: queue<{ amount: number }>(),
  },
  createVars: () => ({ value: 0 }),
  run: async (c) => {
    for await (const message of c.queue.iter()) {
      c.vars.value += message.body.amount;
    }
  },
});

// Client: await handle.send("increment", { amount: 5 });

Use these instead:

Updates and request/reply are plain actions that return a value to the caller. Actions run concurrently, so keep each update atomic in SQLite.
Durable background work is a table in SQLite that the actor drains. See Build a durable job queue.
Durable, replayable streams can use Durable Streams. Waking an actor when a stream receives data is coming soon.

Replace queue messages with actions

Most queues serialized updates to actor data and, with wait: true, returned a result. An action returns its result directly. Actions run concurrently and interleave at every await, so make each update a single SQL statement, as below, or wrap a multi-step update in c.db.transaction.

import { actor } from "rivetkit";
import { db } from "rivetkit/db";

const counter = actor({
  db: db({
    onMigrate: async (db) => {
      await db.execute(`
        CREATE TABLE IF NOT EXISTS counter (
          id INTEGER PRIMARY KEY CHECK (id = 1),
          value INTEGER NOT NULL
        )
      `);
      await db.execute(
        "INSERT OR IGNORE INTO counter (id, value) VALUES (1, 0)",
      );
    },
  }),
  actions: {
    // Actions run concurrently, so the update is a single atomic statement.
    // The write is durable once the action returns.
    increment: async (c, amount: number) => {
      const rows = await c.db.execute<{ value: number }>(
        "UPDATE counter SET value = value + ? WHERE id = 1 RETURNING value",
        amount,
      );
      return rows[0].value;
    },
  },
});

// Client: const value = await handle.increment(5);

Clients call handle.increment(5) instead of handle.send("increment", { amount: 5 }).

Build a durable job queue

When work must survive a crash or sleep and run after the caller returns, store it in SQLite and drain it in the background:

An action inserts a row. The job is durable as soon as the insert resolves.
A drain loop processes rows in order and deletes each one after it succeeds.
c.keepAwake holds the actor awake until the drain finishes.
onWake restarts the drain, so jobs left over from before a sleep or crash still run.

import { actor } from "rivetkit";
import { db, type RawAccess } from "rivetkit/db";

interface Worker {
  // Set when new jobs arrive while a drain is already running
  pending: boolean;
  draining: Promise<void> | null;
}

interface WorkerContext {
  db: RawAccess;
  vars: { worker: Worker };
  keepAwake<T>(promise: Promise<T>): Promise<T>;
  log: { error(fields: Record<string, unknown>): void };
}

// Start a drain unless one is already running. `keepAwake` stops the actor
// from sleeping until every job is processed.
function startDrain(c: WorkerContext): void {
  c.vars.worker.pending = true;
  if (c.vars.worker.draining) return;

  c.vars.worker.draining = c
    .keepAwake(drainJobs(c))
    .catch((error) => {
      // The failed job stays in the table and runs on the next drain
      c.log.error({ msg: "job failed", error });
    })
    .finally(() => {
      c.vars.worker.draining = null;
      // Catch jobs enqueued after the loop last checked `pending`
      if (c.vars.worker.pending) startDrain(c);
    });
}

async function drainJobs(c: WorkerContext): Promise<void> {
  while (c.vars.worker.pending) {
    c.vars.worker.pending = false;

    while (true) {
      const jobs = await c.db.execute<{ id: number; email: string }>(
        "SELECT id, email FROM jobs ORDER BY id LIMIT 1",
      );
      const job = jobs[0];
      if (!job) break;

      await sendWelcomeEmail(job.email);

      // Delete only after the work succeeds, so a crash retries the job
      await c.db.execute("DELETE FROM jobs WHERE id = ?", job.id);
    }
  }
}

const mailer = actor({
  db: db({
    onMigrate: async (db) => {
      await db.execute(`
        CREATE TABLE IF NOT EXISTS jobs (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          email TEXT NOT NULL
        )
      `);
    },
  }),
  createVars: (): { worker: Worker } => ({
    worker: { pending: false, draining: null },
  }),

  // Resume jobs that were still pending when the actor last slept or crashed
  onWake: (c) => startDrain(c),

  actions: {
    // The job is durable once the insert resolves, like a queue send
    enqueue: async (c, email: string) => {
      await c.db.execute("INSERT INTO jobs (email) VALUES (?)", email);
      startDrain(c);
    },
    pendingCount: async (c) => {
      const rows = await c.db.execute<{ count: number }>(
        "SELECT COUNT(*) AS count FROM jobs",
      );
      return rows[0].count;
    },
  },
});

async function sendWelcomeEmail(email: string): Promise<void> {
  console.log("sending welcome email", email);
}

The drain state lives in c.vars, so it resets on every start. The pending flag catches jobs that arrive while a drain is already running. A failed job is logged and left in the table, and the drain stops until the next enqueue or wake. Because a row is deleted only after its work succeeds, a job can run again after a crash, so make the work idempotent.

Move pending messages

Messages that were still waiting in the old queue can be moved into your own table on the next start. Remove the hook once every actor has woken on the new version.

import { actor } from "rivetkit";
import { db } from "rivetkit/db";

const mailer = actor({
  db: db({
    onMigrate: async (db) => {
      await db.execute(`
        CREATE TABLE IF NOT EXISTS jobs (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          email TEXT NOT NULL
        )
      `);
    },
  }),
  onWake: async (c) => {
    // Move messages still waiting in the old queue into the jobs table.
    // Remove this hook once every actor has woken on the new version.
    while (true) {
      const messages = await c.queue.tryNextBatch({ count: 100 });
      if (messages.length === 0) break;

      for (const message of messages) {
        const body = message.body as { email: string };
        await c.db.execute(
          "INSERT INTO jobs (email) VALUES (?)",
          body.email,
        );
      }
    }
  },
  actions: {},
});

A queue message is removed when it is received, so a crash between receiving and inserting it loses that message.

Workflows

ctx.queue.next and ctx.queue.nextBatch are deprecated along with the rest of the queue API. A replacement for delivering messages to a running workflow is coming soon. Until it ships, existing workflows that read from queues keep working.