fix(webapp,clickhouse): stop one un-ingestable JSON row dropping its whole ClickHouse batch

When a single run output, trace span, or payload contained JSON ClickHouse
could not ingest (for example nesting past its depth limit), the whole insert
batch was rejected and those runs and spans silently vanished from the runs
list, traces, and logs.

Recovery is now per-table:

- Runs: follow ClickHouse's failing-row hint to strip just the un-ingestable
  JSON column(s) so the run still lands with its status, up to a configurable
  limit (RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH, default 1). Past the
  limit, land the rest with allow_errors and skip the remainder, so recovery
  cost stays flat on large flushes instead of re-sending the batch per row.
- Trace events and payloads: land the batch with allow_errors so the good rows
  land in one pass and only the un-ingestable rows are skipped.

Reading the failing-row hint needs a patch to @clickhouse/client-common, whose
error parser otherwise discards the row number from the server response.

Author: ericallam

.server-changes/fix-runs-list-batch-drop-on-unparseable-json.md

@@ -3,4 +3,4 @@ area: webapp
 type: fix
 ---
 
-Fixed a rare case where a single run or span carrying data that could not be ingested would make other runs or trace events in the same batch go missing from the runs list, traces, and logs. Now the whole batch is kept: the affected item still appears (a run keeps its status, a span keeps its place in the trace) with only its un-ingestable content dropped, and everything else is stored normally.
+Fixed a rare case where a single run or span carrying data that could not be ingested would make other runs or trace events in the same batch go missing from the runs list, traces, and logs. Now the rest of the batch is always kept: an affected run still appears with its status (only its un-ingestable output is dropped), and an affected trace event or payload is skipped instead of taking down everything around it.

apps/webapp/app/env.server.ts

@@ -1672,6 +1672,7 @@ const EnvironmentSchema = z
     RUN_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(2),
     RUN_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
+    RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH: z.coerce.number().int().default(1),
     RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
     RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
     RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),

apps/webapp/app/services/runsReplicationInstance.server.ts

@@ -117,6 +117,7 @@ function initializeRunsReplicationInstance() {
     maxFlushConcurrency: env.RUN_REPLICATION_MAX_FLUSH_CONCURRENCY,
     flushIntervalMs: env.RUN_REPLICATION_FLUSH_INTERVAL_MS,
     flushBatchSize: env.RUN_REPLICATION_FLUSH_BATCH_SIZE,
+    maxPoisonStripsPerBatch: env.RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH,
     leaderLockTimeoutMs: env.RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS,
     leaderLockExtendIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
     leaderLockAcquireAdditionalTimeMs: env.RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,

apps/webapp/app/services/runsReplicationService.server.ts

@@ -7,7 +7,6 @@ import {
   composeTaskRunVersion,
   getPayloadField,
   getTaskRunField,
-  PAYLOAD_INDEX,
   TASK_RUN_INDEX,
 } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
@@ -44,7 +43,8 @@ import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 import { calculateErrorFingerprint } from "~/utils/errorFingerprinting";
 import { baseWorkerQueue } from "~/runEngine/concerns/workerQueueSplit.server";
 import {
-  insertWithJsonParseRecovery,
+  insertWithBadRowSkip,
+  insertWithLimitedStrip,
   type JsonParseRecoveryOutcome,
 } from "~/v3/eventRepository/sanitizeRowsOnParseError.server";
 
@@ -114,6 +114,7 @@ export type RunsReplicationServiceOptions = {
   insertMaxDelayMs?: number;
   disablePayloadInsert?: boolean;
   disableErrorFingerprinting?: boolean;
+  maxPoisonStripsPerBatch?: number;
 };
 
 type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -1133,14 +1134,24 @@ export class RunsReplicationService {
         return insertResult;
       };
 
-      const outcome = await insertWithJsonParseRecovery({
+      const outcome = await insertWithLimitedStrip({
         rows: taskRunInserts,
         contextLabel: "task_runs_v2",
         logger: this.logger,
         logContext: { attempt },
         insert: (rows) => rawInsert(rows),
-        insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
+        insertSync: (rows) =>
+          rawInsert(rows, { async_insert: 0, input_format_parallel_parsing: 0 }),
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
         stripJsonColumns: stripTaskRunJsonColumns,
+        maxPoisonStrips: this.options.maxPoisonStripsPerBatch,
+        hasMaterializedViews: true,
       });
       this.#recordRecoveryOutcome(outcome, "task_runs_v2", taskRunInserts.length);
       return outcome;
@@ -1176,14 +1187,20 @@ export class RunsReplicationService {
         return insertResult;
       };
 
-      const outcome = await insertWithJsonParseRecovery({
+      const outcome = await insertWithBadRowSkip({
         rows: payloadInserts,
         contextLabel: "raw_task_runs_payload_v1",
         logger: this.logger,
         logContext: { attempt },
         insert: (rows) => rawInsert(rows),
-        insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
-        stripJsonColumns: stripPayloadJsonColumns,
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
+        hasMaterializedViews: false,
       });
       this.#recordRecoveryOutcome(outcome, "raw_task_runs_payload_v1", payloadInserts.length);
       return outcome;
@@ -1587,9 +1604,3 @@ function stripTaskRunJsonColumns(row: TaskRunInsertArray): TaskRunInsertArray {
   stripped[TASK_RUN_INDEX.error] = STRIPPED_JSON;
   return stripped;
 }
-
-function stripPayloadJsonColumns(row: PayloadInsertArray): PayloadInsertArray {
-  const stripped = [...row] as PayloadInsertArray;
-  stripped[PAYLOAD_INDEX.payload] = STRIPPED_JSON;
-  return stripped;
-}

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

@@ -67,7 +67,7 @@ import type {
   TraceSummary,
 } from "./eventRepository.types";
 import {
-  insertWithJsonParseRecovery,
+  insertWithBadRowSkip,
   type JsonParseRecoveryOutcome,
 } from "./sanitizeRowsOnParseError.server";
 
@@ -345,14 +345,19 @@ export class ClickhouseEventRepository implements IEventRepository {
         return insertResult;
       };
 
-      const outcome = await insertWithJsonParseRecovery({
+      const outcome = await insertWithBadRowSkip({
         rows: events,
         contextLabel,
         logger,
         logContext: { flushId, version: this._version },
         insert: (rows) => rawInsert(rows),
-        insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
-        stripJsonColumns: stripTaskEventJsonColumns,
+        insertAllowingBadRows: (rows) =>
+          rawInsert(rows, {
+            async_insert: 0,
+            input_format_parallel_parsing: 0,
+            input_format_allow_errors_num: String(rows.length),
+            input_format_allow_errors_ratio: 1,
+          }),
       });
       this.#recordRecoveryOutcome(outcome, contextLabel, events.length);
 
@@ -377,14 +382,19 @@ export class ClickhouseEventRepository implements IEventRepository {
       return insertResult;
     };
 
-    const outcome = await insertWithJsonParseRecovery({
+    const outcome = await insertWithBadRowSkip({
       rows,
       contextLabel: "llm_metrics_v1",
       logger,
       logContext: { flushId },
       insert: (batch) => rawInsert(batch),
-      insertSync: (batch) => rawInsert(batch, { async_insert: 0 }),
-      stripJsonColumns: (row) => row,
+      insertAllowingBadRows: (batch) =>
+        rawInsert(batch, {
+          async_insert: 0,
+          input_format_parallel_parsing: 0,
+          input_format_allow_errors_num: String(batch.length),
+          input_format_allow_errors_ratio: 1,
+        }),
     });
     this.#recordRecoveryOutcome(outcome, "llm_metrics_v1", rows.length);
 
@@ -2945,7 +2955,3 @@ function formatClickhouseUnsignedIntegerString(value: number | bigint): string {
 
   return Math.floor(value).toString();
 }
-
-function stripTaskEventJsonColumns<T extends { attributes?: unknown }>(row: T): T {
-  return { ...row, attributes: {} };
-}