feat(webapp): re-broadcast LLM pricing to OTLP workers on registry reload

The pricing registry now notifies subscribers after every load/reload (interval + pub/sub);
the OTLP worker pool subscribes and re-broadcasts the compiled model rows to all workers, so
their in-memory pricing stays fresh instead of being a static snapshot from pool creation.
Respawned workers also inherit the latest snapshot via workerData.

Verified on the isolated stack: broadcast fires on init then every reload interval to all
workers; registry-sourced cost enrichment kept landing in ClickHouse across multiple reload
cycles under load (no break).

Author: ericallam

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,37 @@ 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) {
+    listener(llmPricingRegistry.toSerializable());
+  }
+  return () => {
+    pricingReloadListeners.delete(listener);
+  };
+}
+
 async function initRegistry(registry: ModelPricingRegistry) {
   if (env.LLM_PRICING_SEED_ON_STARTUP) {
     await seedLlmPricing(prisma);
@@ -25,16 +57,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 +110,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);
     }
 

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

@@ -32,7 +32,7 @@ import {
 } from "./otlpTransform.server";
 import os from "node:os";
 import { getOtlpWorkerPool } from "./otlpWorkerPool.server";
-import { llmPricingRegistry } from "./llmPricingRegistry.server";
+import { llmPricingRegistry, subscribeToPricingReload } 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.
@@ -54,6 +54,7 @@ class OTLPExporter {
   private readonly _clickhouseFactory: ClickhouseFactory;
   private readonly _verbose: boolean;
   private readonly _spanAttributeValueLengthLimit: number;
+  #pricingSubscribed = false;
 
   constructor(config: OTLPExporterConfig) {
     this._tracer = trace.getTracer("otlp-exporter");
@@ -161,7 +162,13 @@ class OTLPExporter {
     await waitForLlmPricingReady();
     const models =
       llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : [];
-    return getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models);
+    const pool = getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models);
+    if (!this.#pricingSubscribed) {
+      this.#pricingSubscribed = true;
+      // Re-broadcast pricing to workers on every registry reload so their cost math stays fresh.
+      subscribeToPricingReload((updated) => pool.broadcastPricing(updated));
+    }
+    return pool;
   }
 
   async #exportEvents(

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

@@ -133,6 +133,10 @@ export class OtlpWorkerPool {
     for (const worker of this.workers) {
       worker.postMessage({ type: "pricing", models });
     }
+    logger.info("OtlpWorkerPool broadcast pricing", {
+      models: models.length,
+      workers: this.workers.length,
+    });
   }
 
   get queueDepth() {