Mixpanel Worker Integration#

Langfuse exports LLM observability data (traces, generations, scores, events) to Mixpanel on a scheduled basis using a two-stage BullMQ queue pipeline. The integration is enabled per-project and runs on a cron schedule, gated by the QUEUE_CONSUMER_MIXPANEL_INTEGRATION_QUEUE_IS_ENABLED environment variable 1.

Architecture: Two-Stage Queue Pipeline#

The integration uses two BullMQ queues registered in worker/src/app.ts:

1. MixpanelIntegrationQueue — a cron-triggered scheduler queue. Its processor (mixpanelIntegrationProcessor) calls handleMixpanelIntegrationSchedule(), which queries Prisma for all enabled integrations and enqueues one MixpanelIntegrationProcessingJob per project.

2. MixpanelIntegrationProcessingQueue — the per-project worker queue. Its processor (mixpanelIntegrationProcessingProcessor) calls handleMixpanelIntegrationProjectJob(), which fetches data and sends it to Mixpanel.

Both workers run at concurrency 1. The processing queue adds a BullMQ rate limiter of max 1 job per 10 seconds globally 2.

Job Deduplication#

The scheduler uses a static jobId composed of ${projectId}-${lastSyncAt.toISOString()} 3. This prevents the same project from being enqueued twice within the same sync window. Jobs use removeOnFail: true so failed jobs are immediately removed from Redis, unblocking re-queuing on the next cycle 4.

History: An earlier attempt (PR #11988) appended an hourly suffix to the jobId to bypass permanently-stalled jobs, but this broke deduplication by allowing unbounded queue growth. PR #11998 reverted to the static jobId approach and switched to removeOnFail: true as the correct fix.

Per-Project Data Export#

handleMixpanelIntegrationProjectJob() builds an executionConfig with a sync window: minTimestamp from lastSyncAt (defaulting to 2000-01-01 if null) and maxTimestamp set to 30 minutes ago 5. It then runs up to four data streams in parallel via Promise.all depending on the project's exportSource setting 6:

On success, lastSyncAt is updated in Prisma 7. On failure, the error is logged and re-thrown so BullMQ can retry 8.

MixpanelClient: Batching and HTTP#

MixpanelClient is instantiated per data stream. It accumulates events in memory and flushes in chunks of 1,000 events (matching PostHog's flushAt convention, and well within Mixpanel's 2,000-event API limit) 9. Each flush calls the Mixpanel Import API with:

• gzip compression of the JSON body 10
• Basic Auth using the project token as username with an empty password 11
• The target region subdomain (e.g., api, api-eu, api-in) in the URL 12

HTTP error handling: A 400 response is treated as a partial success if num_records_imported > 0; the partial failure is logged as a warning rather than throwing 13. All other non-2xx responses throw an error 14.

Note: The current fetch() call has no explicit timeout. A hung HTTP request will block the stream until BullMQ's stall detection kicks in (see below).

Stall Detection and BullMQ Worker Settings#

The processing worker uses non-default stall settings to tolerate slow HTTP round-trips and CPU wait spikes 15:

These were introduced in PR #11988 to address integration jobs permanently stalling under high CPU load.

Key Files#