AliveMCP

Guide · Job Queues

MCP Tools for BullMQ — job lifecycle states, worker concurrency, retry backoff, flow producers

Three BullMQ behaviours catch developers building MCP job-dispatch tools: completed and failed jobs are not automatically removed — they accumulate in Redis until the queue fills memory, so every Queue must be created with defaultJobOptions: { removeOnComplete: true, removeOnFail: 1000 } or equivalent per-job options, and without this setting a long-lived MCP server will eventually exhaust Redis memory; worker concurrency and rate limiting operate at different layers and interact non-obviously — concurrency is the maximum number of jobs processed simultaneously per worker instance, while limiter applies a rate cap across all worker instances using a Redis token bucket, so setting concurrency: 5 on 3 worker instances gives 15 simultaneous jobs which may still violate a limiter: { max: 10, duration: 1000 } rate cap and cause jobs to stall waiting for the rate window; and the delayed state requires a separate scheduler process in BullMQ v3+ — calling queue.add('job', data, { delay: 5000 }) without a running QueueScheduler (v2) or Queue with autorun: true (v3) means delayed jobs silently remain in the delayed set and are never promoted to waiting.

TL;DR

Setup: new Queue('name', { connection: redisOpts, defaultJobOptions: { removeOnComplete: true, removeOnFail: 1000 } }). Add job: queue.add('jobName', data, { delay, priority, attempts, backoff }). Worker: new Worker('name', async (job) => { ... }, { connection, concurrency: 5 }). Events: new QueueEvents('name', { connection }) for cross-process event listening. Flow: new FlowProducer({ connection }).add({ name, queueName, data, children: [...] }). Health: await queue.getWorkers() + redis.ping().

Queue setup and connection management

BullMQ uses ioredis connections internally. Each Queue, Worker, and QueueEvents instance opens its own Redis connection — for an MCP server with 1 queue and 1 worker listening to events, that's at minimum 3 connections. Use a shared IORedis instance to cap connection count, but note that BullMQ requires connections in non-subscribing mode for queue operations and blocking mode for workers — pass enableReadyCheck: false, maxRetriesPerRequest: null on worker connections.

import { Queue, Worker, QueueEvents, Job } from 'bullmq';
import IORedis from 'ioredis';

const redisConnection = {
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
  password: process.env.REDIS_PASSWORD,
  tls: process.env.REDIS_TLS === 'true' ? {} : undefined,
  // Required for Worker connections — prevents ioredis from retrying indefinitely
  enableReadyCheck: false,
  maxRetriesPerRequest: null,
};

// Shared queue for enqueueing — does NOT need maxRetriesPerRequest: null
const queueConnection = {
  host: redisConnection.host,
  port: redisConnection.port,
  password: redisConnection.password,
  tls: redisConnection.tls,
};

const taskQueue = new Queue('agent-tasks', {
  connection: queueConnection,
  defaultJobOptions: {
    removeOnComplete: true,    // remove completed jobs immediately to prevent memory growth
    removeOnFail: 1000,        // keep last 1000 failed jobs for debugging
    attempts: 3,               // retry failed jobs up to 3 times
    backoff: {
      type: 'exponential',
      delay: 1000,             // 1s, 2s, 4s between retries
    },
  },
});

// Graceful shutdown — close queue to prevent new jobs from being lost
process.on('SIGTERM', async () => {
  await taskQueue.close();
});

For MCP servers deployed in containers, connect to Redis via the service hostname and set REDIS_PASSWORD from a secret. If using Redis Cluster, pass the cluster mode configuration — BullMQ supports Redis Cluster, but each queue name must hash to the same slot, which BullMQ handles by wrapping queue names in hash tags automatically.

Adding jobs — delay, priority, deduplication, and job IDs

BullMQ jobs are added with queue.add(name, data, opts). The job name is a label for grouping — multiple jobs with the same name can exist simultaneously. For MCP tools that dispatch agent tasks, set a deterministic jobId to prevent duplicate job creation when the MCP tool is called multiple times with the same intent. A job with a duplicate jobId returns the existing job without creating a new one.

// Add a simple job
const job = await taskQueue.add('generate-report', {
  userId: 'usr_123',
  reportType: 'monthly',
  month: '2026-07',
});
console.log(`Job ${job.id} queued`);

// Add a delayed job (executes after 30 seconds)
const delayedJob = await taskQueue.add(
  'send-reminder',
  { userId: 'usr_456', message: 'Your trial expires tomorrow' },
  { delay: 30_000 }  // milliseconds
);

// Add with priority (lower number = higher priority; 1 is highest, undefined is lowest)
await taskQueue.add('critical-alert', { alertId: 'alt_789' }, { priority: 1 });
await taskQueue.add('background-sync', { datasetId: 'ds_001' }, { priority: 10 });

// Deduplicate jobs using a deterministic jobId
async function dispatchOnce(userId: string, action: string): Promise {
  const jobId = `${action}:${userId}`;  // deterministic — same inputs → same ID
  const existing = await taskQueue.getJob(jobId);
  if (existing && (await existing.getState()) !== 'failed') {
    return existing.id!;  // reuse existing job
  }
  const newJob = await taskQueue.add(
    action,
    { userId },
    {
      jobId,             // custom ID — duplicate add returns existing job
      removeOnComplete: false,  // keep for deduplication check
    }
  );
  return newJob.id!;
}

// Bulk add for efficiency (single Redis pipeline)
await taskQueue.addBulk([
  { name: 'process-image', data: { imageId: 'img_1' } },
  { name: 'process-image', data: { imageId: 'img_2' } },
  { name: 'process-image', data: { imageId: 'img_3' } },
]);

The delay option requires BullMQ's scheduler to promote delayed jobs to the waiting state. In BullMQ v2, this required a separate QueueScheduler process. In BullMQ v3+, the Queue itself handles promotion when it has autorun: true (default), but at least one queue or worker instance must be running for promotion to happen. In a serverless environment where the MCP server exits between requests, delayed jobs will not be promoted until the next invocation — use queue.run() at startup to trigger promotion of any overdue delayed jobs.

Workers — concurrency, rate limiting, and job progress

A BullMQ Worker processes jobs from a queue. The concurrency option controls how many jobs the worker processes in parallel within a single process. For CPU-bound tasks, set concurrency to the number of CPU cores; for I/O-bound tasks (API calls, database queries), higher concurrency (16–64) improves throughput. The worker processor function receives a Job object with job.data, job.id, and methods like job.updateProgress() and job.log().

const worker = new Worker(
  'agent-tasks',
  async (job: Job) => {
    // job.name tells you which handler to run
    switch (job.name) {
      case 'generate-report':
        return await generateReport(job.data, job);
      case 'send-reminder':
        return await sendReminder(job.data);
      default:
        throw new Error(`Unknown job type: ${job.name}`);
    }
  },
  {
    connection: redisConnection,  // use the worker-specific connection (maxRetriesPerRequest: null)
    concurrency: 10,              // process 10 jobs simultaneously
    limiter: {
      max: 20,                    // max 20 jobs per duration window across ALL worker instances
      duration: 1000,             // 1 second window
    },
  }
);

async function generateReport(data: any, job: Job): Promise {
  // Report progress for long-running jobs (readable via QueueEvents or job.getState())
  await job.updateProgress(0);
  const records = await fetchRecords(data.userId);
  await job.updateProgress(33);
  await job.log(`Fetched ${records.length} records`);

  const report = await buildReport(records, data.reportType);
  await job.updateProgress(66);
  await job.log('Report built, uploading...');

  const url = await uploadReport(report);
  await job.updateProgress(100);

  // Return value is stored as job.returnvalue (available after completion)
  return { url, recordCount: records.length };
}

// Listen to worker events for logging
worker.on('completed', (job, result) => {
  console.log(`Job ${job.id} completed:`, result);
});
worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err.message);
});
worker.on('stalled', (jobId) => {
  // Job was processing when worker crashed — BullMQ auto-requeues stalled jobs
  console.warn(`Job ${jobId} stalled and requeued`);
});

// Graceful shutdown — wait for active jobs to complete
process.on('SIGTERM', async () => {
  await worker.close();
});

Stalled job detection: BullMQ workers send heartbeats to Redis every stalledInterval milliseconds (default 30 seconds). If a worker crashes without completing a job, other workers detect the stale lock and move the job back to waiting after maxStalledCount failures (default 1). Set maxStalledCount: 0 to move stalled jobs directly to failed state rather than retrying — appropriate for non-idempotent operations where re-execution could cause double-processing.

Job state inspection for MCP status tools

MCP tools that dispatch long-running jobs need a corresponding "check status" tool. BullMQ's job.getState() returns one of: waiting, active, completed, failed, delayed, prioritized, waiting-children, or unknown. Cross-process event listening requires QueueEvents — the worker and queue objects emit events only within the same process.

// Status check tool — called by agent to poll job progress
async function getJobStatus(jobId: string): Promise {
  const job = await taskQueue.getJob(jobId);
  if (!job) {
    return { status: 'not_found', jobId };
  }

  const state = await job.getState();
  const progress = job.progress;

  const result: Record = { jobId, state, progress };

  if (state === 'completed') {
    result.returnValue = job.returnvalue;
    result.finishedAt = new Date(job.finishedOn!).toISOString();
  } else if (state === 'failed') {
    result.error = job.failedReason;
    result.attemptsMade = job.attemptsMade;
    result.logs = await job.getLogs();
  } else if (state === 'delayed') {
    result.processAfter = new Date(job.timestamp + job.opts.delay!).toISOString();
  }

  return result;
}

// Queue-level metrics for health dashboards
async function getQueueMetrics(queueName: string) {
  const q = new Queue(queueName, { connection: queueConnection });
  const counts = await q.getJobCounts(
    'waiting', 'active', 'completed', 'failed', 'delayed', 'prioritized'
  );
  const workers = await q.getWorkers();
  await q.close();
  return {
    counts,
    activeWorkers: workers.length,
    workerIds: workers.map(w => w.id),
  };
}

// Cross-process event listener — fires when any worker completes/fails a job
const queueEvents = new QueueEvents('agent-tasks', { connection: queueConnection });
queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed:`, returnvalue);
});
queueEvents.on('failed', ({ jobId, failedReason }) => {
  console.error(`Job ${jobId} failed: ${failedReason}`);
});
queueEvents.on('progress', ({ jobId, data }) => {
  // 'data' is what was passed to job.updateProgress()
  console.log(`Job ${jobId} progress: ${data}%`);
});

		
For MCP tools that need to wait for job completion (synchronous-style dispatch + poll), use job.waitUntilFinished(queueEvents, timeout) which returns a promise that resolves with the return value on completion or rejects on failure. This is useful for short-running jobs where the MCP tool can block; for long-running jobs, return the jobId immediately and let the agent call a status-check tool on the next turn.

	

	

		
Flow producers — dependent job graphs

		
BullMQ FlowProducer creates trees of jobs where parent jobs wait for all children to complete before being promoted to waiting. This is useful for MCP tools that orchestrate multi-step agent pipelines — for example, parallel data fetches whose results feed a synthesis job. The parent job receives each child's return value in job.getChildrenValues().


		
import { FlowProducer } from 'bullmq';

const flow = new FlowProducer({ connection: queueConnection });

// Create a job tree: parent waits for all 3 children to complete
const tree = await flow.add({
  name: 'synthesize-research',
  queueName: 'agent-tasks',
  data: { topic: 'MCP server patterns' },
  children: [
    {
      name: 'fetch-github-data',
      queueName: 'agent-tasks',
      data: { query: 'mcp server github stars' },
    },
    {
      name: 'fetch-npm-data',
      queueName: 'agent-tasks',
      data: { package: '@modelcontextprotocol/sdk' },
    },
    {
      name: 'fetch-docs-data',
      queueName: 'agent-tasks',
      data: { url: 'https://modelcontextprotocol.io/docs' },
    },
  ],
});

// In the parent job processor, access children's results
async function synthesizeResearch(job: Job) {
  // wait-children state: parent is held until all children complete
  const childResults = await job.getChildrenValues();
  // childResults is a Record
  const [github, npm, docs] = Object.values(childResults);
  return buildSynthesis(github, npm, docs);
}


		
Flow jobs use a special waiting-children state — distinct from waiting — that is not processed by workers until all child jobs complete. If any child fails permanently (exhausts retries), the parent job transitions to failed with the reason from the failed child. Use opts.failParentOnFailure: false on individual children to allow partial failures where the parent should still proceed with available data.

	


	

		
Health probe — Redis ping and queue worker count

		
BullMQ's health depends entirely on Redis connectivity. Use redis.ping() as the liveness check and queue.getWorkers() as the readiness check — if no workers are registered, new jobs will queue but never execute. A common production issue is workers restarting without re-registering, causing jobs to accumulate silently.


		
import IORedis from 'ioredis';

async function checkBullMQHealth(queueName: string): Promise {
  const redis = new IORedis(queueConnection);
  try {
    // Liveness: Redis connectivity
    await redis.ping();

    // Readiness: at least one worker is registered
    const q = new Queue(queueName, { connection: queueConnection });
    const workers = await q.getWorkers();
    const counts = await q.getJobCounts('waiting', 'active', 'failed', 'delayed');
    await q.close();
    await redis.quit();

    return {
      alive: true,
      ready: workers.length > 0,
      workerCount: workers.length,
      queueDepth: counts.waiting + counts.delayed,
      activeJobs: counts.active,
      failedJobs: counts.failed,
    };
  } catch (err) {
    await redis.quit().catch(() => {});
    return { alive: false, error: String(err) };
  }
}

		
Register this health check in your MCP server's startup probe. If workerCount === 0 with jobs in the waiting state, the queue is backing up — this is a silent failure mode where MCP tools appear to succeed (jobs are enqueued) but no work is being done. AliveMCP can monitor this pattern by polling the queue depth metric and alerting when it grows without decreasing.

	

	

		
Related guides