@nolag/queue

Realtime job queues with lifecycle events, progress updates and load-balanced workers.

Overview

@nolag/queue distributes work across connected workers and keeps every participant informed about each job. Jobs move through a small state machine: pending, claimed, active, then completed or failed. A failed job goes back to pending until it runs out of attempts. Three roles take part in a queue: producers add jobs, workers claim and process them, and monitors watch. A wrapper instance plays one role, so run one per role. Your app owns one core NoLag client and injects it into NoLagQueue; the wrapper attaches its behaviour to that connection.

Key Features

• Job lifecycle broadcast to every participant: added, claimed, completed, failed, retrying
• Load-balanced delivery: each new job reaches one worker in the group
• Priority label on every job and immediate re-queue on failure
• Progress reporting (0 to 100) on a separate topic
• Capability routing with filters: a job tagged gpu only reaches workers that filter on gpu
• Worker presence per queue, plus an app-wide online lobby

How It Works

NoLagQueue attaches to an injected @nolag/js-sdk client. Calling joinQueue(name) returns a QueueRoom for the room name and subscribes to two topics in it: 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). kraken delivers each message on a load-balanced subscription to exactly one member of the group: on the built-in syn broker that member is picked at random, and with BROKER_BACKEND=mqtt your MQTT broker's shared-subscription strategy picks it. kraken scopes the group name by project and app, so two apps that both use queue-workers-image-processing never share workers. 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 lock. claimJob() is a local state change plus a jobClaimed broadcast: the worker that received jobAdded is the only worker in its group that knows about the job. 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, moves the job back to pending and emits jobRetrying on the same worker in the same call. Nothing re-dispatches the job for you; the worker that failed it claims it again from its jobRetrying handler. priority is a label carried on the job; nothing reorders delivery by it.

Every lifecycle event is published with echo: true, so a participant subscribed to jobs can receive its own event back. Producers drop their own jobAdded (the cache de-duplicates it), but a worker can see jobClaimed or jobCompleted twice for the same job when kraken happens to hand its own broadcast back to it. Keep handlers idempotent.

Topics to grant

Grant these either in kraken's static token file or in a @nolag/core project document. The excerpts below show one worker token and one app; see Static auth and Project document for the full formats.

{
  "tokens": {
    "worker-token-1": {
      "actorTokenId": "worker-1",
      "projectId": "demo-project",
      "organizationId": "demo-org",
      "actorType": "service",
      "allowedTopics": [
        {
          "pattern": "queue/image-processing/#",
          "permission": "pubSub",
          "app_id": "queue-app",
          "room_id": "room-image-processing",
          "room_slug": "image-processing"
        },
        {
          "pattern": "queue/email-dispatch/#",
          "permission": "pubSub",
          "app_id": "queue-app",
          "room_id": "room-email-dispatch",
          "room_slug": "email-dispatch"
        }
      ],
      "allowedLobbies": [{ "lobby_slug": "online", "lobby_id": "queue-online" }]
    }
  }
}

Things that matter with the static file:

• Every token that takes part in the same queue needs the same app_id and projectId. kraken routes wildcard-matched patterns to <app_id>/<pattern> and scopes load-balancing groups by project and app, so a different value puts a token on a different queue.
• Presence, and therefore workerJoined and workerLeft, only works in rooms that have a rule carrying both room_slug and room_id. A catch-all rule such as queue/+/# lets a token use any queue name, but presence stays silent in rooms without their own rule.
• Without the allowedLobbies entry the lobby subscribe is refused: the wrapper emits an error (not_authorized) and ready() only resolves after the core SDK's 10 second lobby timeout.

With @nolag/core, rooms are records: a queue name that is not a room in the app is refused with unknown_topic (42940) on the error event. Add the room to the project document (or create it in core) before a client joins 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 image-processing queue. The tokens are the ones from your auth.json or the credentials @nolag/core minted on import.

import { NoLag } from '@nolag/js-sdk'
import { NoLagQueue } from '@nolag/queue'

const client = NoLag('producer-token', { url: 'ws://localhost:8080/ws' })
const producer = new NoLagQueue({
  client,
  appName: 'queue', // the App Slug you granted
  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)
})

// Later: release the wrapper, then close the socket
producer.detach()
client.disconnect()

Routing jobs by capability

Tag a job with a filter value and let workers declare what they can run. Load balancing still applies among the matching workers, so one of them gets the job.

// Producer: only GPU workers should see this job
queue.addJob({ type: 'render', payload: { sceneId: 's-42' }, filter: 'gpu' })

// Worker: receive only jobs tagged gpu
const gpuQueue = worker.joinQueue('image-processing', { filters: ['gpu'] })

A worker that joins without filters receives every job, including filtered ones. Filters are routing, not access control; see Filters.

Reconnects

When the core client reconnects, the wrapper re-subscribes the online lobby, re-sets its presence in every joined queue and then emits reconnected. It does not re-subscribe the jobs and _progress topics: it was written for a broker that restores subscriptions on its own. Do not rely on that with kraken (see Production); re-issue the subscriptions yourself, with the same options the wrapper used:

worker.on('reconnected', () => {
  for (const room of worker.getQueues()) {
    const ctx = client.setApp('queue').setRoom(room.name)
    const filters = room.filters.length > 0 ? { filters: room.filters } : {}
    // Workers: load balanced, in the same group the wrapper uses
    // (the default below, or the loadBalanceGroup you configured)
    ctx.subscribe('jobs', { loadBalance: true, loadBalanceGroup: `queue-workers-${room.name}`, ...filters })
    ctx.subscribe('_progress', filters)
  }
})

The wrapper's message handlers stay registered on the client, so subscribing again is enough. Subscribing to a topic that is already subscribed just replaces the subscription.

API Reference

NoLagQueue

Constructor Options

Events

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 only emitted on clients that already hold the job in their cache.

Types