fix(webapp): use composite keyset cursor for run pagination

matt-aitken · claude · matt-aitken · commit c6c3b4eead35 · 2026-06-07T16:42:28.000+01:00
listRunIds/listRuns order by the composite key (created_at, run_id) but
the cursor predicate cut on run_id alone. That is only sound when run_id
lexicographic order matches created_at order. When a burst of runs is
created such that the two diverge, keyset pagination both re-includes
already-returned runs (duplicates) and drops runs it should return
(skips). For bulk replay this produced duplicate runs; for the dashboard
and runs.list it could silently skip or repeat runs at page boundaries.

- Encode cursors as the composite (created_at, run_id) key
  (v2_&lt;createdAtMs&gt;_&lt;runId&gt;) and cut on the matching tuple predicate
  ((created_at, run_id) &lt; / &gt; (...)). The ORDER BY is unchanged, so the
  table's primary-key alignment (and query performance) is preserved.
- Cursors are server-issued opaque tokens (the SDK just echoes
  pagination.next/previous back), so this needs no client update. Legacy
  bare-run_id cursors decode to the old run_id-only predicate for
  backwards compatibility with in-flight cursors.
- Add listRunIdsWithCursor for forward-only batch iteration (bulk
  actions) so the created_at component is sourced from the same query
  that orders the rows.
- ClickHouse getTaskRunsQueryBuilder now also selects
  toUnixTimestamp64Milli(created_at) AS created_at_ms.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/.server-changes/bulk-action-cursor-pagination.md b/.server-changes/bulk-action-cursor-pagination.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix run pagination that could duplicate or skip runs: the query orders by `(created_at, run_id)` but the cursor cut on `run_id` alone, which diverges when run_id order doesn't match created_at order (e.g. bulk replay re-processing runs). Cursors now encode the composite key as an opaque token and cut on the matching tuple; legacy bare-run_id cursors stay supported for in-flight pagination.
diff --git a/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts b/apps/webapp/app/presenters/v3/TestTaskPresenter.server.ts
@@ -203,7 +203,7 @@ export class TestTaskPresenter {
       prisma: this.replica as PrismaClient,
     });
 
-    const runIds = await runsRepository.listRunIds({
+    const { runIds } = await runsRepository.listRunIds({
       organizationId: environment.organizationId,
       environmentId: environment.id,
       projectId: environment.projectId,
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.live.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.live.ts
@@ -58,7 +58,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
             return { count: 0, since: newRunsSince };
           }
 
-          const newRunIds = await runsRepository.listRunIds({
+          const { runIds: newRunIds } = await runsRepository.listRunIds({
             organizationId: project.organizationId,
             projectId: project.id,
             environmentId: environment.id,
diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts
@@ -4,12 +4,16 @@ import {
   type FilterRunsOptions,
   type IRunsRepository,
   type ListRunsOptions,
+  type RunIdsPage,
   type RunListInputOptions,
   type RunsRepositoryOptions,
   type TagListOptions,
   convertRunListInputOptionsToFilterRunsOptions,
 } from "./runsRepository.server";
 import parseDuration from "parse-duration";
+import { decodeRunsCursor, encodeRunsCursor } from "./runsCursor.server";
+
+type RunCursorRow = { runId: string; createdAt: number };
 
 export class ClickHouseRunsRepository implements IRunsRepository {
   constructor(private readonly options: RunsRepositoryOptions) {}
@@ -18,25 +22,52 @@ export class ClickHouseRunsRepository implements IRunsRepository {
     return "clickhouse";
   }
 
-  async listRunIds(options: ListRunsOptions) {
+  /**
+   * Runs the keyset-paginated query and returns `{ runId, createdAt }` rows
+   * (one extra beyond `page.size` to signal "has more"). The ordering is always
+   * the composite `(created_at, run_id)`; the cursor predicate must match it.
+   *
+   * Composite cursors carry both components, so we cut on the
+   * `(created_at, run_id)` tuple — sound regardless of how run_id order relates
+   * to created_at order. Legacy bare-run_id cursors fall back to the old
+   * `run_id`-only predicate (knowingly unsound) for backwards compatibility
+   * with in-flight cursors.
+   */
+  private async listRunRows(options: ListRunsOptions): Promise<RunCursorRow[]> {
     const queryBuilder = this.options.clickhouse.taskRuns.queryBuilder();
     applyRunFiltersToQueryBuilder(
       queryBuilder,
       await convertRunListInputOptionsToFilterRunsOptions(options, this.options.prisma)
     );
 
+    const forward = options.page.direction === "forward" || !options.page.direction;
+
     if (options.page.cursor) {
-      if (options.page.direction === "forward" || !options.page.direction) {
-        queryBuilder
-          .where("run_id < {runId: String}", { runId: options.page.cursor })
-          .orderBy("created_at DESC, run_id DESC")
-          .limit(options.page.size + 1);
+      const decoded = decodeRunsCursor(options.page.cursor);
+
+      if (forward) {
+        if (decoded.kind === "composite") {
+          queryBuilder.where(
+            "(created_at, run_id) < (fromUnixTimestamp64Milli({cursorCreatedAt: Int64}), {runId: String})",
+            { cursorCreatedAt: decoded.createdAt, runId: decoded.runId }
+          );
+        } else {
+          queryBuilder.where("run_id < {runId: String}", { runId: decoded.runId });
+        }
+        queryBuilder.orderBy("created_at DESC, run_id DESC");
       } else {
-        queryBuilder
-          .where("run_id > {runId: String}", { runId: options.page.cursor })
-          .orderBy("created_at ASC, run_id ASC")
-          .limit(options.page.size + 1);
+        if (decoded.kind === "composite") {
+          queryBuilder.where(
+            "(created_at, run_id) > (fromUnixTimestamp64Milli({cursorCreatedAt: Int64}), {runId: String})",
+            { cursorCreatedAt: decoded.createdAt, runId: decoded.runId }
+          );
+        } else {
+          queryBuilder.where("run_id > {runId: String}", { runId: decoded.runId });
+        }
+        queryBuilder.orderBy("created_at ASC, run_id ASC");
       }
+
+      queryBuilder.limit(options.page.size + 1);
     } else {
       // Initial page - no cursor provided
       queryBuilder.orderBy("created_at DESC, run_id DESC").limit(options.page.size + 1);
@@ -48,75 +79,89 @@ export class ClickHouseRunsRepository implements IRunsRepository {
       throw queryError;
     }
 
-    const runIds = result.map((row) => row.run_id);
-    return runIds;
+    return result.map((row) => ({ runId: row.run_id, createdAt: row.created_at_ms }));
   }
 
-  async listFriendlyRunIds(options: ListRunsOptions) {
-    // First get internal IDs from ClickHouse
-    const internalIds = await this.listRunIds(options);
+  /**
+   * A keyset-paginated page of run ids ordered by `(created_at, run_id)`, plus
+   * the cursors to page forward/backward. Cursors are composite tokens that
+   * match the ordering, so pagination can't duplicate or skip runs even when
+   * run_id order diverges from created_at order. This is the single source of
+   * cursor construction — `listRuns` and bulk actions both build on it.
+   */
+  async listRunIds(options: ListRunsOptions): Promise<RunIdsPage> {
+    const rows = await this.listRunRows(options);
 
-    if (internalIds.length === 0) {
-      return [];
-    }
-
-    // Then get friendly IDs from Prisma
-    const runs = await this.options.prisma.taskRun.findMany({
-      where: {
-        id: {
-          in: internalIds,
-        },
-      },
-      select: {
-        friendlyId: true,
-      },
-    });
-
-    return runs.map((run) => run.friendlyId);
-  }
-
-  async listRuns(options: ListRunsOptions) {
-    const runIds = await this.listRunIds(options);
+    // listRunRows fetches one extra row beyond page.size to detect "has more".
+    const hasMore = rows.length > options.page.size;
 
-    // If there are more runs than the page size, we need to fetch the next page
-    const hasMore = runIds.length > options.page.size;
+    const cursorFor = (row: RunCursorRow | undefined): string | null =>
+      row ? encodeRunsCursor(row.createdAt, row.runId) : null;
 
     let nextCursor: string | null = null;
     let previousCursor: string | null = null;
 
-    //get cursors for next and previous pages
     const direction = options.page.direction ?? "forward";
     switch (direction) {
       case "forward": {
-        previousCursor = options.page.cursor ? runIds.at(0) ?? null : null;
+        previousCursor = options.page.cursor ? cursorFor(rows.at(0)) : null;
         if (hasMore) {
-          // The next cursor should be the last run ID from this page
-          nextCursor = runIds[options.page.size - 1];
+          // The next cursor is the last run on this page.
+          nextCursor = cursorFor(rows[options.page.size - 1]);
         }
         break;
       }
       case "backward": {
-        const reversedRunIds = [...runIds].reverse();
+        const reversedRows = [...rows].reverse();
         if (hasMore) {
-          previousCursor = reversedRunIds.at(1) ?? null;
-          nextCursor = reversedRunIds.at(options.page.size) ?? null;
+          previousCursor = cursorFor(reversedRows.at(1));
+          nextCursor = cursorFor(reversedRows.at(options.page.size));
         } else {
-          nextCursor = reversedRunIds.at(options.page.size - 1) ?? null;
+          nextCursor = cursorFor(reversedRows.at(options.page.size - 1));
         }
-
         break;
       }
     }
 
-    const runIdsToReturn =
-      options.page.direction === "backward" && hasMore
-        ? runIds.slice(1, options.page.size + 1)
-        : runIds.slice(0, options.page.size);
+    const runIds = (
+      direction === "backward" && hasMore
+        ? rows.slice(1, options.page.size + 1)
+        : rows.slice(0, options.page.size)
+    ).map((row) => row.runId);
+
+    return { runIds, pagination: { nextCursor, previousCursor } };
+  }
+
+  async listFriendlyRunIds(options: ListRunsOptions) {
+    // First get internal IDs from ClickHouse
+    const { runIds } = await this.listRunIds(options);
+
+    if (runIds.length === 0) {
+      return [];
+    }
+
+    // Then get friendly IDs from Prisma
+    const runs = await this.options.prisma.taskRun.findMany({
+      where: {
+        id: {
+          in: runIds,
+        },
+      },
+      select: {
+        friendlyId: true,
+      },
+    });
+
+    return runs.map((run) => run.friendlyId);
+  }
+
+  async listRuns(options: ListRunsOptions) {
+    const { runIds, pagination } = await this.listRunIds(options);
 
     let runs = await this.options.prisma.taskRun.findMany({
       where: {
         id: {
-          in: runIdsToReturn,
+          in: runIds,
         },
       },
       orderBy: {
@@ -163,10 +208,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
 
     return {
       runs,
-      pagination: {
-        nextCursor,
-        previousCursor,
-      },
+      pagination,
     };
   }
 
diff --git a/apps/webapp/app/services/runsRepository/runsCursor.server.ts b/apps/webapp/app/services/runsRepository/runsCursor.server.ts
@@ -0,0 +1,47 @@
+/**
+ * Cursor encoding for keyset pagination over `(created_at, run_id)`.
+ *
+ * The list query orders by the composite key `(created_at, run_id)`, so a sound
+ * cursor must carry BOTH components — cutting on `run_id` alone re-includes and
+ * skips rows whenever `run_id` order diverges from `created_at` order.
+ *
+ * A cursor is an opaque URL-safe base64 token wrapping `{ c: createdAtMs, r:
+ * runId }`. Cursors are server-issued (the SDK just echoes
+ * `pagination.next`/`previous` back), so this format needs no client update.
+ *
+ * Legacy cursors were the bare internal run_id (a cuid). They are detected by
+ * decode failure: a cuid base64-decodes to non-JSON bytes, so it falls through
+ * to `{ kind: "legacy" }` and the old (knowingly unsound) `run_id`-only
+ * predicate. In-flight legacy cursors keep working and drain naturally.
+ */
+
+import { z } from "zod";
+
+export type DecodedRunsCursor =
+  | { kind: "composite"; createdAt: number; runId: string }
+  | { kind: "legacy"; runId: string };
+
+// `c` = created_at (ms since epoch), `r` = run_id. Short keys keep the token small.
+const CompositeCursor = z.object({
+  c: z.number().int(),
+  r: z.string().min(1),
+});
+
+export function encodeRunsCursor(createdAtMs: number, runId: string): string {
+  return Buffer.from(JSON.stringify({ c: createdAtMs, r: runId })).toString("base64url");
+}
+
+export function decodeRunsCursor(cursor: string): DecodedRunsCursor {
+  try {
+    const parsed = CompositeCursor.safeParse(
+      JSON.parse(Buffer.from(cursor, "base64url").toString("utf8"))
+    );
+    if (parsed.success) {
+      return { kind: "composite", createdAt: parsed.data.c, runId: parsed.data.r };
+    }
+  } catch {
+    // JSON.parse threw — not a composite cursor.
+  }
+
+  return { kind: "legacy", runId: cursor };
+}
diff --git a/apps/webapp/app/services/runsRepository/runsRepository.server.ts b/apps/webapp/app/services/runsRepository/runsRepository.server.ts
@@ -127,9 +127,25 @@ export type TagList = {
   tags: string[];
 };
 
+export type CursorPagination = {
+  nextCursor: string | null;
+  previousCursor: string | null;
+};
+
+export type RunIdsPage = {
+  runIds: string[];
+  pagination: CursorPagination;
+};
+
 export interface IRunsRepository {
   name: string;
-  listRunIds(options: ListRunsOptions): Promise<string[]>;
+  /**
+   * A keyset-paginated page of run ids plus the cursors to navigate
+   * forward/backward. The cursors are opaque composite `(created_at, run_id)`
+   * tokens, so pagination can't duplicate or skip runs. This is the single
+   * cursor-aware list primitive — `listRuns` and bulk actions build on it.
+   */
+  listRunIds(options: ListRunsOptions): Promise<RunIdsPage>;
   /** Returns friendly IDs (e.g., run_xxx) instead of internal UUIDs. Used for ClickHouse task_events queries. */
   listFriendlyRunIds(options: ListRunsOptions): Promise<string[]>;
   listRuns(options: ListRunsOptions): Promise<{
@@ -154,7 +170,7 @@ export class RunsRepository implements IRunsRepository {
     return "runsRepository";
   }
 
-  async listRunIds(options: ListRunsOptions): Promise<string[]> {
+  async listRunIds(options: ListRunsOptions): Promise<RunIdsPage> {
     return startActiveSpan(
       "runsRepository.listRunIds",
       async () => this.clickHouseRunsRepository.listRunIds(options),
diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
@@ -159,8 +159,13 @@ export class BulkActionService extends BaseService {
       throw new Error(`Bulk action group has invalid query name: ${group.queryName}`);
     }
 
-    // 2. Get the runs to process in this batch
-    const runIds = await runsRepository.listRunIds({
+    // 2. Get the runs to process in this batch, plus the cursor for the next
+    // batch. The cursor is a composite (created_at, run_id) keyset cursor so the
+    // next batch can't re-include or skip runs.
+    const {
+      runIds: runIdsToProcess,
+      pagination: { nextCursor },
+    } = await runsRepository.listRunIds({
       ...filters,
       page: {
         size: env.BULK_ACTION_BATCH_SIZE,
@@ -172,8 +177,6 @@ export class BulkActionService extends BaseService {
     // 3. Process the runs
     let successCount = 0;
     let failureCount = 0;
-    // Slice because we fetch an extra for the cursor
-    const runIdsToProcess = runIds.slice(0, env.BULK_ACTION_BATCH_SIZE);
 
     switch (group.type) {
       case BulkActionType.CANCEL: {
@@ -274,7 +277,10 @@ export class BulkActionService extends BaseService {
       }
     }
 
-    const isFinished = runIdsToProcess.length === 0;
+    // A null nextCursor means there is no further page — this batch was the
+    // last (or there were no runs at all), so the action is complete. (An empty
+    // batch also yields a null cursor.)
+    const isFinished = nextCursor === null;
 
     logger.debug("Bulk action group processed batch", {
       bulkActionId,
@@ -292,7 +298,8 @@ export class BulkActionService extends BaseService {
     const updatedGroup = await this._prisma.bulkActionGroup.update({
       where: { id: bulkActionId },
       data: {
-        cursor: runIdsToProcess.at(runIdsToProcess.length - 1),
+        // Json column: leave unchanged when there's no next cursor (finished).
+        cursor: nextCursor ?? undefined,
         successCount: {
           increment: successCount,
         },
diff --git a/apps/webapp/test/runsRepository.part2.test.ts b/apps/webapp/test/runsRepository.part2.test.ts
diff --git a/apps/webapp/test/runsRepositoryCursor.test.ts b/apps/webapp/test/runsRepositoryCursor.test.ts
diff --git a/internal-packages/clickhouse/src/taskRuns.ts b/internal-packages/clickhouse/src/taskRuns.ts

Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ export async function loader({ request, params }: LoaderFunctionArgs) {`
`58`	`58`	`return { count: 0, since: newRunsSince };`
`59`	`59`	`}`
`60`	`60`
`61`		`- const newRunIds = await runsRepository.listRunIds({`
	`61`	`+ const { runIds: newRunIds } = await runsRepository.listRunIds({`
`62`	`62`	`organizationId: project.organizationId,`
`63`	`63`	`projectId: project.id,`
`64`	`64`	`environmentId: environment.id,`