BullMQ Worker Lifecycle#

Overview#

All BullMQ queue workers in the Langfuse worker service are managed by the WorkerManager singleton class. It handles registration, per-job metric instrumentation, error event binding, and graceful shutdown. The Express app at worker/src/app.ts owns the startup sequence: it calls WorkerManager.register() once per enabled queue (guarded by QUEUE_CONSUMER_*_IS_ENABLED env vars), then registers onShutdown for SIGINT/SIGTERM 1.

Registration#

WorkerManager.register(queueName, processor, options?) is the single entry point for bringing a worker online:

1. Deduplicates — if a worker for that queue already exists, it logs and returns 2.
2. Creates a dedicated Redis connection via createNewRedisInstance(redisQueueRetryOptions) 3.
3. Wraps the caller-supplied processor with metricWrapper before passing it to BullMQ's Worker constructor 4.
4. Binds "failed" and "error" event listeners that log, call traceException, and increment OTel counters 5.

Workers are registered conditionally in app.ts based on environment variables. Each queue section looks like:

if (env.QUEUE_CONSUMER_<QUEUE>_IS_ENABLED === "true") {
  WorkerManager.register(QueueName.<Queue>, processor, { concurrency, limiter? });
}

Sharded queues (Ingestion, OtelIngestion, TraceUpsert, EvalExecution, LLMAsJudge) iterate over Queue.getShardNames() and register one worker per shard 6 7.

Queue Metrics#

WorkerManager.metricWrapper wraps every job with OTel instrumentation:

• On entry: increments <queue>.request and records <queue>.wait_time (job age since enqueue) 8.
• On completion: samples queue-depth gauges (<queue>.length waiting, <queue>.dlq_length failed, <queue>.active) and records <queue>.processing_time 9.
• Sharded queues sample depth gauges probabilistically via LANGFUSE_QUEUE_METRICS_SAMPLE_RATE to reduce metric volume 10.

Stall Handling#

Long-running jobs (eval execution, PostHog/Mixpanel/BlobStorage integration processing, trace delete) use non-default stall settings to tolerate slow processors 11:

These settings are applied to the EvalExecution family, LLMAsJudge, PostHog/Mixpanel/BlobStorage integration processing, and TraceDelete queues.

Health Check & Readiness Endpoints#

worker/src/api/index.ts mounts two routes on the /api prefix:

checkContainerHealth performs the actual checks. The SIGTERM guard lets Kubernetes stop routing traffic as soon as shutdown begins without waiting for the process to exit.

Stuck Worker (Event Propagation) Health Check#

Added in PR #14680, both /health and /ready accept an optional query param ?failIfEventPropagationStuck=true. When set:

• The EventPropagationQueue worker writes a Redis heartbeat key (langfuse:event-propagation:last-run-started-at) at the start of every job invocation, including no-op runs 14.
• evaluateEventPropagationStuck in health/index.ts reads the key and compares staleness against LANGFUSE_EVENT_PROPAGATION_STUCK_THRESHOLD_MINUTES (default 15 min — sized above the 10-min ClickHouse INSERT timeout).
• If stale beyond the threshold, the endpoint returns 503. A Kubernetes liveness probe configured with this param will trigger a pod restart, which frees the occupied global-concurrency slot.
• If the heartbeat key is absent (job never ran), the endpoint returns 200 (not stuck) to avoid crash-looping a fresh pod.

The response body includes eventPropagation.propagationDelayMs and eventPropagation.lastRunStartedAt for observability, but propagation delay alone does not drive the 503 — a growing backlog usually means slow ClickHouse, not a stuck worker.

Shutdown Sequence#

SIGINT / SIGTERM both invoke onShutdown:

1. setSigtermReceived() — marks the readiness probe as failing immediately 15.
2. server.close() — stops accepting new HTTP connections 16.
3. Batch cleaners are stopped 17.
4. WorkerManager.closeWorkers() — calls worker.close() on all registered workers in parallel 18, following BullMQ production guidance.
5. BackgroundMigrationManager.close() — aborts active migrations 19.
6. ClickhouseWriter.getInstance().shutdown() — flushes pending buffered writes after workers are closed to prevent new writes mid-flush 20.
7. Redis, Prisma, ClickHouse connections are closed; tokenization worker threads are terminated and tokenizer cache freed 21.

Key Files#