feat(webapp): opt-in worker pool for OTLP ingest transform (#4232)

## Summary

Under high OTLP ingest volume, the whole decode, transform, and enrich
pipeline runs on the request event loop, so a single CPU core becomes
the ceiling while the rest sit idle. This adds an opt-in worker pool
that moves decode, transform, and LLM-cost enrichment onto worker
threads, keeping the main thread free for I/O. It is off by default
(`OTEL_TRANSFORM_WORKER_POOL_ENABLED`), so behavior is unchanged unless
enabled.

## Design

Workers do decode, filter, convert, and enrich (including LLM pricing
match). The main thread stays the single database reader: it loads the
pricing registry and broadcasts the compiled model rows to the workers
(re-broadcasting on every reload), so workers never touch the database.
The pure transform is extracted into a dependency-light module (no
Prisma/Redis/ClickHouse imports) so it can run inside a worker.

Importantly, the main thread keeps the existing single consolidated
insert path, so ClickHouse insert batching and part count are unchanged.
The parallelism buys CPU headroom, not more insert streams (which would
add merge pressure).

The worker is bundled as a standalone file at build time and ships in
the existing image with no Dockerfile change. In local load testing the
pool sustained roughly 2.6x the throughput of the single-thread path and
kept the main thread responsive under load.

Author: ericallam

.server-changes/otlp-transform-worker-pool.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Optionally process high-volume telemetry ingestion in parallel for higher throughput under heavy load by setting `OTEL_TRANSFORM_WORKER_POOL_ENABLED=1`. Off by default.

apps/webapp/app/env.server.ts

@@ -1939,6 +1939,13 @@ const EnvironmentSchema = z
     EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000),
     EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000),
 
+    // OTLP ingest transform worker pool (opt-in). When enabled, decode/convert/enrich run in a
+    // worker_threads pool instead of the request event loop; the single consolidated insert path
+    // is unchanged.
+    OTEL_TRANSFORM_WORKER_POOL_ENABLED: BoolEnv.default(false),
+    OTEL_TRANSFORM_WORKER_POOL_SIZE: z.coerce.number().int().optional(),
+    OTEL_TRANSFORM_WORKER_PATH: z.string().optional(),
+
     // Organization data stores registry
     ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce
       .number()

apps/webapp/app/routes/otel.v1.logs.ts

@@ -1,7 +1,7 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { ExportLogsServiceRequest, ExportLogsServiceResponse } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -17,6 +17,15 @@ export async function action({ request }: ActionFunctionArgs) {
     } else if (contentType.startsWith("application/x-protobuf")) {
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportLogsRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportLogsServiceResponse.encode(ExportLogsServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportLogsServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportLogs(exportRequest);

apps/webapp/app/routes/otel.v1.metrics.ts

@@ -4,7 +4,7 @@ import {
   ExportMetricsServiceRequest,
   ExportMetricsServiceResponse,
 } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -21,6 +21,15 @@ export async function action({ request }: ActionFunctionArgs) {
       const exporter = await otlpExporter;
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportMetricsRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportMetricsServiceResponse.encode(ExportMetricsServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportMetricsServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportMetrics(exportRequest);

apps/webapp/app/routes/otel.v1.traces.ts

@@ -1,7 +1,7 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { ExportTraceServiceRequest, ExportTraceServiceResponse } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -17,6 +17,15 @@ export async function action({ request }: ActionFunctionArgs) {
     } else if (contentType.startsWith("application/x-protobuf")) {
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportTracesRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportTraceServiceResponse.encode(ExportTraceServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportTraceServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportTraces(exportRequest);

apps/webapp/app/v3/dynamicFlushScheduler.server.ts

@@ -1,5 +1,6 @@
 import { Logger } from "@trigger.dev/core/logger";
 import { tryCatch } from "@trigger.dev/core/utils";
+import { getMeter, type Counter, type Histogram, type Meter } from "@internal/tracing";
 import { nanoid } from "nanoid";
 import pLimit from "p-limit";
 import { signalsEmitter } from "~/services/signals.server";
@@ -16,6 +17,11 @@ export type DynamicFlushSchedulerConfig<T> = {
   loadSheddingThreshold?: number; // Number of items that triggers load shedding
   loadSheddingEnabled?: boolean;
   isDroppableEvent?: (item: T) => boolean; // Function to determine if an event can be dropped
+  // Self-observability. `name` is the low-cardinality `scheduler` label that separates the
+  // task_events / llm_metrics / otlp_metrics instances in the same process. `meter` defaults to
+  // the global provider; inject one in tests. Instruments are no-op unless metrics are enabled.
+  meter?: Meter;
+  name?: string;
 };
 
 export class DynamicFlushScheduler<T> {
@@ -54,7 +60,21 @@ export class DynamicFlushScheduler<T> {
 
   private readonly logger: Logger = new Logger("EventRepo.DynamicFlushScheduler", "info");
 
+  // Pre-allocated attribute objects (closed label sets) so the hot flush path never allocates.
+  private readonly _metricAttrs: { scheduler: string };
+  private readonly _batchOkAttrs: { scheduler: string; outcome: string };
+  private readonly _batchFailedAttrs: { scheduler: string; outcome: string };
+  private _batchesCounter?: Counter;
+  private _itemsCounter?: Counter;
+  private _flushDurationHistogram?: Histogram;
+  private _batchSizeHistogram?: Histogram;
+  private _droppedEventsCounter?: Counter;
+
   constructor(config: DynamicFlushSchedulerConfig<T>) {
+    const schedulerName = config.name ?? "unknown";
+    this._metricAttrs = { scheduler: schedulerName };
+    this._batchOkAttrs = { scheduler: schedulerName, outcome: "ok" };
+    this._batchFailedAttrs = { scheduler: schedulerName, outcome: "failed" };
     this.batchQueue = [];
     this.currentBatch = [];
     this.BATCH_SIZE = config.batchSize;
@@ -80,6 +100,54 @@ export class DynamicFlushScheduler<T> {
     this.startFlushTimer();
     this.startMetricsReporter();
     this.setupShutdownHandlers();
+    this.#setupOtelMetrics(config.meter, schedulerName);
+  }
+
+  #setupOtelMetrics(meterOverride: Meter | undefined, name: string): void {
+    const meter = meterOverride ?? getMeter("ingest-flush");
+
+    this._batchesCounter = meter.createCounter("ingest.flush.batches", {
+      description: "Batches flushed to the sink, by outcome",
+      unit: "batches",
+    });
+    this._itemsCounter = meter.createCounter("ingest.flush.items", {
+      description: "Items successfully flushed to the sink",
+      unit: "items",
+    });
+    this._flushDurationHistogram = meter.createHistogram("ingest.flush.duration", {
+      description: "Wall-clock duration of a single batch flush",
+      unit: "ms",
+    });
+    this._batchSizeHistogram = meter.createHistogram("ingest.flush.batch_size", {
+      description: "Number of items in a flushed batch",
+      unit: "items",
+    });
+    this._droppedEventsCounter = meter.createCounter("ingest.flush.dropped_events", {
+      description: "Events dropped by load shedding before they reached the sink",
+      unit: "events",
+    });
+
+    // Pull-based gauges: read at export time only, so they add zero hot-path cost.
+    const queueDepthGauge = meter.createObservableGauge("ingest.flush.queue_depth", {
+      description: "Items queued and awaiting flush",
+      unit: "items",
+    });
+    const concurrencyGauge = meter.createObservableGauge("ingest.flush.concurrency", {
+      description: "Current concurrent-flush limit",
+      unit: "flushes",
+    });
+    const loadSheddingGauge = meter.createObservableGauge("ingest.flush.load_shedding", {
+      description: "1 while actively shedding load, otherwise 0",
+    });
+
+    meter.addBatchObservableCallback(
+      (result) => {
+        result.observe(queueDepthGauge, this.totalQueuedItems, this._metricAttrs);
+        result.observe(concurrencyGauge, this.limiter.concurrency, this._metricAttrs);
+        result.observe(loadSheddingGauge, this.isLoadShedding ? 1 : 0, this._metricAttrs);
+      },
+      [queueDepthGauge, concurrencyGauge, loadSheddingGauge]
+    );
   }
 
   addToBatch(items: T[]): void {
@@ -92,6 +160,7 @@ export class DynamicFlushScheduler<T> {
 
       if (dropped.length > 0) {
         this.metrics.droppedEvents += dropped.length;
+        this._droppedEventsCounter?.add(dropped.length, this._metricAttrs);
 
         // Track dropped events by kind if possible
         dropped.forEach((item) => {
@@ -213,6 +282,11 @@ export class DynamicFlushScheduler<T> {
             self.metrics.flushedBatches++;
             self.metrics.totalItemsFlushed += itemCount;
 
+            self._flushDurationHistogram?.record(duration, self._metricAttrs);
+            self._batchSizeHistogram?.record(itemCount, self._metricAttrs);
+            self._itemsCounter?.add(itemCount, self._metricAttrs);
+            self._batchesCounter?.add(1, self._batchOkAttrs);
+
             self.logger.debug("Batch flushed successfully", {
               flushId,
               itemCount,
@@ -253,6 +327,7 @@ export class DynamicFlushScheduler<T> {
           this.logger.error("Error flushing batch", {
             error: flushError,
           });
+          this._batchesCounter?.add(1, this._batchFailedAttrs);
         }
       })
     );

apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts

@@ -8,8 +8,8 @@ import type {
   TaskEventV1Input,
   TaskEventV2Input,
 } from "@internal/clickhouse";
-import type { Attributes, Tracer } from "@internal/tracing";
-import { startSpan, trace } from "@internal/tracing";
+import type { Attributes, Counter, Meter, Tracer } from "@internal/tracing";
+import { getMeter, startSpan, trace } from "@internal/tracing";
 
 import { createJsonErrorObject } from "@trigger.dev/core/v3/errors";
 import { serializeTraceparent } from "@trigger.dev/core/v3/isomorphic";
@@ -104,6 +104,8 @@ export type ClickhouseEventRepositoryConfig = {
   otlpMetricsBatchSize?: number;
   otlpMetricsFlushInterval?: number;
   otlpMetricsMaxConcurrency?: number;
+  /** Inject a meter for self-observability; defaults to the global provider. */
+  meter?: Meter;
 };
 
 /**
@@ -125,14 +127,22 @@ export class ClickhouseEventRepository implements IEventRepository {
    * track the drop count for observability.
    */
   private _permanentlyDroppedBatches = 0;
+  private readonly _droppedBatchesCounter: Counter;
 
   constructor(config: ClickhouseEventRepositoryConfig) {
     this._clickhouse = config.clickhouse;
     this._config = config;
     this._tracer = config.tracer ?? trace.getTracer("clickhouseEventRepo", "0.0.1");
     this._version = config.version ?? "v1";
 
+    const meter = config.meter ?? getMeter("ingest-flush");
+    this._droppedBatchesCounter = meter.createCounter("ingest.flush.batches_dropped", {
+      description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error",
+      unit: "batches",
+    });
+
     this._flushScheduler = new DynamicFlushScheduler({
+      name: `task_events_${this._version}`,
       batchSize: config.batchSize ?? 1000,
       flushInterval: config.flushInterval ?? 1000,
       callback: this.#flushBatch.bind(this),
@@ -149,6 +159,7 @@ export class ClickhouseEventRepository implements IEventRepository {
     });
 
     this._llmMetricsFlushScheduler = new DynamicFlushScheduler({
+      name: "llm_metrics",
       batchSize: config.llmMetricsBatchSize ?? 5000,
       flushInterval: config.llmMetricsFlushInterval ?? 2000,
       callback: this.#flushLlmMetricsBatch.bind(this),
@@ -160,6 +171,7 @@ export class ClickhouseEventRepository implements IEventRepository {
     });
 
     this._otlpMetricsFlushScheduler = new DynamicFlushScheduler({
+      name: "otlp_metrics",
       batchSize: config.otlpMetricsBatchSize ?? 10000,
       flushInterval: config.otlpMetricsFlushInterval ?? 1000,
       callback: this.#flushOtelMetricsBatch.bind(this),
@@ -359,6 +371,7 @@ export class ClickhouseEventRepository implements IEventRepository {
       // exactly the retry storm this wrapper is designed to avoid.
       if (fieldsSanitized === 0) {
         this._permanentlyDroppedBatches += 1;
+        this._droppedBatchesCounter.add(1, { table: contextLabel });
         logger.error(
           "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix",
           {
@@ -390,6 +403,7 @@ export class ClickhouseEventRepository implements IEventRepository {
         if (!isClickHouseJsonParseError(retryError)) throw retryError;
 
         this._permanentlyDroppedBatches += 1;
+        this._droppedBatchesCounter.add(1, { table: contextLabel });
         const retryMessage =
           typeof retryError === "object" && retryError !== null && "message" in retryError
             ? String((retryError as { message?: unknown }).message ?? "")

apps/webapp/app/v3/eventRepository/eventRepository.server.ts

@@ -101,6 +101,7 @@ export class EventRepository implements IEventRepository {
     private readonly _config: EventRepoConfig
   ) {
     this._flushScheduler = new DynamicFlushScheduler({
+      name: "postgres_events",
       batchSize: _config.batchSize,
       flushInterval: _config.batchInterval,
       callback: this.#flushBatch.bind(this),

apps/webapp/app/v3/llmPricingRegistry.server.ts

@@ -1,4 +1,5 @@
 import { ModelPricingRegistry, seedLlmPricing } from "@internal/llm-model-catalog";
+import type { LlmModelWithPricing } from "@internal/llm-model-catalog";
 import { prisma, $replica } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
@@ -7,6 +8,43 @@ import { createRedisClient } from "~/redis.server";
 import { singleton } from "~/utils/singleton";
 import { setLlmPricingRegistry } from "./utils/enrichCreatableEvents.server";
 
+type PricingReloadListener = (models: LlmModelWithPricing[]) => void;
+const pricingReloadListeners = new Set<PricingReloadListener>();
+
+// Notify subscribers (e.g. the OTLP worker pool) after each load/reload so they can rebuild
+// their own in-memory copy. No-op until something subscribes.
+function emitPricingReload() {
+  if (!llmPricingRegistry || !llmPricingRegistry.isLoaded || pricingReloadListeners.size === 0) {
+    return;
+  }
+  const models = llmPricingRegistry.toSerializable();
+  for (const listener of pricingReloadListeners) {
+    try {
+      listener(models);
+    } catch (err) {
+      logger.warn("LLM pricing reload listener failed", {
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
+  }
+}
+
+export function subscribeToPricingReload(listener: PricingReloadListener): () => void {
+  pricingReloadListeners.add(listener);
+  if (llmPricingRegistry?.isLoaded) {
+    try {
+      listener(llmPricingRegistry.toSerializable());
+    } catch (err) {
+      logger.warn("LLM pricing reload listener failed", {
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
+  }
+  return () => {
+    pricingReloadListeners.delete(listener);
+  };
+}
+
 async function initRegistry(registry: ModelPricingRegistry) {
   if (env.LLM_PRICING_SEED_ON_STARTUP) {
     await seedLlmPricing(prisma);
@@ -25,16 +63,21 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => {
   // Wire up the registry so enrichCreatableEvents can use it
   setLlmPricingRegistry(registry);
 
-  initRegistry(registry).catch((err) => {
-    console.error("Failed to initialize LLM pricing registry", err);
-  });
+  initRegistry(registry)
+    .then(() => emitPricingReload())
+    .catch((err) => {
+      console.error("Failed to initialize LLM pricing registry", err);
+    });
 
   // Periodic reload (backstop for the pub/sub path below)
   const reloadInterval = env.LLM_PRICING_RELOAD_INTERVAL_MS;
   const interval = setInterval(() => {
-    registry.reload().catch((err) => {
-      console.error("Failed to reload LLM pricing registry", err);
-    });
+    registry
+      .reload()
+      .then(() => emitPricingReload())
+      .catch((err) => {
+        console.error("Failed to reload LLM pricing registry", err);
+      });
   }, reloadInterval);
 
   // Pub/sub reload is opt-in per process (default off). Without it, the
@@ -73,11 +116,14 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => {
       if (pendingReloadTimer) return;
       pendingReloadTimer = setTimeout(() => {
         pendingReloadTimer = null;
-        registry.reload().catch((err) => {
-          logger.warn("Failed to reload LLM pricing registry from pub/sub", {
-            error: err instanceof Error ? err.message : String(err),
+        registry
+          .reload()
+          .then(() => emitPricingReload())
+          .catch((err) => {
+            logger.warn("Failed to reload LLM pricing registry from pub/sub", {
+              error: err instanceof Error ? err.message : String(err),
+            });
           });
-        });
       }, debounceMs);
     }