fix(webapp): sanitize OTel attributes on ClickHouse JSON parse rejection

ClickHouse's JSONEachRow parser rejects rows containing unpaired UTF-16
surrogates (`Cannot parse JSON object here ... ParallelParsingBlock
InputFormat`), losing the whole 5–10k-row batch through the scheduler's
retry path. Locally reproduced with ~10 KB rows; the 100 MB size-stress
error is distinct (`Size of JSON object is extremely large`), so the
root cause is content quality, not size.

`ClickhouseEventRepository.#flushBatch` and `#flushLlmMetricsBatch` now
retry once after sanitizing every row in the batch — any string with a
lone surrogate is replaced with `"[invalid-utf16]"`. ClickHouse's
`at row N` hint is logged for observability but not used to slice; its
semantics under `input_format_parallel_parsing` aren't reliable, and a
whole-batch scan catches multi-row poisoning in one pass. If the retry
also fails: loud error log with sample row, `permanentlyDroppedBatches`
increments, return normally — deterministic parse failures don't benefit
from the scheduler's transient-retry backoff. Non-parse errors propagate
unchanged.

Detection reuses `detectBadJsonStrings` via `JSON.stringify(value)`,
with a latent regex bug fixed: the low-surrogate nibble matched `[cd]`
instead of `[c-f]`, missing U+DE00–U+DFFF and false-flagging common
emoji pairs (e.g. 😀). Healthy batches pay zero scan cost — the check
only runs when ClickHouse has already rejected.

Author: 0ski

.server-changes/otel-attribute-utf16-sanitization.md

@@ -0,0 +1,21 @@
+---
+area: webapp
+type: fix
+---
+
+Recover from ClickHouse `JSONEachRow` parse failures caused by lone
+UTF-16 surrogates in OTel attribute strings (`Cannot parse JSON object
+here ... ParallelParsingBlockInputFormat`).
+
+`ClickhouseEventRepository.#flushBatch` and `#flushLlmMetricsBatch` now
+retry once after sanitizing every row in the batch: any string value
+containing a lone surrogate is replaced with `"[invalid-utf16]"`.
+If the retry still fails, the batch is dropped,
+`permanentlyDroppedBatches` increments, and an error log with a 1KB
+sample row is emitted. Non-parse errors propagate unchanged.
+
+Detection reuses `detectBadJsonStrings` via `JSON.stringify(value)`,
+with a latent regex bug fixed: the low-surrogate hex nibble matched
+`[cd]` instead of `[c-f]`, missing the U+DE00–U+DFFF half of the range
+and false-flagging common emoji pairs. Healthy batches pay zero scan
+cost — the check only runs when ClickHouse has already rejected.

apps/webapp/app/utils/detectBadJsonStrings.ts

@@ -1,3 +1,17 @@
+/**
+ * Detects unpaired UTF-16 surrogate escape sequences in JSON-encoded text.
+ *
+ * Returns true if the input contains a `\uD8XX`/`\uD9XX`/`\uDAXX`/`\uDBXX`
+ * high-surrogate escape not immediately followed by a `\uDC..`–`\uDF..` low
+ * surrogate, or a `\uDC..`–`\uDF..` low surrogate not immediately preceded by
+ * a high surrogate. Strict JSON parsers (e.g. ClickHouse `JSONEachRow`)
+ * reject input containing such sequences.
+ *
+ * Surrogate hex ranges (case-insensitive — inputs from `JSON.stringify` are
+ * lowercase):
+ *   - High surrogate (U+D800–U+DBFF):  `\uD[8-B][0-9A-F][0-9A-F]`
+ *   - Low surrogate  (U+DC00–U+DFFF):  `\uD[C-F][0-9A-F][0-9A-F]`
+ */
 export function detectBadJsonStrings(jsonString: string): boolean {
   // Fast path: skip everything if no \u
   let idx = jsonString.indexOf("\\u");
@@ -13,7 +27,7 @@ export function detectBadJsonStrings(jsonString: string): boolean {
     if (jsonString[idx + 1] === "u" && jsonString[idx + 2] === "d") {
       const third = jsonString[idx + 3];
 
-      // High surrogate check
+      // High surrogate check — third nibble is 8, 9, a, or b (U+D800–U+DBFF)
       if (
         /[89ab]/.test(third) &&
         /[0-9a-f]/.test(jsonString[idx + 4]) &&
@@ -28,17 +42,17 @@ export function detectBadJsonStrings(jsonString: string): boolean {
           jsonString[idx + 6] !== "\\" ||
           jsonString[idx + 7] !== "u" ||
           jsonString[idx + 8] !== "d" ||
-          !/[cd]/.test(jsonString[idx + 9]) ||
+          !/[c-f]/.test(jsonString[idx + 9]) ||
           !/[0-9a-f]/.test(jsonString[idx + 10]) ||
           !/[0-9a-f]/.test(jsonString[idx + 11])
         ) {
           return true; // Incomplete high surrogate
         }
       }
 
-      // Low surrogate check
+      // Low surrogate check — third nibble is c, d, e, or f (U+DC00–U+DFFF)
       if (
-        (third === "c" || third === "d") &&
+        /[c-f]/.test(third) &&
         /[0-9a-f]/.test(jsonString[idx + 4]) &&
         /[0-9a-f]/.test(jsonString[idx + 5])
       ) {

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

@@ -45,6 +45,11 @@ import {
   removePrivateProperties,
   isEmptyObject,
 } from "./common.server";
+import {
+  isClickHouseJsonParseError,
+  parseRowNumberFromError,
+  sanitizeRows,
+} from "./sanitizeRowsOnParseError.server";
 import type {
   CompleteableTaskRun,
   CreateEventInput,
@@ -104,6 +109,13 @@ export class ClickhouseEventRepository implements IEventRepository {
   private readonly _llmMetricsFlushScheduler: DynamicFlushScheduler<LlmMetricsV1Input>;
   private _tracer: Tracer;
   private _version: "v1" | "v2";
+  /**
+   * Counts batches that hit a ClickHouse JSON parse failure that survived
+   * one sanitize-retry. These batches are dropped on the floor (the scheduler
+   * is told the flush "succeeded" so its queue counter doesn't leak), and we
+   * track the drop count for observability.
+   */
+  private _permanentlyDroppedBatches = 0;
 
   constructor(config: ClickhouseEventRepositoryConfig) {
     this._clickhouse = config.clickhouse;
@@ -147,6 +159,11 @@ export class ClickhouseEventRepository implements IEventRepository {
     return this._config.maximumLiveReloadingSetting ?? 1000;
   }
 
+  /** Exposed for tests and metrics — total batches lost to unrecoverable parse errors. */
+  get permanentlyDroppedBatches() {
+    return this._permanentlyDroppedBatches;
+  }
+
   /**
    * Clamps a start time (in nanoseconds) to now if it's too far in the past.
    * Returns the clamped value as a bigint.
@@ -215,19 +232,32 @@ export class ClickhouseEventRepository implements IEventRepository {
           ? this._clickhouse.taskEventsV2.insert
           : this._clickhouse.taskEvents.insert;
 
-      const [insertError, insertResult] = await insertFn(events, {
-        params: {
-          clickhouse_settings: this.#getClickhouseInsertSettings(),
-        },
-      });
+      const doInsert = async () => {
+        const [insertError, insertResult] = await insertFn(events, {
+          params: {
+            clickhouse_settings: this.#getClickhouseInsertSettings(),
+          },
+        });
+        if (insertError) throw insertError;
+        return insertResult;
+      };
+
+      const outcome = await this.#insertWithJsonParseRecovery(
+        flushId,
+        events,
+        doInsert,
+        `task_events_${this._version}`
+      );
 
-      if (insertError) {
-        throw insertError;
+      if (outcome.kind === "dropped") {
+        // Loud log already emitted; nothing landed in ClickHouse — don't publish to Redis.
+        return;
       }
 
       logger.info("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
         events: events.length,
-        insertResult,
+        insertResult: outcome.insertResult,
+        sanitized: outcome.kind === "sanitized",
         version: this._version,
       });
 
@@ -236,22 +266,112 @@ export class ClickhouseEventRepository implements IEventRepository {
   }
 
   async #flushLlmMetricsBatch(flushId: string, rows: LlmMetricsV1Input[]) {
+    const doInsert = async () => {
+      const [insertError, insertResult] = await this._clickhouse.llmMetrics.insert(rows, {
+        params: {
+          clickhouse_settings: this.#getClickhouseInsertSettings(),
+        },
+      });
+      if (insertError) throw insertError;
+      return insertResult;
+    };
 
-    const [insertError] = await this._clickhouse.llmMetrics.insert(rows, {
-      params: {
-        clickhouse_settings: this.#getClickhouseInsertSettings(),
-      },
-    });
+    const outcome = await this.#insertWithJsonParseRecovery(
+      flushId,
+      rows,
+      doInsert,
+      "llm_metrics_v1"
+    );
 
-    if (insertError) {
-      throw insertError;
+    if (outcome.kind === "dropped") {
+      return;
     }
 
     logger.info("ClickhouseEventRepository.flushLlmMetricsBatch Inserted LLM metrics batch", {
       rows: rows.length,
+      sanitized: outcome.kind === "sanitized",
     });
   }
 
+  /**
+   * Wraps a ClickHouse insert callable with reactive UTF-16 sanitization.
+   *
+   * On a `Cannot parse JSON object` failure:
+   *   1. Sanitize the batch from `max(0, parsedRowN - 1)` onwards (rows
+   *      before the failing one parsed fine — known good).
+   *   2. Retry the insert once with the sanitized batch.
+   *   3. If the retry still fails with the same error class, log loudly,
+   *      increment `permanentlyDroppedBatches`, and return without
+   *      throwing — the scheduler's transient-retry path would just repeat
+   *      the same deterministic failure.
+   *
+   * Non-parse errors propagate unchanged so the scheduler's existing
+   * backoff/retry behaviour still handles transient network or CH issues.
+   */
+  async #insertWithJsonParseRecovery<T extends object>(
+    flushId: string,
+    rows: T[],
+    doInsert: () => Promise<unknown>,
+    contextLabel: string
+  ): Promise<
+    | { kind: "inserted"; insertResult: unknown }
+    | { kind: "sanitized"; insertResult: unknown }
+    | { kind: "dropped" }
+  > {
+    try {
+      return { kind: "inserted", insertResult: await doInsert() };
+    } catch (firstError) {
+      if (!isClickHouseJsonParseError(firstError)) throw firstError;
+
+      const firstMessage =
+        typeof firstError === "object" && firstError !== null && "message" in firstError
+          ? String((firstError as { message?: unknown }).message ?? "")
+          : String(firstError);
+
+      // Sanitize the whole batch. ClickHouse's `at row N` index is logged
+      // for observability but not used to slice — its semantics under
+      // parallel parsing are not stable enough to safely skip rows.
+      const rowHint = parseRowNumberFromError(firstMessage);
+      const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
+
+      logger.warn("Sanitizing batch after ClickHouse JSON parse error", {
+        flushId,
+        contextLabel,
+        batchSize: rows.length,
+        clickhouseRowHint: rowHint,
+        rowsTouched,
+        fieldsSanitized,
+        clickhouseError: firstMessage.split("\n")[0],
+      });
+
+      try {
+        return { kind: "sanitized", insertResult: await doInsert() };
+      } catch (retryError) {
+        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+
+        this._permanentlyDroppedBatches += 1;
+        const retryMessage =
+          typeof retryError === "object" && retryError !== null && "message" in retryError
+            ? String((retryError as { message?: unknown }).message ?? "")
+            : String(retryError);
+        logger.error(
+          "Dropped batch after sanitize-retry still hit ClickHouse JSON parse error",
+          {
+            flushId,
+            contextLabel,
+            batchSize: rows.length,
+            permanentlyDroppedBatches: this._permanentlyDroppedBatches,
+            sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
+            firstError: firstMessage.split("\n")[0],
+            retryError: retryMessage.split("\n")[0],
+          }
+        );
+
+        return { kind: "dropped" };
+      }
+    }
+  }
+
   #createLlmMetricsInput(event: CreateEventInput): LlmMetricsV1Input {
     const llmMetrics = event._llmMetrics!;
 

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

@@ -0,0 +1,118 @@
+import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
+
+/**
+ * Replacement string we substitute for any attribute value that contains
+ * a lone UTF-16 surrogate. JSON-safe, distinctly recognisable in logs and
+ * the dashboard so operators can spot affected rows.
+ */
+export const INVALID_UTF16_SENTINEL = "[invalid-utf16]";
+
+export type SanitizeResult = {
+  /** How many rows had at least one string field replaced. */
+  rowsTouched: number;
+  /** Total count of string fields replaced across all sanitized rows. */
+  fieldsSanitized: number;
+};
+
+/**
+ * Recognises ClickHouse's "Cannot parse JSON object" rejection — the
+ * deterministic-failure class our sanitizer is designed for. Bubbles up
+ * from `@clickhouse/client` as an `InsertError` whose `.message` retains
+ * the original ClickHouse error text.
+ */
+export function isClickHouseJsonParseError(err: unknown): boolean {
+  if (!err) return false;
+  const message =
+    typeof err === "object" && err !== null && "message" in err
+      ? String((err as { message?: unknown }).message ?? "")
+      : String(err);
+  return message.includes("Cannot parse JSON object");
+}
+
+/**
+ * Extracts the row index ClickHouse reported as the first to fail
+ * (`(at row N)`). Returns `null` if the message doesn't include one —
+ * caller should treat that as "sanitize from row 0".
+ */
+export function parseRowNumberFromError(errorMessage: string): number | null {
+  const match = errorMessage.match(/at row (\d+)/);
+  return match ? Number.parseInt(match[1], 10) : null;
+}
+
+/**
+ * Walks `value` recursively and replaces any string leaf that contains a
+ * lone UTF-16 surrogate with `INVALID_UTF16_SENTINEL`. Mutates objects
+ * and arrays in place; primitives are returned unchanged.
+ *
+ * Caller passes anything: a row object, a single field, an unknown JSON
+ * payload. The walker doesn't depend on the row's schema — it sanitizes
+ * every string in the structure, which is exactly what ClickHouse cares
+ * about when parsing the row's JSON form.
+ */
+export function sanitizeUnknownInPlace(value: unknown): { value: unknown; fixed: number } {
+  if (typeof value === "string") {
+    // `detectBadJsonStrings` works on JSON-escaped text — feed it the
+    // serialized form so any lone UTF-16 surrogate in the JS string is
+    // emitted as a `\uXXXX` escape it can spot. Valid surrogate pairs
+    // (e.g. emoji) are emitted as raw characters by JSON.stringify and
+    // exit at the function's fast path.
+    if (detectBadJsonStrings(JSON.stringify(value))) {
+      return { value: INVALID_UTF16_SENTINEL, fixed: 1 };
+    }
+    return { value, fixed: 0 };
+  }
+
+  if (Array.isArray(value)) {
+    let fixed = 0;
+    for (let i = 0; i < value.length; i++) {
+      const result = sanitizeUnknownInPlace(value[i]);
+      value[i] = result.value;
+      fixed += result.fixed;
+    }
+    return { value, fixed };
+  }
+
+  if (value !== null && typeof value === "object") {
+    let fixed = 0;
+    const obj = value as Record<string, unknown>;
+    for (const k of Object.keys(obj)) {
+      const result = sanitizeUnknownInPlace(obj[k]);
+      obj[k] = result.value;
+      fixed += result.fixed;
+    }
+    return { value, fixed };
+  }
+
+  return { value, fixed: 0 };
+}
+
+/**
+ * Sanitizes every row in `rows`, mutating each in place so callers can
+ * hand the same array to the retry insert.
+ *
+ * Rationale for scanning the whole batch (instead of starting from the
+ * row index ClickHouse reports): `at row N` semantics under
+ * `input_format_parallel_parsing` aren't well-defined — N can be
+ * chunk-relative rather than batch-global, and 0-vs-1 indexing differs
+ * between formats. Whole-batch scanning is robust to those quirks and
+ * also catches multiple bad rows in one pass (so a single retry covers
+ * the entire failure even if more than one row is poisoned).
+ *
+ * The cost is bounded: this only runs on the rare ClickHouse-rejection
+ * path, and `detectBadJsonStrings` exits in O(1) for clean strings
+ * (the fast `indexOf("\\u")` check), so healthy attributes are effectively
+ * free even when included in the walk.
+ */
+export function sanitizeRows<T extends object>(rows: T[]): SanitizeResult {
+  const result: SanitizeResult = { rowsTouched: 0, fieldsSanitized: 0 };
+
+  for (let i = 0; i < rows.length; i++) {
+    const { fixed } = sanitizeUnknownInPlace(rows[i]);
+    if (fixed > 0) {
+      result.rowsTouched++;
+      result.fieldsSanitized += fixed;
+    }
+  }
+
+  return result;
+}

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

@@ -39,7 +39,6 @@ import { startSpan } from "./tracing.server";
 import { enrichCreatableEvents } from "./utils/enrichCreatableEvents.server";
 import { waitForLlmPricingReady } from "./llmPricingRegistry.server";
 import { env } from "~/env.server";
-import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 import { singleton } from "~/utils/singleton";
 
 class OTLPExporter {