@piperadar/bullmq

Drop-in pipeline failure & latency monitoring for BullMQ queues. Add one line to your worker process and PipeRadar tracks job failures, retries, and latency — surfacing incidents, failure groups, and alerts in your dashboard.

It hooks BullMQ's native QueueEvents, so there is zero overhead inside your job-processing path. Monitoring must never break the host app, so every failure path in the SDK is swallowed.

Install

npm install @piperadar/bullmq

bullmq is a peer dependency (>=3.0.0) — you already have it.

What PipeRadar collects

PipeRadar never sends your job payloads. It observes BullMQ's native queue events — it never reads job.data, return values, your Redis, or your environment. Turn on transparency mode (DEBUG=piperadar) to watch the exact bytes leave your process.

| ✅ Sent to PipeRadar | ❌ Never sent | |---|---| | Queue name | Job data / payload / arguments | | Job name & job ID | Return values / results | | Status (completed / failed / retrying) | Redis contents or credentials | | Latency (ms) | Environment variables | | Attempt / retry count | Secrets, tokens, API keys | | Timestamp | Customer objects / PII | | environment (e.g. production) & optional service name | Raw error text (unless you opt in) | | SDK version (sdk_version) | | | Error message — scrubbed by default (see below) | |

On job IDs and queue names: job IDs are sent (they let you find a specific failure in the dashboard). Queue and job names are sent verbatim, so don't encode PII in them — name a queue send-invoice, not [email protected].

Quick start

import { Queue } from 'bullmq'
import { PipeRadar } from '@piperadar/bullmq'

const emailQueue = new Queue('email')
const paymentQueue = new Queue('payments')

const pr = PipeRadar({ apiKey: process.env.PIPERADAR_API_KEY! })
pr.watch(emailQueue)
pr.watch(paymentQueue)

// On graceful shutdown, flush buffered events and stop listeners:
process.on('SIGTERM', () => pr.destroy())

That's it. watch() is idempotent per queue, so calling it twice is harmless.

Options

PipeRadar({
  apiKey: 'pr_live_...',     // required
  environment,               // tags every event (default: process.env.NODE_ENV)
  service,                   // service/app name, e.g. 'billing-worker' (optional)
  batchSize,                 // events buffered before a flush (default: 25)
  flushInterval,             // flush cadence in ms (default: 5000)
  maxBufferedEvents,         // retry-buffer cap during an outage (default: 1000)
  enabled,                   // set false to disable, e.g. in tests (default: true)
  errorMessages,             // 'scrub' (default) | 'raw' | 'off'  — see below
  errorTransformer,          // (raw) => string | undefined        — full control
  advanced: { apiUrl },      // ingestion base URL (default: https://api.piperadar.dev)
})

environment and service travel with every event so you can tell a production failure from a staging one, or attribute events when several services share a queue. Each event also carries the SDK version (sdk_version) for compatibility diagnostics — you don't set it.

apiUrl moved under advanced. 99% of workers never point the SDK anywhere but production, so the base URL now lives in advanced.apiUrl to keep setup to a single apiKey. The old top-level apiUrl still works (deprecated) — pass advanced.apiUrl in new code.

Flushing

Events are batched and flushed automatically, but in short-lived processes (scripts, serverless handlers, tests) call await pr.flush() before you exit so nothing buffered is lost. flush() never throws.

await pr.flush()   // force-send everything buffered, now

On a long-running worker, prefer await pr.destroy() on shutdown — it flushes and closes the queue-event listeners.

Error messages — safe by default

Developers throw errors like new Error(`User [email protected] failed payment`) or throw new Error(JSON.stringify(req.body)), which can carry PII or secrets. So PipeRadar scrubs error text by default before it ever leaves your process, while keeping enough of the message intact for the backend to group identical failures.

• errorMessages: 'scrub' (default) — strips emails, JWTs, Bearer tokens, sk_live_…/pr_test_…-style keys, GitHub tokens, UUIDs, IPs, phone numbers, and long digit runs (card numbers, long IDs), replacing each with a [redacted-…] marker. User [email protected] failed → User [redacted-email] failed.
• errorMessages: 'raw' — send the message verbatim. Only if you're certain your errors never contain sensitive data.
• errorMessages: 'off' — never send error text at all. Failures still group by queue + job.
• errorTransformer: (raw) => string | undefined — full control. Return exactly what should be sent, or undefined to omit it. Overrides errorMessages; a throwing transformer is treated as "omit" and never breaks monitoring.

Messages are also capped at 500 characters.

Transparency mode

Run your worker with DEBUG=piperadar (or DEBUG=*) and the SDK prints the exact JSON body of every ingest request to stderr — so you can see for yourself that job payloads never leave your infrastructure.

DEBUG=piperadar node worker.js

How it behaves

• Latency is computed from BullMQ's active → completed/failed transitions. "Started" events are never sent — only terminal outcomes — which keeps your event quota low.
• Retries vs. failures. A failure that BullMQ will retry is reported as retrying; only the final attempt is a terminal failed.
• Batching & at-least-once delivery. Events are batched and flushed on an interval (or when a batch fills). Each batch carries a stable idempotency key that is reused across retries, so the backend dedupes a retried batch instead of double-counting it.
• Resilient under outages. Failed batches stay in a bounded in-memory retry buffer (maxBufferedEvents; oldest dropped when over cap). Permanent 4xx rejections are dropped so a poison batch can't wedge the queue. (On-disk buffering across process restarts is future work.)
• Graceful shutdown. destroy() closes listeners and flushes what's buffered.

Live example

The repo ships one end-to-end example: examples/harness.ts. It drives the real SDK against a local Redis — three realistic queues (payment-processor, email-sender, webhook-dispatch) with real BullMQ workers producing a continuous mix of successes, retried failures, and terminal failures. The SDK picks those up through its native QueueEvents hooks, computes latency, batches, and ships them — watch queue health, failure groups, and KPIs update live in your dashboard.

You need two things: a running Redis, and an ingest key from your PipeRadar dashboard.

PIPERADAR_API_KEY=pr_...
npm run example:harness  # runnable integration example — flood/spike demo (examples/harness.ts)

It generates continuous synthetic traffic against your PipeRadar project until you stop it. Stop with Ctrl-C — it flushes buffered events and shuts down cleanly.

Tune it with environment variables:

| Variable | Default | What it does | |---|---|---| | PIPERADAR_API_KEY | — (required) | ingest key from your PipeRadar dashboard | | PIPERADAR_API_URL | https://api.piperadar.dev | point at a local or self-hosted backend | | REDIS_URL | redis://localhost:6379 | the Redis instance BullMQ connects to | | RATE_MS | 700 | milliseconds between job submissions | | SPIKE | off | set to 1 to force a high failure rate on payment-processor — simulates an incident to demo failure grouping and alerts | | DEBUG | off | set to 1 to log worker errors and every failed job attempt (with its error message) to stderr |

DEBUG=1 controls the harness's own logging. The SDK's transparency mode is separate — run with DEBUG=piperadar to print every ingest request body instead.

Using it as a template for your own app

The harness is deliberately structured like a production worker process, so the integration parts copy straight into your own codebase:

1. Create the client once per process — PipeRadar({ apiKey, service, advanced: { apiUrl } }). (The harness uses a small batchSize and fast flushInterval so events appear quickly while demoing; in production the defaults are usually right.)
2. Watch each queue — pr.watch(queue) right after you construct it. That's the entire integration; your worker code doesn't change.
3. Shut down gracefully — on SIGINT/SIGTERM, close your workers, then await pr.destroy() to flush buffered events and stop the listeners.

Everything else in the file (the queue specs, the producer loop, the simulated latencies and failures) exists only to generate realistic traffic — drop it.

Development

npm run build            # tsc → dist/
npm test                 # node --test via tsx (no jest)
npm run example:harness  # live end-to-end demo (see "Live example" above)

License

MIT