Documentation / @super-line/plugin-queue

@super-line/plugin-queue

Durable, at-least-once jobs and cluster-wide cron schedules backed by super-line collections.

import * as z from 'zod'
import { defineContract } from '@super-line/core'
import { queue } from '@super-line/plugin-queue'
import { createSuperLineServer } from '@super-line/server'

const queueKit = queue({
  queues: {
    sendEmail: {
      input: z.object({ to: z.email() }),
      result: z.object({ messageId: z.string() }),
      concurrency: 3,
      worker: async ({ to }, { signal }) => {
        const messageId = await sendEmail(to, { signal })
        return { messageId }
      },
    },
  },
})

const contract = defineContract({
  roles: { user: {} },
  plugins: [queueKit.contract],
})

const server = createSuperLineServer(contract, {
  // ...
  plugins: [queueKit.plugin],
})

await queueKit.enqueue('sendEmail', { to: 'hello@example.com' })
await queueKit.schedules.create({
  queue: 'sendEmail',
  cron: '0 9 * * *',
  timezone: 'Europe/Berlin',
  input: { to: 'hello@example.com' },
})

Workers are server-only. queue() is constructed once and returns both the contract fragment and runtime plugin. Concurrency is declarative and enforced with durable slot rows; there is no imperative concurrency control plane. Claims, lease renewals, and completion use atomic conditional batches with a fencing runId.

worker is optional, so a queue can be declared beside the contract and bound where its implementation lives:

queueKit.queue('sendEmail').setWorker(await loadSendEmailWorker())

Binding is per node and takes effect immediately; the last one wins. A node never claims a queue it has not bound — those jobs stay queued for a node that has one rather than failing, so a process that binds nothing can enqueue and observe while its peers execute. queue(name) is also that queue's namespace: hasWorker, enqueue, a scoped list, and scoped schedules.create/list. Job-id operations (get, cancel, retry) stay on the kit.

The memory and SQLite collection backends coordinate one node. Use the PGlite/Postgres collection backend for cluster-wide concurrency and cron scheduling; its conditional batches serialize on central Postgres. The adapter wake channel only reduces latency—durable polling remains the correctness path.

Execution is at least once. Workers should be idempotent and honor WorkerContext.signal. Queue collections have deny-all client policies; server code and the privileged inspector can access them.

@super-line/plugin-queue ​

Guides ​

Classes ​

Interfaces ​

Type Aliases ​

Variables ​

Functions ​

@super-line/plugin-queue

Guides

Classes

Interfaces

Type Aliases

Variables

Functions