fix(webapp): only count a dropped batch when nothing landed, and assert the isolation cap bounds inserts

The whole-batch-dropped counter fired whenever a recovered batch stripped no
rows but dropped at least one, which mislabels a batch where clean rows landed
alongside one unrecoverable row. Gate it on rowsDropped === batchSize (nothing
landed) instead. The cap unit test now also asserts the bounded insert count so
a broken budget cannot pass unnoticed.

Author: ericallam

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

@@ -1142,7 +1142,7 @@ export class RunsReplicationService {
         insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
         stripJsonColumns: stripTaskRunJsonColumns,
       });
-      this.#recordRecoveryOutcome(outcome, "task_runs_v2");
+      this.#recordRecoveryOutcome(outcome, "task_runs_v2", taskRunInserts.length);
       return outcome;
     });
   }
@@ -1185,12 +1185,16 @@ export class RunsReplicationService {
         insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
         stripJsonColumns: stripPayloadJsonColumns,
       });
-      this.#recordRecoveryOutcome(outcome, "raw_task_runs_payload_v1");
+      this.#recordRecoveryOutcome(outcome, "raw_task_runs_payload_v1", payloadInserts.length);
       return outcome;
     });
   }
 
-  #recordRecoveryOutcome(outcome: JsonParseRecoveryOutcome, contextLabel: string) {
+  #recordRecoveryOutcome(
+    outcome: JsonParseRecoveryOutcome,
+    contextLabel: string,
+    batchSize: number
+  ) {
     if (outcome.kind !== "recovered") {
       return;
     }
@@ -1211,7 +1215,7 @@ export class RunsReplicationService {
     if (outcome.rowsDropped > 0) {
       this._permanentlyDroppedRows += outcome.rowsDropped;
       this._rowsDroppedCounter.add(outcome.rowsDropped, { table: contextLabel });
-      if (outcome.rowsStripped === 0) {
+      if (outcome.rowsDropped === batchSize) {
         this._permanentlyDroppedBatches += 1;
         this._droppedBatchesCounter.add(1, { table: contextLabel });
       }

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

@@ -354,7 +354,7 @@ export class ClickhouseEventRepository implements IEventRepository {
         insertSync: (rows) => rawInsert(rows, { async_insert: 0 }),
         stripJsonColumns: stripTaskEventJsonColumns,
       });
-      this.#recordRecoveryOutcome(outcome, contextLabel);
+      this.#recordRecoveryOutcome(outcome, contextLabel, events.length);
 
       logger.debug("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
         events: events.length,
@@ -386,15 +386,19 @@ export class ClickhouseEventRepository implements IEventRepository {
       insertSync: (batch) => rawInsert(batch, { async_insert: 0 }),
       stripJsonColumns: (row) => row,
     });
-    this.#recordRecoveryOutcome(outcome, "llm_metrics_v1");
+    this.#recordRecoveryOutcome(outcome, "llm_metrics_v1", rows.length);
 
     logger.debug("ClickhouseEventRepository.flushLlmMetricsBatch Inserted LLM metrics batch", {
       rows: rows.length,
       outcome: outcome.kind,
     });
   }
 
-  #recordRecoveryOutcome(outcome: JsonParseRecoveryOutcome, contextLabel: string) {
+  #recordRecoveryOutcome(
+    outcome: JsonParseRecoveryOutcome,
+    contextLabel: string,
+    batchSize: number
+  ) {
     if (outcome.kind !== "recovered") {
       return;
     }
@@ -415,7 +419,7 @@ export class ClickhouseEventRepository implements IEventRepository {
     if (outcome.rowsDropped > 0) {
       this._permanentlyDroppedRows += outcome.rowsDropped;
       this._rowsDroppedCounter.add(outcome.rowsDropped, { table: contextLabel });
-      if (outcome.rowsStripped === 0) {
+      if (outcome.rowsDropped === batchSize) {
         this._permanentlyDroppedBatches += 1;
         this._droppedBatchesCounter.add(1, { table: contextLabel });
       }

apps/webapp/test/sanitizeRowsOnParseError.test.ts

@@ -27,14 +27,16 @@ function parseError() {
  */
 function makeHarness() {
   const landed: FakeRow[] = [];
+  let insertCalls = 0;
   const insertSync = async (rows: FakeRow[]) => {
+    insertCalls += 1;
     if (rows.some((r) => r.poison || r.unstrippable)) throw parseError();
     landed.push(...rows);
     return undefined;
   };
   const stripJsonColumns = (row: FakeRow): FakeRow =>
     row.unstrippable ? row : { ...row, poison: false, stripped: true };
-  return { landed, insertSync, stripJsonColumns };
+  return { landed, insertSync, stripJsonColumns, insertCalls: () => insertCalls };
 }
 
 describe("isClickHouseJsonParseError", () => {
@@ -379,7 +381,7 @@ describe("insertWithJsonParseRecovery", () => {
   });
 
   it("caps recovery cost on a poison flood by stripping the remainder in one insert", async () => {
-    const { landed, insertSync, stripJsonColumns } = makeHarness();
+    const { landed, insertSync, stripJsonColumns, insertCalls } = makeHarness();
     const rows = Array.from({ length: 8 }, (_, id) => ({ id, poison: true as const }));
 
     const outcome = await insertWithJsonParseRecovery({
@@ -400,6 +402,7 @@ describe("insertWithJsonParseRecovery", () => {
     }
     expect(landed).toHaveLength(8);
     expect(landed.every((r) => r.stripped)).toBe(true);
+    expect(insertCalls()).toBeLessThanOrEqual(6);
   });
 
   it("rethrows non-parse errors so the caller's transient-retry path handles them", async () => {