fix(webapp): don't notify trace subscribers when nothing landed

Swallowing an unrecoverable batch instead of throwing lost the one thing the
throw was still doing: skipping the Redis publish. The flush therefore told live
trace and log views that new events had arrived, and logged the batch as
inserted, when ClickHouse had rejected every row. Subscribers refetched and found
nothing.

Both event flush paths now check the outcome and return before publishing when it
reports the whole batch dropped. The check requires an exact dropped count, so a
partial drop on a table whose count is only a floor still publishes the rows that
did land.

Author: ericallam

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

@@ -69,6 +69,7 @@ import type {
 import {
   insertWithBadRowSkip,
   type JsonParseRecoveryOutcome,
+  landedNothing,
 } from "./sanitizeRowsOnParseError.server";
 
 export type ClickhouseEventRepositoryConfig = {
@@ -330,6 +331,10 @@ export class ClickhouseEventRepository implements IEventRepository {
       });
       this.#recordRecoveryOutcome(outcome, contextLabel, events.length);
 
+      if (landedNothing(outcome, events.length)) {
+        return;
+      }
+
       logger.debug("ClickhouseEventRepository.flushBatch Inserted batch into clickhouse", {
         events: events.length,
         outcome: outcome.kind,
@@ -367,6 +372,10 @@ export class ClickhouseEventRepository implements IEventRepository {
     });
     this.#recordRecoveryOutcome(outcome, "llm_metrics_v1", rows.length);
 
+    if (landedNothing(outcome, rows.length)) {
+      return;
+    }
+
     logger.debug("ClickhouseEventRepository.flushLlmMetricsBatch Inserted LLM metrics batch", {
       rows: rows.length,
       outcome: outcome.kind,

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

@@ -232,6 +232,20 @@ export type JsonParseRecoveryOutcome =
       bailReason?: RecoveryBailReason;
     };
 
+/**
+ * True when a recovery landed no rows at all, so nothing reached ClickHouse.
+ *
+ * Callers must not treat such a flush as a successful insert: there is no new
+ * data for subscribers to read, so notifying them only causes refetches that
+ * find nothing. Requires an exact dropped count, so a batch whose partial-drop
+ * count is only a floor is never mistaken for a total loss.
+ */
+export function landedNothing(outcome: JsonParseRecoveryOutcome, batchSize: number): boolean {
+  return (
+    outcome.kind === "recovered" && outcome.rowsDroppedExact && outcome.rowsDropped >= batchSize
+  );
+}
+
 /**
  * Default number of poison rows to isolate-and-strip precisely before bailing
  * to a single `allow_errors` skip insert. One covers the common case (a single

apps/webapp/test/sanitizeRowsOnParseError.test.ts

@@ -4,6 +4,7 @@ import {
   insertWithBadRowSkip,
   insertWithLimitedStrip,
   isClickHouseJsonParseError,
+  landedNothing,
   parseRowNumberFromError,
   parseStrippableRowNumber,
   sanitizeRows,
@@ -107,6 +108,55 @@ describe("parseRowNumberFromError", () => {
   });
 });
 
+describe("landedNothing", () => {
+  it("is true only when an exact dropped count covers the whole batch", () => {
+    expect(
+      landedNothing(
+        {
+          kind: "recovered",
+          rowsStripped: 0,
+          rowsDropped: 3,
+          rowsDroppedExact: true,
+          capped: true,
+        },
+        3
+      )
+    ).toBe(true);
+    expect(
+      landedNothing(
+        {
+          kind: "recovered",
+          rowsStripped: 0,
+          rowsDropped: 2,
+          rowsDroppedExact: true,
+          capped: false,
+        },
+        3
+      )
+    ).toBe(false);
+  });
+
+  it("is false when the dropped count is only a floor, so a partial drop is never read as total loss", () => {
+    expect(
+      landedNothing(
+        {
+          kind: "recovered",
+          rowsStripped: 0,
+          rowsDropped: 1,
+          rowsDroppedExact: false,
+          capped: false,
+        },
+        1
+      )
+    ).toBe(false);
+  });
+
+  it("is false for a healthy or sanitized insert", () => {
+    expect(landedNothing({ kind: "inserted", insertResult: {} }, 5)).toBe(false);
+    expect(landedNothing({ kind: "sanitized", insertResult: {} }, 5)).toBe(false);
+  });
+});
+
 describe("parseStrippableRowNumber", () => {
   it("reads the parenthesised position ClickHouse appends", () => {
     expect(