# Deprecated Queues

> **WARNING:** 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`.
- The `maxQueueSize` and `maxQueueMessageSize` actor options.
- `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.
- `POST /gateway/{actor}/queue/{name}` on the HTTP API.

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

examples/docs/actors-deprecated-queues/before.ts:

```ts
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](/actors/docs/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](/actors/docs/sqlite) that the actor drains. See [Build a durable job queue](#build-a-durable-job-queue).
- **Durable, replayable streams** can use [Durable Streams](/durable-streams/docs). 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`.

examples/docs/actors-deprecated-queues/actions.ts:

```ts
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:

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

examples/docs/actors-deprecated-queues/mini-queue.ts:

```ts
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.

examples/docs/actors-deprecated-queues/drain-legacy.ts:

```ts
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.

While a workflow waits on `ctx.queue.next`, the actor can still sleep. The workflow resumes when the actor wakes.

## Reference

These details apply to queues until you migrate.

### Limits

| Name | Default | Description |
| --- | --- | --- |
| Max queue size | 1,000 messages | Maximum number of messages in the queue before new messages are rejected. Configurable with the `maxQueueSize` actor option. |
| Max queue message size | 64 KiB | Maximum size of each message. Configurable with the `maxQueueMessageSize` actor option. The usable payload is slightly smaller after serialization overhead. Messages sent over HTTP are also limited by `maxIncomingMessageSize`. |

### `canPublish`

`queues.<name>.canPublish` allows or denies an inbound publish. It can be async and must return a boolean: `true` allows the message, and `false` rejects it with `forbidden`. Returning `undefined`, `null`, or any other non-boolean throws an internal error. Return `false` unless the caller is explicitly allowed.

`canPublish` applies only to queue names declared in `queues`. Messages for undeclared queues are ignored, and the publish still reports as completed.

### Send over HTTP

`POST /gateway/{actor}/queue/{name}` sends a message to the queue `name`. `{actor}` accepts the same selectors as the other gateway endpoints. See [Actor Routing](/docs/api/actor-routing) and [Authentication](/docs/api/authentication).

```bash
curl -X POST "https://rivet.example.com/gateway/counter/queue/increment?rvt-namespace=$RIVET_NAMESPACE&rvt-method=getOrCreate&rvt-key=my-counter&rvt-pool=default" \
  -H "x-rivet-token: $RIVET_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"body": {"amount": 1}, "wait": true, "timeout": 5000}'
```

| Field | Type | Required | Description |
| --- | --- | --- | --- |
| `body` | `any` | Yes | Message payload. Any JSON value. |
| `wait` | `boolean` | No | Wait for the actor to complete the message and return its response. |
| `timeout` | `integer` | No | How long to wait for completion, in milliseconds. Only used with `wait`. |

By default, the request returns once the message is enqueued. With `wait: true`, it waits for `message.complete(...)`, and a timeout returns `status: "timedOut"` without canceling the queued work. The response always includes `status`, and includes `response` when the actor passed a value to `complete()`:

```json
{ "status": "completed", "response": { "count": 1 } }
```