@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.

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

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

npm install @nolag/queue @nolag/js-sdk

Quick Start

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

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

API Reference

NoLagQueue

Constructor Options

NoLagQueue Events

QueueRoom

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.

Types