# The Job Object Inside a Worker Processor

What a bunqueue processor receives: job data and metadata, progress updates, per-job logs, lock extension and access to child results from a flow.

Canonical: https://bunqueue.dev/guide/worker/job-object/

---

import { Tabs, TabItem } from '@astrojs/starlight/components';

<div class="bq-wrap bq-hero">
  <span class="bq-eyebrow">guide · worker</span>
  <h1 class="bq-hero-h1 bq-bench-h1">Everything the processor <em>gets handed.</em></h1>
  <p class="bq-hero-sub">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.</p>
</div>

## Use the job object

Inside the processor you get the full job:

<Tabs syncKey="lang">
<TabItem label="Bun">

```typescript
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; // Attempts consumed so far (0 on the first attempt)
  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 });
```

</TabItem>
<TabItem label="Node.js / Deno">

```typescript
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; // Attempts consumed so far (0 on the first attempt)
  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: false });
```

</TabItem>
<TabItem label="Python">

```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
```

</TabItem>
<TabItem label="PHP">

```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;
});
```

</TabItem>
<TabItem label="Go">

```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{})
```

</TabItem>
<TabItem label="Rust">

```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(),
);
```

</TabItem>
<TabItem label="Elixir">

```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)
```

</TabItem>
</Tabs>

## Pull and process a job manually

Both TypeScript packages can disable the automatic loop and explicitly acquire one job:

```typescript
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.

## Native batch members (Bun)

When the Worker has `batch: { size, ... }`, the processor is invoked once with
the leading job. Read all independently leased members through `getBatch()`:

```typescript
const worker = new Worker('imports', async (leadingJob) => {
  const jobs = leadingJob.getBatch?.() ?? [leadingJob];
  for (const job of jobs) {
    try {
      await importRow(job.data);
    } catch (error) {
      job.setAsFailed?.(error instanceof Error ? error : new Error(String(error)));
    }
  }
  return { processed: jobs.length };
}, { batch: { size: 50, minSize: 10, timeout: 100 } });
```

`getBatch` and `setAsFailed` are optional in the public Job type because an
ordinary one-job delivery does not carry them. Every returned member has both
methods. `setAsFailed` records a selective failure for that member after the
shared processor invocation completes; all unmarked members receive the
processor's common result.

## 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

| Guide | What it covers |
|---|---|
| [Worker](/guide/worker/) | Create a worker and process your first job |
| [Worker Concurrency and Batch Pulling](/guide/worker/concurrency/) | Run jobs in parallel and pull them in batches |
| [Worker Events](/guide/worker/events/) | completed, failed, stalled and the rest |
| [Worker Error Handling, Retries and Backoff](/guide/worker/errors/) | Retries, backoff, timeouts and giving up |
| [Worker Lifecycle](/guide/worker/lifecycle/) | Pause, resume and shut down without losing work |
| [Heartbeats, Stall Detection and Lock Ownership](/guide/worker/stalls/) | Heartbeats, stall recovery and lock ownership |
| [SandboxedWorker](/guide/worker/sandboxed/) | Experimental isolation for CPU-heavy handlers |
| [WorkerOptions Reference](/guide/worker/options/) | Every WorkerOptions field, with defaults |