---
title: Queue SDK
description: Real-time job queues with lifecycle tracking, progress updates, and worker management.
---

# @nolag/queue

Real-time job queues with lifecycle tracking, progress updates, and worker management.

## Overview

Distribute work across real-time workers with full lifecycle visibility. Jobs follow a state machine: `pending`, `claimed`, `active`, then `completed` or `failed`. Every state transition is broadcast to queue participants. A failed job goes back to `pending` until `maxAttempts` is exhausted. Three roles participate in a queue: **producers** add jobs, **workers** claim and process them, and **monitors** observe queue state. A wrapper plays one role; run one wrapper per role. Your app owns one core NoLag client and injects it into `NoLagQueue`; the wrapper attaches its behaviour to that connection.

### Key Features

- State machine lifecycle: pending, claimed, active, completed, failed
- Load-balanced delivery: each new job reaches exactly one worker in the group
- Priority label on every job and automatic re-queue on failure
- Real-time progress reporting (0-100) broadcast to all participants
- Ephemeral progress channel keeps durable storage lean
- Worker online/offline presence via lobby

## How It Works

`NoLagQueue` attaches to an injected `@nolag/js-sdk` client and maintains a lobby for worker presence. Calling `joinQueue(name)` returns a `QueueRoom` that subscribes to two topics: `jobs` for lifecycle events and `_progress` for progress updates. Workers subscribe to `jobs` with `loadBalance: true` in the group `queue-workers-<queue>` (or the `loadBalanceGroup` you set), so the broker hands each job event to exactly one worker in the group. Producers and monitors subscribe without load balancing and see every event. `_progress` is never load balanced.

Exclusivity comes from that delivery, not from a server-side claim. `claimJob()` is a local state transition plus a broadcast of `jobClaimed`: the worker that received `jobAdded` is the only worker in its group that holds the job, and `claimJob()` returns `null` only when the job is unknown locally or is not `pending`. A retry is immediate: `failJob()` records the attempt, and while `attempts < maxAttempts` it moves the job back to `pending` and emits `jobRetrying` in the same call, with no delay or backoff. Nothing re-dispatches the job for you; the worker that failed it claims it again from its `jobRetrying` handler. `priority` is a label stored on the job and carried in every event; nothing reorders delivery by it. The app owns the socket lifecycle; the wrapper never opens or closes it.

| Topic | Purpose |
| --- | --- |
| `jobs` | Job lifecycle events: added, claimed, completed, failed, retrying |
| `_progress` | In-flight progress updates (0-100), not persisted |

Because workers subscribe with `loadBalance: true`, a worker group whose actors hold a persistent session (`agent` or `orchestrator`) is the one case where the hosted broker replays what was dispatched while the group was away; see [Replay](/docs/concepts/replay) for the exact conditions. Replay progress is reported by the core client (`client.on('replay:start')`, `client.on('replay:end')`), not by the wrapper. Producers and monitors never trigger replay.

**Before you start.** Create an app from the `nolag-queue-sdk` blueprint. That seeds the rooms this SDK expects (`image-processing`, `email-dispatch`) and the `online` lobby, and returns an app slug with a random suffix. That slug is the `appName` you pass to the wrapper. You can also do this in the portal: Apps, New App, pick the blueprint.

```typescript [Setup (server side, once)]
import { NoLagApi } from "@nolag/js-sdk";

const api = new NoLagApi(process.env.NOLAG_API_KEY); // nlg_live_...
const app = await api.apps.create({ name: "My App", blueprintId: "nolag-queue-sdk" });
console.log(app.slug); // e.g. "my-app-a3f9": this is your appName

const actor = await api.actors.create({ name: "web-client", actorType: "user" });
console.log(actor.accessToken); // shown once; keep it
```

## Installation

```bash [Terminal]
npm install @nolag/queue @nolag/js-sdk
```

**Shared connection.** One core NoLag client can back several wrapper SDKs at once, for example a job queue, a dashboard, and notify on a single socket, as long as each wrapper uses a distinct `appName`. Each wrapper attaches its handlers on construction and releases them with `detach()`, and never touches the socket itself. Your app owns `connect()` and `disconnect()`.

## Quick Start

Each role is its own wrapper, typically its own process. All three join the seeded `image-processing` queue.

```typescript [Producer]
import { NoLag } from '@nolag/js-sdk'
import { NoLagQueue } from '@nolag/queue'

const client = NoLag(PRODUCER_TOKEN)
const producer = new NoLagQueue({
  client,
  appName: APP_SLUG, // the slug returned when you created the app
  role: 'producer',
})

await client.connect()   // the app owns the connection
await producer.ready()   // wrapper setup complete

const queue = producer.joinQueue('image-processing')

// Enqueue a job. `type` is required; the id is generated for you.
const job = queue.addJob({
  type: 'resize',
  payload: { imageId: 'img_xyz', width: 1280 },
  priority: 'high',   // 'low' | 'normal' | 'high' | 'critical'; a label, not an ordering
  maxAttempts: 3,     // default 3
})
console.log('Added job:', job.id, 'status:', job.status) // 'pending'

// Producers receive every lifecycle event
queue.on('jobCompleted', (done) => console.log(`Job ${done.id} done:`, done.result))
queue.on('jobFailed', (failed) => {
  console.warn(`Job ${failed.id} failed (attempt ${failed.attempts}/${failed.maxAttempts}):`, failed.error)
})

producer.detach()
client.disconnect()
```
```typescript [Worker]
import { NoLag } from '@nolag/js-sdk'
import { NoLagQueue, type Job } from '@nolag/queue'

const client = NoLag(WORKER_TOKEN)
const worker = new NoLagQueue({
  client,
  appName: APP_SLUG, // the slug returned when you created the app
  role: 'worker',
  concurrency: 4,
})

await client.connect()
await worker.ready()

const queue = worker.joinQueue('image-processing')

// Workers subscribe to `jobs` with loadBalance: true, so each new job reaches
// exactly one worker in the group. claimJob is a local transition plus a
// broadcast; it returns null if the job is unknown here or not pending.
async function process(job: Job) {
  const claimed = queue.claimJob(job.id)
  if (!claimed) return

  try {
    queue.reportProgress(job.id, 25)
    const outputUrl = await doExpensiveWork(job.payload)
    queue.reportProgress(job.id, 100)
    queue.completeJob(job.id, { outputUrl })
  } catch (err) {
    // Retry is immediate: while attempts < maxAttempts the job goes back to
    // pending and jobRetrying fires on this worker, which claims it again.
    queue.failJob(job.id, err instanceof Error ? err.message : String(err))
  }
}

queue.on('jobAdded', process)
queue.on('jobRetrying', (job) => {
  console.log(`Retrying job ${job.id} (attempt ${job.attempts + 1}/${job.maxAttempts})`)
  process(job)
})

worker.detach()
client.disconnect()
```
```typescript [Monitor]
import { NoLag } from '@nolag/js-sdk'
import { NoLagQueue } from '@nolag/queue'

const client = NoLag(MONITOR_TOKEN)
const monitor = new NoLagQueue({
  client,
  appName: APP_SLUG, // the slug returned when you created the app
  role: 'monitor',   // the default
})

await client.connect()
await monitor.ready()

const queue = monitor.joinQueue('image-processing')

queue.on('jobProgress', ({ jobId, progress, workerId }) => {
  console.log(`Job ${jobId} is ${progress}% complete on ${workerId}`)
})

queue.on('jobClaimed', (job) => console.log(`Job ${job.id} claimed by ${job.claimedBy}`))
queue.on('jobCompleted', (job) => console.log(`Job ${job.id} done:`, job.result))
queue.on('jobFailed', (job) => console.warn(`Job ${job.id} failed:`, job.error))

// Queue depth from the local cache (plain getters). A monitor only knows about
// jobs whose events it has received since joining.
console.log('Pending:', queue.pendingCount)
console.log('Active:', queue.activeCount)
console.log('Failed:', queue.getJobs({ status: 'failed' }).length)

monitor.on('workerOnline', (w) => console.log('Worker online:', w.workerId))
monitor.on('workerOffline', (w) => console.log('Worker offline:', w.workerId))

monitor.detach()
client.disconnect()
```

## API Reference

### NoLagQueue

#### Constructor Options

| Option | Type | Description |
| --- | --- | --- |
| `client` | `NoLagSocket` | **Required.** The injected core NoLag client the app owns and connects. |
| `appName` | `string` | The app slug used as the topic prefix. Pass the suffixed slug returned when you created the app; the default `'queue'` will not match a hosted app. |
| `workerId` | `string` | Stable worker ID for this client (auto-generated if omitted). Recorded as `createdBy` and `claimedBy` on jobs. |
| `role` | `'producer' \| 'worker' \| 'monitor'` | Role this wrapper plays (default `'monitor'`). Only workers subscribe with load balancing. |
| `concurrency` | `number` | Advertised in worker presence as `QueueWorker.concurrency` (default `1`). The wrapper does not limit claims by it. |
| `metadata` | `Record<string, unknown>` | Optional custom data attached to worker presence. |
| `queues` | `string[]` | Queues to join once the wrapper is ready. |
| `loadBalanceGroup` | `string` | Subscribe-level load-balance group for workers (default `'queue-workers-<queueName>'`). Set a custom group to partition workers by region or capability. |
| `maxJobCache` | `number` | Max jobs cached in memory per queue (default `1000`). |
| `debug` | `boolean` | Enable wrapper debug logging (default `false`). |

| Method | Returns | Description |
| --- | --- | --- |
| `ready()` | `Promise<void>` | Resolves once wrapper setup completed. Join methods throw before this resolves. |
| `detach()` | `void` | Release this wrapper's handlers and topics; terminal, never closes the socket. |
| `joinQueue(name, opts?)` | `QueueRoom` | Subscribe to a named queue and return it. Synchronous; returns the existing instance if already joined. `opts.filters` declares the job filter values this participant wants (for a worker, its capabilities). |
| `leaveQueue(name)` | `void` | Unsubscribe from a queue and release its resources. |
| `getQueues()` | `QueueRoom[]` | All currently joined queues. |
| `getOnlineWorkers()` | `QueueWorker[]` | Workers currently present in the `online` lobby. |

### NoLagQueue Events

| Event | Payload | Description |
| --- | --- | --- |
| `connected` | none | Wrapper setup completed on a live connection. |
| `disconnected` | `reason: string` | Connection closed. |
| `reconnecting` | none | The core client is attempting to reconnect. |
| `reconnected` | none | Connection restored; queue membership and presence are restored automatically. |
| `error` | `error: Error` | A transport or protocol error occurred. |
| `workerOnline` | `worker: QueueWorker` | A worker joined the lobby. |
| `workerOffline` | `worker: QueueWorker` | A worker left the lobby. |

### QueueRoom

| Method | Role | Returns | Description |
| --- | --- | --- | --- |
| `addJob(opts)` | Producer | `Job` | Enqueue a job. `opts` is `{ type, payload?, priority?, maxAttempts?, filter?, filters? }`; `type` is required, the id is generated. |
| `claimJob(jobId)` | Worker | `Job \| null` | Move a `pending` job to `claimed` and broadcast it. `null` if the job is unknown locally or not pending. |
| `reportProgress(jobId, progress)` | Worker | `void` | Broadcast a progress value (0-100) on `_progress`. |
| `completeJob(jobId, result?)` | Worker | `Job \| null` | Mark the job `completed` with an optional result. `null` if the job is unknown locally or not in `claimed` or `active`. |
| `failJob(jobId, error?)` | Worker | `Job \| null` | Mark the job `failed` and increment `attempts`. While `attempts < maxAttempts` it returns to `pending` and `jobRetrying` fires immediately. |
| `getJob(id)` | Any | `Job \| undefined` | Read a job from the local cache. |
| `getJobs(filter?)` | Any | `Job[]` | Read cached jobs, optionally filtered by `{ status?, type?, priority? }`. Does not change what the server sends. |
| `pendingCount` | Any | `number` | Getter: cached jobs in the `pending` state. |
| `activeCount` | Any | `number` | Getter: cached jobs in the `active` state. |
| `getWorkers()` | Any | `QueueWorker[]` | Workers present in this queue. |
| `setFilters(values)` / `addFilters(values)` / `removeFilters(values)` | Any | `void` | Change which job filter values this participant receives. Filters compose with load balancing: one matching worker gets each job. |

### QueueRoom Events

All lifecycle events carry the full `Job`. Producers and monitors receive every `jobAdded`; within a worker group only one worker does. The later events are emitted only on clients that already hold the job in their cache.

| Event | Payload | Description |
| --- | --- | --- |
| `jobAdded` | `Job` | A new job was added. Delivered to one worker per group, and to every producer and monitor. |
| `jobClaimed` | `Job` | A worker claimed a job; `claimedBy` names it. |
| `jobProgress` | `JobProgress` | A worker reported progress: `{ jobId, progress, workerId, timestamp }`. |
| `jobCompleted` | `Job` | A job completed; `result` carries the payload. |
| `jobFailed` | `Job` | A job failed; `error`, `attempts` and `maxAttempts` describe the state. |
| `jobRetrying` | `Job` | A failed job was returned to `pending` for another attempt. |
| `workerJoined` | `worker: QueueWorker` | A worker joined this queue. |
| `workerLeft` | `worker: QueueWorker` | A worker left this queue. |

### Types

| Type | Shape |
| --- | --- |
| `Job` | `{ id, type, payload?, priority, status, progress, result?, error?, attempts, maxAttempts, claimedBy?, createdBy, createdAt, updatedAt, completedAt?, filter?, isReplay }` |
| `JobStatus` | `'pending' \| 'claimed' \| 'active' \| 'completed' \| 'failed'` |
| `JobPriority` | `'low' \| 'normal' \| 'high' \| 'critical'` (default `'normal'`) |
| `QueueWorker` | `{ workerId, actorTokenId, role, activeJobs, concurrency, metadata?, joinedAt, isLocal }` |