queue

Work queue: a limit on how many tasks run together, retries with growing backoff and a notification on completion. It combines what @convex-dev/workpool and @convex-dev/action-retrier do, on top of ctx.scheduler.

Enabling

// convex/schema.ts
import { queueTables } from "notusbase-modules/queue/schema";
export default defineSchema({ ...queueTables, /* yours */ });

// convex/queue.ts
export * from "notusbase-modules/queue";

Using

import { Queue, vOnCompleteArgs } from "notusbase-modules/queue";
import { internal } from "./_generated/api";

const emails = new Queue("emails", { maxParallelism: 5, retry: true });

export const send = mutation({
  args: { to: v.string() },
  handler: async (ctx, { to }): Promise<string> => {
    return emails.enqueueAction(ctx, internal.email.send, { to }, {
      onComplete: internal.email.sent,
      context: { to },
    });
  },
});

export const sent = internalMutation({
  args: vOnCompleteArgs(v.object({ to: v.string() })),
  handler: async (ctx, { workId, context, result }) => {
    if (result.kind === "failed") console.error(context.to, result.error);
  },
});

Constructor options: maxParallelism (default 10), retry, sweepMs (default 30 s, sweep for lost tasks), retentionMs (default 1 day, how long a finished task stays in the table). Parallelism belongs to the queue: use the same value everywhere that names it, because the last call's value wins.

enqueue* options: retry, onComplete, context, runAfter, runAt.

retry is true ({ maxAttempts: 5, initialBackoffMs: 250, base: 2 }), false or an object with those fields, counting the first execution. The wait is initialBackoffMs * base^(failures - 1), multiplied by a number between 0.5 and 1.5. Without retry, it runs once.

onComplete receives { workId, context, result }, with result as { kind: "success", returnValue }, { kind: "failed", error } or { kind: "canceled" }.

Enqueueing is part of the caller's transaction: if the mutation rolls back, the task never existed.

Guarantees

• A mutation runs once, in the same transaction that marks the task as done. One that fails undoes everything it did.
• An action without retry runs at most once. If the server dies midway, the sweep marks it failed.
• An action with retry runs at least once: a crash counts as a failure and it runs again, even if it had already done the work. It has to be idempotent.
• cancel on an action in progress does not interrupt it.
• The notification is scheduled in the transaction that ends the task. If it fails or no longer exists, the task stays finished, and the error stays in the log and in _scheduled_functions.
• A task that ends failed (attempts ran out, or it was lost) leaves one line in the failure log and one failure on the functions screen, under queue:reportFailure: queue <name>: task <id> (<function>) failed for good after N attempts: <error>. That function only throws: the mutation that gives up has to commit, so it schedules the report. The line has the task's function and the first 300 characters of its error, which is text the task's code wrote and may carry whatever that code put in it; queue.status has the whole error. The arguments are never in the line.

Disabling

Remove convex/queue.ts, the Queue instances in the code and ...queueTables from the schema. The queue_tasks and queue_state tables stay. With the module gone, the queue's jobs fail because they cannot find the function, and pending tasks wait. When you enable it again, they run when the first new task is enqueued.

How it differs from the original

• new Queue("name", options), without components.workpool. Several queues with different names coexist in the same tables.
• One class for both. enqueueQuery, enqueueActionBatch, enqueueMutationBatch and pause do not exist. To enqueue in bulk, call enqueue* in a loop, in the same mutation.
• The function goes in by name (getFunctionName), because NotusBase has no createFunctionHandle. Renaming a function with pending tasks makes those tasks fail.
• retry instead of retryActionsByDefault, and it applies to mutations if you ask for it.
• No queue partitioned into segments nor logLevel. Dispatch uses one row per queue in queue_state, so enqueueing is serial on the same queue; different queues do not get in each other's way. The limit is exact.
• The Action Retrier's action timeout, heartbeat and waitForAction do not exist. Whoever wants to wait writes the onComplete.
• A lost task is found by the job's state in _scheduled_functions, every sweepMs, and not by heartbeat.

Depends on the scheduler

Everything here goes through jobs.go:

• A mutation job that fails because the function is missing (old bundle with a new binary, or the module removed) does not come back. The queue recovers on the next enqueue, after 2 * sweepMs, or with a manual dispatch: notusbase run queue:dispatch '{"queue":"emails"}'.
• The scheduler limits simultaneous actions to 256, a fixed number. A queue with a larger maxParallelism waits in the scheduler.
• Jobs stay 7 days in the table. A running task with a deleted job counts as lost, which would only happen to a task stuck for longer than that.