Advanced
Change a task payload
Change the Zod payload schema of a VitNode queue task without failing queued rows, and see what happens to a stored payload the new schema rejects.
Queued tasks are durable. A row in core_queue keeps the JSON it was dispatched with, and the worker parses it with the task's current schema right before the handler runs. So a schema change also applies to rows queued by the previous version of your plugin. This guide adds a field to the announce-note task without failing the rows already waiting.
Before you begin
- The
announce-notetask from Queue, registered in a top-level module'squeueTasks.
Rules for a compatible change
| Change | Safe for queued rows? | Do instead |
|---|---|---|
Add a field with .optional() or .default() | Yes | - |
| Remove a field | Yes | z.object drops unknown keys, so old rows still parse |
| Add a required field | No | Give it a .default(), or make it optional and handle undefined |
Narrow a field, such as .min(3) on a string | No | Accept the old values, or drain the queue first |
| Rename the task, or move it to another plugin | No | Keep name and the plugin id. Rows are matched to the handler by both |
| Change what the payload means | No | Register a new task with a new name, and keep the old one until its rows are gone |
Add an optional field
Give the task a channel that old rows do not have. .default() fills it in when the worker parses a row queued before the change:
export const announceNoteTask = buildQueueTask({
name: "announce-note",
payload: z.object({
noteId: z.number().int().positive(),
channel: z.enum(["log", "email"]).default("log"),
}),
maxAttempts: 5,
handler: async (c, { noteId, channel }) => {
// channel is "log" for rows queued before this change
},
});Keep name exactly as it was. The change takes effect when the API restarts; nothing is migrated in the database, and rows keep their JSON.
Dispatch the new field
dispatch accepts the schema's input, so channel is optional where you queue the task:
await c
.get("queue")
.dispatch(announceNoteTask, { noteId: id, channel: "email" });Check the result
-
Build the plugin (a running
pnpm devis enough), then runpnpm vitnode inspect jobsinside it. Theannounce-noterow showspayload schemain the Contract column andsite-notesas its Owner. The command never prints payloads. -
Dispatch the task and run the processor, as in Check the result. It completes as before.
-
Optional: insert a row the schema rejects, with
noteIdas a string. With the Docker database from Setup, run this from the workspace root:docker compose -p my-vitnode-vitnode-dev exec -T database psql -U root -d vitnode -c "INSERT INTO core_queue (\"pluginId\", name, \"maxAttempts\", payload) VALUES ('site-notes', 'announce-note', 5, '{\"noteId\": \"3\"}');"After the next processor run, Debug Panel ā Queue shows it as failed with
1/5attempts andInvalid payload: payload.noteId: invalid_type (expected number).
What happens to an invalid stored payload
The worker parses each row with the task's schema right before the handler runs. When the schema rejects it, the row becomes failed at once, without using the remaining attempts or the backoff, because retrying cannot change stored data. The row is kept, so you can find it in the Debug Panel and dispatch a corrected task. The error names each issue's path and code, never the value, so a payload that holds an email address does not leak into logs.
A schema that throws, for example an async refinement whose database query fails, is not a validation failure: the attempt fails and is retried like any handler error.
A row whose task is no longer registered at all, for example after a rename, is retried with No handler registered for queue task "site-notes:announce-note" until it runs out of attempts.
Ownership and registration
A task belongs to the plugin whose bound builder created it. buildApiPlugin throws QueueTaskRegistrationError when another plugin registers it, when an entry in queueTasks is not a task definition, or when two different tasks of one plugin share a name. dispatch(task, payload) throws the same error when the task is not registered in the running API, so a dispatch can never queue work nothing would run.
Related
- Queue reference: every option, type and error.
- Queue: retries, leases, deduplication and the worker.