Skip to main content

Producing jobs

All producing goes through pq itself. Every method takes one options object — with the queue named in it — and returns a result object.

You do not need to declare a queue to produce onto it: addJob addresses it by name and creates its row on first use, which is why a process that only produces, the dashboard and the CLI can all enqueue without knowing anything about how the queue is processed. Declaring a queue with pq.createQueue({ … }) is how a process takes on processing it — see Consuming jobs. Most apps do both from the same process, and that is a fine place to stay; splitting producers from consumers is available when your deployment wants it, not a prerequisite.

Add a job​

addJob names the queue, carries the payload under data, and hands back a result object:

const emailJob = await pq.addJob({
  queue: 'emails',
  data: { to: 'a@b.com', template: 'welcome' },
});

emailJob.jobId; // 'b0f1…' — the row that was inserted
emailJob.status; // JobStatus.PENDING — waiting to be claimed
emailJob.startAfter; // Date — when it becomes eligible (now, unless delayed)
emailJob.deduplicated; // false — no idempotency/singleton key matched

The queue's row is created on first use, so the call above works whether or not this process declared emails — a process that only produces just calls addJob.

Need to enqueue several at once? pq.addJobs({ queue, jobs }) inserts the whole batch in one statement and returns a result per entry, in order. Each entry still carries its own options explicitly — nothing is silently shared between them:

const results = await pq.addJobs({
  queue: 'emails',
  jobs: [{ data: { to: 'a@b.com' } }, { data: { to: 'c@d.com' }, priority: 10 }],
});

addJobs is for one-off jobs only: a recurring entry is rejected, because each recurrence needs its own addJob call to return a scheduleId.

AddJobResult carries jobId, queue, status, deduplicated, startAfter, and — for scheduled adds — scheduleId / nextRunAt.

Everything below is an option on that same addJob call. data is the only required key; the rest are optional and default sensibly.

Scheduling​

All timing lives under the schedule group — pass exactly one mode. Durations are whole seconds:

await pq.addJob({ queue: 'emails', data, schedule: { startAfterSeconds: 86_400 } }); // 24 h, in seconds
await pq.addJob({ queue: 'emails', data, schedule: { startEverySeconds: 30 } }); // recurring interval, in seconds
await pq.addJob({ queue: 'emails', data, schedule: { cron: '0 8 * * 1-5' } }); // recurring cron (UTC)
await pq.addJob({
  queue: 'emails',
  data,
  schedule: { cron: { pattern: '0 8 * * 1-5', tz: 'America/New_York' } },
});
  • schedule.startAfterSeconds — delay a one-off job by N seconds from now.
  • schedule.startEverySeconds — a recurring fixed interval, in seconds between runs.
  • schedule.cron — a recurring cron expression, a bare string (UTC) or { pattern, tz }.

Pick exactly one recurring mode; a one-off delay can't be combined with a recurring mode. Omit schedule to run immediately. For a recurring mode, every option on the add (priority, retry/backoff, expireInSeconds, retention, …) becomes the template stamped onto each spawned run — it shapes the produced jobs, not the schedule. See Delayed & recurring jobs for the full recurring story (dedup, timezones, unscheduling).

Priority​

await pq.addJob({ queue: 'emails', data, priority: 10 }); // higher values are claimed first

priority orders ready jobs — higher values run sooner (default 0). It composes with scheduling: a job becomes eligible at its scheduled start time, then priority breaks ties among the ready set.

Automatic retries​

retry sets the per-job retry policy: how many attempts, the base delay, and the backoff curve.

await pq.addJob({
  queue: 'emails',
  data,
  retry: {
    maxAttempts: 5, // total attempts incl. the first (default 1 = no retry)
    delaySeconds: 2, // base delay before a retry, in seconds
    backoff: {
      strategy: BackoffStrategy.EXPONENTIAL,
      factor: 2,
      maxDelaySeconds: 300,
      jitter: true,
    },
  },
});

A throwing handler is retried per this policy until attempts are exhausted, then the job goes FAILED. Which thrown errors skip retry is a consumer concern (dontAutoretryFor) — see Consuming jobs → error handling. Full field list in the API reference.

Expiry​

await pq.addJob({ queue: 'emails', data, expireInSeconds: 3600 }); // give up if it hasn't started within an hour

expireInSeconds drops the job to EXPIRED if it hasn't started within that many seconds — useful for time-sensitive work that's worthless if delayed. It bounds the wait to start, not the handler's own run time.

Retention​

await pq.addJob({
  queue: 'emails',
  data,
  retention: { ttlSeconds: 604_800 }, // delete 7 days after it finishes
});

retention overrides the queue's default retention for this one job — how long the finished row is kept before automatic pruning. See Retention for queue-level defaults and the trade-offs.

Preferred server​

// Runs only on 8 GB servers that also have 4 cores; '.*' skips whatever sits between the tags.
await pq.addJob({ queue: 'emails', data, preferredInstanceKeyPattern: 'ram-8gb.*cpu-4cor' });

preferredInstanceKeyPattern is a regular-expression source string matched against instance keys — only servers whose instanceKey matches will claim the job. It requires the producing process to have an instanceKey. The match is unanchored (a plain "key contains this" test) and case-sensitive, so a single capability tag like 'ram-8gb' is often the whole pattern you need; add ^/$ when you want to match the key exactly. Prefer broad, capability-based patterns over pinning one host. See Multi-server / instance.

Retry a job​

Requeue a settled job to run again with retryJob. It re-arms that job's own row — status back to PENDING, retry counters, lock, return value and timestamps reset — and returns an AddJobResult (never a bare id). There are no retry "modes": one job id, one behaviour.

const result = await pq.retryJob({ jobId });
// → AddJobResult for the requeued job, or null if it couldn't be re-armed

null means the id is unknown or the job is ACTIVE. A running job is never re-armed: clearing its lock would let a second worker claim it, so the same job would run twice at once. Cancel it first with cancelJob and retry once it has settled.

To start a fresh run from scratch instead, don't retry at all — just add a new job to the queue.

Cancel a job​

Stop a one-off job — an immediate or a delayed one — from running, without deleting its row.

await pq.cancelJob({ jobId });

If the job is still pending/delayed, it's marked CANCELLED so it never runs. If it's already running on this instance, its handler's AbortSignal is fired instead — the handler must honour job.signal to actually stop. See Canceling jobs for how to write abort-aware handlers.

Recurring jobs are different

cancelJob acts on a single job. To stop a recurring (interval/cron) job, cancel or remove its schedule instead — see Remove a schedule below (removeSchedule halts the recurrence so no new jobs are produced).

Remove a schedule​

await pq.removeSchedule({ scheduleId }); // delete a schedule row

removeSchedule deletes a row from the schedule table, so no new jobs are produced from that recurrence — this is how you stop a recurring (interval/cron) job. Jobs it has already produced are untouched; remove those separately with removeJob / removeJobs if you want them gone too.

Remove jobs​

Removal deletes rows — it is for cleanup, not for stopping work:

await pq.removeJob({ jobId }); // delete one job row (accepts tx)
await pq.removeJobs({ jobIds: [a, b] }); // delete several by id, in one statement (any queue)
await pq.removeJobs({ queue: 'emails', statuses: [JobStatus.COMPLETED] }); // by status, one queue

removeJob / removeJobs delete rows from the jobs table. The job is gone — its history, result, and failure message go with it. To merely stop a job rather than delete it, use cancelJob; to stop a recurring job, remove its schedule instead — see Remove a schedule above. Reach for removeJob / removeJobs only when you actually want the record gone.

Removal never touches a running job

A job that is currently ACTIVE (being processed) is never deleted — cancel it first with cancelJob, then remove it once it has settled. Concretely:

  • removeJob({ jobId }) on an active job skips it and returns false.
  • removeJobs({ jobIds }) skips any active ids and returns a count that excludes them.
  • removeJobs({ queue, statuses }) throws if you pass JobStatus.ACTIVE — you can't bulk-delete running jobs by status.
Prefer retention over manual deletes

Hand-deleting rows is easy to get wrong — you can lose failures you'd have wanted to triage. Prefer configuring retention at the queue level or per-job on addJob, and let PortQueue prune finished jobs for you. Keep manual remove* for deliberate one-offs.

Pause & resume the queue​

Persisted and cluster-wide — every instance honours it:

await pq.pauseQueue({ queue: 'emails', mode: PauseMode.PROCESSING }); // stop claiming; still accepts adds (default)
await pq.pauseQueue({ queue: 'emails', mode: PauseMode.ALL }); // also refuse new adds (drain)
await pq.resumeQueue({ queue: 'emails' });

Idempotency & singleton​

Both keys are queue-scoped, but they solve different problems:

  • idempotencyKey — a first-writer-wins dedup identity. The first add with a given key creates the job (or schedule); any later add with the same key on that queue returns the existing entity unchanged (deduplicated: true, deduplicationReason: 'IDEMPOTENCY') and creates nothing. Dedup is per entity: a one-off add is matched against the jobs table, a recurring add against the schedule table (they never dedup against each other). Use it to guarantee an operation is enqueued at most once ever (per key).
  • singletonKey — at most one active/pending job at a time for that key. Unlike idempotencyKey, it does not pin a permanent identity: once the job finishes, a new add with the same key can enqueue again. Use it to prevent a second concurrent run (e.g. one reindex per tenant at a time), while still allowing the next run later.
const paymentsJob = await pq.addJob({
  queue: 'payments',
  data: { orderId },
  idempotencyKey: `charge:${orderId}`,
});
if (paymentsJob.deduplicated) logger.info('already queued', { jobId: paymentsJob.jobId }); // charged at most once

await pq.addJob({ queue: 'reindex', data, singletonKey: `reindex:${tenantId}` }); // one live reindex per tenant

Transactional enqueue​

Pass a tx adapter so the insert joins your own transaction — the job exists iff your business write commits, closing the classic dual-write gap where a crash between the write and the enqueue leaves one without the other. PortQueue never opens or commits the transaction; your code stays the single owner of commit/rollback, and the NOTIFY that wakes workers fires only after commit.

An adapter ships from @portqueue/core/tx for every supported client — query builders, ORMs, and raw drivers alike. Pick your tab:

import { fromKnex } from '@portqueue/core/tx';

await knex.transaction(async (trx) => {
  await trx('order').insert(order);
  await pq.addJob({ queue: 'fulfillment', data: { orderId }, tx: fromKnex(trx) });
}); // the developer's trx owns commit/rollback

Every job-manipulating method that writes to the database (addJob, addJobs, retryJob, cancelJob, removeJob, removeJobs, removeSchedule) accepts tx; omit it and the method runs autonomously. addJobs is one insert for the whole batch, so every entry must share the same tx (or none of them set one). See Transactions for the full list of adapters and the reasoning behind the design.

Inspect​

Read-side views of what you've produced. A per-queue read names its queue; a by-id read doesn't need one, because an id already identifies exactly one row:

const emailsJob = await pq.getJob({ jobId }); // JobSnapshot | null
const emailsStats = await pq.getQueueStats({ queue: 'emails' }); // per-status counts

// The two list reads take the same window — `limit` (default 50, max 500) and `offset` — and
// return `{ items, total }`. `total` counts every row matching the filter, not just this page.
const failed = await pq.getJobs({
  queue: 'emails',
  statuses: [JobStatus.FAILED],
  limit: 20,
  offset: 0,
});
failed.items; // JobSnapshot[] — this page, newest first
failed.total; // e.g. 143 — every FAILED job on the queue

const schedules = await pq.getSchedules({ queue: 'emails', enabled: true, limit: 20, offset: 0 });
schedules.items; // ScheduleInfo[] — each with the scheduleId you pass to removeSchedule
schedules.total; // every armed schedule on the queue

// The client-level list reads take the same window and return the same `{ items, total }`:
const queues = await pq.getQueues({ limit: 20, offset: 0 }); // QueueInfo[] + the total across all queues

There is no hasMore flag on a page: another page follows when offset + items.length < total. Walk a whole list by stepping offset until it doesn't.

for (let offset = 0; ; offset += 100) {
  const page = await pq.getJobs({
    queue: 'emails',
    statuses: [JobStatus.FAILED],
    limit: 100,
    offset,
  });
  for (const job of page.items) await triage(job);
  if (offset + page.items.length >= page.total) break;
}

There is no updateJob or updateSchedule — by design​

PortQueue has no update* method on any entity. Three reasons:

  • A job's settings are a set, not independent fields. They were resolved together at insert time — startAfter and expireAt are derived instants, the retry counters are read against retryMaxAttempts. Editing one can produce a combination the queue never creates on its own, like a job that expires before it becomes eligible.
  • A row can change underneath the update. Status is owned by whichever worker holds the lock; between your read and your write the job can be claimed, retried, completed or reclaimed as stalled.
  • Updating destroys history. Throughput, failure rates and "what did we actually run" are all read from these rows.

What to do instead​

Add a new entity rather than editing an old one. The right move depends on where the job is:

SituationDo this
Finished job (COMPLETED, FAILED, CANCELLED, EXPIRED)Leave it. It's your history, and retention deletes it on schedule. Add a new job for what you need now.
Pending or delayed job that shouldn't run as configuredpq.cancelJob({ jobId }) so it never becomes eligible, then addJob with the settings you want.
Recurring schedule that needs a different cron, interval or payloadpq.removeSchedule({ scheduleId }) so it stops spawning jobs, then addJob with the new definition. See Changing a schedule.
Active jobNothing to update — it's running. Let it finish, or cancel cooperatively (the handler observes job.signal) and add a replacement.

Cancel-then-add and remove-then-add cost one extra call, and in exchange every row stays a truthful record of one attempt at one configuration. Give the replacement an idempotencyKey if it must be created exactly once.