feat(webapp): cap row-isolation recovery cost against a poison flood

Bound the per-batch isolation work so a burst of un-ingestable rows cannot lag
the shared replication stream. Once a batch exceeds the isolation-insert budget
the remaining rows are stripped and landed in one insert instead of bisected
further. The ceiling is transparent for normal recovery and small run batches
(they still isolate precisely); it only bites on a large batch that is heavily
poisoned. Adds a recovery_cap_hits counter as the flood signal.

Author: ericallam

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

@@ -192,6 +192,14 @@ export class RunsReplicationService {
    */
   private _rowIsolationRecoveries = 0;
 
+  /**
+   * Counts batches whose isolation blew the per-batch insert budget (a poison
+   * flood), so the remaining rows were stripped in one insert instead of
+   * bisected further. A burst signal; clean rows in those batches also lose
+   * their JSON.
+   */
+  private _recoveryCapHits = 0;
+
   /**
    * Counts rows that landed with their un-ingestable JSON column(s) stripped
    * (the run kept its status, only the output/payload content was lost).
@@ -213,6 +221,7 @@ export class RunsReplicationService {
   private _insertRetriesCounter: Counter;
   private _eventsProcessedCounter: Counter;
   private _rowIsolatedBatchesCounter: Counter;
+  private _recoveryCapHitsCounter: Counter;
   private _rowsStrippedCounter: Counter;
   private _rowsDroppedCounter: Counter;
   private _droppedBatchesCounter: Counter;
@@ -278,6 +287,12 @@ export class RunsReplicationService {
       }
     );
 
+    this._recoveryCapHitsCounter = this._meter.createCounter("runs_replication.recovery_cap_hits", {
+      description:
+        "Batches whose row isolation hit the per-batch insert budget and stripped the remainder in one insert (poison-flood signal)",
+      unit: "batches",
+    });
+
     this._rowsStrippedCounter = this._meter.createCounter("runs_replication.rows_stripped", {
       description:
         "Rows landed with their un-ingestable JSON stripped (kept the row, lost only the JSON content)",
@@ -472,6 +487,11 @@ export class RunsReplicationService {
     return this._rowIsolationRecoveries;
   }
 
+  /** Exposed for tests and metrics — batches whose isolation hit the per-batch insert budget. */
+  get recoveryCapHits() {
+    return this._recoveryCapHits;
+  }
+
   /** Exposed for tests and metrics — rows that landed with their un-ingestable JSON stripped. */
   get rowsStripped() {
     return this._rowsStripped;
@@ -1178,6 +1198,11 @@ export class RunsReplicationService {
     this._rowIsolationRecoveries += 1;
     this._rowIsolatedBatchesCounter.add(1, { table: contextLabel });
 
+    if (outcome.capped) {
+      this._recoveryCapHits += 1;
+      this._recoveryCapHitsCounter.add(1, { table: contextLabel });
+    }
+
     if (outcome.rowsStripped > 0) {
       this._rowsStripped += outcome.rowsStripped;
       this._rowsStrippedCounter.add(outcome.rowsStripped, { table: contextLabel });

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

@@ -137,6 +137,13 @@ export class ClickhouseEventRepository implements IEventRepository {
   private _rowIsolationRecoveries = 0;
   private readonly _rowIsolatedBatchesCounter: Counter;
 
+  /**
+   * Counts batches whose isolation blew the per-batch insert budget (a poison
+   * flood), so the remaining rows were stripped in one insert. A burst signal.
+   */
+  private _recoveryCapHits = 0;
+  private readonly _recoveryCapHitsCounter: Counter;
+
   /**
    * Counts rows that landed with their un-ingestable JSON stripped (the span
    * kept its place in the trace, only the attributes content was lost).
@@ -167,6 +174,11 @@ export class ClickhouseEventRepository implements IEventRepository {
         "Batches recovered by isolating un-ingestable rows (landed the rest) after a ClickHouse JSON parse error",
       unit: "batches",
     });
+    this._recoveryCapHitsCounter = meter.createCounter("ingest.flush.recovery_cap_hits", {
+      description:
+        "Batches whose row isolation hit the per-batch insert budget and stripped the remainder in one insert (poison-flood signal)",
+      unit: "batches",
+    });
     this._rowsStrippedCounter = meter.createCounter("ingest.flush.rows_stripped", {
       description:
         "Rows landed with their un-ingestable JSON stripped (kept the row, lost only the JSON content)",
@@ -236,6 +248,11 @@ export class ClickhouseEventRepository implements IEventRepository {
     return this._rowIsolationRecoveries;
   }
 
+  /** Exposed for tests and metrics — batches whose isolation hit the per-batch insert budget. */
+  get recoveryCapHits() {
+    return this._recoveryCapHits;
+  }
+
   /** Exposed for tests and metrics — rows that landed with their un-ingestable JSON stripped. */
   get rowsStripped() {
     return this._rowsStripped;
@@ -385,6 +402,11 @@ export class ClickhouseEventRepository implements IEventRepository {
     this._rowIsolationRecoveries += 1;
     this._rowIsolatedBatchesCounter.add(1, { table: contextLabel });
 
+    if (outcome.capped) {
+      this._recoveryCapHits += 1;
+      this._recoveryCapHitsCounter.add(1, { table: contextLabel });
+    }
+
     if (outcome.rowsStripped > 0) {
       this._rowsStripped += outcome.rowsStripped;
       this._rowsStrippedCounter.add(outcome.rowsStripped, { table: contextLabel });

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

@@ -177,7 +177,21 @@ export type JsonParseRecoveryLogger = {
 export type JsonParseRecoveryOutcome =
   | { kind: "inserted"; insertResult: unknown }
   | { kind: "sanitized"; insertResult: unknown }
-  | { kind: "recovered"; rowsStripped: number; rowsDropped: number };
+  | { kind: "recovered"; rowsStripped: number; rowsDropped: number; capped: boolean };
+
+/**
+ * Default per-batch ceiling on the number of isolation inserts. Bisecting a
+ * batch tops out around ~3x its row count, so this leaves small batches (e.g.
+ * the ~50-row run-replication flushes) to isolate precisely even when heavily
+ * poisoned. It bites on the case that actually needs bounding: a large batch
+ * (e.g. thousands of trace events) with many un-ingestable rows, where precise
+ * isolation would otherwise run thousands of inserts and lag the shared
+ * replication stream. Once exceeded, the remaining subtree is stripped in one
+ * insert instead of bisected further.
+ */
+export const DEFAULT_MAX_ISOLATION_INSERTS = 256;
+
+type IsolationBudget = { inserts: number; capped: boolean };
 
 /**
  * Shared ClickHouse insert recovery for `Cannot parse JSON object` rejections,
@@ -195,7 +209,11 @@ export type JsonParseRecoveryOutcome =
  *      keeps its place in the trace) and only the un-ingestable content is
  *      lost. Clean rows land in full. A row that can't parse even stripped is
  *      dropped and counted.
- *   4. Non-parse errors propagate unchanged so the caller's transient-retry
+ *   4. Isolation is bounded by `maxIsolationInserts`. If a batch is so poisoned
+ *      that it blows the budget, the remaining rows are stripped and landed in
+ *      one insert (`capped`) rather than bisected further, so a poison flood
+ *      degrades gracefully instead of lagging the whole stream.
+ *   5. Non-parse errors propagate unchanged so the caller's transient-retry
  *      path still handles them.
  */
 export async function insertWithJsonParseRecovery<T extends object>(params: {
@@ -206,6 +224,7 @@ export async function insertWithJsonParseRecovery<T extends object>(params: {
   insert: (rows: T[]) => Promise<unknown>;
   insertSync: (rows: T[]) => Promise<unknown>;
   stripJsonColumns: (row: T) => T;
+  maxIsolationInserts?: number;
 }): Promise<JsonParseRecoveryOutcome> {
   const { rows, contextLabel, logger, logContext, insert, insertSync, stripJsonColumns } = params;
 
@@ -236,26 +255,39 @@ export async function insertWithJsonParseRecovery<T extends object>(params: {
       }
     }
 
+    const budget: IsolationBudget = {
+      inserts: params.maxIsolationInserts ?? DEFAULT_MAX_ISOLATION_INSERTS,
+      capped: false,
+    };
     const { rowsStripped, rowsDropped } = await isolateAndStripPoisonRows(
       rows,
       insertSync,
-      stripJsonColumns
+      stripJsonColumns,
+      budget
     );
 
-    logger.info(
-      "Isolated un-ingestable rows after ClickHouse JSON parse error — landed the batch with poison JSON stripped",
-      {
-        ...logContext,
-        contextLabel,
-        batchSize: rows.length,
-        rowsStripped,
-        rowsDropped,
-        sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
-        clickhouseError: firstMessage.split("\n")[0],
-      }
-    );
+    const meta = {
+      ...logContext,
+      contextLabel,
+      batchSize: rows.length,
+      rowsStripped,
+      rowsDropped,
+      sampleRow: JSON.stringify(rows[0] ?? null).slice(0, 1024),
+      clickhouseError: firstMessage.split("\n")[0],
+    };
+    if (budget.capped) {
+      logger.warn(
+        "Recovery cost cap hit — stripped the remaining rows in one insert (clean rows in the capped subtree also lost their JSON)",
+        meta
+      );
+    } else {
+      logger.info(
+        "Isolated un-ingestable rows after ClickHouse JSON parse error — landed the batch with poison JSON stripped",
+        meta
+      );
+    }
 
-    return { kind: "recovered", rowsStripped, rowsDropped };
+    return { kind: "recovered", rowsStripped, rowsDropped, capped: budget.capped };
   }
 }
 
@@ -266,19 +298,34 @@ export async function insertWithJsonParseRecovery<T extends object>(params: {
  * lands. A row that can't parse even stripped is dropped. All inserts are
  * synchronous so each subset's parse error surfaces. The two halves are probed
  * concurrently so recovery wall-clock stays ~O(log n) round-trips rather than
- * O(n), which matters when a poison burst would otherwise lag the shared
- * replication stream. Returns exact counts.
+ * O(n). When `budget.inserts` is exhausted the remaining subtree is stripped
+ * and landed in a single insert (setting `budget.capped`) so a poison flood
+ * cannot do unbounded work. Returns exact counts.
  */
 async function isolateAndStripPoisonRows<T extends object>(
   rows: T[],
   insertSync: (rows: T[]) => Promise<unknown>,
-  stripJsonColumns: (row: T) => T
+  stripJsonColumns: (row: T) => T,
+  budget: IsolationBudget
 ): Promise<{ rowsStripped: number; rowsDropped: number }> {
   if (rows.length === 0) {
     return { rowsStripped: 0, rowsDropped: 0 };
   }
 
+  if (budget.inserts <= 0) {
+    budget.capped = true;
+    budget.inserts -= 1;
+    try {
+      await insertSync(rows.map(stripJsonColumns));
+      return { rowsStripped: rows.length, rowsDropped: 0 };
+    } catch (error) {
+      if (!isClickHouseJsonParseError(error)) throw error;
+      return { rowsStripped: 0, rowsDropped: rows.length };
+    }
+  }
+
   if (rows.length === 1) {
+    budget.inserts -= 1;
     try {
       await insertSync([stripJsonColumns(rows[0])]);
       return { rowsStripped: 1, rowsDropped: 0 };
@@ -291,12 +338,13 @@ async function isolateAndStripPoisonRows<T extends object>(
   const mid = rows.length >> 1;
 
   const recoverHalf = async (half: T[]) => {
+    budget.inserts -= 1;
     try {
       await insertSync(half);
       return { rowsStripped: 0, rowsDropped: 0 };
     } catch (error) {
       if (!isClickHouseJsonParseError(error)) throw error;
-      return isolateAndStripPoisonRows(half, insertSync, stripJsonColumns);
+      return isolateAndStripPoisonRows(half, insertSync, stripJsonColumns, budget);
     }
   };
 

apps/webapp/test/runsReplicationJsonRecoveryBenchmark.test.ts

@@ -211,6 +211,7 @@ async function runScenario(
   console.log(`Repl dur:     ${replication.duration.toFixed(0)}ms`);
   console.log(`Throughput:   ${throughput.toFixed(0)} runs/sec`);
   console.log(`Row-isolation recoveries: ${service.rowIsolationRecoveries}`);
+  console.log(`Recovery cap hits: ${service.recoveryCapHits}`);
   console.log(`Rows stripped (kept row, lost JSON): ${service.rowsStripped}`);
   console.log(`Rows dropped (unrecoverable): ${service.permanentlyDroppedRows}`);
   console.log(`Dropped batches (whole): ${service.permanentlyDroppedBatches}`);
@@ -275,6 +276,7 @@ describe("RunsReplicationService JSON-recovery ELU benchmark", () => {
       expect(poisoned.replication.count).toBe(poisoned.expectedLanded);
       expect(poisoned.service.permanentlyDroppedBatches).toBe(0);
       expect(poisoned.service.permanentlyDroppedRows).toBe(0);
+      expect(poisoned.service.recoveryCapHits).toBe(0);
       expect(poisoned.service.rowIsolationRecoveries).toBeGreaterThanOrEqual(1);
       expect(poisoned.service.rowsStripped).toBe(poisoned.producerStats.poisoned);
     }