Monitoring BullMQ workers on a toy crypto exchange

On 24 October at 21:20:11 UTC I ran FLUSHALL against the Redis behind a small exchange's job queues. The worker process stayed up. It logged nothing. The balance check that runs every minute did not run again until 21:24:00, after I restarted the API, and the only component that noticed was a monitor outside the system, at 21:22:00.

The exchange is a toy I built for this post. It takes euro and bitcoin deposits and withdrawals, sends a confirmation email after each one, and reconciles its ledger against a fake hot wallet once a minute. Everything a user does returns within a second; everything that touches money or a third party happens later, in a worker, through BullMQ. This post builds that system from the request handler outward, shows five ways it goes quiet without an error, and adds the forty lines that make each of them page someone. The timings are from the run above.

Prerequisites#

Node 22, Postgres 16, Redis 7 started with --maxmemory-policy noeviction --appendonly yes, and bullmq@6.4.2 (published the day I ran this). BullMQ 6 no longer depends on ioredis; install ioredis@5 yourself and pass a client, or the Worker refuses to start with "BullMQ could not load the optional 'ioredis' package". The HTTP layer is Fastify 5, the database driver is pg, and Mailpit stands in for an SMTP provider. The custodian, the bank and the chain watcher are stubs that sleep for two or three seconds. You also need a free Drumbeats project. The repository is linked at the end if you would rather read than type.

Three processes and four queues#

The exchange is one Docker image started three ways.

api answers HTTP. POST /withdrawals reserves the amount on the user's balance, inserts a transactions row and returns 202. It never talks to the custodian, because the custodian takes three seconds and sometimes fails, and a request handler is the wrong place to retry.

worker is a second Node process. It opens a BullMQ Worker on each queue and runs one function per job name. A queue, in this context, is a named set of Redis structures that BullMQ manages; a job is a name, a JSON payload and an id.

relay is the third process and the subject of the next section.

The queues are deposits (jobs deposit-fiat-confirm, deposit-crypto-confirm), withdrawals (withdrawal-fiat, withdrawal-crypto), emails (email-transaction) and balance-checks, which has no producer. Its single job, balance-check, comes from a Job Scheduler:

// src/queues.ts
import { Queue, Worker, type Processor } from 'bullmq';
import { Redis } from 'ioredis';
export const connection = new Redis({ host: process.env.REDIS_HOST ?? 'localhost', port: 6379, maxRetriesPerRequest: null });

export const withdrawals   = new Queue('withdrawals',    { connection });
export const balanceChecks = new Queue('balance-checks', { connection });
// deposits and emails are declared the same way

export function startWorker(queueName: string, processors: Record<string, Processor>, concurrency = 5) {
  const worker = new Worker(queueName, async (job) => {
    const run = processors[job.name];
    if (!run) throw new Error(`no processor for ${job.name}`);
    return run(job);
  }, { connection, concurrency });
  worker.on('error', (err) => console.error(`[${queueName}]`, err));
  const stop = async () => { await worker.close(); process.exit(0); };
  process.once('SIGINT', stop); process.once('SIGTERM', stop);
  return worker;
}

// src/scheduler.ts, called once when api boots; calling it again is a no-op
export async function ensureScheduler(balanceChecks: Queue) {
  await balanceChecks.upsertJobScheduler(
    'balance-check-minutely',
    { pattern: '* * * * *', tz: 'UTC' },
    { name: 'balance-check', data: {}, opts: { removeOnComplete: 60, removeOnFail: 100 } },
  );
}

// src/queues.ts
import { Queue, Worker, type Processor } from 'bullmq';
import { Redis } from 'ioredis';
export const connection = new Redis({ host: process.env.REDIS_HOST ?? 'localhost', port: 6379, maxRetriesPerRequest: null });

export const withdrawals   = new Queue('withdrawals',    { connection });
export const balanceChecks = new Queue('balance-checks', { connection });
// deposits and emails are declared the same way

export function startWorker(queueName: string, processors: Record<string, Processor>, concurrency = 5) {
  const worker = new Worker(queueName, async (job) => {
    const run = processors[job.name];
    if (!run) throw new Error(`no processor for ${job.name}`);
    return run(job);
  }, { connection, concurrency });
  worker.on('error', (err) => console.error(`[${queueName}]`, err));
  const stop = async () => { await worker.close(); process.exit(0); };
  process.once('SIGINT', stop); process.once('SIGTERM', stop);
  return worker;
}

// src/scheduler.ts, called once when api boots; calling it again is a no-op
export async function ensureScheduler(balanceChecks: Queue) {
  await balanceChecks.upsertJobScheduler(
    'balance-check-minutely',
    { pattern: '* * * * *', tz: 'UTC' },
    { name: 'balance-check', data: {}, opts: { removeOnComplete: 60, removeOnFail: 100 } },
  );
}

The scheduler is the part worth reading the docs on. They say: "The scheduler will only generate new jobs when the last job begins processing." There is no timer in any process. The scheduler is a Redis key plus one delayed job, and a worker promotes that delayed job when its time comes and creates the next one as it starts. Jobs produced this way get the id repeat:balance-check-minutely:<epoch ms>, which matters later.

Why the API cannot call queue.add() itself#

A withdrawal has to do two things: write a row in Postgres and tell a worker. Postgres and Redis do not share a transaction, so the order you pick decides which failure you get. Enqueue before COMMIT and a worker can pick up a job whose row does not exist yet, or the commit can fail after the custodian was already called. Enqueue after COMMIT and a crash or a Redis timeout in between leaves a committed withdrawal that nobody will ever process. With money, both are unacceptable.

The transactional outbox pattern stores the intent to enqueue as a row in the same transaction as the withdrawal, and a separate process moves those rows to the queue. The topic column becomes the job name:

// src/withdrawals.ts
export async function requestWithdrawal(accountId: string, asset: 'BTC' | 'EUR', amount: string, destination: string) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rowCount } = await client.query(
      `UPDATE balances SET available = available - $3, reserved = reserved + $3
       WHERE account_id = $1 AND asset = $2 AND available >= $3`, [accountId, asset, amount]);
    if (!rowCount) throw new Error('insufficient balance');
    const { rows: [tx] } = await client.query(
      `INSERT INTO transactions (account_id, kind, asset, amount) VALUES ($1,'withdrawal',$2,$3) RETURNING id`,
      [accountId, asset, amount]);
    const topic = asset === 'EUR' ? 'withdrawal-fiat' : 'withdrawal-crypto';
    await client.query('INSERT INTO outbox (topic, payload) VALUES ($1,$2)',
      [topic, { withdrawalId: tx.id, accountId, asset, amount, destination }]);
    await client.query('COMMIT');
    return tx.id as string;
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}

// src/withdrawals.ts
export async function requestWithdrawal(accountId: string, asset: 'BTC' | 'EUR', amount: string, destination: string) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rowCount } = await client.query(
      `UPDATE balances SET available = available - $3, reserved = reserved + $3
       WHERE account_id = $1 AND asset = $2 AND available >= $3`, [accountId, asset, amount]);
    if (!rowCount) throw new Error('insufficient balance');
    const { rows: [tx] } = await client.query(
      `INSERT INTO transactions (account_id, kind, asset, amount) VALUES ($1,'withdrawal',$2,$3) RETURNING id`,
      [accountId, asset, amount]);
    const topic = asset === 'EUR' ? 'withdrawal-fiat' : 'withdrawal-crypto';
    await client.query('INSERT INTO outbox (topic, payload) VALUES ($1,$2)',
      [topic, { withdrawalId: tx.id, accountId, asset, amount, destination }]);
    await client.query('COMMIT');
    return tx.id as string;
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}

The outbox table is id uuid, topic text, payload jsonb, created_at, published_at nullable, with a partial index on created_at where published_at is null.

The relay#

The relay loops once a second, locks up to 100 unpublished rows, adds one job per row and marks the rows published in the same transaction. The job id is the outbox row id. BullMQ ignores an add whose id already exists, so if the relay crashes between add and COMMIT the retry is harmless:

// src/relay.ts
async function relayOnce(): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query(
      `SELECT id, topic, payload FROM outbox WHERE published_at IS NULL
       ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED`);
    for (const r of rows) {
      await queueFor(r.topic).add(r.topic, r.payload, {
        jobId: r.id, attempts: 5, backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: { age: 3600, count: 1000 }, removeOnFail: { age: 86400 },
      });
    }
    if (rows.length) await client.query(
      'UPDATE outbox SET published_at = now() WHERE id = ANY($1::uuid[])', [rows.map(r => r.id)]);
    await client.query('COMMIT');
    return rows.length;
  } catch (e) { await client.query('ROLLBACK'); throw e; }
  finally { client.release(); }
}

export async function runRelay(onTick: () => void) {
  for (;;) { try { await relayOnce(); onTick(); } catch (e) { console.error('relay', e); } await sleep(1000); }
}

// src/relay.ts
async function relayOnce(): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query(
      `SELECT id, topic, payload FROM outbox WHERE published_at IS NULL
       ORDER BY created_at LIMIT 100 FOR UPDATE SKIP LOCKED`);
    for (const r of rows) {
      await queueFor(r.topic).add(r.topic, r.payload, {
        jobId: r.id, attempts: 5, backoff: { type: 'exponential', delay: 1000 },
        removeOnComplete: { age: 3600, count: 1000 }, removeOnFail: { age: 86400 },
      });
    }
    if (rows.length) await client.query(
      'UPDATE outbox SET published_at = now() WHERE id = ANY($1::uuid[])', [rows.map(r => r.id)]);
    await client.query('COMMIT');
    return rows.length;
  } catch (e) { await client.query('ROLLBACK'); throw e; }
  finally { client.release(); }
}

export async function runRelay(onTick: () => void) {
  for (;;) { try { await relayOnce(); onTick(); } catch (e) { console.error('relay', e); } await sleep(1000); }
}

I chose one second and 100 rows because the exchange is tiny; the shape is what matters. FOR UPDATE SKIP LOCKED lets a second relay run without the two fighting over rows. attempts: 5 with exponential backoff retries a job at 1, 2, 4 and 8 seconds before it lands in the failed set.

The worker's deposit and withdrawal processors call the stub custodian, mark the transaction settled, move the fake wallet balance, and insert another outbox row with topic email-transaction. The relay picks that up too, and the email worker sends through Mailpit.

Where this goes quiet#

Each failure below ties to one of the three processes, and in none of them does anything throw.

I ran the third one. With the relay stopped at 21:20:21 I submitted a withdrawal and a deposit; both returned 202. Two minutes later:

$ psql -c "select count(*) from outbox where published_at is null"
 2
$ redis-cli llen bull:withdrawals:wait ; redis-cli llen bull:withdrawals:active ; redis-cli zcard bull:withdrawals:failed
0
0
0
$ docker compose logs --since 3m worker
(nothing)

$ psql -c "select count(*) from outbox where published_at is null"
 2
$ redis-cli llen bull:withdrawals:wait ; redis-cli llen bull:withdrawals:active ; redis-cli zcard bull:withdrawals:failed
0
0
0
$ docker compose logs --since 3m worker
(nothing)

Bull Board would have shown the same zeros. Its own documentation says it is a viewer and that alerting belongs in worker code. The docs page on repeatable jobs says the rest: "If there are no workers running, repeatable jobs will not accumulate next time a worker is online."

Seven monitors from one prompt#

A Drumbeats monitor is addressed by project id and slug, and a ping to a slug that does not exist returns "Monitor not found". So the monitors come first. There are seven: one per job name, plus one for the relay, and I created them from Claude Code with the Drumbeats MCP server connected:

In Drumbeats, create a project called "Tulip Exchange" and these monitors in it, slug equal to the name: balance-check as a cron monitor on * UTC with a 60 s grace; outbox-relay as a heartbeat every 1m with a 30 s grace; deposit-crypto-confirm, deposit-fiat-confirm, withdrawal-crypto, withdrawal-fiat and email-transaction as event-driven monitors, max duration 300 s for the withdrawals, 120 s for the deposits and 60 s for the email, failure tolerance 3 for the email and 1 for the rest. Tag them tulip and bullmq, then list the project's monitors.

Seven create_monitor calls and one list_monitors later I had the project id for the environment file. The types are the decision. balance-check is the only job with a schedule to miss, so it gets cron job monitoring; everything else is a heartbeat or event-driven:

The same server lists incidents and acknowledges them, which I used from the terminal during the drills instead of the dashboard.

What the worker and the relay report#

Every BullMQ Worker emits active, completed, failed and stalled, and the job object carries name, id and processedOn. That is enough to describe a run to Drumbeats without touching any processor:

// src/drumbeats.ts
const slugOf = (name: string) => name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 50);

export function attachDrumbeats(worker: Worker, opts: { projectId: string; baseUrl?: string }) {
  const base = opts.baseUrl ?? 'https://api.drumbeats.io';
  const names = new Map<string, string>();                       // jobId -> job.name (for 'stalled')
  const chains = new Map<string, Promise<unknown>>();            // jobId -> last ping, so start lands before finish
  const send = (slug: string, event: string, runId: string, body: { duration_ms?: number; payload?: unknown } = {}) => {
    const url = `${base}/v1/s-ping/${opts.projectId}/${slug}/${event}?run_id=${encodeURIComponent(runId)}`;
    const payload = body.payload === undefined ? undefined : JSON.stringify(body.payload);   // Drumbeats stores payload as a string
    return fetch(url, { method: 'POST', headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ duration_ms: body.duration_ms, payload }), signal: AbortSignal.timeout(3000) }).catch(() => {});
  };
  const ping = (slug: string, event: string, runId: string, body?: { duration_ms?: number; payload?: unknown }) => {
    const next = (chains.get(runId) ?? Promise.resolve()).then(() => send(slug, event, runId, body));
    chains.set(runId, next);
    if (event !== 'start') void next.finally(() => { if (chains.get(runId) === next) chains.delete(runId); });
  };
  const elapsed = (job: Job) => (job.processedOn ? Date.now() - job.processedOn : undefined);

  worker.on('active', (job) => { names.set(String(job.id), job.name); ping(slugOf(job.name), 'start', String(job.id)); });
  worker.on('completed', (job) => { names.delete(String(job.id));
    ping(slugOf(job.name), 'success', String(job.id), { duration_ms: elapsed(job), payload: { name: job.name, keys: Object.keys(job.data ?? {}) } }); });
  worker.on('failed', (job, err) => { if (!job) return; names.delete(String(job.id));
    ping(slugOf(job.name), 'failure', String(job.id), { duration_ms: elapsed(job), payload: { reason: err.message, attempt: job.attemptsMade } }); });
  worker.on('stalled', (jobId) => { const name = names.get(jobId); if (!name) return;
    ping(slugOf(name), 'log', jobId, { payload: 'stalled: lock lost, requeued by BullMQ' }); });
}

// src/drumbeats.ts
const slugOf = (name: string) => name.toLowerCase().replace(/[^a-z0-9]+/g, '-').replace(/^-+|-+$/g, '').slice(0, 50);

export function attachDrumbeats(worker: Worker, opts: { projectId: string; baseUrl?: string }) {
  const base = opts.baseUrl ?? 'https://api.drumbeats.io';
  const names = new Map<string, string>();                       // jobId -> job.name (for 'stalled')
  const chains = new Map<string, Promise<unknown>>();            // jobId -> last ping, so start lands before finish
  const send = (slug: string, event: string, runId: string, body: { duration_ms?: number; payload?: unknown } = {}) => {
    const url = `${base}/v1/s-ping/${opts.projectId}/${slug}/${event}?run_id=${encodeURIComponent(runId)}`;
    const payload = body.payload === undefined ? undefined : JSON.stringify(body.payload);   // Drumbeats stores payload as a string
    return fetch(url, { method: 'POST', headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ duration_ms: body.duration_ms, payload }), signal: AbortSignal.timeout(3000) }).catch(() => {});
  };
  const ping = (slug: string, event: string, runId: string, body?: { duration_ms?: number; payload?: unknown }) => {
    const next = (chains.get(runId) ?? Promise.resolve()).then(() => send(slug, event, runId, body));
    chains.set(runId, next);
    if (event !== 'start') void next.finally(() => { if (chains.get(runId) === next) chains.delete(runId); });
  };
  const elapsed = (job: Job) => (job.processedOn ? Date.now() - job.processedOn : undefined);

  worker.on('active', (job) => { names.set(String(job.id), job.name); ping(slugOf(job.name), 'start', String(job.id)); });
  worker.on('completed', (job) => { names.delete(String(job.id));
    ping(slugOf(job.name), 'success', String(job.id), { duration_ms: elapsed(job), payload: { name: job.name, keys: Object.keys(job.data ?? {}) } }); });
  worker.on('failed', (job, err) => { if (!job) return; names.delete(String(job.id));
    ping(slugOf(job.name), 'failure', String(job.id), { duration_ms: elapsed(job), payload: { reason: err.message, attempt: job.attemptsMade } }); });
  worker.on('stalled', (jobId) => { const name = names.get(jobId); if (!name) return;
    ping(slugOf(name), 'log', jobId, { payload: 'stalled: lock lost, requeued by BullMQ' }); });
}

Call attachDrumbeats(worker, { projectId }) once per worker. The job name becomes the slug, so one listener covers every queue, and the job id becomes run_id. For relayed jobs that id is the outbox row id, so an incident links back to the database row and from there to the transaction. For scheduler jobs it is repeat:balance-check-minutely:1792877040000, which Drumbeats stores as given.

Two details cost me an hour each. Drumbeats keeps payload only when it is a string, so the adapter stringifies the object. And a 20 ms job sends completed before its start request has left the socket; in my first version the success ping arrived 6 ms before the start ping, Drumbeats saw a start with no finish and, 120 s later, opened a failure on a job that had succeeded. The per-job promise chain fixes the order. A stall is reported as a log entry rather than a failure, because BullMQ is about to retry the job; if the retry also stalls, failed fires with "job stalled more than allowable limit" and the failure handler reports it.

The relay is not a worker, so it gets one line in its loop: a throttled GET to /v1/s-ping/<project>/outbox-relay/success at most every 30 seconds. If the loop dies, the pings stop by construction.

The drills#

Four drills, each run once, with Drumbeats' own incident timestamps.

After the first drill the Redis completed set held the 21:16 and 21:19 jobs and the 21:17 job, which ran at 21:18:54 when the worker came back. There was never a 21:18 job. That is the sentence from the docs, observed.

The second drill is the one I built the exchange to show. Nothing restarted, nothing crashed, no log line was written, and Bull Board would have shown four idle queues. The first signal anywhere was the missed check-in, one minute and forty-nine seconds after the flush.

One failure I did not plan. My first version of the worker had a Postgres statement with two untyped parameters, amount + $1 $2, and every deposit and withdrawal failed five times with operator is not unique: unknown unknown. Drumbeats opened a FAILED incident on deposit-fiat-confirm at 21:11:21.3, about 100 ms after the first failure ping and before I had opened a log. The payload on each ping carried the error and the attempt number.

Letting Drumbeats fix the easy one#

The Redis case has a one-call fix, so Drumbeats can apply it. A webhook notification channel posts the incident to the API, and the API re-upserts the scheduler:

// src/api.ts
app.post<{ Body: { event: string; incidentId?: string; monitor: { slug: string }; incident?: { event: string } } }>(
  '/hooks/drumbeats', async (req, reply) => {
    if (req.headers['x-hook-secret'] !== process.env.HOOK_SECRET) return reply.code(401).send();
    const { event, monitor, incident } = req.body;
    if (event !== 'DOWN') return { ok: true, ignored: true };
    if (monitor.slug === 'balance-check' && incident?.event === 'MISSED') {
      await ensureScheduler(balanceChecks);                 // recreates the scheduler Redis lost; a no-op otherwise
      return { ok: true, action: 'scheduler-upserted' };
    }
    return { ok: true };
  });

// src/api.ts
app.post<{ Body: { event: string; incidentId?: string; monitor: { slug: string }; incident?: { event: string } } }>(
  '/hooks/drumbeats', async (req, reply) => {
    if (req.headers['x-hook-secret'] !== process.env.HOOK_SECRET) return reply.code(401).send();
    const { event, monitor, incident } = req.body;
    if (event !== 'DOWN') return { ok: true, ignored: true };
    if (monitor.slug === 'balance-check' && incident?.event === 'MISSED') {
      await ensureScheduler(balanceChecks);                 // recreates the scheduler Redis lost; a no-op otherwise
      return { ok: true, action: 'scheduler-upserted' };
    }
    return { ok: true };
  });

Act once per incidentId, or an API that crashes on start will loop. A stopped relay or worker needs the platform to restart a container (Coolify, for example, exposes POST /api/v1/applications/{uuid}/restart), and that belongs in the receiver rather than in the exchange. I did not run this route from a real webhook: the demo ran on my laptop, which Drumbeats cannot reach.

What the pings do not tell you#

Queue depth and the failed set need polling. queue.getJobCounts() on a schedule, or an uptime check against a small /queue-health route that returns 503 above a threshold, covers that. The pings describe runs; they say nothing about jobs that were never enqueued because a producer has a bug, and nothing about a worker that is alive but too slow for the backlog.

I did not test Redis Cluster, sandboxed processors, BullMQ Pro groups, or more than one worker replica, so the stalled path in the table above is from the docs and the source defaults, not from a drill.

Where the code is#

The exchange, the adapter and the compose file are at github.com/drumbeats-io/tulip-jobs, pinned to BullMQ 6.4.2, Node 22 and Postgres 16. The drill logs and the Drumbeats ping history they produced are in the repository under docs/run-2026-10-24.md.