fix(webapp,clickhouse): keep customer run data out of insert error messages and logs

ClickHouse insert errors no longer include the rejected row's data in their message, and run replication logs identify runs by ID instead of logging their payload or output.

Mono-RevId: 9fc4fb20b388b657d630a1082a7833c0d1b13972

Author: Mariefay

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

@@ -1,5 +1,6 @@
 import type { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
 import {
+  insertErrorServerText,
   type ClickHouse,
   type ClickHouseSettings,
   type PayloadInsertArray,
@@ -1086,7 +1087,7 @@ export class RunsReplicationService {
 
   // Retry all errors except known permanent ones
   #isRetryableError(error: Error): boolean {
-    const errorMessage = error.message.toLowerCase();
+    const errorMessage = insertErrorServerText(error).toLowerCase();
 
     // Permanent errors that should NOT be retried
     const permanentErrorPatterns = [
@@ -1155,7 +1156,8 @@ export class RunsReplicationService {
           },
         });
         if (insertError) {
-          this.logger.error("Error inserting task run inserts attempt", {
+          // Warn: recovery may still land the batch; a real loss logs at error.
+          this.logger.warn("Error inserting task run inserts attempt", {
             error: insertError,
             attempt,
           });
@@ -1168,6 +1170,7 @@ export class RunsReplicationService {
       const outcome = await insertWithLimitedStrip({
         rows: taskRunInserts,
         contextLabel: "task_runs_v2",
+        rowId: (row) => getTaskRunField(row, "run_id"),
         logger: this.logger,
         logContext: { attempt },
         insert: (rows) => rawInsert(rows),
@@ -1208,7 +1211,8 @@ export class RunsReplicationService {
           }
         );
         if (insertError) {
-          this.logger.error("Error inserting payload inserts attempt", {
+          // Warn: recovery may still land the batch; a real loss logs at error.
+          this.logger.warn("Error inserting payload inserts attempt", {
             error: insertError,
             attempt,
           });
@@ -1310,7 +1314,7 @@ export class RunsReplicationService {
     event: "insert" | "update" | "delete",
     _version: bigint
   ): Promise<TaskRunInsertArray> {
-    const output = await this.#prepareJson(run.output, run.outputType);
+    const output = await this.#prepareJson(run.id, run.output, run.outputType);
     const errorData = { data: run.error };
 
     // Calculate error fingerprint for failed runs
@@ -1406,7 +1410,7 @@ export class RunsReplicationService {
   }
 
   async #preparePayloadInsert(run: TaskRun, _version: bigint): Promise<PayloadInsertArray> {
-    const payload = await this.#prepareJson(run.payload, run.payloadType);
+    const payload = await this.#prepareJson(run.id, run.payload, run.payloadType);
 
     // Return array matching PAYLOAD_COLUMNS order
     return [
@@ -1417,6 +1421,7 @@ export class RunsReplicationService {
   }
 
   async #prepareJson(
+    runId: string,
     data: string | undefined | null,
     dataType: string
   ): Promise<{ data: unknown }> {
@@ -1430,23 +1435,22 @@ export class RunsReplicationService {
 
     if (detectBadJsonStrings(data)) {
       this.logger.warn("Detected bad JSON strings", {
-        data,
+        runId,
         dataType,
+        length: data.length,
       });
       return { data: undefined };
     }
 
-    const packet = {
-      data,
-      dataType,
-    };
-
-    const [parseError, parsedData] = await tryCatch(parsePacketAsJson(packet));
+    const [parseError, parsedData] = await tryCatch(parsePacketAsJson({ data, dataType }));
 
+    // Parser messages quote the input, so only the error name is logged.
     if (parseError) {
       this.logger.error("Error parsing packet", {
-        error: parseError,
-        packet,
+        runId,
+        dataType,
+        length: data.length,
+        errorName: parseError instanceof Error ? parseError.name : typeof parseError,
       });
 
       return { data: undefined };

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

@@ -1,5 +1,5 @@
 import type { ClickHouse, SessionInsertArray } from "@internal/clickhouse";
-import { getSessionField } from "@internal/clickhouse";
+import { insertErrorServerText, getSessionField } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
 import {
   LogicalReplicationClient,
@@ -716,7 +716,7 @@ export class SessionsReplicationService {
 
   // Retry all errors except known permanent ones
   #isRetryableError(error: Error): boolean {
-    const errorMessage = error.message.toLowerCase();
+    const errorMessage = insertErrorServerText(error).toLowerCase();
 
     // Permanent errors that should NOT be retried
     const permanentErrorPatterns = [

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

@@ -3,7 +3,7 @@ import type {
   WebhookDeliveryInsertArray,
   WebhookEndpointInsertArray,
 } from "@internal/clickhouse";
-import { getWebhookDeliveryField } from "@internal/clickhouse";
+import { insertErrorServerText, getWebhookDeliveryField } from "@internal/clickhouse";
 import { type RedisOptions } from "@internal/redis";
 import {
   LogicalReplicationClient,
@@ -759,7 +759,7 @@ export class WebhookDeliveriesReplicationService {
 
   // Retry all errors except known permanent ones
   #isRetryableError(error: Error): boolean {
-    const errorMessage = error.message.toLowerCase();
+    const errorMessage = insertErrorServerText(error).toLowerCase();
 
     // Permanent errors that should NOT be retried
     const permanentErrorPatterns = [

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

@@ -1,3 +1,4 @@
+import { insertErrorServerText, insertErrorType } from "@internal/clickhouse";
 import { detectBadJsonStrings } from "~/utils/detectBadJsonStrings";
 
 /**
@@ -58,17 +59,12 @@ export type SanitizeResult = {
 
 /**
  * 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.
+ * deterministic-failure class our sanitizer is designed for. Reads the full
+ * server text, since `InsertError.message` only carries the error type.
  */
 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");
+  return insertErrorServerText(err).includes("Cannot parse JSON object");
 }
 
 /**
@@ -183,19 +179,7 @@ export function sanitizeRows<T extends object>(rows: T[]): SanitizeResult {
   return result;
 }
 
-function errorMessage(err: unknown): string {
-  return typeof err === "object" && err !== null && "message" in err
-    ? String((err as { message?: unknown }).message ?? "")
-    : String(err);
-}
-
-function rawErrorMessage(err: unknown): string {
-  if (typeof err === "object" && err !== null) {
-    const raw = (err as { rawMessage?: unknown }).rawMessage;
-    if (typeof raw === "string" && raw.length > 0) return raw;
-  }
-  return errorMessage(err);
-}
+type RowIdAccessor<T> = (row: T) => string | undefined;
 
 /**
  * `error` is part of the contract because losing a whole batch is a genuine
@@ -297,6 +281,7 @@ export async function insertWithLimitedStrip<T extends object>(params: {
   stripJsonColumns: (row: T) => T;
   maxPoisonStrips?: number;
   hasMaterializedViews?: boolean;
+  rowId?: RowIdAccessor<T>;
 }): Promise<JsonParseRecoveryOutcome> {
   const { rows, contextLabel, logger, logContext, insert, insertSync, insertAllowingBadRows } =
     params;
@@ -309,7 +294,7 @@ export async function insertWithLimitedStrip<T extends object>(params: {
   } catch (firstError) {
     if (!isClickHouseJsonParseError(firstError)) throw firstError;
 
-    const firstMessage = errorMessage(firstError);
+    const clickhouseErrorType = insertErrorType(firstError);
     const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
 
     if (fieldsSanitized > 0) {
@@ -319,7 +304,7 @@ export async function insertWithLimitedStrip<T extends object>(params: {
         batchSize: rows.length,
         rowsTouched,
         fieldsSanitized,
-        clickhouseError: firstMessage.split("\n")[0],
+        clickhouseErrorType,
       });
 
       try {
@@ -331,6 +316,8 @@ export async function insertWithLimitedStrip<T extends object>(params: {
 
     const working = rows.slice();
     const stripped = new Array(working.length).fill(false);
+    // Row hints come from insertSync (no parallel parsing), so these IDs are exact.
+    const strippedRunIds: string[] = [];
     let rowsStripped = 0;
     let bailReason: RecoveryBailReason = "strip_attempts_exhausted";
 
@@ -347,7 +334,8 @@ export async function insertWithLimitedStrip<T extends object>(params: {
               contextLabel,
               batchSize: rows.length,
               rowsStripped,
-              clickhouseError: firstMessage.split("\n")[0],
+              strippedRunIds,
+              clickhouseErrorType,
             }
           );
         }
@@ -368,14 +356,16 @@ export async function insertWithLimitedStrip<T extends object>(params: {
         break;
       }
 
-      const hint = parseStrippableRowNumber(rawErrorMessage(parseError));
+      const hint = parseStrippableRowNumber(insertErrorServerText(parseError));
       const index = hint === null ? -1 : hint - 1;
 
       if (index < 0 || index >= working.length || stripped[index]) {
         bailReason = "row_not_locatable";
         break;
       }
 
+      const runId = params.rowId?.(working[index]);
+      if (runId) strippedRunIds.push(runId);
       working[index] = stripJsonColumns(working[index]);
       stripped[index] = true;
       rowsStripped += 1;
@@ -395,7 +385,8 @@ export async function insertWithLimitedStrip<T extends object>(params: {
         rowsStripped,
         capped: true,
         bailReason,
-        firstMessage,
+        clickhouseErrorType,
+        strippedRunIds,
         skipError,
       });
     }
@@ -411,7 +402,8 @@ export async function insertWithLimitedStrip<T extends object>(params: {
       rowsDropped: dropped.rows,
       rowsDroppedExact: dropped.exact,
       landedRows: writtenRowCount(insertResult),
-      clickhouseError: firstMessage.split("\n")[0],
+      strippedRunIds,
+      clickhouseErrorType,
     });
 
     return {
@@ -464,7 +456,8 @@ function wholeBatchDropped<T extends object>(params: {
   rowsStripped: number;
   capped: boolean;
   bailReason?: RecoveryBailReason;
-  firstMessage: string;
+  clickhouseErrorType?: string;
+  strippedRunIds?: string[];
   skipError: unknown;
 }): JsonParseRecoveryOutcome {
   const { rows, contextLabel, logger, logContext, rowsStripped, capped, bailReason } = params;
@@ -476,8 +469,9 @@ function wholeBatchDropped<T extends object>(params: {
     batchSize: rows.length,
     rowsStripped,
     rowsDropped: rows.length,
-    clickhouseError: params.firstMessage.split("\n")[0],
-    skipInsertError: errorMessage(params.skipError).split("\n")[0],
+    strippedRunIds: params.strippedRunIds,
+    clickhouseErrorType: params.clickhouseErrorType,
+    skipInsertErrorType: insertErrorType(params.skipError),
   });
 
   return {
@@ -577,7 +571,7 @@ export async function insertWithBadRowSkip<T extends object>(params: {
   } catch (firstError) {
     if (!isParseError(firstError)) throw firstError;
 
-    const firstMessage = errorMessage(firstError);
+    const clickhouseErrorType = insertErrorType(firstError);
     const { rowsTouched, fieldsSanitized } = sanitizeRows(rows);
 
     if (fieldsSanitized > 0) {
@@ -587,7 +581,7 @@ export async function insertWithBadRowSkip<T extends object>(params: {
         batchSize: rows.length,
         rowsTouched,
         fieldsSanitized,
-        clickhouseError: firstMessage.split("\n")[0],
+        clickhouseErrorType,
       });
 
       try {
@@ -611,7 +605,7 @@ export async function insertWithBadRowSkip<T extends object>(params: {
         logContext,
         rowsStripped: 0,
         capped: false,
-        firstMessage,
+        clickhouseErrorType,
         skipError,
       });
     }
@@ -627,7 +621,7 @@ export async function insertWithBadRowSkip<T extends object>(params: {
         rowsDropped: dropped.rows,
         rowsDroppedExact: dropped.exact,
         landedRows: writtenRowCount(insertResult),
-        clickhouseError: firstMessage.split("\n")[0],
+        clickhouseErrorType,
       }
     );