MCP Tools for Job Queues: Delivery Model Spectrum, Retry and Failure Handling, and Scheduling Composition

Pattern 1: The Delivery Model Spectrum

The five systems in this arc represent five fundamentally different answers to the question of how a message gets from the publisher to the processor. Understanding the delivery model is the prerequisite for everything else — retry behaviour, health probes, and deployment requirements all follow from it.

BullMQ: Redis List Polling with Worker Lock Acquisition

BullMQ stores jobs as Redis hash objects and tracks their state using Redis sorted sets and lists. Adding a job with queue.add() pushes an entry into the waiting set; workers claim jobs by atomically moving entries from the waiting set to the active set via BRPOP-style commands. The job remains in the active set while it is being processed — the "lock" is a Redis key with an expiry that the worker must renew every lockDuration milliseconds. When the processor function returns, BullMQ moves the job to the completed set (or the failed set on exception). The completed and failed sets are the source of the most common BullMQ integration mistake: they accumulate indefinitely unless removeOnComplete and removeOnFail are configured at Queue creation.

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

// Queue connection — does NOT need maxRetriesPerRequest: null
const queueConnection = {
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
  password: process.env.REDIS_PASSWORD,
};

// Worker connection — must have these two options (BullMQ requirement)
const workerConnection = {
  ...queueConnection,
  enableReadyCheck: false,
  maxRetriesPerRequest: null,
};

// Queue: always configure removeOnComplete and removeOnFail
const taskQueue = new Queue('mcp-tasks', {
  connection: queueConnection,
  defaultJobOptions: {
    removeOnComplete: true,     // prevent memory exhaustion from completed jobs
    removeOnFail: 500,          // keep last 500 failed jobs for debugging
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
  },
});

// Worker: concurrency=10 means 10 simultaneous active jobs per worker process
const worker = new Worker(
  'mcp-tasks',
  async (job) => {
    switch (job.name) {
      case 'generate-report': return generateReport(job.data);
      case 'process-webhook': return processWebhook(job.data);
      default: throw new Error(`Unknown job type: ${job.name}`);
    }
  },
  { connection: workerConnection, concurrency: 10 }
);

// QueueEvents required for cross-process event listening
// (Worker events only fire in the same process)
const queueEvents = new QueueEvents('mcp-tasks', { connection: workerConnection });
queueEvents.on('completed', ({ jobId, returnvalue }) => {
  console.log(`Job ${jobId} completed`, returnvalue);
});

// Health probe: workers registered + Redis responding
async function bullmqHealth() {
  const workers = await taskQueue.getWorkers();
  if (workers.length === 0) throw new Error('No workers registered');
  await (new IORedis(queueConnection)).ping();
}

process.on('SIGTERM', async () => {
  await worker.close();
  await taskQueue.close();
});

The connection requirement split is the most common BullMQ setup mistake: enableReadyCheck: false and maxRetriesPerRequest: null are required on Worker connections because workers use blocking commands that ioredis's default connection mode does not support. Queue connections used only for add() calls do not need these options. If you pass the worker connection options to the Queue constructor, it works — but if you pass queue connection options to the Worker constructor, the worker silently fails to start processing jobs.

For MCP servers, the key implication is that BullMQ requires a long-lived worker process running alongside the MCP server process. A short-lived serverless MCP invocation can enqueue jobs via queue.add(), but it cannot process them. The worker must be a separate persistent process or a container that stays running between requests.

Celery: Broker-Agnostic Task Dispatch with Configurable Acknowledgement

Celery decouples the transport layer (broker) from the result storage layer (backend). The broker carries task messages from the caller to the worker; the backend stores return values so callers can retrieve them later. Both can be the same Redis instance, but they serve different purposes and have different TTL requirements. The default pickle serializer is a security risk — an attacker who can write to the broker can execute arbitrary code on every worker by crafting a malicious pickle payload. Set task_serializer='json' unconditionally.

from celery import Celery
import os

app = Celery(
    'mcp_tasks',
    broker=os.environ['CELERY_BROKER_URL'],    # redis://host:6379/0
    backend=os.environ['CELERY_RESULT_BACKEND'], # redis://host:6379/1
)
app.conf.update(
    task_serializer='json',       # NEVER use default pickle
    accept_content=['json'],
    result_serializer='json',
    result_expires=3600 * 24 * 7, # 7 days; without this, default varies by backend
    timezone='UTC',
    enable_utc=True,
    task_acks_late=True,          # ack after completion, not at receive (crash-safe)
    task_reject_on_worker_lost=True, # requeue if worker dies holding the task
)

@app.task(bind=True, max_retries=3, queue='mcp-tasks')
def generate_report(self, user_id: str, report_type: str) -> dict:
    try:
        records = fetch_records(user_id)
        report = build_report(records, report_type)
        return {'url': upload_report(report), 'record_count': len(records)}
    except TemporaryError as exc:
        # Exponential backoff: 60s, 120s, 240s
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

# MCP tool: dispatch and return task ID for polling
def mcp_generate_report(user_id: str, report_type: str) -> dict:
    task = generate_report.apply_async(args=[user_id, report_type])
    return {'task_id': task.id, 'status': 'queued'}

# MCP tool: poll for result
def mcp_get_report_status(task_id: str) -> dict:
    from celery.result import AsyncResult
    result = AsyncResult(task_id, app=app)
    # PENDING state is ambiguous: waiting OR never-existed OR expired result
    # Always store task IDs in your own DB to distinguish "queued" from "unknown"
    if result.state == 'PENDING':
        return {'status': 'pending_or_expired'}
    elif result.state == 'SUCCESS':
        return {'status': 'success', 'result': result.result}
    elif result.state == 'FAILURE':
        return {'status': 'failed', 'error': str(result.result)}
    return {'status': result.state.lower()}

The acknowledgement mode determines crash safety. With the default task_acks_early=True, the broker removes the task message from the queue the moment a worker receives it — if the worker crashes during processing, the task is lost. With task_acks_late=True, the broker holds the message until the processor returns (or raises); combined with task_reject_on_worker_lost=True, a crashed worker causes the message to be requeued rather than discarded. The cost of late acknowledgement is that a task can be executed twice if the worker crashes after completing the work but before sending the ack — tasks must be idempotent.

Inngest: HTTP Push to Your Endpoint with Serverless-Native Memoization

Inngest inverts the delivery model entirely. Instead of your worker polling a queue, Inngest's cloud calls your HTTP endpoint when an event triggers a function. Your endpoint must be publicly reachable. The function is stateless — after each step.run() completes, Inngest serializes the accumulated state to its cloud and the serverless invocation exits. On the next step, Inngest calls your endpoint again with the full state, injecting previously completed step results. This means all code outside step.run() blocks executes on every continuation — side effects like counters, API calls, and logging that live outside step blocks fire once per step, not once per function.

import { Inngest } from 'inngest';
import { serve } from 'inngest/express';
import express from 'express';

export const inngest = new Inngest({
  id: 'mcp-agent-server',
  signingKey: process.env.INNGEST_SIGNING_KEY,
  eventKey: process.env.INNGEST_EVENT_KEY,
});

// Function: all side effects MUST be inside step.run() blocks
export const processAgentTaskFn = inngest.createFunction(
  { id: 'process-agent-task', retries: 3 },
  { event: 'mcp/agent.task.requested' },
  async ({ event, step }) => {
    // WARNING: this line runs on EVERY step resume, not just the first
    // Put logging here; never put API calls, DB writes, or counters here
    const taskId = event.data.taskId;  // safe to read — event data is immutable

    // step.run is memoized — if this step already ran, the callback is skipped
    // and the saved result is injected. Retries are per-step, independent.
    const rawData = await step.run('fetch-source-data', async () => {
      return fetchFromExternalAPI(event.data.sourceUrl);
    });

    const transformed = await step.run('transform-data', async () => {
      return transformData(rawData);  // rawData comes from memoized step 1
    });

    // step.sleep suspends without a timer process or cron job
    await step.sleep('rate-limit-pause', '3s');

    const result = await step.run('send-results', async () => {
      return sendToDestination(transformed, event.data.destinationId);
    });

    return { taskId, status: 'completed', resultId: result.id };
  }
);

// MCP tool: dispatch event and return immediately
async function mcpDispatchTask(taskData: object): Promise<{ runId: string }> {
  const [{ id: runId }] = await inngest.send({
    name: 'mcp/agent.task.requested',
    data: taskData,
  });
  return { runId };
}

const app = express();
app.use('/api/inngest', serve({ client: inngest, functions: [processAgentTaskFn] }));
app.listen(3000);

The event-based trigger means Inngest functions are decoupled from the MCP tool call — the MCP tool sends an event and returns a run ID immediately; the function executes asynchronously. This is the correct pattern for long-running agent tasks (LLM chains, multi-step data pipelines, approval workflows). For local development, the Inngest Dev Server (npx inngest-cli@latest dev) replaces Inngest's cloud and runs on localhost, providing a UI at http://localhost:8288 showing function runs, event history, and step-level execution details.

Azure Service Bus: AMQP Long-Poll with Lease-Based Message Locking

Azure Service Bus uses the AMQP 1.0 protocol for message transport. The @azure/service-bus SDK maintains a persistent AMQP connection to the Service Bus namespace. When a receiver calls getMessageIterator(), it long-polls — the AMQP link stays open and messages arrive as they are enqueued, without the receiver repeatedly polling the API. Processing a message requires holding its lock for the duration: the queue's LockDuration (default 60 seconds) is a timer that starts when the message is delivered. If the processor does not complete — or call renewMessageLock() — before the timer expires, the message is unlocked and becomes visible to other consumers. This causes silent double-processing for any operation that takes longer than LockDuration.

import { ServiceBusClient, ServiceBusAdministrationClient } from '@azure/service-bus';
import { DefaultAzureCredential } from '@azure/identity';

const namespace = `${process.env.SERVICE_BUS_NAMESPACE}.servicebus.windows.net`;
const sbClient = new ServiceBusClient(namespace, new DefaultAzureCredential());
const adminClient = new ServiceBusAdministrationClient(namespace, new DefaultAzureCredential());

// Create queue with appropriate LockDuration for your processing time
async function ensureQueue(name: string): Promise<void> {
  if (!await adminClient.queueExists(name)) {
    await adminClient.createQueue(name, {
      lockDuration: 'PT5M',               // 5 minutes — increase for slow processors
      maxDeliveryCount: 5,                // DLQ after 5 failed attempts
      deadLetteringOnMessageExpiration: true,
      requiresDuplicateDetection: true,
      duplicateDetectionHistoryTimeWindow: 'PT10M',
    });
  }
}

// Receiver: must complete or renewMessageLock before LockDuration expires
async function startWorker(queueName: string): Promise<void> {
  const receiver = sbClient.createReceiver(queueName, {
    receiveMode: 'peekLock',  // default — holds lock until complete/abandon/deadletter
  });

  for await (const message of receiver.getMessageIterator()) {
    // autoCompleteMessages:false gives manual control
    const renewInterval = setInterval(async () => {
      // Renew before LockDuration expires (every 4 min if LockDuration=5min)
      await receiver.renewMessageLock(message);
    }, 4 * 60 * 1000);

    try {
      await processTask(message.body);
      clearInterval(renewInterval);
      await receiver.completeMessage(message);  // removes from queue
    } catch (err) {
      clearInterval(renewInterval);
      if (isRecoverable(err)) {
        await receiver.abandonMessage(message);  // re-delivery; increments deliveryCount
      } else {
        await receiver.deadLetterMessage(message, {
          deadLetterReason: 'ProcessingError',
          deadLetterErrorDescription: String(err),
        });
      }
    }
  }
}

// Health probe: queue runtime properties
async function serviceBusHealth(queueName: string): Promise<void> {
  const props = await adminClient.getQueueRuntimeProperties(queueName);
  // activeMessageCount does NOT include DLQ messages
  // Check deadLetterMessageCount separately to surface silent failures
  if (props.deadLetterMessageCount > 100) {
    throw new Error(`DLQ has ${props.deadLetterMessageCount} messages — investigate`);
  }
}

process.on('SIGTERM', async () => { await sbClient.close(); });

Standard tier supports only queues. Topics and subscriptions — which enable fan-out (one message delivered to multiple subscribers) — require Premium tier. The error when you attempt to create a topic on Standard tier is MessagingEntityAlreadyExistsError, which is confusing because it implies the topic already exists rather than that the tier does not support it. Premium tier also enables VNET integration, availability zones, and message sizes up to 100 MB.

QStash: HTTP Push Delivery — No Poll, No Worker, No Persistent Connection

QStash (by Upstash) requires no worker process, no broker connection, and no infrastructure in your deployment. The publisher sends a message to the QStash API with a destination URL; QStash delivers it via HTTP POST to that URL, retrying on non-2xx responses. The destination can be any publicly reachable endpoint — a serverless function, a Next.js API route, a webhook handler. The critical security requirement is that all receiving endpoints must verify the Upstash-Signature header using QStash's HMAC-SHA256 signing key. Without verification, anyone who discovers the endpoint URL can trigger execution by sending a crafted POST request.

import { Client, Receiver } from '@upstash/qstash';
import express from 'express';

const qstash = new Client({ token: process.env.QSTASH_TOKEN! });

// Both keys required — nextSigningKey handles rotation window
const receiver = new Receiver({
  currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
  nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
});

// Publisher: dispatch with retry, failure callback, deduplication
async function dispatchTask(taskId: string, data: object): Promise<string> {
  const response = await qstash.publishJSON({
    url: `${process.env.WORKER_BASE_URL}/api/tasks/process`,
    body: data,
    retries: 3,
    timeout: 30,
    deduplicationId: taskId,  // same taskId within 24h = single delivery (silent drop of 2nd)
    // ALWAYS set failureCallback — without it, permanently failed messages disappear
    failureCallback: `${process.env.WORKER_BASE_URL}/api/tasks/failure`,
  });
  return response.messageId;
}

// Receiving endpoint: verify signature BEFORE JSON parse
const app = express();
app.post('/api/tasks/process',
  express.raw({ type: '*/*' }),  // express.raw() not express.json() — sig covers raw bytes
  async (req, res) => {
    const rawBody = req.body.toString('utf-8');
    const signature = req.headers['upstash-signature'] as string;

    if (!signature) return res.status(401).json({ error: 'Missing signature' });

    try {
      await receiver.verify({ signature, body: rawBody, clockTolerance: 5 });
    } catch {
      return res.status(401).json({ error: 'Invalid signature' });
    }

    // Parse only after verification
    const data = JSON.parse(rawBody);

    try {
      await processTask(data);
      res.json({ status: 'processed' });  // 2xx stops retries
    } catch (err) {
      res.status(500).json({ error: 'Processing failed' });  // non-2xx triggers retry
    }
  }
);

// CRON schedule: survives server restarts — managed by QStash
async function createWeeklyReport(): Promise<string> {
  const schedule = await qstash.schedules.create({
    destination: `${process.env.WORKER_BASE_URL}/api/reports/weekly`,
    cron: '0 9 * * 1',  // every Monday 9:00 UTC
    body: JSON.stringify({ type: 'weekly-summary' }),
  });
  return schedule.scheduleId;
}

The most common production failure with QStash is a stale signing key after rotating keys in the Upstash console. The Receiver class tries both currentSigningKey and nextSigningKey — if you update only one of them in your environment variables, the verification window during rotation still covers both keys. Failing to update both keys after the rotation window closes causes all QStash deliveries to fail verification until the environment variables are updated and the service is redeployed.

Pattern 2: Retry and Failure Handling

Retry semantics, failure isolation, and dead-letter handling differ significantly across the five systems. The failure modes that catch teams are not the obvious ones (like not configuring retries at all) but the subtle ones: Celery's PENDING state ambiguity, BullMQ's stall detection gaps for non-idempotent operations, Inngest's null return from waitForEvent on timeout, Azure Service Bus's invisible DLQ accumulation, and QStash's silent deduplication window.

BullMQ: Per-Job Retry with Exponential Backoff and Stall Detection

BullMQ retries are configured per-job (in defaultJobOptions or in the per-job opts argument). When a job processor throws, BullMQ checks the job's remaining attempts: if attempts remain, the job transitions to the delayed state with a backoff wait before re-entering waiting; if attempts are exhausted, the job transitions to failed. The failed set is the BullMQ dead-letter equivalent — failed jobs stay there until removed by removeOnFail or manual cleanup.

Stall detection is separate from explicit failures. BullMQ workers send a heartbeat every stalledInterval milliseconds (default 30 seconds). If a worker stops sending heartbeats (process crash, OS kill, network partition), BullMQ's stalledInterval timer in the Queue detects the stalled active jobs and requeues them. For non-idempotent operations — operations that must not execute twice — set maxStalledCount: 0 on the Worker to move stalled jobs directly to the failed set instead of requeuing them.

// Per-job retry with exponential backoff
const job = await taskQueue.add('process-payment', paymentData, {
  attempts: 5,
  backoff: {
    type: 'exponential',
    delay: 2000,  // 2s, 4s, 8s, 16s, 32s
  },
  removeOnFail: { count: 500 },  // keep last 500 failed jobs, remove older ones
});

// Worker with stall detection for non-idempotent operations
const worker = new Worker('mcp-tasks', processor, {
  connection: workerConnection,
  concurrency: 5,
  stalledInterval: 30_000,  // check every 30s
  maxStalledCount: 0,       // non-idempotent: stalled → failed (not requeued)
});

// FlowProducer: parent waits until ALL children complete
const flow = await new FlowProducer({ connection: queueConnection }).add({
  name: 'synthesize-report',
  queueName: 'synthesis-tasks',
  data: { reportId: 'rpt_001' },
  children: [
    { name: 'fetch-source-a', queueName: 'fetch-tasks', data: { source: 'a' } },
    { name: 'fetch-source-b', queueName: 'fetch-tasks', data: { source: 'b' } },
    { name: 'fetch-source-c', queueName: 'fetch-tasks', data: { source: 'c' } },
    // failParentOnFailure: false (default true) — set false for partial-result tolerance
  ],
});
// Parent job enters 'waiting-children' state; moves to 'waiting' only when all children complete

Celery: Self-Retry with Countdown, chord_propagates, and Result Expiry Traps

Celery retries are triggered explicitly inside the task function by calling self.retry(exc=exc, countdown=N). There is no automatic retry on exception by default — the max_retries on the task decorator sets the ceiling, but raising self.retry() is what triggers the retry. A task that raises a non-SoftTimeLimitExceeded exception without calling self.retry() goes directly to the failure state. The result backend stores the exception as the task result with state FAILURE.

The PENDING state ambiguity is a persistent source of bugs in MCP status-check tools. AsyncResult(task_id).state == 'PENDING' means one of three things: the task is waiting to be processed; the task ID was never created (you have a bug); or the result expired from the backend TTL and was deleted. Celery cannot distinguish these cases from the result backend alone. The workaround is to store task IDs in your own database at dispatch time — if an ID exists in your DB but returns PENDING from Celery, it is either queued or expired, not unknown.

from celery import chord, group, chain
from celery.result import AsyncResult

# Chord: parallel group → single callback
# REQUIRES chord_propagates=True (Celery 4+ default) or False = callback fires on failure
@app.task(bind=True, name='tasks.synthesize')
def synthesize_all(self, results: list) -> dict:
    # results is a list of return values from the group tasks
    return {'combined': merge(results)}

@app.task(name='tasks.fetch')
def fetch_from_source(source_id: str) -> dict:
    return {'data': pull(source_id)}

def mcp_dispatch_parallel_fetch(source_ids: list[str]) -> dict:
    # chord = group of fetch tasks → synthesize callback
    workflow = chord(
        group(fetch_from_source.s(sid) for sid in source_ids)
    )(synthesize_all.s())
    # With chord_propagates=True (default), if any group task fails,
    # synthesize_all receives ChordError — handle it explicitly
    return {'chord_id': workflow.id}

# Result polling with PENDING ambiguity workaround
def mcp_get_status(task_id: str, known_task_ids: set) -> dict:
    if task_id not in known_task_ids:
        return {'status': 'unknown', 'reason': 'task_id_not_in_registry'}
    result = AsyncResult(task_id, app=app)
    if result.state == 'PENDING':
        # Could be: (a) actually queued, (b) result expired, (c) ID typo
        # Cannot distinguish without checking task creation timestamp in own DB
        return {'status': 'pending_or_expired'}
    return {'status': result.state.lower(), 'result': result.result if result.ready() else None}

Inngest: Per-Step Independent Retries and waitForEvent Null Handling

Inngest retries operate at two levels: the whole-function retries config (applies to non-step code and uncaught exceptions) and per-step retry counts (passed as the third argument to step.run()). When a step.run() callback throws, Inngest retries only that step — previously completed steps are not re-executed. This makes Inngest particularly suitable for multi-step LLM pipelines where early steps (embedding generation, context retrieval) are expensive and the later steps (LLM completion, post-processing) are more likely to fail.

The step.waitForEvent null return is the most common silent failure in Inngest MCP integrations. When the function suspends waiting for an event and the timeout expires, waitForEvent returns null — it does not throw. Code that destructures the return value without checking for null proceeds silently with undefined properties, producing incorrect output without any error.

export const approvalWorkflowFn = inngest.createFunction(
  { id: 'mcp-approval-workflow', retries: 2 },
  { event: 'mcp/approval.requested' },
  async ({ event, step }) => {
    // Step 1: Send approval request (retried independently up to 3 times)
    await step.run('send-approval-request', async () => {
      await sendSlackApprovalRequest({
        requestId: event.data.requestId,
        approverSlackId: event.data.approverSlackId,
      });
    }, { retries: 3 });  // per-step retry count, independent of function-level retries

    // Step 2: Wait for approval event (timeout returns null — NOT throw)
    const approvalEvent = await step.waitForEvent('wait-for-approval', {
      event: 'mcp/approval.responded',
      timeout: '24h',
      // CEL expression to correlate — must match exactly
      if: 'event.data.requestId == async.data.requestId',
    });

    // ALWAYS check for null — timeout returns null, not exception
    if (approvalEvent === null) {
      await step.run('handle-timeout', async () => {
        await markRequestExpired(event.data.requestId);
      });
      return { requestId: event.data.requestId, status: 'expired' };
    }

    if (approvalEvent.data.approved) {
      const result = await step.run('execute-approved-action', async () => {
        return executeAction(event.data.actionPayload);
      });
      return { requestId: event.data.requestId, status: 'approved', result };
    }

    return { requestId: event.data.requestId, status: 'rejected' };
  }
);

Azure Service Bus: maxDeliveryCount, Lock Expiry, and Invisible DLQ

Azure Service Bus moves messages to the dead-letter queue (DLQ) automatically when a message's deliveryCount reaches the queue's maxDeliveryCount limit (default 10). Each abandonMessage() call increments the delivery count. A message that is abandoned enough times ends up in the DLQ at the path {queue}/$DeadLetterQueue. The critical operational trap: DLQ messages do not appear in the queue's activeMessageCount — the queue looks empty while failures accumulate silently in the DLQ. Monitoring deadLetterMessageCount separately from activeMessageCount is mandatory for any production integration.

// DLQ consumer — drain and inspect permanently failed messages
async function drainDLQ(queueName: string): Promise<void> {
  const dlqReceiver = sbClient.createReceiver(queueName, {
    subQueueType: 'deadLetter',
  });

  for await (const msg of dlqReceiver.getMessageIterator()) {
    console.log({
      messageId: msg.messageId,
      deliveryCount: msg.deliveryCount,
      deadLetterReason: msg.deadLetterReason,
      deadLetterErrorDescription: msg.deadLetterErrorDescription,
      body: msg.body,
    });
    // After inspection, complete to remove from DLQ (or abandon to retry via normal queue)
    await dlqReceiver.completeMessage(msg);
  }
}

// Scheduled message: delivered at specific time
async function scheduleMessage(queueName: string, body: object, deliverAt: Date): Promise<void> {
  const sender = sbClient.createSender(queueName);
  const seqNos = await sender.scheduleMessages(
    [{ body, messageId: crypto.randomUUID() }],
    deliverAt
  );
  await sender.close();
  console.log('Scheduled message sequence numbers:', seqNos);
}

// Session-enabled queue: ordered processing per session ID (e.g., per-tenant, per-user)
// Requires sessionId property on each message and acceptSession() on receiver
async function processSessionQueue(queueName: string, sessionId: string): Promise<void> {
  const sessionReceiver = await sbClient.acceptSession(queueName, sessionId, {
    maxAutoLockRenewalDurationInMs: 5 * 60 * 1000,  // auto-renew lock up to 5min
  });
  for await (const msg of sessionReceiver.getMessageIterator()) {
    await processInOrder(msg.body);
    await sessionReceiver.completeMessage(msg);
  }
}

QStash: Retry Exhaustion, failureCallback as DLQ, and Deduplication Window

QStash retries failed deliveries with exponential backoff — 3 attempts by default, configurable up to the plan limit. After all retries are exhausted, the message is discarded unless a Upstash-Failure-Callback (or SDK equivalent failureCallback) was configured on the original publish. The failure callback receives the failed message body and delivery metadata, allowing you to implement a DLQ pattern by writing to a database or alerting system. Without it, permanently failed messages vanish without trace.

The deduplicationId window is 24 hours for queued messages and longer for CRON schedules. Two publishes with the same deduplicationId within that window result in a single delivery — the second publish is acknowledged as successful (returns the same messageId) but produces no additional delivery. This is intentional for idempotent dispatch (prevent duplicate deliveries when an MCP tool is called multiple times with the same intent) but will silently discard intentional second sends if the deduplication window has not expired.

// Named queue with rate limiting: useful for rate-limited downstream APIs
const openAIQueue = qstash.queue({ queueName: 'openai-requests' });

async function dispatchToOpenAIQueue(data: object, taskId: string): Promise<string> {
  const response = await openAIQueue.enqueueJSON({
    url: `${process.env.WORKER_BASE_URL}/api/openai-proxy`,
    body: data,
    retries: 5,
    deduplicationId: taskId,
    // failureCallback receives POSTed JSON with:
    // { sourceMessageId, topicName, url, method, header, body, maxRetries, lastDeliveryAttempt }
    failureCallback: `${process.env.WORKER_BASE_URL}/api/tasks/failure`,
  });
  return response.messageId;
}

// Queue with parallelism:1 for FIFO sequential delivery
async function createSequentialQueue(): Promise<void> {
  await qstash.queues.upsert({
    queueName: 'sequential-tasks',
    parallelism: 1,  // FIFO: one delivery at a time to the destination URL
  });
}

// Check message status
async function getMessageStatus(messageId: string): Promise<object> {
  const msg = await qstash.messages.get(messageId);
  return {
    state: msg.state,           // 'DELIVERED' | 'FAILED' | 'RETRY' | 'ERROR'
    deliveredAt: msg.deliveredAt,
    responseStatus: msg.responseStatus,
  };
}

Pattern 3: Scheduling and Workflow Composition

The five systems also diverge significantly in how they handle delayed dispatch, recurring schedules, and multi-step workflow composition. These patterns are central to MCP server implementations that orchestrate agent workflows — where a tool invocation triggers a multi-step process that spans minutes or hours rather than milliseconds.

BullMQ: Priority Queues, Delayed Jobs, and FlowProducer Dependency Trees

BullMQ supports three scheduling primitives: priority (integer on the job, lower value = higher priority), delay (job enters delayed state and is promoted to waiting after the specified millisecond offset), and cron-style repeatable jobs via queue.add(name, data, { repeat: { pattern: '0 9 * * *' } }). Delayed jobs in BullMQ v3+ are promoted by the Queue itself when autorun: true (the default) — but promotion only happens while at least one Queue or Worker instance is running. In a serverless context where all processes exit between requests, delayed jobs will not be promoted until the next process startup.

FlowProducer enables dependency trees — a parent job that waits for all of its children to complete before it becomes eligible for processing. The parent enters the waiting-children state and transitions to waiting only when all children have reached the completed state. If a child fails and its retry attempts are exhausted, the parent defaults to failing as well (controlled by failParentOnFailure per child).

import { FlowProducer, Queue } from 'bullmq';

// Priority queue: urgent tasks preempt background tasks
const reportQueue = new Queue('reports', { connection: queueConnection });
await reportQueue.add('urgent-report', data, { priority: 1 });      // highest
await reportQueue.add('background-report', data, { priority: 100 }); // lower priority

// Delayed job: execute after 10 minutes
await reportQueue.add('scheduled-summary', data, { delay: 10 * 60 * 1000 });

// Repeatable job: runs on CRON pattern (persists in Redis, survives restarts)
await reportQueue.add('weekly-digest', {}, {
  repeat: { pattern: '0 9 * * 1' },  // every Monday 09:00
  jobId: 'weekly-digest',             // deduplicate to prevent duplicate CRON registrations
});

// FlowProducer: fan-out then synthesize
const flow = new FlowProducer({ connection: queueConnection });
await flow.add({
  name: 'generate-weekly-report',
  queueName: 'report-synthesis',
  data: { weekStart: '2026-07-20' },
  children: [
    {
      name: 'fetch-metrics',
      queueName: 'data-fetch',
      data: { source: 'metrics', week: '2026-07-20' },
      opts: { failParentOnFailure: true },  // parent fails if this child fails
    },
    {
      name: 'fetch-alerts',
      queueName: 'data-fetch',
      data: { source: 'alerts', week: '2026-07-20' },
      opts: { failParentOnFailure: false }, // partial result tolerated
    },
    {
      name: 'fetch-uptime',
      queueName: 'data-fetch',
      data: { source: 'uptime', week: '2026-07-20' },
    },
  ],
});

Celery: ETA with Clock Sync, Chord/Group/Chain, and the PENDING Ambiguity Workaround

Celery's eta parameter schedules a task for execution at a specific UTC datetime. The broker holds the task message until the ETA arrives, then delivers it to an available worker. The critical requirement: the broker and worker system clocks must be synchronized. If a worker's clock is behind the broker's by more than a few seconds and the ETA is in the near future, the task executes immediately rather than waiting. This is a silent correctness failure with no error — the task simply runs earlier than intended.

The three composition primitives — chain, group, and chord — cover the full space of sequential, parallel, and parallel-then-sequential workflows. The si() vs s() distinction is the most common composition bug: task.s(arg) creates a signature that prepends the previous task's return value to the arguments; task.si(arg) creates an immutable signature that ignores the previous result. Using s() in a chain when the task expects only its own arguments causes a type error or incorrect argument binding that is often only visible at runtime.

from celery import chain, group, chord
from datetime import datetime, timezone, timedelta

# ETA scheduling: clock sync is critical
def schedule_task_at(user_id: str, run_at: datetime) -> str:
    # run_at must be timezone-aware; worker and broker clocks must match
    task = generate_report.apply_async(
        args=[user_id],
        eta=run_at,  # datetime in UTC
    )
    return task.id

# Chain: task_b.s() prepends task_a result to task_b's args
# task_b.si() ignores task_a's result (use for tasks with fixed args)
sequential = chain(
    fetch_data.s(source_id),           # returns: {'data': [...]}
    normalize_data.s(),               # receives {'data': [...]} as first positional arg
    store_result.si(destination_id),  # ignores normalize result — takes only destination_id
)
result = sequential.apply_async()

# Group: parallel, returns list of results in order
parallel = group(fetch_from_source.s(sid) for sid in ['a', 'b', 'c'])
group_result = parallel.apply_async()
# group_result.get() blocks until all complete

# Chord: parallel group → single callback (chord_propagates=True is default since Celery 4)
workflow = chord(
    group(fetch_from_source.s(sid) for sid in ['a', 'b', 'c'])
)(synthesize.s())
# synthesize receives list of group results when all complete
# If any group task fails with chord_propagates=True, synthesize receives ChordError

Inngest: Durable Waits with step.sleep, Fan-Out, and Function Config Controls

Inngest's scheduling primitives do not require cron jobs, timers, or external schedulers. step.sleep('label', duration) suspends the function for the specified duration — the serverless invocation exits, Inngest schedules a resumption, and the function continues from the next step when the duration expires. step.waitForEvent('label', { event, timeout, if }) suspends until a matching event arrives or the timeout expires. Neither requires a running process during the wait — Inngest manages the resumption schedule externally.

Concurrency, throttle, and debounce are defined in the function configuration object and enforced by Inngest's cloud, not by your code. concurrency: { limit: 5 } means Inngest will not run more than 5 instances of the function simultaneously. debounce: { period: '5s' } means Inngest waits 5 seconds after the last matching event before triggering the function — useful for rate-limiting MCP tools that should not fire on every event in a burst.

// Function with durable scheduling, fan-out, and concurrency config
export const orchestratorFn = inngest.createFunction(
  {
    id: 'mcp-orchestrator',
    retries: 2,
    concurrency: { limit: 10 },           // max 10 concurrent runs
    throttle: { count: 50, period: '1m' }, // max 50 new runs per minute
    debounce: { period: '3s' },            // wait 3s after last event before triggering
  },
  { event: 'mcp/orchestration.requested' },
  async ({ event, step }) => {
    // Fan-out: send multiple events in one call (buffered until step completes)
    const sourceIds: string[] = event.data.sourceIds;
    await step.run('dispatch-parallel-fetches', async () => {
      await inngest.send(
        sourceIds.map(id => ({
          name: 'mcp/fetch.requested',
          data: { sourceId: id, parentRunId: event.data.runId },
        }))
      );
      // inngest.send inside step.run — events delivered after step.run completes
    });

    // Wait for all parallel fetches to signal completion
    // In practice: poll a DB record or use a counter with step.waitForEvent
    await step.sleep('allow-fetches-to-complete', '30s');

    // Durable wait: suspend until approval or timeout
    const approval = await step.waitForEvent('wait-for-approval', {
      event: 'mcp/approval.granted',
      timeout: '48h',
      if: `event.data.runId == async.data.runId`,
    });

    if (approval === null) {
      return { status: 'expired' };  // timeout — not throw
    }

    const result = await step.run('execute-approved-workflow', async () => {
      return executeWorkflow(event.data.payload);
    });

    return { status: 'completed', result };
  }
);

Azure Service Bus: scheduleMessages, Session Queues, and Topic Filters

Service Bus supports deferred message delivery at the protocol level via sender.scheduleMessages(messages, enqueueTimeUtc). Scheduled messages are held in the Service Bus namespace and become visible at the specified time without requiring a running client. This is the correct approach for time-sensitive agent workflows that must fire at a specific UTC datetime regardless of client uptime — unlike BullMQ's delayed jobs, which require a running Queue instance to promote the delayed job to the waiting state.

Session-enabled queues implement ordered per-entity processing. Messages with the same sessionId are delivered in order and only to one receiver at a time — the receiver that calls acceptSession(queueName, sessionId) holds the session until it releases it. This pattern is natural for per-tenant or per-user ordered workflows in MCP servers that manage multiple independent agent sessions.

// scheduleMessages: fire at specific time, no client required during the wait
async function scheduleAgentTask(taskData: object, runAt: Date): Promise<bigint[]> {
  const sender = sbClient.createSender('agent-tasks');
  const seqNos = await sender.scheduleMessages(
    [{
      body: taskData,
      messageId: crypto.randomUUID(),
      subject: 'agent-task',
    }],
    runAt  // Date object in UTC
  );
  await sender.close();
  return seqNos;  // sequence numbers can be used to cancel via cancelScheduledMessages()
}

// Topic subscription with SQL filter (Premium tier only)
async function setupFilteredSubscription(topicName: string): Promise<void> {
  if (!await adminClient.topicExists(topicName)) {
    await adminClient.createTopic(topicName);
  }
  if (!await adminClient.subscriptionExists(topicName, 'critical-only')) {
    await adminClient.createSubscription(topicName, 'critical-only', {
      lockDuration: 'PT2M',
      maxDeliveryCount: 3,
    });
    // SQL filter on ApplicationProperties
    await adminClient.createRule(topicName, 'critical-only', 'severity-filter', {
      sqlExpression: "severity = 'critical' OR severity = 'high'",
    });
    await adminClient.deleteRule(topicName, 'critical-only', '$Default');
  }
}

QStash: External CRON Schedules, Named Queue Rate Control, and Agent-Managed Idempotency

QStash CRON schedules are managed by Upstash's infrastructure — they survive server restarts, deployments, and process exits. Creating a schedule via qstash.schedules.create() registers it in the Upstash cloud, which triggers deliveries on the specified cron pattern regardless of whether your server is running. The schedule persists until explicitly deleted with qstash.schedules.delete(scheduleId). This is fundamentally different from BullMQ's repeatable jobs (stored in Redis, lost if Redis is flushed) or Celery's celerybeat (requires a separate beating process).

Named queues with parallelism control are QStash's rate-limiting primitive for downstream APIs. Setting parallelism: 1 on a queue means QStash delivers one message at a time to the destination URL — the next delivery waits for the previous one to receive a 2xx response. This implements FIFO delivery with automatic back-pressure, useful for MCP servers that fan out to a rate-limited external API (LLM providers, payment processors, email services).

// CRON schedule management — persists in Upstash cloud, not your server
async function setupWeeklyReports(): Promise<string> {
  // List existing schedules to avoid duplicates
  const existing = await qstash.schedules.list();
  const weeklyExists = existing.find(s => s.destination === `${WORKER_URL}/api/reports/weekly`);
  if (weeklyExists) return weeklyExists.scheduleId;

  const schedule = await qstash.schedules.create({
    destination: `${WORKER_URL}/api/reports/weekly`,
    cron: '0 9 * * 1',
    retries: 3,
    body: JSON.stringify({ type: 'weekly-mcp-registry-summary' }),
    failureCallback: `${WORKER_URL}/api/tasks/failure`,
  });
  return schedule.scheduleId;
}

// Named queue: FIFO rate-limited delivery to OpenAI
async function dispatchToLLMQueue(data: object, requestId: string): Promise<string> {
  // Ensure queue exists with parallelism:1 before first use
  await qstash.queues.upsert({ queueName: 'llm-requests', parallelism: 1 });

  const q = qstash.queue({ queueName: 'llm-requests' });
  const response = await q.enqueueJSON({
    url: `${WORKER_URL}/api/llm-proxy`,
    body: data,
    deduplicationId: requestId,          // idempotent dispatch from MCP tool
    failureCallback: `${WORKER_URL}/api/tasks/failure`,
  });
  return response.messageId;
}

Composite Health Probe Design

Each system requires a different health probe because the failure modes differ. A probe that checks only TCP connectivity or an HTTP ping misses the most common production failures — jobs accumulating in the DLQ, workers not registered, result backends unavailable, or destination endpoints failing signature verification.

For BullMQ specifically, the most important operational metric beyond the health probe is the failedCount on the queue: await queue.getFailedCount() tells you how many jobs have exceeded their retry attempts. If this count grows without a corresponding decrease from removeOnFail, it indicates a systematic failure mode rather than isolated transient errors.

For Celery, the app.control.inspect().active() call returns the currently executing tasks per worker. If workers are active but results are consistently PENDING in the status checker, the result backend is unavailable or the result TTL has expired — check the backend connection separately from the broker connection.

For Azure Service Bus, the deadLetterMessageCount not appearing in activeMessageCount is the most dangerous health probe gap. Build a separate monitoring routine that checks props.deadLetterMessageCount on each queue and fires an alert when it exceeds a threshold. The MCP monitoring tool pattern applies here: the probe must exercise the full delivery path, not just the connectivity layer.

System Selection Guide for MCP Server Use Cases

The serverless vs persistent process distinction is the highest-stakes choice. BullMQ and Celery require persistent worker processes — deploying them in a container or VM that stays running is a prerequisite, not a configuration option. Inngest and QStash both work with short-lived serverless functions, but they require publicly reachable endpoints and impose different constraints on function structure (Inngest's step memoization model vs QStash's signature verification requirement).

For MCP servers in particular, the distinction between dispatching jobs (which can be done in a stateless invocation) and processing them (which requires a persistent process in BullMQ/Celery) shapes the entire deployment architecture. An MCP server that only enqueues work can be stateless; an MCP server that processes its own queued work must maintain a worker process alongside the protocol handler.

The AliveMCP public dashboard monitors MCP endpoints across all five job queue integrations as part of its registry crawl. The most common failure pattern: MCP servers that return HTTP 200 on the protocol handshake while all background job processing has silently stopped — no workers registered in BullMQ, result backend unavailable in Celery, destination endpoint failing QStash signature verification. The protocol layer is alive; the work layer is dead.

The Five Design Decisions Behind These Systems

Each architectural choice in this arc solves a real problem while creating a predictable failure mode. Understanding the tradeoff makes the failure modes less surprising.

BullMQ's completed job accumulation default exists because some applications need to inspect completed job return values — job.returnvalue is only available while the job is in the completed set. Making removeOnComplete: false the default preserves this option. The cost: every BullMQ deployment that doesn't read the documentation eventually hits a Redis memory exhaustion event, and the failure is silent until Redis starts rejecting writes.

Celery's PENDING state ambiguity exists because the result backend stores only terminal states (SUCCESS, FAILURE, REVOKED) and currently-executing states (STARTED, if configured with task_track_started=True). There is no "queued" state stored in the backend — PENDING is the state returned when the backend has no record of the task ID, which could mean queued, not-yet-started, or expired. The cost: every MCP status-check tool that relies on PENDING to mean "still waiting" will eventually tell users their task is in progress when it has actually been discarded.

Inngest's re-execution of pre-step code exists because step functions are designed to be stateless and deterministic — re-executing the function from the top with memoized step results lets Inngest's cloud reconstruct the full function state from an append-only log without storing the entire call frame. The cost: developers who add logging, counters, or API calls outside step blocks see them execute once per step continuation rather than once per function invocation, often only discovering this in production when log volumes are unexpectedly high.

Azure Service Bus's invisible DLQ exists because the DLQ is architecturally a separate queue — it has its own message count, receivers, and ACK path. The activeMessageCount property reflects only the main queue because DLQ monitoring and remediation is the consumer's responsibility. The cost: teams that monitor only activeMessageCount see "queue empty" while the DLQ fills with unprocessed failures, often discovered during an incident review.

QStash's 24-hour deduplication window exists because the deduplication lookup is a Redis SETNX operation against the deduplicationId key — the window is long enough to cover typical retry storms (where an MCP tool is called multiple times due to a timeout or client-side error) while expiring quickly enough to allow re-use of the same ID across scheduling cycles. The cost: tasks with intentional second dispatches within 24 hours are silently dropped, and the publishJSON call returns HTTP 200 with the same messageId — indistinguishable from a successful new dispatch.

For MCP server uptime monitoring, the job queue layer is one of the most important health signals to surface. An MCP server whose protocol endpoint returns 200 while its job queue worker has stopped, its result backend has disconnected, or its QStash destination is rejecting signatures will appear healthy to any probe that checks only the protocol layer. The composite probe that exercises the full delivery path — from enqueue through to result retrieval — is the only probe that catches these failures before your users do.