Queue

The queue moves work out of the request. A route stores a task in the core_queue table and answers at once, and a background worker runs it a little later, retrying it if it fails. In this guide the site-notes plugin queues an announcement for a note and you watch the task complete.

Before you beginCopy Anchor Link

• A queue processor. Tasks run only when core's process-queue cron job runs, every minute. Configure a cron adapter or an external trigger. During development you can trigger it by hand, as shown in Check the result.
• The site-notes plugin with its site_notes table and the Create notes permission, from the beginner path up to Permissions. Any plugin with a top-level API module and a guarded route works the same way.

import { eq } from "drizzle-orm";
import { z } from "zod";

import { buildQueueTask } from "@/api/builders";
import { site_notes } from "@/database/notes";

export const announceNoteTask = buildQueueTask({
  name: "announce-note",
  payload: z.object({ noteId: z.number().int().positive() }),
  maxAttempts: 5,
  handler: async (c, { noteId }) => {
    const [note] = await c
      .get("db")
      .select({ title: site_notes.title })
      .from(site_notes)
      .where(eq(site_notes.id, noteId));

    if (!note) return;

    await c.get("log").debug(`Announced note "${note.title}"`);
  },
});

import { buildModule } from "@/api/builders";
import { announceNoteTask } from "@/api/tasks/announce-note.task"; 

import { listNotesRoute } from "./routes/list.route";
import { showNoteRoute } from "./routes/show.route";

export const notesModule = buildModule({
  name: "notes",
  routes: [listNotesRoute, showNoteRoute],
  queueTasks: [announceNoteTask], 
});

import { z } from "@hono/zod-openapi";

import { buildRoute } from "@/api/builders";
import { announceNoteTask } from "@/api/tasks/announce-note.task";
import { staffPermissions } from "@/staff-permissions";

export const announceNoteRoute = buildRoute({
  adminStaffPermission: staffPermissions.admin.notes.canCreate,
  route: {
    method: "post",
    path: "/{id}/announce",
    request: {
      params: z.object({ id: z.coerce.number().int().positive() }),
    },
    responses: {
      202: {
        content: {
          "application/json": {
            schema: z.object({
              deduplicated: z.boolean(),
              taskId: z.number().nullable(),
            }),
          },
        },
        description: "The announcement is queued, or already waiting.",
      },
    },
  },
  handler: async c => {
    const { id } = c.req.valid("param");
    const { deduplicated, id: taskId } = await c
      .get("queue")
      .dispatch(
        announceNoteTask,
        { noteId: id },
        { dedupeKey: `announce-note:${id}` },
      );

    return c.json({ deduplicated, taskId }, 202);
  },
});

Check the resultCopy Anchor Link

1. Add a temporary button that queues an announcement for note 3 with fetcher. Create src/pages/check-announce-note.tsx:

   Copy Text

   import { Button } from "@vitnode/core/components/ui/button";
   import { fetcher } from "@vitnode/core/tanstack/fetcher";
   import { useState } from "react";
   
   import { manifest } from "@/manifest.gen";
   
   export const CheckAnnounceNote = () => {
     const [result, setResult] = useState("");
     const [isPending, setIsPending] = useState(false);
   
     const onClick = async () => {
       setIsPending(true);
       try {
         const response = await fetcher({
           plugin: manifest.id,
           method: "post",
           module: "admin/notes",
           path: "/{id}/announce",
           args: { params: { id: 3 } },
         });
         setResult(
           response.ok
             ? JSON.stringify(await response.json())
             : String(response.status),
         );
       } finally {
         setIsPending(false);
       }
     };
   
     return (
       <div className="flex items-center gap-3">
         <Button isLoading={isPending} onClick={onClick}>
           Announce note 3
         </Button>
         {result ? <output>{result}</output> : null}
       </div>
     );
   };

   Render <CheckAnnounceNote /> in src/pages/notes-page.tsx, sign in to the AdminCP in the same browser, open /site-notes/notes and press the button twice. The first press shows {"deduplicated":false,"taskId":…} with the new row's id. The second shows {"deduplicated":true,"taskId":null}: it is dropped because a task with the same dedupeKey is still waiting.

2. Open Debug Panel → Queue (/admin/core/debug/queue). The announce-note task from site-notes is pending, because nothing has run it yet.

3. Run the processor. With a cron adapter, wait for the next minute. During development you can trigger cron yourself; CRON_SECRET is in your .env:

   Copy Text

   curl -X POST http://localhost:3000/api/@vitnode/core/cron \
     -H "Authorization: Bearer $CRON_SECRET" \
     -H "Content-Type: application/json"

   It answers Cron jobs processed. Executed N jobs.

4. Reload Debug Panel → Queue. The task is completed, with 1/5 attempts. Debug Panel → Log explorer has the debug entry Announced note "New coffee machine". It is listed under @vitnode/core, because the worker runs outside any plugin's route. Remove CheckAnnounceNote when you are done.

Run a task laterCopy Anchor Link

availableAt is the earliest time the task may run. The default is now. This dispatch runs the announcement in an hour at the earliest:

await c.get("queue").dispatch(
  announceNoteTask,
  { noteId: id },
  { availableAt: new Date(Date.now() + 60 * 60 * 1000) },
);

The task still waits for the next processor run after that time. priority (higher first, default 0) orders tasks that are due at the same time.

Retries and backoffCopy Anchor Link

A thrown error fails the attempt. Until maxAttempts is reached, the task returns to pending with an exponential delay. The first retry waits about 10 seconds, and each later one waits twice as long, up to one hour. After the last attempt it is failed, with the error in lastError.

maxAttempts comes from the dispatch, then the task definition, then 3. To wait without using an attempt, for example on a rate limit, throw QueueDeferError from @vitnode/core/api/lib/queue:

throw new QueueDeferError(new Date(Date.now() + 60_000), "Provider rate limit");

A failed task stays failed. Fix the cause and dispatch again.

A stored payload the schema rejects, for example a row queued before an incompatible schema change, fails at once without a retry, because no attempt can fix stored data. See What happens to an invalid stored payload.

Leases and abandoned tasksCopy Anchor Link

A worker that claims a task holds it for the task's leaseSeconds, 600 by default. If the worker dies, or the handler is still running when the lease ends, the task counts as abandoned. It returns to pending, or becomes failed when it has no attempts left, and another run picks it up.

Keep handlers well inside their lease, or raise leaseSeconds on the task. A handler that outlives its lease may run twice at once. For large jobs, save progress and dispatch a follow-up task, as notification fan-out does.

DeduplicationCopy Anchor Link

While a task with the same plugin id and dedupeKey is pending or processing, a new dispatch is dropped and returns { deduplicated: true, id: null }. Once the first task completes or fails, the same key can be queued again. Without a dedupeKey, every dispatch creates a task.

Enqueue inside a transactionCopy Anchor Link

Pass tx to insert the task with the transaction that writes your data, so it is only queued if that write commits:

await c.get("db").transaction(async tx => {
  const [note] = await tx.insert(site_notes).values(values).returning();
  if (!note) throw new Error("Inserting the note returned no row.");

  await c.get("queue").dispatch(announceNoteTask, { noteId: note.id }, { tx });
});

Without tx, a task dispatched before a rollback would still run, for a note that was never stored. Write a complete operation shows the whole pattern, with the cache and events after the commit.

How the worker runsCopy Anchor Link

Each process-queue run claims tasks in batches of 25, runs up to 5 at once, highest priority first, and keeps draining for about 45 seconds. Tasks on the ai queue are drained by core's AI job instead. Completed and failed tasks stay in core_queue for 7 days; the hourly cleanup removes older ones.

Debug Panel → Queue lists every task with its queue, status, attempts, next run, creation time and last error, filterable by status. It is read-only and needs the queue → can_view staff permission.

Every option, type and error is listed in the queue reference.

RelatedCopy Anchor Link

• Change a task payload: evolve a task's schema without breaking queued rows.
• Cron: recurring jobs, and the adapters that run the queue.
• Events: react to something that happened, inside the request.