fix(webapp): anchor the row hint and stop rethrowing a dead batch

Two problems with the recovery path, both found in review.

The failing-row index came from the first `at row N` match anywhere in the
untruncated error text, but ClickHouse embeds a snippet of the offending row's
own data ahead of its real position suffix. A task whose output happened to
contain text like "failed at row 7" therefore redirected the strip: an unrelated
run landed with its output emptied, the un-ingestable run stayed broken, and with
the default budget of one strip the batch then skipped that run entirely. The
strip loop now reads the parenthesised `(at row N)` form and takes the last
occurrence, so the server's own suffix wins over anything in user data, and gives
up rather than guessing when no position is present. The existing helper keeps
its first-match contract for the logging-only caller.

A batch that ClickHouse rejected even with `allow_errors` also propagated out of
both helpers. That failure is deterministic, so the run-replication retry layer
re-ran the whole recovery on bytes that cannot change, and in the event
repository a terminal throw skipped the flush scheduler's queue-depth decrement,
leaking a counter that feeds its load-shedding and memory-pressure decisions.
Such a batch is now counted as wholly dropped instead. Transient failures of that
insert still rethrow so the retry layer keeps working.

Author: ericallam

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

@@ -81,6 +81,26 @@ export function parseRowNumberFromError(errorMessage: string): number | null {
   return match ? Number.parseInt(match[1], 10) : null;
 }
 
+/**
+ * Extracts the failing-row index for the strip loop, where a wrong answer empties
+ * an innocent row's JSON instead of the poison one.
+ *
+ * ClickHouse reports the position as a parenthesised `(at row N)` suffix, but the
+ * same message embeds a snippet of the offending row's own data ahead of it
+ * (`Cannot parse JSON object here: {...}: (while reading ...): (at row N)`), so a
+ * task whose output merely contains the text `at row 7` would otherwise win the
+ * match. Requiring the parentheses and taking the LAST occurrence keeps the
+ * server's own suffix authoritative, since user data appears before it.
+ *
+ * Returns `null` when no parenthesised position is present; the caller treats
+ * that as "row not locatable" and bails to the skip insert rather than guessing.
+ */
+export function parseStrippableRowNumber(errorMessage: string): number | null {
+  const matches = [...errorMessage.matchAll(/\(at row (\d+)\)/g)];
+  if (matches.length === 0) return null;
+  return Number.parseInt(matches[matches.length - 1][1], 10);
+}
+
 /**
  * Walks `value` recursively and replaces any string leaf that contains a
  * lone UTF-16 surrogate with `INVALID_UTF16_SENTINEL`. Mutates objects
@@ -328,7 +348,7 @@ export async function insertWithLimitedStrip<T extends object>(params: {
         break;
       }
 
-      const hint = parseRowNumberFromError(rawErrorMessage(parseError));
+      const hint = parseStrippableRowNumber(rawErrorMessage(parseError));
       const index = hint === null ? -1 : hint - 1;
 
       if (index < 0 || index >= working.length || stripped[index]) {
@@ -341,7 +361,25 @@ export async function insertWithLimitedStrip<T extends object>(params: {
       rowsStripped += 1;
     }
 
-    const insertResult = await insertAllowingBadRows(working);
+    const [skipError, insertResult] = await tryInsertAllowingBadRows(
+      insertAllowingBadRows,
+      working
+    );
+
+    if (skipError) {
+      return wholeBatchDropped({
+        rows,
+        contextLabel,
+        logger,
+        logContext,
+        rowsStripped,
+        capped: true,
+        bailReason,
+        firstMessage,
+        skipError,
+      });
+    }
+
     const dropped = droppedRowCount(insertResult, working.length, hasMaterializedViews);
 
     logger.warn("Landed the batch via allow_errors and skipped the remaining un-ingestable rows", {
@@ -367,6 +405,69 @@ export async function insertWithLimitedStrip<T extends object>(params: {
   }
 }
 
+/**
+ * Runs the `allow_errors` skip insert, separating the two ways it can fail.
+ *
+ * A surviving `Cannot parse JSON object` is deterministic: the same bytes will
+ * fail again, so it is returned for the caller to swallow and count. Anything
+ * else (a connection drop, a server restart) is transient and rethrown so the
+ * caller's retry layer still gets its chance.
+ */
+async function tryInsertAllowingBadRows<T extends object>(
+  insertAllowingBadRows: (rows: T[]) => Promise<unknown>,
+  rows: T[]
+): Promise<[unknown, undefined] | [undefined, unknown]> {
+  try {
+    return [undefined, await insertAllowingBadRows(rows)];
+  } catch (error) {
+    if (!isClickHouseJsonParseError(error)) throw error;
+    return [error, undefined];
+  }
+}
+
+/**
+ * Reports a batch that could not land even with `allow_errors`, without throwing.
+ *
+ * Handing a deterministic parse failure back to the caller's retry layer only
+ * burns the whole recovery again on bytes that cannot change, and in the event
+ * repository a terminal throw also skips the scheduler's queue-depth decrement,
+ * leaking a counter that feeds its load-shedding and memory-pressure decisions.
+ * Counting the batch as wholly dropped keeps both layers honest instead.
+ */
+function wholeBatchDropped<T extends object>(params: {
+  rows: T[];
+  contextLabel: string;
+  logger: JsonParseRecoveryLogger;
+  logContext?: Record<string, unknown>;
+  rowsStripped: number;
+  capped: boolean;
+  bailReason?: RecoveryBailReason;
+  firstMessage: string;
+  skipError: unknown;
+}): JsonParseRecoveryOutcome {
+  const { rows, contextLabel, logger, logContext, rowsStripped, capped, bailReason } = params;
+
+  logger.warn("Dropped the whole batch: ClickHouse rejected it even with allow_errors", {
+    ...logContext,
+    contextLabel,
+    bailReason,
+    batchSize: rows.length,
+    rowsStripped,
+    rowsDropped: rows.length,
+    clickhouseError: params.firstMessage.split("\n")[0],
+    skipInsertError: errorMessage(params.skipError).split("\n")[0],
+  });
+
+  return {
+    kind: "recovered",
+    rowsStripped,
+    rowsDropped: rows.length,
+    rowsDroppedExact: true,
+    capped,
+    bailReason,
+  };
+}
+
 function writtenRowCount(insertResult: unknown): number | null {
   if (typeof insertResult === "object" && insertResult !== null) {
     const summary = (insertResult as { summary?: { written_rows?: unknown } }).summary;
@@ -472,7 +573,21 @@ export async function insertWithBadRowSkip<T extends object>(params: {
       }
     }
 
-    const insertResult = await insertAllowingBadRows(rows);
+    const [skipError, insertResult] = await tryInsertAllowingBadRows(insertAllowingBadRows, rows);
+
+    if (skipError) {
+      return wholeBatchDropped({
+        rows,
+        contextLabel,
+        logger,
+        logContext,
+        rowsStripped: 0,
+        capped: false,
+        firstMessage,
+        skipError,
+      });
+    }
+
     const dropped = droppedRowCount(insertResult, rows.length, hasMaterializedViews);
 
     logger.warn(

apps/webapp/test/sanitizeRowsOnParseError.test.ts

@@ -5,6 +5,7 @@ import {
   insertWithLimitedStrip,
   isClickHouseJsonParseError,
   parseRowNumberFromError,
+  parseStrippableRowNumber,
   sanitizeRows,
   sanitizeUnknownInPlace,
 } from "~/v3/eventRepository/sanitizeRowsOnParseError.server";
@@ -106,6 +107,37 @@ describe("parseRowNumberFromError", () => {
   });
 });
 
+describe("parseStrippableRowNumber", () => {
+  it("reads the parenthesised position ClickHouse appends", () => {
+    expect(
+      parseStrippableRowNumber(
+        "Cannot parse JSON object here: { ... }: (while reading the value of key attributes): (at row 1942)\n: While executing ParallelParsingBlockInputFormat."
+      )
+    ).toBe(1942);
+  });
+
+  it("ignores an `at row N` that appears inside the offending row's own data", () => {
+    const withDecoyInPayload =
+      'Cannot parse JSON object here: {"output":{"dbError":"syntax error at row 7"}}: (while reading the value of key output): (at row 3)';
+
+    expect(parseStrippableRowNumber(withDecoyInPayload)).toBe(3);
+    expect(parseRowNumberFromError(withDecoyInPayload)).toBe(7);
+  });
+
+  it("takes the last parenthesised position when the payload fakes that form too", () => {
+    expect(
+      parseStrippableRowNumber(
+        'Cannot parse JSON object here: {"msg":"failed (at row 99)"}: (at row 4)'
+      )
+    ).toBe(4);
+  });
+
+  it("returns null when no parenthesised position is present, so the caller bails", () => {
+    expect(parseStrippableRowNumber("Cannot parse JSON object here: {...}: at row 5")).toBeNull();
+    expect(parseStrippableRowNumber("Cannot parse JSON object, no position at all")).toBeNull();
+  });
+});
+
 describe("sanitizeUnknownInPlace", () => {
   it("returns the string unchanged when it has no surrogates", () => {
     const result = sanitizeUnknownInPlace("hello world");
@@ -515,6 +547,98 @@ describe("insertWithLimitedStrip", () => {
       })
     ).rejects.toThrow("Connection refused");
   });
+
+  it("strips the row ClickHouse pointed at, not one named by the payload's own text", async () => {
+    const landed: FakeRow[] = [];
+    const insertSync = async (rows: FakeRow[]) => {
+      const badIndex = rows.findIndex((r) => r.poison);
+      if (badIndex >= 0) {
+        throw new Error(
+          `Cannot parse JSON object here: {"output":"failed at row 1"}: (at row ${badIndex + 1})`
+        );
+      }
+      landed.push(...rows);
+      return { summary: { written_rows: String(rows.length) } };
+    };
+    const insertAllowingBadRows = async (rows: FakeRow[]) => {
+      const good = rows.filter((r) => !r.poison);
+      landed.push(...good);
+      return { summary: { written_rows: String(good.length) } };
+    };
+
+    const outcome = await insertWithLimitedStrip({
+      rows: [clean(0), clean(1), { id: 2, poison: true }],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns: (row) => ({ ...row, poison: false, stripped: true }),
+    });
+
+    expect(outcome).toEqual({
+      kind: "recovered",
+      rowsStripped: 1,
+      rowsDropped: 0,
+      rowsDroppedExact: true,
+      capped: false,
+    });
+    expect(landed.map((r) => r.id).sort((a, b) => a - b)).toEqual([0, 1, 2]);
+    expect(landed.find((r) => r.id === 2)?.stripped).toBe(true);
+    expect(landed.filter((r) => r.id !== 2).every((r) => !r.stripped)).toBe(true);
+  });
+
+  it("counts the batch as dropped instead of throwing when even allow_errors is rejected", async () => {
+    const insertSync = async () => {
+      throw parseErrorAtRow(1);
+    };
+    let allowCalls = 0;
+    const insertAllowingBadRows = async () => {
+      allowCalls += 1;
+      throw new Error("Cannot parse JSON object here: {...}: (at row 1)");
+    };
+
+    const outcome = await insertWithLimitedStrip({
+      rows: [clean(0), { id: 1, poison: true }],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert: insertSync,
+      insertSync,
+      insertAllowingBadRows,
+      stripJsonColumns: (row) => row,
+    });
+
+    expect(outcome).toEqual({
+      kind: "recovered",
+      rowsStripped: 1,
+      rowsDropped: 2,
+      rowsDroppedExact: true,
+      capped: true,
+      bailReason: "strip_budget_spent",
+    });
+    expect(allowCalls).toBe(1);
+  });
+
+  it("rethrows a transient failure of the allow_errors insert so the retry layer still runs", async () => {
+    const insertSync = async () => {
+      throw parseErrorAtRow(1);
+    };
+    const insertAllowingBadRows = async () => {
+      throw new Error("Connection refused");
+    };
+
+    await expect(
+      insertWithLimitedStrip({
+        rows: [clean(0), { id: 1, poison: true }],
+        contextLabel: "test",
+        logger: silentLogger,
+        insert: insertSync,
+        insertSync,
+        insertAllowingBadRows,
+        stripJsonColumns: (row) => row,
+      })
+    ).rejects.toThrow("Connection refused");
+  });
 });
 
 /**
@@ -713,4 +837,51 @@ describe("insertWithBadRowSkip", () => {
       })
     ).rejects.toThrow("Connection refused");
   });
+
+  it("counts the batch as dropped instead of throwing when even allow_errors is rejected", async () => {
+    const insert = async () => {
+      throw parseErrorAtRow(1);
+    };
+    let allowCalls = 0;
+    const insertAllowingBadRows = async () => {
+      allowCalls += 1;
+      throw new Error("Cannot parse JSON object here: {...}: (at row 1)");
+    };
+
+    const outcome = await insertWithBadRowSkip({
+      rows: [clean(0), { id: 1, poison: true }, clean(2)],
+      contextLabel: "test",
+      logger: silentLogger,
+      insert,
+      insertAllowingBadRows,
+    });
+
+    expect(outcome).toEqual({
+      kind: "recovered",
+      rowsStripped: 0,
+      rowsDropped: 3,
+      rowsDroppedExact: true,
+      capped: false,
+    });
+    expect(allowCalls).toBe(1);
+  });
+
+  it("rethrows a transient failure of the allow_errors insert so the retry layer still runs", async () => {
+    const insert = async () => {
+      throw parseErrorAtRow(1);
+    };
+    const insertAllowingBadRows = async () => {
+      throw new Error("Connection refused");
+    };
+
+    await expect(
+      insertWithBadRowSkip({
+        rows: [clean(0), { id: 1, poison: true }],
+        contextLabel: "test",
+        logger: silentLogger,
+        insert,
+        insertAllowingBadRows,
+      })
+    ).rejects.toThrow("Connection refused");
+  });
 });