fix(webapp): preserve log search batches with invalid JSON escapes

Prevent log search batches from being lost when a message contains an
invalid JSON escape. Search inserts now use the same lenient escape
setting as source inserts and share the existing bounded
sanitize/row-skip recovery, so an unrepairable row does not discard its
neighbors.

Drop counters start at zero for each reason, and recovered batches count
only skipped rows as dropped.

Mono-RevId: 353071db24c77a392baaa7cb96340f6ab64d8e3e

Author: carderne

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

@@ -71,6 +71,7 @@ import type {
   TraceEventOptions,
   TraceSummary,
 } from "./eventRepository.types";
+import { insertLogsSearchRows } from "./insertLogsSearchRows.server";
 import {
   insertWithBadRowSkip,
   type JsonParseRecoveryOutcome,
@@ -233,6 +234,19 @@ export class ClickhouseEventRepository implements IEventRepository {
       "logs_search.dual_write.rows_dropped",
       { unit: "rows" }
     );
+    for (const reason of [
+      "limiter_full",
+      "shutdown",
+      "source_recovered",
+      "mapping_failed",
+      "insert_failed",
+      "parse_failed",
+    ]) {
+      this._logsSearchRowsDroppedCounter.add(0, {
+        ...this._logsSearchMetricAttributes,
+        reason,
+      });
+    }
     this._logsSearchBatchesCounter = meter.createCounter("logs_search.dual_write.batches", {
       unit: "batches",
     });
@@ -575,34 +589,51 @@ export class ClickhouseEventRepository implements IEventRepository {
 
   async #insertLogsSearchRows(flushId: string, rows: TaskEventSearchV2Input[]): Promise<void> {
     const startedAt = Date.now();
-    let lastError: { clickhouseErrorType?: string } | undefined;
+    let lastError: unknown;
 
     for (let attempt = 1; attempt <= 2; attempt++) {
-      const [error] = await this._logsSearchClickhouse.taskEventsSearch.insert(rows, {
-        params: {
-          clickhouse_settings: {
-            async_insert: 0,
-            insert_deduplication_token: flushId,
-          },
-        },
-      });
-
-      if (!error) {
-        this._logsSearchRowsLandedCounter.add(rows.length, this._logsSearchMetricAttributes);
+      try {
+        const outcome = await insertLogsSearchRows(
+          this._logsSearchClickhouse.taskEventsSearch.insert,
+          flushId,
+          rows,
+          logger
+        );
+        const dropped = outcome.kind === "recovered" ? outcome.rowsDropped : 0;
+        if (dropped > 0) {
+          this._logsSearchRowsDroppedCounter.add(dropped, {
+            ...this._logsSearchMetricAttributes,
+            reason: "parse_failed",
+          });
+        }
+        if (outcome.kind !== "recovered" || outcome.rowsDroppedExact) {
+          this._logsSearchRowsLandedCounter.add(
+            rows.length - dropped,
+            this._logsSearchMetricAttributes
+          );
+        }
         this._logsSearchBatchesCounter.add(1, {
           ...this._logsSearchMetricAttributes,
-          outcome: attempt === 1 ? "ok" : "retried_ok",
+          outcome: landedNothing(outcome, rows.length)
+            ? "failed"
+            : outcome.kind === "recovered"
+              ? outcome.rowsDroppedExact
+                ? "recovered"
+                : "recovered_unknown"
+              : attempt === 1 && outcome.kind === "inserted"
+                ? "ok"
+                : "retried_ok",
         });
         this._logsSearchFlushDurationHistogram.record(
           Date.now() - startedAt,
           this._logsSearchMetricAttributes
         );
         return;
-      }
-
-      lastError = error;
-      if (attempt === 1) {
-        await new Promise((resolve) => setTimeout(resolve, 500));
+      } catch (error) {
+        lastError = error;
+        if (attempt === 1) {
+          await new Promise((resolve) => setTimeout(resolve, 500));
+        }
       }
     }
 
@@ -621,7 +652,7 @@ export class ClickhouseEventRepository implements IEventRepository {
     logger.error("Logs search dual-write insert failed", {
       flushId,
       rows: rows.length,
-      clickhouseErrorType: lastError?.clickhouseErrorType,
+      error: lastError,
     });
   }
 

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

@@ -0,0 +1,48 @@
+import type { ClickHouse, ClickHouseSettings, TaskEventSearchV2Input } from "@internal/clickhouse";
+import {
+  insertWithBadRowSkip,
+  isClickHouseJsonParseError,
+  type JsonParseRecoveryLogger,
+} from "./sanitizeRowsOnParseError.server";
+
+export function insertLogsSearchRows(
+  insertRows: ClickHouse["taskEventsSearch"]["insert"],
+  flushId: string,
+  rows: TaskEventSearchV2Input[],
+  logger: JsonParseRecoveryLogger
+) {
+  const insert = async (batch: TaskEventSearchV2Input[], settings?: ClickHouseSettings) => {
+    const [error, result] = await insertRows(batch, {
+      params: {
+        clickhouse_settings: {
+          async_insert: 0,
+          insert_deduplication_token: flushId,
+          ...settings,
+        },
+      },
+    });
+    if (error) throw error;
+    return result;
+  };
+
+  return insertWithBadRowSkip({
+    rows,
+    contextLabel: "task_events_search_v2",
+    logger,
+    logContext: { flushId },
+    hasMaterializedViews: false,
+    isParseError: (error) =>
+      isClickHouseJsonParseError(error) ||
+      (typeof error === "object" &&
+        error !== null &&
+        "clickhouseErrorType" in error &&
+        error.clickhouseErrorType === "CANNOT_PARSE_ESCAPE_SEQUENCE"),
+    insert: (batch) => insert(batch),
+    insertAllowingBadRows: (batch) =>
+      insert(batch, {
+        input_format_parallel_parsing: 0,
+        input_format_allow_errors_num: String(batch.length),
+        input_format_allow_errors_ratio: 1,
+      }),
+  });
+}

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

@@ -435,12 +435,13 @@ export async function insertWithLimitedStrip<T extends object>(params: {
  */
 async function tryInsertAllowingBadRows<T extends object>(
   insertAllowingBadRows: (rows: T[]) => Promise<unknown>,
-  rows: T[]
+  rows: T[],
+  isParseError: (error: unknown) => boolean = isClickHouseJsonParseError
 ): Promise<[unknown, undefined] | [undefined, unknown]> {
   try {
     return [undefined, await insertAllowingBadRows(rows)];
   } catch (error) {
-    if (!isClickHouseJsonParseError(error)) throw error;
+    if (!isParseError(error)) throw error;
     return [error, undefined];
   }
 }
@@ -565,14 +566,16 @@ export async function insertWithBadRowSkip<T extends object>(params: {
   insert: (rows: T[]) => Promise<unknown>;
   insertAllowingBadRows: (rows: T[]) => Promise<unknown>;
   hasMaterializedViews?: boolean;
+  isParseError?: (error: unknown) => boolean;
 }): Promise<JsonParseRecoveryOutcome> {
   const { rows, contextLabel, logger, logContext, insert, insertAllowingBadRows } = params;
   const hasMaterializedViews = params.hasMaterializedViews ?? true;
+  const isParseError = params.isParseError ?? isClickHouseJsonParseError;
 
   try {
     return { kind: "inserted", insertResult: await insert(rows) };
   } catch (firstError) {
-    if (!isClickHouseJsonParseError(firstError)) throw firstError;
+    if (!isParseError(firstError)) throw firstError;
 
     const firstMessage = errorMessage(firstError);
     const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
@@ -590,11 +593,15 @@ export async function insertWithBadRowSkip<T extends object>(params: {
       try {
         return { kind: "sanitized", insertResult: await insert(rows) };
       } catch (retryError) {
-        if (!isClickHouseJsonParseError(retryError)) throw retryError;
+        if (!isParseError(retryError)) throw retryError;
       }
     }
 
-    const [skipError, insertResult] = await tryInsertAllowingBadRows(insertAllowingBadRows, rows);
+    const [skipError, insertResult] = await tryInsertAllowingBadRows(
+      insertAllowingBadRows,
+      rows,
+      isParseError
+    );
 
     if (skipError) {
       return wholeBatchDropped({

apps/webapp/test/clickhouseEventRepositoryDualWrite.test.ts

@@ -1,11 +1,22 @@
-import { ClickHouse, type TaskEventV2Input } from "@internal/clickhouse";
+import {
+  ClickHouse,
+  TASK_EVENT_SEARCH_V2_INSERT_COLUMNS,
+  toTaskEventSearchV2Row,
+  type TaskEventSearchV2Input,
+  type TaskEventV2Input,
+} from "@internal/clickhouse";
 import { clickhouseTest } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
 import { z } from "zod";
 import {
   ClickhouseEventRepository,
   logsSearchRolloutSelectedRowCount,
 } from "~/v3/eventRepository/clickhouseEventRepository.server";
+import { insertLogsSearchRows } from "~/v3/eventRepository/insertLogsSearchRows.server";
+import {
+  INVALID_UTF16_SENTINEL,
+  insertWithBadRowSkip,
+} from "~/v3/eventRepository/sanitizeRowsOnParseError.server";
 import { latestMetrics, metricSum } from "./otlpMetrics.helpers";
 import { createInMemoryMetrics } from "./utils/tracing";
 
@@ -86,7 +97,10 @@ describe("ClickhouseEventRepository logs search dual writer", () => {
 
       try {
         const allowedEvents = Array.from({ length: 1_000 }, (_, index) =>
-          event({ span_id: `span_allowed_${index}` })
+          event({
+            span_id: `span_allowed_${index}`,
+            message: index === 500 ? "broken \uD800 escape" : "source write survives",
+          })
         );
         (repository as any).addToBatch([
           ...allowedEvents,
@@ -127,6 +141,63 @@ describe("ClickhouseEventRepository logs search dual writer", () => {
     60_000
   );
 
+  clickhouseTest(
+    "recovers a strict search insert without losing neighboring rows",
+    async ({ clickhouseContainer }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        logLevel: "error",
+      });
+      const strictInsert = clickhouse.writer.insertUnsafe<TaskEventSearchV2Input>({
+        name: "strict-search-insert",
+        table: "trigger_dev.task_events_search_v2",
+        columns: TASK_EVENT_SEARCH_V2_INSERT_COLUMNS,
+        settings: { input_format_json_throw_on_bad_escape_sequence: 1 },
+      });
+      const rows = ["clean prefix", "broken \uD800 escape", "clean suffix"].map((message, index) =>
+        toTaskEventSearchV2Row(event({ message, span_id: `span_${index}` }), new Date())
+      );
+      const readMessages = clickhouse.reader.query({
+        name: "read-recovered-search-messages",
+        query: `SELECT message FROM trigger_dev.task_events_search_v2
+          WHERE environment_id = {environmentId: String} ORDER BY span_id`,
+        params: z.object({ environmentId: z.string() }),
+        schema: z.object({ message: z.string() }),
+      });
+
+      try {
+        const [error] = await strictInsert(rows);
+        expect(error?.clickhouseErrorType).toBe("CANNOT_PARSE_ESCAPE_SEQUENCE");
+        const insert = async (batch: TaskEventSearchV2Input[]) => {
+          const [insertError, result] = await strictInsert(batch);
+          if (insertError) throw insertError;
+          return result;
+        };
+        await expect(
+          insertWithBadRowSkip({
+            rows,
+            contextLabel: "default-recovery",
+            logger: console,
+            insert,
+            insertAllowingBadRows: insert,
+          })
+        ).rejects.toMatchObject({ clickhouseErrorType: "CANNOT_PARSE_ESCAPE_SEQUENCE" });
+        const outcome = await insertLogsSearchRows(strictInsert, "strict-recovery", rows, console);
+        expect(outcome.kind).toBe("sanitized");
+        const [queryError, messages] = await readMessages({ environmentId: "env_dual_write_test" });
+        expect(queryError).toBeNull();
+        expect(messages).toEqual([
+          { message: "clean prefix" },
+          { message: INVALID_UTF16_SENTINEL },
+          { message: "clean suffix" },
+        ]);
+      } finally {
+        await clickhouse.close();
+      }
+    },
+    60_000
+  );
+
   clickhouseTest(
     "records mapping failures as dropped search rows",
     async ({ clickhouseContainer }) => {
@@ -154,6 +225,12 @@ describe("ClickhouseEventRepository logs search dual writer", () => {
       });
 
       try {
+        expect(
+          metricSum(await latestMetrics(metrics), "logs_search.dual_write.rows_dropped", {
+            table: "task_events_search_v2",
+            reason: "mapping_failed",
+          })
+        ).toBe(0);
         (repository as any).addToBatch([
           event({ start_time: new Date().toISOString().replace("T", " ").replace("Z", "") }),
         ]);

apps/webapp/test/sanitizeRowsOnParseError.test.ts

@@ -74,6 +74,18 @@ describe("isClickHouseJsonParseError", () => {
     expect(isClickHouseJsonParseError(err)).toBe(true);
   });
 
+  it("does not classify escape errors as native JSON parse errors", () => {
+    expect(
+      isClickHouseJsonParseError({ type: "CANNOT_PARSE_ESCAPE_SEQUENCE", message: "bad escape" })
+    ).toBe(false);
+    expect(
+      isClickHouseJsonParseError({
+        clickhouseErrorType: "CANNOT_PARSE_ESCAPE_SEQUENCE",
+        message: "bad escape",
+      })
+    ).toBe(false);
+  });
+
   it("returns false for unrelated errors", () => {
     expect(isClickHouseJsonParseError(new Error("Connection refused"))).toBe(false);
     expect(

internal-packages/clickhouse/src/taskEventsSearch.ts

@@ -120,6 +120,9 @@ export function insertTaskEventsSearchV2(
     name: "insertTaskEventsSearchV2",
     table: "trigger_dev.task_events_search_v2",
     columns: TASK_EVENT_SEARCH_V2_INSERT_COLUMNS,
+    settings: {
+      input_format_json_throw_on_bad_escape_sequence: 0,
+    },
   });
 }