perf(webapp,clickhouse): preserve early exit when hiding log retries

Fetch bounded extra rows and remove duplicate projection identities in the application. Keep exact keyset pagination while background merges collapse physical copies.

Author: carderne

apps/webapp/app/presenters/v3/LogsListPresenter.server.ts

@@ -18,8 +18,10 @@ import { ServiceValidationError } from "~/v3/services/baseService.server";
 import {
   escapeClickHouseLike,
   hasMinimumLogsSearchLength,
+  LOGS_SEARCH_RETRY_OVERFETCH_FACTOR,
   MIN_LOGS_SEARCH_LENGTH,
   normalizeLogsSearchTerm,
+  prepareLogsSearchPage,
 } from "~/utils/logSearch";
 
 export type { LogLevel };
@@ -66,7 +68,7 @@ export type LogEntry = LogsList["logs"][0];
 
 // Bump when the cursor shape changes so stale cursors are ignored (reset to the first page)
 // rather than misparsed.
-const LOG_CURSOR_VERSION = 3;
+const LOG_CURSOR_VERSION = 4;
 
 // Cursor is a base64 encoded JSON of the pagination keys
 type LogCursor = {
@@ -76,6 +78,7 @@ type LogCursor = {
   triggeredTimestamp: string; // DateTime64(9) string
   traceId: string;
   spanId: string;
+  projectionFingerprint?: string;
 };
 
 const LogCursorSchema = z.object({
@@ -85,6 +88,7 @@ const LogCursorSchema = z.object({
   triggeredTimestamp: z.string(),
   traceId: z.string(),
   spanId: z.string(),
+  projectionFingerprint: z.string().optional(),
 });
 
 function encodeCursor(cursor: LogCursor): string {
@@ -222,6 +226,10 @@ export class LogsListPresenter extends BasePresenter {
     }
 
     const effectivePageSize = Math.min(pageSize, env.LOGS_LIST_MAX_PAGE_SIZE);
+    const usesV2Search = env.LOGS_SEARCH_TABLE_VERSION === "v2";
+    const queryLimit = usesV2Search
+      ? (effectivePageSize + 1) * LOGS_SEARCH_RETRY_OVERFETCH_FACTOR
+      : effectivePageSize + 1;
 
     // Only honor a cursor scoped to this org+env; one copied from another scope would shift the
     // pagination anchor instead of resetting to the first page.
@@ -238,10 +246,9 @@ export class LogsListPresenter extends BasePresenter {
     const clampedTo = effectiveTo !== undefined ? (effectiveTo > now ? now : effectiveTo) : now;
 
     const rawSearchTerm = search?.trim() ?? "";
-    const normalizedSearchTerm =
-      env.LOGS_SEARCH_TABLE_VERSION === "v2"
-        ? normalizeLogsSearchTerm(rawSearchTerm)
-        : rawSearchTerm.toLocaleLowerCase();
+    const normalizedSearchTerm = usesV2Search
+      ? normalizeLogsSearchTerm(rawSearchTerm)
+      : rawSearchTerm.toLocaleLowerCase();
     if (rawSearchTerm !== "" && !hasMinimumLogsSearchLength(normalizedSearchTerm)) {
       throw new ServiceValidationError(
         `Log searches must be at least ${MIN_LOGS_SEARCH_LENGTH} characters.`
@@ -286,7 +293,7 @@ export class LogsListPresenter extends BasePresenter {
       }
 
       if (searchTerm !== undefined) {
-        if (env.LOGS_SEARCH_TABLE_VERSION === "v2") {
+        if (usesV2Search) {
           // One predicate lets the text index answer substring searches without an OR across
           // independently indexed columns.
           queryBuilder.where("search_text LIKE {searchPattern: String}", {
@@ -327,26 +334,37 @@ export class LogsListPresenter extends BasePresenter {
         queryBuilder.whereOr(conditions);
       }
 
-      // Keyset pagination over the full sort key. ORDER BY is DESC, so the next page is the rows
-      // that sort after the cursor (strictly less-than). (triggered_timestamp, trace_id) is not
-      // unique because spans of a trace share both, so span_id is the final tiebreaker; without
-      // it rows at a tie boundary could be skipped or duplicated across pages.
+      // Keyset pagination over the sort key. ORDER BY is DESC, so the next page is the rows
+      // that sort after the cursor (strictly less-than). V2 adds the projection identity as the
+      // final tiebreaker so retry copies and distinct rows at a span boundary paginate safely.
       if (decodedCursor) {
+        const cursorParams = {
+          cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp,
+          cursorTraceId: decodedCursor.traceId,
+          cursorSpanId: decodedCursor.spanId,
+          ...(usesV2Search && decodedCursor.projectionFingerprint
+            ? { cursorProjectionFingerprint: decodedCursor.projectionFingerprint }
+            : {}),
+        };
         queryBuilder.where(
-          `(triggered_timestamp < {cursorTriggeredTimestamp: String}
-            OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String})
-            OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}))`,
-          {
-            cursorTriggeredTimestamp: decodedCursor.triggeredTimestamp,
-            cursorTraceId: decodedCursor.traceId,
-            cursorSpanId: decodedCursor.spanId,
-          }
+          usesV2Search && decodedCursor.projectionFingerprint
+            ? `(triggered_timestamp < {cursorTriggeredTimestamp: String}
+              OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String})
+              OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String})
+              OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id = {cursorSpanId: String} AND projection_fingerprint < {cursorProjectionFingerprint: UInt128}))`
+            : `(triggered_timestamp < {cursorTriggeredTimestamp: String}
+              OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id < {cursorTraceId: String})
+              OR (triggered_timestamp = {cursorTriggeredTimestamp: String} AND trace_id = {cursorTraceId: String} AND span_id < {cursorSpanId: String}))`,
+          cursorParams
         );
       }
 
-      queryBuilder.orderBy("triggered_timestamp DESC, trace_id DESC, span_id DESC");
-      // Limit + 1 to check if there are more results
-      queryBuilder.limit(effectivePageSize + 1);
+      queryBuilder.orderBy(
+        usesV2Search
+          ? "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC"
+          : "triggered_timestamp DESC, trace_id DESC, span_id DESC"
+      );
+      queryBuilder.limit(queryLimit);
 
       return queryBuilder.execute();
     };
@@ -360,8 +378,14 @@ export class LogsListPresenter extends BasePresenter {
     // marker. Keep the default throw behavior so the product never presents truncated results as
     // complete.
     const results = queryResult ?? [];
-    const hasMore = results.length > effectivePageSize;
-    const logs = results.slice(0, effectivePageSize);
+    const page = usesV2Search
+      ? prepareLogsSearchPage(results, effectivePageSize, queryLimit)
+      : {
+          rows: results.slice(0, effectivePageSize),
+          hasMore: results.length > effectivePageSize,
+        };
+    const hasMore = page.hasMore;
+    const logs = page.rows;
 
     // Build next cursor from the last item
     let nextCursor: string | undefined;
@@ -374,6 +398,7 @@ export class LogsListPresenter extends BasePresenter {
         triggeredTimestamp: lastLog.triggered_timestamp,
         traceId: lastLog.trace_id,
         spanId: lastLog.span_id,
+        projectionFingerprint: lastLog.projection_fingerprint_string,
       });
     }
 

apps/webapp/app/utils/logSearch.test.ts

@@ -3,6 +3,7 @@ import {
   escapeClickHouseLike,
   hasMinimumLogsSearchLength,
   normalizeLogsSearchTerm,
+  prepareLogsSearchPage,
 } from "./logSearch";
 
 describe("log search normalization", () => {
@@ -22,4 +23,33 @@ describe("log search normalization", () => {
     expect(hasMinimumLogsSearchLength("abc")).toBe(true);
     expect(hasMinimumLogsSearchLength("日本語")).toBe(true);
   });
+
+  it("removes projector retry copies after bounded overfetch", () => {
+    const row = (fingerprint: string) => ({
+      projection_fingerprint_string: fingerprint,
+      trace_id: `trace_${fingerprint}`,
+      span_id: `span_${fingerprint}`,
+      run_id: `run_${fingerprint}`,
+      start_time: "2026-08-14 12:00:00.000000000",
+    });
+    const page = prepareLogsSearchPage([row("a"), row("a"), row("b"), row("c"), row("d")], 2, 5);
+
+    expect(page.rows.map((item) => item.projection_fingerprint_string)).toEqual(["a", "b"]);
+    expect(page.hasMore).toBe(true);
+  });
+
+  it("keeps pagination open when retries fill the overfetch bound", () => {
+    const duplicate = {
+      projection_fingerprint_string: "same",
+      trace_id: "trace",
+      span_id: "span",
+      run_id: "run",
+      start_time: "2026-08-14 12:00:00.000000000",
+    };
+
+    expect(prepareLogsSearchPage([duplicate, duplicate, duplicate, duplicate], 2, 4)).toEqual({
+      rows: [duplicate],
+      hasMore: true,
+    });
+  });
 });

apps/webapp/app/utils/logSearch.ts

@@ -1,4 +1,34 @@
 export const MIN_LOGS_SEARCH_LENGTH = 3;
+export const LOGS_SEARCH_RETRY_OVERFETCH_FACTOR = 4;
+
+type ProjectedLogIdentity = {
+  projection_fingerprint_string?: string;
+  trace_id: string;
+  span_id: string;
+  run_id: string;
+  start_time: string;
+};
+
+export function prepareLogsSearchPage<T extends ProjectedLogIdentity>(
+  rows: T[],
+  pageSize: number,
+  queryLimit: number
+): { rows: T[]; hasMore: boolean } {
+  const seen = new Set<string>();
+  const uniqueRows = rows.filter((row) => {
+    const identity =
+      row.projection_fingerprint_string ??
+      JSON.stringify([row.trace_id, row.span_id, row.run_id, row.start_time]);
+    if (seen.has(identity)) return false;
+    seen.add(identity);
+    return true;
+  });
+
+  return {
+    rows: uniqueRows.slice(0, pageSize),
+    hasMore: uniqueRows.length > pageSize || rows.length === queryLimit,
+  };
+}
 
 export function hasMinimumLogsSearchLength(value: string): boolean {
   return [...value.trim()].length >= MIN_LOGS_SEARCH_LENGTH;

internal-packages/clickhouse/schema/039_schedule_task_events_search_v2.sql

@@ -27,11 +27,8 @@ CREATE TABLE trigger_dev.task_events_search_v2_projector
   status LowCardinality(String) CODEC(ZSTD(1)),
   duration UInt64 CODEC(ZSTD(1)),
   parent_span_id String CODEC(ZSTD(1)),
-  projection_fingerprint FixedString(16) DEFAULT sipHash128(
-    trace_id,
-    span_id,
-    run_id,
-    start_time
+  projection_fingerprint UInt128 DEFAULT reinterpretAsUInt128(
+    sipHash128(trace_id, span_id, run_id, start_time)
   ),
 
   INDEX idx_run_id run_id TYPE bloom_filter(0.001) GRANULARITY 1,

internal-packages/clickhouse/src/client/queryBuilder.ts

@@ -148,7 +148,6 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
   private params: QueryParams = {};
   private orderByClause: string | null = null;
   private limitClause: string | null = null;
-  private limitByClause: string | null = null;
   private groupByClause: string | null = null;
 
   constructor(
@@ -243,11 +242,6 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
     return this;
   }
 
-  limitBy(limit: number, expression: string): this {
-    this.limitByClause = `LIMIT ${limit} BY ${expression}`;
-    return this;
-  }
-
   execute(): ReturnType<ClickhouseQueryFunction<void, TOutput>> {
     const { query, params } = this.build();
 
@@ -296,9 +290,6 @@ export class ClickhouseQueryFastBuilder<TOutput extends Record<string, any>> {
     if (this.orderByClause) {
       query += ` ORDER BY ${this.orderByClause}`;
     }
-    if (this.limitByClause) {
-      query += ` ${this.limitByClause}`;
-    }
     if (this.limitClause) {
       query += ` ${this.limitClause}`;
     }

internal-packages/clickhouse/src/taskEvents.ts

@@ -299,6 +299,7 @@ export const LogsSearchListResult = z.object({
   status: z.string(),
   duration: z.number().or(z.string()),
   triggered_timestamp: z.string(),
+  projection_fingerprint_string: z.string().optional(),
 });
 
 export type LogsSearchListResult = z.output<typeof LogsSearchListResult>;
@@ -335,17 +336,21 @@ export function getLogsSearchListQueryBuilder(
       "status",
       "duration",
       "triggered_timestamp",
+      ...(version === "v2"
+        ? [
+            {
+              name: "projection_fingerprint_string",
+              expression: "toString(projection_fingerprint)",
+            },
+          ]
+        : []),
     ],
     settings: {
       use_query_condition_cache: 1,
     },
   });
 
-  return (options?: Parameters<typeof createBuilder>[0]) => {
-    const builder = createBuilder(options);
-    if (version === "v2") builder.limitBy(1, "projection_fingerprint");
-    return builder;
-  };
+  return createBuilder;
 }
 
 // Single log detail query builder (for side panel)

internal-packages/clickhouse/src/taskEventsSearch.test.ts

@@ -56,7 +56,9 @@ async function project(ch: ClickHouse, start: Date, end: Date) {
 function searchRows(ch: ClickHouse) {
   const builder = ch.taskEventsSearch.logsListQueryBuilder("v2");
   builder.where("organization_id = {organizationId: String}", { organizationId: ORG });
-  builder.orderBy("triggered_timestamp DESC, trace_id DESC, span_id DESC");
+  builder.orderBy(
+    "triggered_timestamp DESC, trace_id DESC, span_id DESC, projection_fingerprint DESC"
+  );
   builder.limit(50);
   return builder.execute();
 }
@@ -111,9 +113,9 @@ describe("task events search v2", () => {
       expect(Number(firstProjection.summary?.written_rows)).toBe(1);
       expect(Number(retryProjection.summary?.written_rows)).toBe(1);
 
-      const [readError, rows] = await searchRows(ch);
-      expect(readError).toBeNull();
-      expect(rows).toHaveLength(1);
+      const [preMergeReadError, preMergeRows] = await searchRows(ch);
+      expect(preMergeReadError).toBeNull();
+      expect([1, 2]).toContain(preMergeRows?.length);
       const rawQuery = ch.reader.query({
         name: "count-raw-search-v2-fixture",
         query: `SELECT count() AS count FROM trigger_dev.task_events_search_v2
@@ -123,7 +125,7 @@ describe("task events search v2", () => {
       });
       let [rawError, rawRows] = await rawQuery({ organizationId: ORG });
       expect(rawError).toBeNull();
-      expect(rawRows?.[0].count).toBe(2);
+      expect([1, 2]).toContain(rawRows?.[0].count);
 
       const optimize = ch.writer.command({
         name: "merge-search-v2-retry-fixture",
@@ -134,6 +136,9 @@ describe("task events search v2", () => {
       [rawError, rawRows] = await rawQuery({ organizationId: ORG });
       expect(rawError).toBeNull();
       expect(rawRows?.[0].count).toBe(1);
+      const [readError, rows] = await searchRows(ch);
+      expect(readError).toBeNull();
+      expect(rows).toHaveLength(1);
 
       expect(rows?.[0].message.toLowerCase()).toContain(
         "typeerror: zahlungsübersicht failed, retrying /api/orders/42"
@@ -145,7 +150,7 @@ describe("task events search v2", () => {
         query: `SELECT search_text, error_message
           FROM trigger_dev.task_events_search_v2
           WHERE organization_id = {organizationId: String}
-          LIMIT 1 BY projection_fingerprint`,
+          LIMIT 1`,
         params: z.object({ organizationId: z.string() }),
         schema: z.object({ search_text: z.string(), error_message: z.string() }),
       });
@@ -189,7 +194,7 @@ describe("task events search v2", () => {
         query: `SELECT length(search_text) AS search_length
           FROM trigger_dev.task_events_search_v2
           WHERE organization_id = {organizationId: String}
-          LIMIT 1 BY projection_fingerprint`,
+          LIMIT 1`,
         params: z.object({ organizationId: z.string() }),
         schema: z.object({ search_length: z.number() }),
       });

internal-packages/clickhouse/src/taskEventsSearchProjector.ts

@@ -44,12 +44,12 @@ const projectedColumns = `
   duration,
   parent_span_id`;
 
-const projectionFingerprint = (alias: string) => `sipHash128(
+const projectionFingerprint = (alias: string) => `reinterpretAsUInt128(sipHash128(
   ${alias}.trace_id,
   ${alias}.span_id,
   ${alias}.run_id,
   ${alias}.start_time
-)`;
+))`;
 
 const projectionSql = `
 INSERT INTO trigger_dev.task_events_search_v2