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 counters, wait/processing time, and sampled old-style depth metrics: worker/src/queues/workerManager.ts.
• Queue depth 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:

Prefer the newer tagged 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>.time{env:<env>,type:wait}
<metric_base>.time{env:<env>,type:processing}

For sharded queues, use the shard tag when present. shard:all is emitted by the depth runner for aggregate depth across shards.

Backward-compatible metrics may still appear:

<metric_base>.length
<metric_base>.dlq_length
<metric_base>.active
<metric_base>.request
<metric_base>.failed
<metric_base>.error
<metric_base>.wait_time
<metric_base>.processing_time

For non-BullMQ internal write buffering, ClickhouseWriter emits langfuse.queue.clickhouse_writer.* metrics, but it is not a QueueName consumer.

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} or old .request metrics 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.