fix(webapp): harden the OTLP transform worker pool

Address review feedback on the opt-in worker pool:

- Validate the pool config through env.server and clamp the worker count to at
  least 1, so a zero or non-numeric OTEL_TRANSFORM_WORKER_POOL_SIZE can no longer
  leave an empty pool that hangs every request.
- Add a per-task timeout (reject and reap a stuck worker) and a bounded queue
  (shed when full), so a wedged worker or overload can't hang requests forever or
  grow memory without limit.
- Respawn crashed workers with exponential backoff instead of a tight loop.
- Guard the immediate pricing-reload listener replay so a throwing listener can't
  escape subscription setup.

Author: ericallam

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

@@ -3,4 +3,4 @@ area: webapp
 type: improvement
 ---
 
-Optionally offload OTLP ingest processing (decode, transform, and LLM-cost enrichment) to a worker pool, freeing the main event loop under high telemetry volume. Off by default; opt in with the OTEL_TRANSFORM_WORKER_POOL_ENABLED environment variable.
+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/v3/llmPricingRegistry.server.ts

@@ -32,7 +32,13 @@ function emitPricingReload() {
 export function subscribeToPricingReload(listener: PricingReloadListener): () => void {
   pricingReloadListeners.add(listener);
   if (llmPricingRegistry?.isLoaded) {
-    listener(llmPricingRegistry.toSerializable());
+    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);

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

@@ -20,7 +20,6 @@ import { eventRepository } from "./eventRepository/eventRepository.server";
 import type { CreateEventInput, IEventRepository } from "./eventRepository/eventRepository.types";
 import { startSpan } from "./tracing.server";
 import { enrichCreatableEvents } from "./utils/enrichCreatableEvents.server";
-import { waitForLlmPricingReady } from "./llmPricingRegistry.server";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
 import {
@@ -32,16 +31,22 @@ import {
 } from "./otlpTransform.server";
 import os from "node:os";
 import { getOtlpWorkerPool } from "./otlpWorkerPool.server";
-import { llmPricingRegistry, subscribeToPricingReload } from "./llmPricingRegistry.server";
+import {
+  llmPricingRegistry,
+  subscribeToPricingReload,
+  waitForLlmPricingReady,
+} from "./llmPricingRegistry.server";
 
 // When enabled, decode+convert+enrich run in a worker pool; the main thread keeps the single
 // consolidated insert path (batching/part-count unchanged). Off = today's single-thread path.
-export const otlpTransformWorkerPoolEnabled =
-  process.env.OTEL_TRANSFORM_WORKER_POOL_ENABLED === "1";
+export const otlpTransformWorkerPoolEnabled = env.OTEL_TRANSFORM_WORKER_POOL_ENABLED;
 
-const OTEL_TRANSFORM_WORKER_POOL_SIZE = process.env.OTEL_TRANSFORM_WORKER_POOL_SIZE
-  ? parseInt(process.env.OTEL_TRANSFORM_WORKER_POOL_SIZE, 10)
-  : Math.max(1, (os.cpus()?.length ?? 2) - 2);
+// Always at least 1 worker: a 0/negative override must not silently disable the pool while the
+// flag is on (that would hang every raw-export request on an empty pool).
+const OTEL_TRANSFORM_WORKER_POOL_SIZE = Math.max(
+  1,
+  env.OTEL_TRANSFORM_WORKER_POOL_SIZE ?? (os.cpus()?.length ?? 2) - 2
+);
 
 type OTLPExporterConfig = {
   clickhouseFactory: ClickhouseFactory;
@@ -143,12 +148,20 @@ class OTLPExporter {
   }
 
   async #exportRawEvents(kind: "traces" | "logs", payload: Uint8Array): Promise<void> {
-    await startSpan(this._tracer, kind === "traces" ? "exportTracesRaw" : "exportLogsRaw", async (span) => {
-      const pool = await this.#pool();
-      const { eventsWithStores } = await pool.runTransform(kind, payload, this.#transformConfig());
-      const eventCount = await this.#exportEvents(eventsWithStores, true);
-      span.setAttribute("event_count", eventCount);
-    });
+    await startSpan(
+      this._tracer,
+      kind === "traces" ? "exportTracesRaw" : "exportLogsRaw",
+      async (span) => {
+        const pool = await this.#pool();
+        const { eventsWithStores } = await pool.runTransform(
+          kind,
+          payload,
+          this.#transformConfig()
+        );
+        const eventCount = await this.#exportEvents(eventsWithStores, true);
+        span.setAttribute("event_count", eventCount);
+      }
+    );
   }
 
   #transformConfig() {
@@ -162,7 +175,11 @@ class OTLPExporter {
     await waitForLlmPricingReady();
     const models =
       llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : [];
-    const pool = getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models);
+    const pool = getOtlpWorkerPool(
+      OTEL_TRANSFORM_WORKER_POOL_SIZE,
+      models,
+      env.OTEL_TRANSFORM_WORKER_PATH
+    );
     if (!this.#pricingSubscribed) {
       this.#pricingSubscribed = true;
       // Re-broadcast pricing to workers on every registry reload so their cost math stays fresh.

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

@@ -4,29 +4,38 @@ import { logger } from "~/services/logger.server";
 
 export type TransformKind = "traces" | "logs" | "metrics";
 
-type Pending = { resolve: (r: any) => void; reject: (e: Error) => void };
-
-type QueueItem = {
-  message: {
-    id: number;
-    kind: TransformKind;
-    payload: Uint8Array;
-    spanAttributeValueLengthLimit: number;
-    defaultEventStore: string;
-  };
+type TaskMessage = {
+  id: number;
+  kind: TransformKind;
+  payload: Uint8Array;
+  spanAttributeValueLengthLimit: number;
+  defaultEventStore: string;
+};
+
+type Task = {
+  message: TaskMessage;
   transfer: ArrayBuffer[];
-  pending: Pending;
+  resolve: (r: any) => void;
+  reject: (e: Error) => void;
+  timer: NodeJS.Timeout;
+  worker?: Worker;
 };
 
+const TASK_TIMEOUT_MS = 30_000;
+const MAX_QUEUE_DEPTH = 2_000;
+const RESPAWN_BASE_MS = 500;
+const RESPAWN_MAX_MS = 30_000;
+
 // Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run
 // fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers.
 export class OtlpWorkerPool {
   private readonly workers: Worker[] = [];
   private readonly idle: Worker[] = [];
-  private readonly queue: QueueItem[] = [];
-  private readonly pending = new Map<number, Pending>();
+  private readonly queue: number[] = [];
+  private readonly tasks = new Map<number, Task>();
   private readonly busyByWorker = new Map<Worker, number>();
   private nextId = 1;
+  private consecutiveFailures = 0;
   private latestPricingModels: unknown[];
 
   constructor(
@@ -35,9 +44,7 @@ export class OtlpWorkerPool {
     pricingModels: unknown[]
   ) {
     this.latestPricingModels = pricingModels;
-    for (let i = 0; i < size; i++) {
-      this.spawn();
-    }
+    for (let i = 0; i < size; i++) this.spawn();
     logger.info("OtlpWorkerPool started", { size, workerPath });
   }
 
@@ -47,12 +54,14 @@ export class OtlpWorkerPool {
     });
 
     worker.on("message", (msg: { id: number; ok: boolean; result?: any; error?: string }) => {
-      const pending = this.pending.get(msg.id);
-      this.pending.delete(msg.id);
+      this.consecutiveFailures = 0;
       this.busyByWorker.delete(worker);
-      if (pending) {
-        if (msg.ok) pending.resolve(msg.result);
-        else pending.reject(new Error(msg.error ?? "otlp worker error"));
+      const task = this.tasks.get(msg.id);
+      if (task) {
+        clearTimeout(task.timer);
+        this.tasks.delete(msg.id);
+        if (msg.ok) task.resolve(msg.result);
+        else task.reject(new Error(msg.error ?? "otlp worker error"));
       }
       this.release(worker);
     });
@@ -72,22 +81,35 @@ export class OtlpWorkerPool {
     this.idle.push(worker);
   }
 
-  // On crash: fail its in-flight task, drop it, and spawn a replacement.
+  // On crash/timeout: fail the worker's in-flight task (if still pending), drop the worker, and
+  // respawn with exponential backoff so a persistently failing worker can't tight-loop.
   private reap(worker: Worker, error: Error) {
     const inFlightId = this.busyByWorker.get(worker);
+    this.busyByWorker.delete(worker);
     if (inFlightId !== undefined) {
-      const pending = this.pending.get(inFlightId);
-      this.pending.delete(inFlightId);
-      this.busyByWorker.delete(worker);
-      pending?.reject(error);
+      const task = this.tasks.get(inFlightId);
+      if (task) {
+        clearTimeout(task.timer);
+        this.tasks.delete(inFlightId);
+        task.reject(error);
+      }
     }
     const wi = this.workers.indexOf(worker);
     if (wi !== -1) this.workers.splice(wi, 1);
     const ii = this.idle.indexOf(worker);
     if (ii !== -1) this.idle.splice(ii, 1);
     void worker.terminate().catch(() => {});
-    if (this.workers.length < this.size) this.spawn();
-    this.drain();
+    this.scheduleRespawn();
+  }
+
+  private scheduleRespawn() {
+    if (this.workers.length >= this.size) return;
+    const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS);
+    this.consecutiveFailures++;
+    setTimeout(() => {
+      if (this.workers.length < this.size) this.spawn();
+      this.drain();
+    }, delay);
   }
 
   private release(worker: Worker) {
@@ -98,21 +120,43 @@ export class OtlpWorkerPool {
   private drain() {
     while (this.queue.length > 0 && this.idle.length > 0) {
       const worker = this.idle.pop()!;
-      const item = this.queue.shift()!;
-      this.pending.set(item.message.id, item.pending);
-      this.busyByWorker.set(worker, item.message.id);
-      worker.postMessage(item.message, item.transfer);
+      const id = this.queue.shift()!;
+      const task = this.tasks.get(id);
+      if (!task) continue;
+      task.worker = worker;
+      this.busyByWorker.set(worker, id);
+      worker.postMessage(task.message, task.transfer);
     }
   }
 
+  private onTimeout(id: number) {
+    const task = this.tasks.get(id);
+    if (!task) return;
+    this.tasks.delete(id);
+    const err = new Error(`otlp worker task timed out after ${TASK_TIMEOUT_MS}ms`);
+    if (task.worker) {
+      // Dispatched to a stuck worker: reap it. The task is already removed, so reap won't
+      // double-reject.
+      this.reap(task.worker, err);
+    } else {
+      const qi = this.queue.indexOf(id);
+      if (qi !== -1) this.queue.splice(qi, 1);
+    }
+    task.reject(err);
+  }
+
   runTransform(
     kind: TransformKind,
     payload: Uint8Array,
     config: { spanAttributeValueLengthLimit: number; defaultEventStore: string }
   ): Promise<any> {
+    if (this.queue.length >= MAX_QUEUE_DEPTH) {
+      return Promise.reject(new Error("otlp worker pool queue is full"));
+    }
     const id = this.nextId++;
     return new Promise((resolve, reject) => {
-      this.queue.push({
+      const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS);
+      this.tasks.set(id, {
         message: {
           id,
           kind,
@@ -122,8 +166,11 @@ export class OtlpWorkerPool {
         },
         // Zero-copy the payload into the worker; the request owns a fresh ArrayBuffer.
         transfer: [payload.buffer as ArrayBuffer],
-        pending: { resolve, reject },
+        resolve,
+        reject,
+        timer,
       });
+      this.queue.push(id);
       this.drain();
     });
   }
@@ -146,12 +193,14 @@ export class OtlpWorkerPool {
 
 let pool: OtlpWorkerPool | undefined;
 
-export function getOtlpWorkerPool(size: number, pricingModels: unknown[]): OtlpWorkerPool {
+export function getOtlpWorkerPool(
+  size: number,
+  pricingModels: unknown[],
+  workerPath?: string
+): OtlpWorkerPool {
   if (!pool) {
-    const workerPath =
-      process.env.OTEL_TRANSFORM_WORKER_PATH ??
-      path.join(process.cwd(), "build", "otlpTransformWorker.cjs");
-    pool = new OtlpWorkerPool(size, workerPath, pricingModels);
+    const resolvedPath = workerPath ?? path.join(process.cwd(), "build", "otlpTransformWorker.cjs");
+    pool = new OtlpWorkerPool(size, resolvedPath, pricingModels);
   }
   return pool;
 }