Queue Consumers

Use this reference when a task asks which Langfuse queues exist, whether a consumer is running, how much work a queue has, or how to query queue processor spans.

Source Of Truth

• Queue names and job names: packages/shared/src/server/queues.ts (QueueName, QueueJobs).
• Queue producer classes and shard naming: packages/shared/src/server/redis/*.ts.
• Worker consumer registration and feature gates: worker/src/app.ts.
• Worker consumer env vars: worker/src/env.ts (QUEUE_CONSUMER_*_IS_ENABLED plus feature-specific gates).
• Worker registration, request/error/stalled rates, and wait/processing time: worker/src/queues/workerManager.ts.
• Queue depth and DLQ oldest-age background reporter: worker/src/features/queue-metrics-runner/index.ts.
• Metric name conversion: packages/shared/src/server/instrumentation/index.ts (convertQueueNameToMetricName).
• Sharded queue registry: worker/src/queues/shardedQueueRegistry.ts.
• BullMQ tracing setup: worker/src/instrumentation.ts (BullMQInstrumentation).

Queue Inventory

Current QueueName values:

Sharded queues use the base queue for shard 0 and append -1, -2, etc. for additional shards. The sharded base queues are:

• trace-upsert
• evaluation-execution-queue
• secondary-evaluation-execution-queue
• llm-as-a-judge-execution-queue
• otel-ingestion-queue
• secondary-otel-ingestion-queue
• ingestion-queue
• secondary-ingestion-queue

Consumer Gates

Consumer registration is in worker/src/app.ts. Some gates register multiple queues or every shard for a sharded queue.

Query Consumer Spans

Start with aggregate spans on worker services:

env:<env> (service:worker OR service:worker-cpu) operation_name:bullmq.process

Then group by resource_name, queue facets such as bullmq.queue or messaging.*, and error fields. Facet names can differ between Datadog sites, so inspect one sample span before relying on a specific facet.

Queue-specific starter query:

env:<env> (service:worker OR service:worker-cpu) operation_name:bullmq.process (resource_name:"process otel-ingestion-queue" OR resource_name:"Worker.run otel-ingestion-queue" OR bullmq.queue:otel-ingestion-queue)

For sharded queues, query the base queue and shard suffixes:

env:<env> (service:worker OR service:worker-cpu) operation_name:bullmq.process resource_name:"*otel-ingestion-queue*"

If a queue file wraps the handler with instrumentAsync, also search the domain-specific resource name. Examples:

Useful aggregations:

• Count by env, service, and resource_name.
• Count by queue facet and status.
• Count by error.type and error.message.
• p50, p95, and p99 duration by queue or shard.
• Count by messaging.bullmq.job.input.projectId when the processor attaches project IDs to the current span.

Query Queue Metrics

Metric base names come from convertQueueNameToMetricName(queueName):

langfuse.queue.<queue-name-with-hyphens-replaced-by-underscores-and-trailing-_queue-removed>

Examples:

Use the tagged queue metrics:

<metric_base>.depth{env:<env>,type:waiting}
<metric_base>.depth{env:<env>,type:failed}
<metric_base>.depth{env:<env>,type:active}
<metric_base>.rate{env:<env>,type:request}
<metric_base>.rate{env:<env>,type:failed}
<metric_base>.rate{env:<env>,type:error}
<metric_base>.rate{env:<env>,type:stalled}
<metric_base>.time_distribution{env:<env>,type:wait}
<metric_base>.time_distribution{env:<env>,type:processing}
<metric_base>.dlq_oldest_age{env:<env>}

Metric emission and shard handling differ by source:

• WorkerManager emits .rate and .time_distribution. A non-sharded queue has no shard tag. A sharded worker uses the base queue's metric name and sets shard to the full physical queue name. It does not emit shard:all.
• QueueMetricsRunner emits .depth and .dlq_oldest_age only for queue instances with a worker registered in the current process. It does not add a shard tag for a non-sharded queue. For a sharded queue, it emits one series per registered physical shard plus shard:all.

For .depth, shard:all is the sum across registered shards when every poll succeeds. If only some shard polls succeed, the runner estimates the total by scaling the successful sum by registered shard count / successful poll count. For .dlq_oldest_age, shard:all is the maximum age returned by the successful shard polls; failed polls are omitted and are not extrapolated.

Exclude langfuse.queue.clickhouse_writer.* from BullMQ consumer checks. Despite the metric prefix, those metrics come from worker/src/services/ClickhouseWriter/index.ts and describe its in-memory buffers and ClickHouse flushes, not a BullMQ queue.

Consumer Running Checklist

To establish whether a consumer is running in production:

1. Check queue depth metrics for waiting, failed, and active counts.
2. Check rate{type:request} for recent processing.
3. Search BullMQ processor spans on worker and worker-cpu.
4. Search worker logs for the queue name or processor-specific log prefix.
5. If all signals are empty, verify the relevant QUEUE_CONSUMER_*_IS_ENABLED gate and any feature-specific gates in worker/src/app.ts.

The queue metrics runner only polls queues with registered workers. Missing depth metrics can mean the consumer is not registered on that worker, queue metrics are disabled, or the data is on the other Datadog site.