Queue Tasks

Cron jobs handle recurring work on a schedule. Queue tasks handle one-off background work you enqueue on demand — "send this email later", "process this import", "retry this webhook". Tasks are stored in the database (core_queue) and drained every minute by a built-in cron job, so a slow task never blocks the request that scheduled it.

Queue tasks build on the cron system: the process-queue job runs on the same per-minute tick, so you need a cron adapter configured (for example Node CRON) for tasks to be processed automatically.

How it works

c.get("queue").dispatch({ name, payload })  ─►  INSERT core_queue (status = pending)

cron tick "* * * * *" ─► process-queue job
   1. optional Redis lock — one instance drains per tick when Redis is configured
   2. claim a batch  (UPDATE … status = processing … FOR UPDATE SKIP LOCKED)
   3. run each task's handler, matched by pluginId:name
   4. success → completed · error → retry with backoff → failed after maxAttempts
   5. prune completed/failed rows older than 7 days

Concurrency is safe by default: batches are claimed with Postgres FOR UPDATE SKIP LOCKED, so many instances (or many ticks) can drain the queue at once without ever running the same task twice.

Defining a task

A task is a named handler, registered like a cron job. The handler receives the request context and the JSON payload the task was dispatched with.

import { buildQueueTask } from "@vitnode/core/api/lib/queue";

export const sendDigestTask = buildQueueTask({
  name: "send-digest",
  description: "Send the weekly digest email to a user",
  // Optional — defaults to 3
  maxAttempts: 5,
  handler: async (c, payload) => {
    const { userId } = payload as { userId: number };

    // ...do the work; throwing schedules a retry with backoff
    await c.get("log").debug(`Sending digest to user ${userId}`);
  },
});

import { buildModule } from "@vitnode/core/api/lib/module";
import { CONFIG_PLUGIN } from "@/config";

import { sendDigestTask } from "./tasks/send-digest.task";

export const newsletterModule = buildModule({
  pluginId: CONFIG_PLUGIN.pluginId,
  name: "newsletter",
  routes: [],
  queueTasks: [sendDigestTask],
});

Enqueuing work

Dispatch a task from any route or model with c.get("queue"). The name must match a registered task; the row is picked up on the next cron tick.

export const subscribeRoute = buildRoute({
  pluginId: CONFIG_PLUGIN.pluginId,
  route: {},
  handler: async c => {
    await c.get("queue").dispatch({
      name: "send-digest",
      payload: { userId: c.get("user")?.id },
    });

    return c.json({ queued: true });
  },
});

dispatch options

Retries and backoff

If a handler throws, the task is retried with an exponential backoff (10s, 20s, 40s, … capped at one hour) until maxAttempts is reached, after which its status becomes failed and the error is stored in lastError. A task with no registered handler fails immediately.

Completed and failed rows are pruned automatically after 7 days, so the table does not grow without bound.

Optional: Redis coordination

The queue works with just Postgres. When you run multiple instances, adding Redis lets a single instance drain the queue per tick (an advisory lock), which reduces database contention. Correctness never depends on it — FOR UPDATE SKIP LOCKED already prevents double processing.

There is nothing extra to enable: set REDIS_URL (see the Redis guide) and the queue uses it automatically. Without Redis, every instance simply claims its own disjoint batch.

Monitoring in the AdminCP

• Core → Advanced → Queue Tasks — the full, paginated list of tasks with a status filter, attempt counts, and the last error. Requires the queue:can_view staff permission.
• Debug Panel — a live view of currently active (pending/processing) tasks plus per-status counts.
• Core → System → Integrations — a card summarizing registered task handlers and the number of pending/running tasks.