Skip to content

1.3 bullmq

Jump to bottom
wiki[bot] edited this page Apr 27, 2026 · 1 revision

1.3 BullMQ Async Processing

Architectural Rationale

Synchronous waiting on Ollama inference would block the Node.js event loop for seconds to minutes, depending on image size, model load state, and GPU availability. The system therefore adopts an asynchronous command-bus pattern: HTTP requests are accepted, validated, immediately acknowledged (202 Accepted), and delegated to BullMQ backed by Redis/KeyDB. Workers run in separate processes (conceptually) and emit results via Socket.IO as they become available.

flowchart TB
    subgraph Producers["Job Producers"]
        ClassicCtrl[ClassicController]
        McpCtrl[JsonRpcController]
    end

    subgraph BullMQSystem["BullMQ System"]
        QueueMgr[Queue Manager]
        Q1[image-describe]
        Q2[image-compare]
        Q3[image-ocr]
        Sched[Job Scheduler]
    end

    subgraph Workers["Workers"]
        W1[VisionsDescribeProcessor]
        W2[VisionsCompareProcessor]
        W3[VisionsOCRProcessor]
        Base[VisionsProcessor Base]
        Preproc[ImagePreprocessingService]
    end

    subgraph Redis["KeyDB / Redis"]
        Jobs[(Job Storage)]
        Events[(Event Stream)]
    end

    ClassicCtrl -->|addJob| QueueMgr
    McpCtrl -->|addJob| QueueMgr

    QueueMgr --> Q1 --> W1 --> Base
    QueueMgr --> Q2 --> W2 --> Base
    QueueMgr --> Q3 --> W3 --> Base

    W1 --> Preproc
    W2 --> Preproc
    W3 --> Preproc

    BullMQSystem -->|persist| Jobs
    BullMQSystem -->|emit| Events
Loading

Queue Topology

Queue Name Task Enum Processor Concern
describe describe VisionsDescribeProcessor Single-image or multi-image descriptive analysis
compare compare VisionsCompareProcessor Cross-image similarity and delta detection
ocr ocr VisionsOCRProcessor Text extraction with optional preprocessing enhancement

Each queue is configured with identical retry, backoff, and retention policies through BullmqConfigService.

Job Data Structure

interface VisionJob {
  name: string;
  data: {
    buffers: Buffer[];        // Image data (pre-preprocessing; raw buffers from controller)
    meta: ImageMeta[];        // File metadata per image (variant field populated by worker)
    filters: {
      requestId: string;      // Correlation ID
      roomId?: string;        // Socket.IO room subscription
      stream?: boolean;       // Streaming mode flag
      vLLM: string;           // Target Ollama model tag
      task: VisionTask;       // 'describe' | 'compare' | 'ocr'
      prompt?: Prompt[];      // Optional user-provided prompt
      numCtx?: number;        // Context window override
      preprocessing?: ImagePreprocessingOptions;  // Forwarded to worker-side preprocessor
    };
  };
}

interface ImageMeta {
  name: string;
  type: string;
  hash: string;
  requestId: string;
  variant?: string;           // Set by ImagePreprocessingService inside worker
}

Retry Strategy

BullMQ exponential backoff is configured with an initial delay of 5,270 ms and a maximum of 15 attempts.

Attempt Cumulative Delay Rationale
1 0 ms Immediate first try
2 ~5.3 s Brief backoff for transient network errors
3 ~15.8 s GPU warm-up time for freshly loaded models
4 ~36.9 s Ollama cold-start latency accommodation
15 ~5.8 hours Upper bound for deeply degraded infrastructure

Non-Retryable Failures

Certain failure classes bypass the retry mechanism via UnrecoverableError, preventing wasted cycles on deterministic failures:

Failure Source Action
model not found OllamaService Immediate fatal; client must select a different model
invalid image format Sharp preprocessor Immediate fatal; corrupted or unsupported MIME type
job canceled JobTrackingService Unrecoverable; worker terminates mid-stream
socket emit failure SocketIOService Logged but does not abort inference

Worker Architecture

Base Processor: VisionsProcessor

All task processors extend an abstract VisionsProcessor that encapsulates common concerns:

abstract class VisionsProcessor extends WorkerHost {
  protected async handleVision(
    buffers: Buffer[],
    filters: VisionFilters,
    onChunk?: (chunk: string) => void
  ): Promise<ChatResponse>;

  protected async emitToSocket(
    roomId: string,
    event: string,
    data: VisionResponse
  ): Promise<void>;
}

The base class manages:

  • Ollama chat() invocation with streaming callback
  • Socket.IO emission with null-safe adapter guards
  • Job cancellation polling via JobTrackingService.isCanceled()
  • Worker event logging via @ehildt/nestjs-bullmq-logger

Task-Specific Implementations

class VisionsDescribeProcessor extends VisionsProcessor {
  async process(job: Job): Promise<void> {
    const { buffers, meta, filters } = job.data;
    const systemPrompt = this.config.systemPrompts.DESCRIBE;

    await this.handleVision(buffers, filters, (chunk) => {
      this.emitToSocket(filters.roomId, filters.event, {
        meta, task: 'describe',
        message: { role: 'assistant', content: chunk },
        done: false
      });
    });
  }
}

Each subclass injects its own system prompt (defined in OllamaConfigService.systemPrompts) and task identifier but reuses the streaming loop and cancellation logic from the base class.

keep_alive Mapping

The OllamaConfigService returns the keepAlive property in camelCase per TypeScript convention. The VisionsProcessor maps this to keep_alive (snake_case) when building the Ollama chat() request payload:

return {
  messages,
  options: { num_ctx: filters.numCtx },
  stream: filters.stream,
  model: filters.vLLM,
  keep_alive: this.ollamaConfigService.config.keepAlive,
};

This indirection isolates the internal configuration schema from external SDK expectations, allowing future Ollama API version upgrades without cascading breaking changes.

Job Lifecycle

stateDiagram-v2
    [*] --> Waiting: addJob()
    Waiting --> Delayed: Delayed
    Delayed --> Waiting: Delay Expired
    Waiting --> Active: Worker Available
    Active --> Completed: Success
    Active --> Failed: Error
    Failed --> Waiting: Retry (attempts < max)
    Failed --> Dead: Max Retries Exceeded
    Completed --> [*]
    Dead --> [*]
Loading
State Transitions Observable Events
waiting addJob() queue.on('waiting')
active Worker pickup queue.on('active')
completed Success queue.on('completed')
failed Error, retryable queue.on('failed')
delayed Backoff expiry queue.on('delayed')

Job Events

The BullMQLoggerService subscribes to worker events to emit structured logs:

@OnWorkerEvent('active')
protected async onActive(job) { this.bullMQLogger.log(job); this.jobTracking.setActive(job.name); }

@OnWorkerEvent('completed')
protected async onCompleted(job) { this.bullMQLogger.log(job); this.jobTracking.remove(job.name); }

@OnWorkerEvent('failed')
protected async onFailed(job) {
  if (failedReason.includes('canceled')) this.bullMQLogger.log(job, 'canceled');
  else this.bullMQLogger.error(job);
  this.jobTracking.remove(job.name);
}

Configuration Reference

BULLMQ_HOST: keydb
BULLMQ_PORT: 6379
BULLMQ_USER: default
BULLMQ_PASS: redis
BULLMQ_JOB_ATTEMPTS: 15
BULLMQ_BACKOFF_TYPE: exponential
BULLMQ_BACKOFF_DELAY: 5270
BULLMQ_REMOVE_ON_COMPLETED_AGE: 604800000   # 7 days
BULLMQ_REMOVE_ON_COMPLETED_COUNT: 1000
BULLMQ_REMOVE_ON_FAIL_AGE: 604800000
BULLMQ_REMOVE_ON_FAIL_COUNT: 1000

Buffer Serialization

BullMQ serializes job data into Redis, transforming Buffer instances into the JSON-safe representation { type: 'Buffer', data: [byte, ...] }. The ImagePreprocessingService rehydrates these back into native Buffer instances via ensureBuffer() before passing them to Sharp:

private ensureBuffer(buffer: Buffer | SerializedBuffer): Buffer {
  if (isNodeBuffer(buffer)) return buffer;
  if (isSerializedBuffer(buffer)) return Buffer.from(buffer.data);
  throw new Error('Invalid buffer format');
}

Without this rehydration step, Sharp would receive a plain object and throw a TypeError on pipeline construction.

Performance Considerations

Metric Impact Mitigation
Memory per worker Holds image buffers + preprocessing variants Limit concurrency; resize before enqueue
GPU contention Multiple queues share one Ollama instance Tune keep_alive to prevent model eviction
Redis latency Job state transitions Co-locate KeyDB within the same datacenter
Cold start First job after idle loads model to GPU Pre-warm via health check pings if budget allows

Scaling Workers

Horizontal scaling follows this model:

Queue Suggested Concurrency Rationale
describe 2 Lightweight, CPU-bound description
compare 1 GPU-intensive multi-image inference
ocr 2 Fast processing with optional preprocessing overhead

Concurrency is configured via @ehildt/nestjs-bullmq queue registration:

QueueProcessor.forRoot({
  queues: [
    { name: 'describe', processors: [{ path: './describe.processor', concurrency: 2 }] },
    { name: 'compare', processors: [{ path: './compare.processor', concurrency: 1 }] },
  ],
});

Related Documentation

Clone this wiki locally