How Project N.O.M.A.D. Uses Background Jobs for Asynchronous Processing

Project N.O.M.A.D. off-loads time-consuming operations to BullMQ-powered background job queues using a three-layer architecture: a centralized QueueService, declarative Job classes, and a CLI worker command that processes tasks asynchronously.

The open-source Project N.O.M.A.D. (Nomad) repository implements a robust asynchronous processing pipeline to handle heavy I/O operations without blocking user-facing requests. By leveraging the BullMQ library with Redis, the system decouples long-running tasks—such as file downloads, AI model provisioning, and document embedding—from the main application thread. This article examines the exact implementation of background jobs for asynchronous processing in the Crosstalk-Solutions/project-nomad codebase.

Architecture Overview

The system organizes asynchronous work into three distinct components that separate queue management from business logic.

Queue Service: The Centralized Factory

Located in admin/app/services/queue_service.ts, the QueueService acts as a thin wrapper around BullMQ's Queue instances. It creates and caches queue objects using the Redis connection defined in admin/config/queue.ts, hiding implementation details from the rest of the application. This centralized factory ensures that all job producers interact with queues through a single, swappable interface.

Job Classes: Declarative Work Units

Each asynchronous workflow is encapsulated as a class in the admin/app/jobs/ directory, such as RunDownloadJob, DownloadModelJob, and EmbedFileJob. Every job class declares a queue name, a unique job key, a static dispatch() helper for enqueueing, and an instance handle() method that executes the actual work. This pattern provides type-safe job definitions and consistent retry behavior across the codebase.

Worker Command: The Processing Engine

The queue:work command in admin/commands/queue/work.ts spins up BullMQ Worker instances for each configured queue. It registers the appropriate job handlers, manages graceful shutdowns, and applies per-queue concurrency limits—such as allowing three concurrent downloads but only one benchmark job at a time.

The Asynchronous Processing Flow

Understanding how a job moves from dispatch to completion reveals the system's fault-tolerance and scalability.

1. Enqueue: Application code calls JobClass.dispatch(params), which obtains a Queue from QueueService, computes a deterministic jobId using SHA-256 hashing, and adds the job to Redis with configured retry and back-off settings.

2. Worker Startup: Operators run node ace queue:work (or containers start it automatically), which loads all job classes and creates a Worker per queue.

3. Processing: BullMQ delivers jobs to the worker, which invokes the job's handle() method. Long-running tasks execute here, with progress reported via job.updateProgress() and job.updateData().

4. Completion: On success, the worker returns results; on failure, BullMQ automatically retries according to the job's attempts and backoff policy.

5. Maintenance: The worker also schedules nightly jobs like CheckUpdateJob to keep the system updated without manual intervention.

Code Examples

Enqueueing a File Download

To queue a ZIM file download from a controller or service:

import { RunDownloadJob } from '#jobs/run_download_job'

await RunDownloadJob.dispatch({
  url: 'https://example.com/resource.zim',
  filepath: '/data/resources/resource.zim',
  timeout: 300_000,
  allowedMimeTypes: ['application/zim'],
  forceNew: false,
  filetype: 'zim',
  resourceMetadata: {
    resource_id: '123',
    version: '1.2.3',
    collection_ref: 'wikipedia',
  },
})

The dispatch method in admin/app/jobs/run_download_job.ts creates a deterministic job ID and adds the job to the downloads queue with retry settings.

Queueing AI Model Downloads

For asynchronous model provisioning with Ollama:

import { DownloadModelJob } from '#jobs/download_model_job'

await DownloadModelJob.dispatch({ modelName: 'llama2:7b' })

This places the job on the model-downloads queue with up to 40 attempts, allowing the Ollama service time to become ready before failing permanently.

Starting the Worker Process

Process all configured queues from the command line:

node ace queue:work --all

When --all is supplied, the command in admin/commands/queue/work.ts starts a Worker for each defined queue, applying the per-queue concurrency map defined in the configuration.

Reporting Progress from Within a Job

Jobs can update their progress status for real-time monitoring:

// Inside DownloadModelJob.handle()
await ollamaService.downloadModel(modelName, (percent) => {
  if (percent) {
    job.updateProgress(Math.floor(percent))
    job.updateData({ ...job.data, progress: percent })
  }
})

These updates are stored in Redis, enabling clients to query job status via BullMQ's API while the asynchronous processing continues in the background.

Key Configuration Files

The following files define the backbone of the asynchronous system:

• admin/config/queue.ts: Redis connection configuration for BullMQ.
• admin/app/services/queue_service.ts: Factory that lazily creates and caches BullMQ Queue instances.
• admin/commands/queue/work.ts: CLI command that initializes workers, loads job handlers, and manages concurrency.
• admin/app/jobs/run_download_job.ts: Example implementation showing file downloads, database updates, and follow-up job dispatching.
• admin/app/jobs/download_model_job.ts: Demonstrates long-running retries and progress reporting for AI model management.

Summary

• Project N.O.M.A.D. uses BullMQ with Redis to handle background jobs for asynchronous processing, preventing heavy I/O from blocking user requests.
• The QueueService in admin/app/services/queue_service.ts centralizes queue creation and provides a clean abstraction over BullMQ.
• Job classes encapsulate work units with static dispatch() methods for enqueueing and instance handle() methods for execution.
• The queue:work command spins up workers with configurable concurrency, automatic retries, and graceful shutdown handling.
• Progress reporting and maintenance scheduling are built into the worker lifecycle, providing visibility and automation for long-running tasks.

Frequently Asked Questions

What queue system does Project N.O.M.A.D. use for background jobs?

Project N.O.M.A.D. uses BullMQ, a Redis-based queue system for Node.js. The implementation in admin/config/queue.ts configures the Redis connection, while admin/app/services/queue_service.ts manages Queue instances and provides a centralized factory for the application.

How do I add a new background job in Project N.O.M.A.D.?

Create a new class in admin/app/jobs/ that defines a queue name, a unique jobKey, a static dispatch() method for enqueueing, and an async handle() method for processing. The dispatch method should use QueueService to obtain a queue instance and add the job with appropriate retry settings. Register the job in the worker command if it requires special concurrency settings.

How does the system handle failed background jobs?

Each job class defines attempts and backoff settings in its dispatch() method. When a job's handle() method throws an error, BullMQ automatically retries the job according to these policies, with exponential back-off strategies available for transient failures like network timeouts or service unavailability.

Can I monitor the progress of asynchronous jobs?

Yes. Job handlers can call job.updateProgress() to update percentage completion and job.updateData() to store custom metadata. These values are persisted in Redis and can be queried through BullMQ's API to provide real-time status updates to clients while background processing continues.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}