---
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` → `completed` or `failed`. Every state transition is broadcast instantly to all queue participants. Failed jobs are re-queued automatically until `maxRetries` is exhausted. Three roles participate in a queue: **producers** add jobs, **workers** claim and process them, and **monitors** observe queue state. A single client can fulfil multiple roles simultaneously. 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
- Atomic job claiming, so exactly one worker receives each job
- Configurable priority and automatic retry with backoff
- Real-time progress reporting (0-100) broadcast to all participants
- 7-day job history replay on reconnect
- 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 durable lifecycle events and `_progress` for ephemeral progress updates. Job claiming uses a server-side atomic operation so concurrent claim attempts from multiple workers resolve to exactly one winner. The app owns the socket lifecycle; the wrapper never opens or closes it.

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

## Installation

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

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

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

// The app owns one core client. In a browser, pass a token provider so the
// SDK can mint fresh short-lived client tokens from your backend.
const client = NoLag(async () => (await (await fetch('/api/nolag-token')).json()).token)

// Inject the client into the queue wrapper (role: 'producer', 'worker', or 'monitor')
const queueClient = new NoLagQueue({ client, role: 'worker', concurrency: 4 })

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

const queue = await queueClient.joinQueue('video-encoding')

// ── Producer ─────────────────────────────────────────────────────────────────
const job = await queue.addJob({
  id: 'job-101',           // optional, auto-generated if omitted
  payload: { videoId: 'vid_xyz', preset: '1080p' },
  priority: 10,            // higher = processed sooner
  maxRetries: 3,
})
console.log('Added job:', job.id, 'status:', job.status) // 'pending'

// ── Worker ───────────────────────────────────────────────────────────────────
queue.on('jobAdded', async ({ job }) => {
  // Atomically claim the job (only one worker succeeds)
  const claimed = await queue.claimJob(job.id)
  if (!claimed) return  // another worker got it first

  console.log('Claimed job:', job.id)

  try {
    // Report progress (0–100)
    await queue.reportProgress(job.id, 25)
    await doExpensiveWork(job.payload)
    await queue.reportProgress(job.id, 100)

    // Mark complete with result
    await queue.completeJob(job.id, { outputUrl: 'https://cdn.example.com/vid.mp4' })
  } catch (err) {
    // Mark failed (retried automatically if maxRetries not exhausted)
    await queue.failJob(job.id, err instanceof Error ? err.message : String(err))
  }
})

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

queue.on('jobCompleted', ({ jobId, result }) => {
  console.log(`Job ${jobId} done:`, result)
})

queue.on('jobFailed', ({ jobId, error, attempt, maxRetries }) => {
  console.warn(`Job ${jobId} failed (attempt ${attempt}/${maxRetries}):`, error)
})

queue.on('jobRetrying', ({ jobId, attempt, maxRetries }) => {
  console.log(`Retrying job ${jobId} (${attempt}/${maxRetries})...`)
})

// Queue depth monitoring
console.log('Pending:', await queue.pendingCount)
console.log('Active:', await queue.activeCount)

// Teardown: the wrapper releases its handlers; the app closes the socket.
queueClient.detach()
client.disconnect()
```

## API Reference

### NoLagQueue

#### Constructor Options

| Option | Type | Description |
| --- | --- | --- |
| `client` | `NoLagSocket` | **Required.** The injected core NoLag client the app owns and connects. |
| `workerId` | `string` | Stable worker ID for this client (auto-generated if omitted). |
| `role` | `'producer' \| 'worker' \| 'monitor'` | Role this client plays in the queue (default `'monitor'`). |
| `concurrency` | `number` | Max jobs to process concurrently (default `1`). |
| `metadata` | `Record<string, unknown>` | Optional custom data attached to worker presence. |
| `appName` | `string` | NoLag app for topic prefixes (default `'queue'`). |
| `queues` | `string[]` | Queue names this client should participate in. |
| `loadBalanceGroup` | `string` | Subscribe-level load-balance group; workers in the same group receive jobs round-robin (default `'queue-workers-{queueName}'`). |
| `maxJobCache` | `number` | Max jobs cached in memory (default `1000`). |
| `debug` | `boolean` | Enable wrapper debug logging (default `false`). |

| Method | Returns | Description |
| --- | --- | --- |
| `ready()` | `Promise<void>` | Resolves once wrapper setup completed. |
| `detach()` | `void` | Release this wrapper's handlers and topics; terminal, never closes the socket. |
| `joinQueue(name)` | `Promise<QueueRoom>` | Subscribe to a named queue. Returns the room instance. |
| `leaveQueue(name)` | `Promise<void>` | Unsubscribe from a queue and release its resources. |
| `getOnlineWorkers()` | `Worker[]` | Return all workers currently present in the lobby. |

### NoLagQueue Events

| Event | Payload | Description |
| --- | --- | --- |
| `connected` | none | WebSocket connection established. |
| `disconnected` | `reason: string` | Connection closed. |
| `reconnected` | none | Connection restored; job history replay begins automatically. |
| `error` | `error: Error` | A transport or protocol error occurred. |
| `workerOnline` | `worker: Worker` | A worker joined the lobby. |
| `workerOffline` | `worker: Worker` | A worker left the lobby. |

### QueueRoom

| Method | Role | Returns | Description |
| --- | --- | --- | --- |
| `addJob(opts)` | Producer | `Promise<Job>` | Enqueue a new job. Accepts `id?`, `payload`, `priority?`, `maxRetries?`. |
| `claimJob(jobId)` | Worker | `Promise<Job \| null>` | Atomically claim a pending job. Returns `null` if another worker claimed it first. |
| `reportProgress(jobId, progress)` | Worker | `Promise<void>` | Broadcast a progress value (0-100) on the ephemeral channel. |
| `completeJob(jobId, result?)` | Worker | `Promise<void>` | Mark the job as completed with an optional result payload. |
| `failJob(jobId, error)` | Worker | `Promise<void>` | Mark the job as failed. Triggers retry if `maxRetries` has not been reached. |
| `getJob(id)` | Monitor | `Promise<Job \| null>` | Fetch the current state of a job by ID. |
| `getJobs(filter?)` | Monitor | `Promise<Job[]>` | Fetch jobs filtered by status (`'pending' \| 'active' \| 'completed' \| 'failed'`). |
| `pendingCount` | Monitor | `Promise<number>` | Number of jobs currently in the pending state. |
| `activeCount` | Monitor | `Promise<number>` | Number of jobs currently being processed by workers. |

### QueueRoom Events

| Event | Payload | Description |
| --- | --- | --- |
| `jobAdded` | `{ job: Job }` | A new job was added to the queue. |
| `jobClaimed` | `{ jobId, workerId }` | A worker successfully claimed a job. |
| `jobProgress` | `{ jobId, progress: number, workerId }` | A worker reported a progress update. Delivered via the ephemeral channel. |
| `jobCompleted` | `{ jobId, result?, workerId }` | A job was completed successfully. |
| `jobFailed` | `{ jobId, error, attempt, maxRetries, workerId }` | A job failed. If `attempt < maxRetries` the job will be retried. |
| `jobRetrying` | `{ jobId, attempt, maxRetries }` | A failed job has been re-queued for another attempt. |
| `workerJoined` | `worker: Worker` | A worker joined this queue. |
| `workerLeft` | `worker: Worker` | A worker left this queue. |
| `replayStart` | `{ count: number }` | Job history replay has begun. |
| `replayEnd` | `{ replayed: number }` | Job history replay has completed. |