@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fixed a bug where a burst of telemetry could leave OpenTelemetry ingest rejecting every batch for a long time. Batches that wait too long are now dropped individually instead of restarting the processing workers, so ingest recovers as soon as the burst passes.
@@ -1,5 +1,6 @@
 import { Worker } from "node:worker_threads";
 import path from "node:path";
+import { performance } from "node:perf_hooks";
 import {
   getMeter,
   type Counter,
@@ -31,15 +32,40 @@ type Task = {
   // Wall-clock stamp at enqueue; the task-duration histogram measures enqueue -> terminal state
   // (queue wait + worker compute), so the gap from the worker-reported compute time is queue wait.
   enqueuedAt: number;
+  /** Monotonic stamps (performance.now) drive every deadline decision so a wall-clock step can't shed or reap. */
+  enqueuedAtMono: number;
+  dispatchedAtMono?: number;
 };
 
 type ReapReason = "error" | "exit" | "timeout";
 
+export type ReapEvent = {
+  reason: ReapReason;
+  taskAgeMs?: number;
+  sinceDispatchMs?: number;
+  queueDepth: number;
+  aliveWorkers: number;
+};
+
+export type OtlpWorkerPoolOptions = {
+  taskTimeoutMs?: number;
+  respawnBaseMs?: number;
+  respawnMaxMs?: number;
+  /**
+   * Observer for every reap, carrying the same fields as the warn log. It must not affect pool
+   * liveness: it runs after the worker is terminated and the respawn is scheduled, and any
+   * exception it throws or promise it rejects is logged and swallowed.
+   */
+  onReap?: (event: ReapEvent) => void | Promise<void>;
+};
+
 const TASK_TIMEOUT_MS = 30_000;
 const MAX_QUEUE_DEPTH = 2_000;
 const RESPAWN_BASE_MS = 500;
 const RESPAWN_MAX_MS = 30_000;
 const SHUTDOWN_DRAIN_MS = 5_000;
+const STALE_FLOOR_MAX_MS = 1_000;
+const SHED_LOG_INTERVAL_MS = 1_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.
@@ -48,11 +74,21 @@ export class OtlpWorkerPool {
   private readonly idle: Worker[] = [];
   private readonly queue: number[] = [];
   private readonly tasks = new Map<number, Task>();
-  private readonly busyByWorker = new Map<Worker, number>();
+  private readonly busyByWorker = new Map<Worker, Task>();
+  private readonly computeTimers = new Map<Worker, NodeJS.Timeout>();
   private nextId = 1;
   private consecutiveFailures = 0;
   private isShuttingDown = false;
   private latestPricingModels: unknown[];
+  private readonly taskTimeoutMs: number;
+  private readonly respawnBaseMs: number;
+  private readonly respawnMaxMs: number;
+  private readonly staleFloorMs: number;
+  private readonly onReap?: (event: ReapEvent) => void | Promise<void>;
+  private shedInWindow = 0;
+  private shedWindowStartMono = 0;
+  private lastShedMono = 0;
+  private shedFlushTimer?: NodeJS.Timeout;
 
   // Pre-allocated per-kind {kind} attribute objects so the per-task record path never allocates.
   private readonly _kindAttrs: Record<TransformKind, { kind: TransformKind }> = {
@@ -69,9 +105,15 @@ export class OtlpWorkerPool {
     private readonly size: number,
     private readonly workerPath: string,
     pricingModels: unknown[],
-    meter?: Meter
+    meter?: Meter,
+    options?: OtlpWorkerPoolOptions
   ) {
     this.latestPricingModels = pricingModels;
+    this.taskTimeoutMs = options?.taskTimeoutMs ?? TASK_TIMEOUT_MS;
+    this.respawnBaseMs = options?.respawnBaseMs ?? RESPAWN_BASE_MS;
+    this.respawnMaxMs = options?.respawnMaxMs ?? RESPAWN_MAX_MS;
+    this.staleFloorMs = Math.min(STALE_FLOOR_MAX_MS, this.taskTimeoutMs / 10);
+    this.onReap = options?.onReap;
     this.#setupOtelMetrics(meter);
     for (let i = 0; i < size; i++) this.spawn();
     logger.info("OtlpWorkerPool started", { size, workerPath });
@@ -131,6 +173,15 @@ export class OtlpWorkerPool {
     }
   }
 
+  /**
+   * A task that already timed out for its caller recorded its outcome and duration then; its
+   * compute time only becomes known when the worker finally replies, and skipping it would drop
+   * exactly the slow samples from the compute histogram.
+   */
+  #recordLateCompute(task: Task, computeMs: number): void {
+    this._computeDurationHistogram?.record(computeMs, this._kindAttrs[task.message.kind]);
+  }
+
   private spawn() {
     const worker = new Worker(this.workerPath, {
       workerData: { pricingModels: this.latestPricingModels },
@@ -141,7 +192,9 @@ export class OtlpWorkerPool {
       (msg: { id: number; ok: boolean; result?: any; error?: string; computeMs?: number }) => {
         if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker
         this.consecutiveFailures = 0;
+        const inFlight = this.busyByWorker.get(worker);
         this.busyByWorker.delete(worker);
+        this.#clearComputeTimer(worker);
         const task = this.tasks.get(msg.id);
         if (task) {
           clearTimeout(task.timer);
@@ -153,6 +206,8 @@ export class OtlpWorkerPool {
             this.#recordTaskEnd(task, "error", msg.computeMs);
             task.reject(new Error(msg.error ?? "otlp worker error"));
           }
+        } else if (inFlight?.message.id === msg.id && msg.computeMs !== undefined) {
+          this.#recordLateCompute(inFlight, msg.computeMs);
         }
         this.release(worker);
       }
@@ -182,30 +237,92 @@ export class OtlpWorkerPool {
 
     const ii = this.idle.indexOf(worker);
     if (ii !== -1) this.idle.splice(ii, 1);
+    this.#clearComputeTimer(worker);
 
-    const inFlightId = this.busyByWorker.get(worker);
+    const now = performance.now();
+    const inFlight = this.busyByWorker.get(worker);
     this.busyByWorker.delete(worker);
-    if (inFlightId !== undefined) {
-      const task = this.tasks.get(inFlightId);
-      if (task) {
-        clearTimeout(task.timer);
-        this.tasks.delete(inFlightId);
-        // A timed-out task already recorded its own end + was removed from the map, so this only
-        // fires for a crash that killed a task mid-flight.
-        this.#recordTaskEnd(task, "crash");
-        task.reject(error);
-      }
+    if (inFlight !== undefined && this.tasks.has(inFlight.message.id)) {
+      clearTimeout(inFlight.timer);
+      this.tasks.delete(inFlight.message.id);
+      this.#recordTaskEnd(inFlight, "crash");
+      inFlight.reject(error);
     }
 
     this._respawnsCounter?.add(1, { reason });
+    const event: ReapEvent = {
+      reason,
+      queueDepth: this.queue.length,
+      aliveWorkers: this.workers.length,
+      taskAgeMs: inFlight === undefined ? undefined : Math.round(now - inFlight.enqueuedAtMono),
+      sinceDispatchMs:
+        inFlight?.dispatchedAtMono === undefined
+          ? undefined
+          : Math.round(now - inFlight.dispatchedAtMono),
+    };
+    logger.warn("OtlpWorkerPool reaped worker", { ...event, error: error.message });
     void worker.terminate().catch(() => {});
     this.scheduleRespawn();
+    this.#notifyReap(event);
+  }
+
+  #notifyReap(event: ReapEvent): void {
+    if (this.onReap === undefined) return;
+    try {
+      const result = this.onReap(event);
+      if (result !== undefined && typeof result.then === "function") {
+        result.then(undefined, (thrown) => this.#logObserverError(thrown));
+      }
+    } catch (thrown) {
+      this.#logObserverError(thrown);
+    }
+  }
+
+  #logObserverError(thrown: unknown): void {
+    logger.error("OtlpWorkerPool onReap observer threw", {
+      error: thrown instanceof Error ? thrown.message : String(thrown),
+    });
+  }
+
+  #clearComputeTimer(worker: Worker) {
+    const timer = this.computeTimers.get(worker);
+    if (timer === undefined) return;
+    clearTimeout(timer);
+    this.computeTimers.delete(worker);
+  }
+
+  /**
+   * A dispatched task whose caller deadline passed is not evidence the worker is stuck: it may
+   * have sat in the queue for most of its budget. Give the worker the rest of a full compute
+   * budget for it, and only reap if it still hasn't replied by then. The worker's reply (or a
+   * crash reap) clears this timer.
+   */
+  #armComputeTimer(worker: Worker, task: Task, delayMs: number) {
+    this.#clearComputeTimer(worker);
+    const timer = setTimeout(
+      () => {
+        this.computeTimers.delete(worker);
+        if (this.busyByWorker.get(worker) !== task) return;
+        const remainingMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono!);
+        if (remainingMs > 0) {
+          this.#armComputeTimer(worker, task, remainingMs);
+          return;
+        }
+        this.reap(
+          worker,
+          new Error(`otlp worker stuck for ${this.taskTimeoutMs}ms on a single task`),
+          "timeout"
+        );
+      },
+      Math.max(1, Math.ceil(delayMs))
+    );
+    this.computeTimers.set(worker, timer);
   }
 
   private scheduleRespawn() {
     if (this.isShuttingDown) return;
     if (this.workers.length >= this.size) return;
-    const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS);
+    const delay = Math.min(this.respawnBaseMs * 2 ** this.consecutiveFailures, this.respawnMaxMs);
     this.consecutiveFailures++;
     setTimeout(() => {
       if (this.isShuttingDown) return;
@@ -219,28 +336,83 @@ export class OtlpWorkerPool {
     this.drain();
   }
 
+  /**
+   * Hand queued tasks to idle workers in FIFO order, shedding any task whose remaining budget is
+   * below the stale floor so a free worker starts on something it can still finish. The floor is
+   * a small fraction of the timeout, so only tasks whose deadline is effectively already here are
+   * dropped; everything else keeps its FIFO turn.
+   */
   private drain() {
+    const now = performance.now();
     while (this.queue.length > 0 && this.idle.length > 0) {
-      const worker = this.idle.pop()!;
       const id = this.queue.shift()!;
       const task = this.tasks.get(id);
       if (!task) continue;
+      const remainingMs = task.enqueuedAtMono + this.taskTimeoutMs - now;
+      if (remainingMs < this.staleFloorMs) {
+        this.#shed(task, remainingMs, now);
+        continue;
+      }
+      const worker = this.idle.pop()!;
       task.worker = worker;
-      this.busyByWorker.set(worker, id);
+      task.dispatchedAtMono = now;
+      this.busyByWorker.set(worker, task);
       worker.postMessage(task.message, task.transfer);
     }
   }
 
+  /**
+   * Sheds are aggregated into at most one debug line per second; each line carries the count and
+   * the span between the first and last shed it covers.
+   */
+  #shed(task: Task, remainingMs: number, now: number) {
+    clearTimeout(task.timer);
+    this.tasks.delete(task.message.id);
+    this.#recordTaskEnd(task, "stale");
+    task.reject(
+      new Error(
+        `otlp worker task shed after ${Math.round(now - task.enqueuedAtMono)}ms in queue with ${Math.max(
+          0,
+          Math.round(remainingMs)
+        )}ms of ${this.taskTimeoutMs}ms budget left`
+      )
+    );
+    if (this.shedInWindow === 0) this.shedWindowStartMono = now;
+    this.lastShedMono = now;
+    this.shedInWindow++;
+    if (this.shedFlushTimer !== undefined) return;
+    this.shedFlushTimer = setTimeout(() => {
+      this.shedFlushTimer = undefined;
+      logger.debug("OtlpWorkerPool shed stale tasks", {
+        shed: this.shedInWindow,
+        windowMs: Math.round(this.lastShedMono - this.shedWindowStartMono),
+        queueDepth: this.queue.length,
+        aliveWorkers: this.workers.length,
+      });
+      this.shedInWindow = 0;
+    }, SHED_LOG_INTERVAL_MS);
+    this.shedFlushTimer.unref();
+  }
+
+  /**
+   * The caller's budget (queue wait + compute) is spent, so reject it now. Whether the worker is
+   * at fault depends on how long it has held the task: reap only once it has had a full timeout
+   * of compute time on this one task, otherwise let it finish via the compute timer and return to
+   * the idle set on reply. The task is removed here, so neither path can double-reject.
+   */
   private onTimeout(id: number) {
     const task = this.tasks.get(id);
     if (!task) return;
     this.tasks.delete(id);
     this.#recordTaskEnd(task, "timeout");
-    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, "timeout");
+    const err = new Error(`otlp worker task timed out after ${this.taskTimeoutMs}ms`);
+    if (task.worker !== undefined && task.dispatchedAtMono !== undefined) {
+      const remainingComputeMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono);
+      if (remainingComputeMs <= 0) {
+        this.reap(task.worker, err, "timeout");
+      } else {
+        this.#armComputeTimer(task.worker, task, remainingComputeMs);
+      }
     } else {
       const qi = this.queue.indexOf(id);
       if (qi !== -1) this.queue.splice(qi, 1);
@@ -262,7 +434,7 @@ export class OtlpWorkerPool {
     }
     const id = this.nextId++;
     return new Promise((resolve, reject) => {
-      const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS);
+      const timer = setTimeout(() => this.onTimeout(id), this.taskTimeoutMs);
       this.tasks.set(id, {
         message: {
           id,
@@ -277,6 +449,7 @@ export class OtlpWorkerPool {
         reject,
         timer,
         enqueuedAt: Date.now(),
+        enqueuedAtMono: performance.now(),
       });
       this.queue.push(id);
       this.drain();
@@ -319,6 +492,10 @@ export class OtlpWorkerPool {
     this.idle.length = 0;
     this.queue.length = 0;
     this.busyByWorker.clear();
+    for (const timer of this.computeTimers.values()) clearTimeout(timer);
+    this.computeTimers.clear();
+    clearTimeout(this.shedFlushTimer);
+    this.shedFlushTimer = undefined;
     // Reject anything that didn't drain within the deadline.
     for (const [, task] of this.tasks) {
       clearTimeout(task.timer);
 
@@ -0,0 +1,6 @@
+/**
+ * Real worker that never replies, to exercise the pool's stuck-worker reap path.
+ */
+const { parentPort } = require("node:worker_threads");
+
+parentPort.on("message", () => {});
@@ -0,0 +1,22 @@
+/**
+ * Real worker that takes a fixed wall-clock time per task, so pool tests can drive queue wait past
+ * the task deadline while the worker itself stays healthy. Delay comes from
+ * OTLP_SLOW_WORKER_DELAY_MS (default 100). When OTLP_SLOW_WORKER_HANG_AFTER is set the worker
+ * answers that many tasks and then never replies again, to model a worker that wedges mid-run.
+ */
+const { parentPort } = require("node:worker_threads");
+
+const delayMs = Number(process.env.OTLP_SLOW_WORKER_DELAY_MS ?? 100);
+const hangAfter = process.env.OTLP_SLOW_WORKER_HANG_AFTER
+  ? Number(process.env.OTLP_SLOW_WORKER_HANG_AFTER)
+  : Infinity;
+let answered = 0;
+
+parentPort.on("message", (message) => {
+  if (message && message.type === "pricing") return;
+  if (answered >= hangAfter) return;
+  answered++;
+  setTimeout(() => {
+    parentPort.postMessage({ id: message.id, ok: true, result: { rows: [] }, computeMs: delayMs });
+  }, delayMs);
+});