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

.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

@@ -1937,6 +1937,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);
     }