guide · worker

Everything the processor gets handed.

The job is more than its payload: attempt counts, timestamps, progress reporting, its own log stream, and the results of any children that ran before it.

Use the job object

Inside the processor you get the full job:

Bun

const worker = new Worker('queue', async (job) => {
  job.id;           // Job ID
  job.name;         // Job name
  job.data;         // Job data (typed if you use Worker<T>)
  job.attemptsMade; // Current attempt number
  job.timestamp;    // When the job was created

  await job.updateProgress(50, 'Halfway done');  // Report progress
  await job.log('Processing step 1');             // Attach a log line

  return result;
}, { embedded: true });

Node.js / Deno

const worker = new Worker('queue', async (job) => {
  job.id;         // Job ID
  job.name;       // Job name
  job.data;       // Job data (typed if you use Worker<T>)
  job.attempts;   // Attempts consumed so far
  job.timestamp;  // When the job was created (epoch ms)

  await job.updateProgress(50, 'Halfway done');  // Report progress
  await job.log('Processing step 1');            // Attach a log line

  return result;
});

Python

def process(job):
    job.id          # Job ID
    job.name        # Job name
    job.data        # Job data
    job.attempts    # Attempts consumed so far
    job.created_at  # When the job was created (epoch ms)

    job.update_progress(50, "Halfway done")  # Report progress
    job.log("Processing step 1")             # Attach a log line

    return result

PHP

$worker = new Worker('queue', function (Bunqueue\Job $job) {
    $job->id();            // Job ID
    $job->name();          // Job name
    $job->data();          // User payload, including any user-owned name key
    $job->attemptsMade();  // Attempts consumed so far

    $job->updateProgress(50, 'Halfway done');  // Report progress
    $job->log('Processing step 1');            // Attach a log line

    return $result;
});

Go

worker := bunqueue.NewWorker("queue", func(job *bunqueue.Job) (any, error) {
    job.ID()           // Job ID
    job.Name()         // Job name
    job.Data()         // User payload, including any user-owned name key
    job.AttemptsMade() // Attempts consumed so far

    job.UpdateProgress(50, "Halfway done") // Report progress
    job.Log("Processing step 1", "")       // Attach a log line

    return result, nil
}, bunqueue.WorkerOptions{})

Rust

let worker = Worker::new(
    "queue",
    |job| {
        job.id();            // Job ID
        job.name();          // Job name
        job.data();          // User payload, including any user-owned name key
        job.attempts_made(); // Attempts consumed so far

        let _ = job.update_progress(50.0, Some("Halfway done")); // Report progress
        let _ = job.log("Processing step 1", None);              // Attach a log line

        Ok(result)
    },
    WorkerOptions::default(),
);

Elixir

worker =
  Bunqueue.Worker.new("queue", fn job ->
    job.id             # Job ID
    job.name           # Job name
    job.data           # Job data
    job.attempts_made  # Attempts consumed so far

    Bunqueue.Job.update_progress(job, 50, "Halfway done")  # Report progress
    Bunqueue.Job.log(job, "Processing step 1")             # Attach a log line

    {:ok, result}
  end)

Pull and process a job manually

The Bun client can disable the automatic loop and explicitly acquire one job:

const worker = new Worker<{ value: number }, number>(
  'calculations',
  async (job) => job.data.value * 2,
  { autorun: false }
);

const job = await worker.getNextJob();
if (job) {
  job.name;  // the job name, separate from user data
  job.data;  // { value: number }
  job.token; // broker lease token when useLocks is enabled

  await worker.processJobManually(job);
}

processJobManually(job) reuses the lease tracked by getNextJob(), so the token argument can be omitted. If you pass an explicit token, it must match the current delivery.

Ownership-changing methods inside a processor

When a Bun processor calls retry(), changeDelay(), moveToWait(), moveToDelayed(), or moveToWaitingChildren(), the confirmed broker transition consumes that delivery generation. Returning or throwing afterward does not send a second ACK or FAIL and does not emit a contradictory local terminal event. A rejected transition still follows normal processor failure handling.

job.discard() is synchronous for API compatibility, but the Worker tracks and awaits its broker command internally. Graceful shutdown therefore waits for the discard, repeated calls send one command, and a stale processor’s lease token cannot discard a newer active generation. An already-retired job is silent; an actual discard transport or engine failure emits one Worker error with context: 'discard' and leaves broker recovery in charge.

Where to go next

Worker	Create a worker and process your first job
Worker Concurrency and Batch Pulling	Run jobs in parallel and pull them in batches
Worker Events	completed, failed, stalled and the rest
Worker Error Handling, Retries and Backoff	Retries, backoff, timeouts and giving up
Worker Lifecycle	Pause, resume and shut down without losing work
Heartbeats, Stall Detection and Lock Ownership	Heartbeats, stall recovery and lock ownership
SandboxedWorker	Experimental isolation for CPU-heavy handlers
WorkerOptions Reference	Every WorkerOptions field, with defaults