Inspecting & controlling jobs

Read job state, control jobs from a queue, and report progress and logs from inside a processor. Every call talks to NextMQ over HTTP.

Fetching jobs#

const job = await reportQueue.getJob(jobId)
const state = await job.getState()    // 'waiting' | 'active' | 'completed' | ...
await job.refresh()                   // re-fetch latest state

const recent = await reportQueue.getJobs(['waiting', 'active'], 0, 20)
const failed = await reportQueue.getFailed(0, 20)
const counts = await reportQueue.getJobCounts() // { waiting, active, failed, ... }
const total  = await reportQueue.count()

Reporting from inside a processor#

Inside a worker processor, use the job handle to stream progress and logs back as the work runs.

export const exportWorker = new Worker('exports', async (job) => {
  await job.log('starting export')
  await job.updateProgress(10)

  const rows = await fetchRows(job.data)
  await job.updateProgress(60)

  await job.updateData({ ...job.data, rowCount: rows.length })
  return { url: await uploadCsv(rows) }
})

Read retained logs later from the queue:

const { logs, count } = await reportQueue.getJobLogs(jobId, 0, 99)

Acting on jobs#

Method	What it does
`retry()`	Move a failed job back to waiting to try again.
`promote()`	Promote a delayed job so it runs now.
`changeDelay(ms)`	Reschedule a delayed job.
`changePriority(opts)`	Change a waiting job's priority.
`remove(opts?)`	Delete the job from the queue; pass { removeChildren: true } for flow trees.
`updateData(data)`	Replace the job's data payload.
`updateProgress(n)`	Report progress (number or object).
`log(line)`	Append a log line (retained up to keepLogs).

const job = await reportQueue.getJob(jobId)

await job.changePriority({ priority: 1 })
await job.promote()
await job.retry()
await job.remove({ removeChildren: true })

Job instances also expose state helpers such as isCompleted(), isFailed(), isDelayed(), isActive(), isWaiting(), and isWaitingChildren(). The full Job surface — every property and method — is in the SDK reference.

Parents & child values

For flow parents, read child return values with getChildrenValues(). A child's parent reference is on job.parent and job.parentKey.

const values = await job.getChildrenValues()
// { '<childQueue>:<childJobId>': returnValue, ... }

const parentRef = job.parent     // { id, queueKey } | undefined

Waiting for completion#

waitUntilFinished()polls until the job reaches a terminal state. It's a convenience over re-fetching the job — there's no long-lived connection involved.

const job = await reportQueue.add('csv', { reportId })

const result = await job.waitUntilFinished({
  timeoutMs: 30_000,
  pollIntervalMs: 1000,
})

PreviousWorkers & processing NextSchedulers & repeatables