refactor(webapp,clickhouse): route runs-list filters through PREWHERE instead of a created_at clamp

Replaces the created_at window clamp with PREWHERE routing on the runs-list
query. Immutable and additive-only filters (tags, task_identifier, and the rest)
move into PREWHERE so ClickHouse filters, and uses the tags skip index, before
FINAL reconciles versions and before materialising the wide columns. This bounds
the memory a filtered runs-list query uses without dropping any rows, unlike the
date clamp which hid older runs from the list and the runs.list API.

status stays in WHERE (post-FINAL): it is the one lifecycle-mutable filter, so
PREWHERE-ing it would keep a stale version and drop the winning one.

Author: ericallam

.server-changes/runs-list-read-isolation.md

@@ -3,4 +3,4 @@ area: webapp
 type: improvement
 ---
 
-The runs list and the runs.list API are more resilient: a single expensive query can no longer slow the runs list down for everyone. The list now loads from a bounded recent time window, which keeps it fast at scale.
+Improved the performance and reliability of the runs list and the runs.list API, especially for large projects and filtered views.

apps/webapp/app/env.server.ts

@@ -2240,15 +2240,6 @@ const EnvironmentSchema = z
     RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE_FOR_USER: z.coerce.number().int().optional(),
     RUNS_LIST_CLICKHOUSE_MAX_CONCURRENT_QUERIES_FOR_USER: z.coerce.number().int().optional(),
     RUNS_LIST_CLICKHOUSE_READONLY: z.enum(["0", "1", "2"]).default("2"),
-    /**
-     * Hard cap on how far back the runs list / runs.list API `created_at` lower bound may reach,
-     * in milliseconds. The display list adds `created_at >= now - this` so an unbounded filter
-     * can't scan all partitions. `0` disables the cap. Does not apply to count queries.
-     */
-    RUNS_LIST_MAX_CREATED_AT_AGE_MS: z.coerce
-      .number()
-      .int()
-      .default(30 * 24 * 60 * 60 * 1000),
     /**
      * Dedicated ClickHouse service for queue metrics: the ingestion consumer's inserts and every
      * queue-metrics read (dashboards, queue pages, run inspector, health report) go through it, so

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

@@ -256,7 +256,6 @@ export class NextRunListPresenter {
     const runsRepository = new RunsRepository({
       clickhouse: this.clickhouse,
       prisma: this.replica as PrismaClient,
-      maxCreatedAtAgeMs: env.RUNS_LIST_MAX_CREATED_AT_AGE_MS,
       readThrough: this.readThroughDeps
         ? {
             newClient: this.readThroughDeps.newClient ?? this.replica,

apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts

@@ -127,8 +127,7 @@ export class ClickHouseRunsRepository implements IRunsRepository {
         options,
         this.options.prisma,
         this.options.runStore ?? runStore
-      ),
-      this.options.maxCreatedAtAgeMs
+      )
     );
 
     const forward = options.page.direction === "forward" || !options.page.direction;
@@ -336,12 +335,6 @@ export class ClickHouseRunsRepository implements IRunsRepository {
     };
   }
 
-  /**
-   * Deliberately NOT passed `maxCreatedAtAgeMs`: the only callers are billing limit checks and
-   * bulk actions, which must count runs of any age (a queued/delayed run older than the window
-   * still counts). Clamping here would undercount. Runaway counts are bounded instead by the
-   * read pool's server-side `max_execution_time`, not by a date cap.
-   */
   async countRuns(options: RunListInputOptions) {
     const queryBuilder = this.options.clickhouse.taskRuns.countQueryBuilder();
     applyRunFiltersToQueryBuilder(
@@ -419,14 +412,18 @@ export class ClickHouseRunsRepository implements IRunsRepository {
 }
 
 /**
- * Builds the shared WHERE clauses for the runs list. `maxCreatedAtAgeMs` (when > 0) floors the
- * `created_at` lower bound to `now - maxCreatedAtAgeMs`; it is ANDed with any period/from filter,
- * so the tighter bound wins, and it keeps an unbounded filter from scanning every partition.
+ * Builds the shared filter clauses for the runs list against `task_runs_v2 FINAL`.
+ *
+ * Immutable / additive-only columns go in PREWHERE so ClickHouse filters (and, for `tags`, uses
+ * the skip index) before FINAL reconciles versions and before materialising the wide columns,
+ * which is what bounds memory on these scans. `status` is the one lifecycle-mutable filter, so it
+ * stays in WHERE (post-FINAL): PREWHERE-ing it would keep a stale version and drop the winner. The
+ * `(organization_id, project_id, environment_id)` primary-key prefix and the `created_at` range
+ * stay in WHERE so they keep driving primary-key and partition pruning.
  */
 function applyRunFiltersToQueryBuilder<T>(
   queryBuilder: ClickhouseQueryBuilder<T>,
-  options: FilterRunsOptions,
-  maxCreatedAtAgeMs?: number
+  options: FilterRunsOptions
 ) {
   queryBuilder
     .where("organization_id = {organizationId: String}", {
@@ -439,36 +436,10 @@ function applyRunFiltersToQueryBuilder<T>(
       environmentId: options.environmentId,
     });
 
-  if (typeof maxCreatedAtAgeMs === "number" && maxCreatedAtAgeMs > 0) {
-    queryBuilder.where("created_at >= fromUnixTimestamp64Milli({createdAtFloor: Int64})", {
-      createdAtFloor: Date.now() - maxCreatedAtAgeMs,
-    });
-  }
-
-  if (options.tasks && options.tasks.length > 0) {
-    queryBuilder.where("task_identifier IN {tasks: Array(String)}", { tasks: options.tasks });
-  }
-
-  if (options.versions && options.versions.length > 0) {
-    queryBuilder.where("task_version IN {versions: Array(String)}", {
-      versions: options.versions,
-    });
-  }
-
   if (options.statuses && options.statuses.length > 0) {
     queryBuilder.where("status IN {statuses: Array(String)}", { statuses: options.statuses });
   }
 
-  if (options.tags && options.tags.length > 0) {
-    // Both hasAny and hasAll are served by the tags bloom_filter skip index.
-    const tagsFn = options.tagsMatch === "all" ? "hasAll" : "hasAny";
-    queryBuilder.where(`${tagsFn}(tags, {tags: Array(String)})`, { tags: options.tags });
-  }
-
-  if (options.scheduleId) {
-    queryBuilder.where("schedule_id = {scheduleId: String}", { scheduleId: options.scheduleId });
-  }
-
   // Period is a number of milliseconds duration
   if (options.period) {
     queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", {
@@ -486,49 +457,71 @@ function applyRunFiltersToQueryBuilder<T>(
     queryBuilder.where("created_at <= fromUnixTimestamp64Milli({to: Int64})", { to: options.to });
   }
 
+  if (options.tasks && options.tasks.length > 0) {
+    queryBuilder.prewhere("task_identifier IN {tasks: Array(String)}", { tasks: options.tasks });
+  }
+
+  if (options.versions && options.versions.length > 0) {
+    queryBuilder.prewhere("task_version IN {versions: Array(String)}", {
+      versions: options.versions,
+    });
+  }
+
+  if (options.tags && options.tags.length > 0) {
+    // Both hasAny and hasAll are served by the tags bloom_filter skip index.
+    const tagsFn = options.tagsMatch === "all" ? "hasAll" : "hasAny";
+    queryBuilder.prewhere(`${tagsFn}(tags, {tags: Array(String)})`, { tags: options.tags });
+  }
+
+  if (options.scheduleId) {
+    queryBuilder.prewhere("schedule_id = {scheduleId: String}", {
+      scheduleId: options.scheduleId,
+    });
+  }
+
   if (typeof options.isTest === "boolean") {
-    queryBuilder.where("is_test = {isTest: Boolean}", { isTest: options.isTest });
+    queryBuilder.prewhere("is_test = {isTest: Boolean}", { isTest: options.isTest });
   }
 
   if (options.rootOnly) {
-    queryBuilder.where("root_run_id = ''");
+    queryBuilder.prewhere("root_run_id = ''");
   }
 
   if (options.batchId) {
-    queryBuilder.where("batch_id = {batchId: String}", { batchId: options.batchId });
+    queryBuilder.prewhere("batch_id = {batchId: String}", { batchId: options.batchId });
   }
 
   if (options.bulkId) {
-    queryBuilder.where("hasAny(bulk_action_group_ids, {bulkActionGroupIds: Array(String)})", {
+    queryBuilder.prewhere("hasAny(bulk_action_group_ids, {bulkActionGroupIds: Array(String)})", {
       bulkActionGroupIds: [options.bulkId],
     });
   }
 
   if (options.runId && options.runId.length > 0) {
     // it's important that in the query it's "runIds", otherwise it clashes with the cursor which is called "runId"
-    queryBuilder.where("friendly_id IN {runIds: Array(String)}", {
+    queryBuilder.prewhere("friendly_id IN {runIds: Array(String)}", {
       runIds: options.runId.map((runId) => RunId.toFriendlyId(runId)),
     });
   }
 
   if (options.queues && options.queues.length > 0) {
-    queryBuilder.where("queue IN {queues: Array(String)}", { queues: options.queues });
+    queryBuilder.prewhere("queue IN {queues: Array(String)}", { queues: options.queues });
   }
 
   if (options.regions && options.regions.length > 0) {
-    queryBuilder.where("if(region != '', region, worker_queue) IN {regions: Array(String)}", {
+    queryBuilder.prewhere("if(region != '', region, worker_queue) IN {regions: Array(String)}", {
       regions: options.regions,
     });
   }
 
   if (options.machines && options.machines.length > 0) {
-    queryBuilder.where("machine_preset IN {machines: Array(String)}", {
+    queryBuilder.prewhere("machine_preset IN {machines: Array(String)}", {
       machines: options.machines,
     });
   }
 
   if (options.errorId) {
-    queryBuilder.where("error_fingerprint = {errorFingerprint: String}", {
+    queryBuilder.prewhere("error_fingerprint = {errorFingerprint: String}", {
       errorFingerprint: ErrorId.toId(options.errorId),
     });
   }
@@ -539,11 +532,11 @@ function applyRunFiltersToQueryBuilder<T>(
     const effectiveKinds = includesStandard ? [...options.taskKinds, ""] : options.taskKinds;
 
     if (effectiveKinds.length === 1) {
-      queryBuilder.where("task_kind = {taskKind: String}", {
+      queryBuilder.prewhere("task_kind = {taskKind: String}", {
         taskKind: effectiveKinds[0]!,
       });
     } else {
-      queryBuilder.where("task_kind IN {taskKinds: Array(String)}", {
+      queryBuilder.prewhere("task_kind IN {taskKinds: Array(String)}", {
         taskKinds: effectiveKinds,
       });
     }

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

@@ -33,14 +33,6 @@ export type RunsRepositoryOptions = {
     // Resolved boot constant; when false the split branch is never entered.
     splitEnabled?: boolean;
   };
-
-  /**
-   * Hard cap on how far back the run-listing `created_at` lower bound may reach, in ms. When set
-   * and > 0, the list queries add `created_at >= now - maxCreatedAtAgeMs` so an unbounded filter
-   * can't scan every partition. Omitted / 0 => no cap. Applies to `listRuns`/`listRunIds` only,
-   * never to `countRuns` (billing and bulk counts must count runs of any age).
-   */
-  maxCreatedAtAgeMs?: number;
 };
 
 const RunStatus = z.enum(Object.values(TaskRunStatus) as [TaskRunStatus, ...TaskRunStatus[]]);

apps/webapp/test/runsListCreatedAtClamp.test.ts

@@ -0,0 +1,77 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { containerTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
+import {
+  createRun,
+  insertTaskRunV2Rows,
+  seedParents,
+} from "./helpers/apiRunListPresenterTestHelpers";
+
+vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
+
+vi.setConfig({ testTimeout: 90_000 });
+
+const DAY_MS = 24 * 60 * 60 * 1000;
+
+describe("runs list query shape (PREWHERE routing under FINAL)", () => {
+  containerTest(
+    "keeps status post-FINAL and returns old pending runs (no date clamp)",
+    async ({ clickhouseContainer, prisma }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "query-shape-test",
+      });
+
+      const ctx = await seedParents(prisma, "shape");
+
+      const completed = await createRun(prisma, ctx, { friendlyId: "run_completed" });
+      const pendingRecent = await createRun(prisma, ctx, { friendlyId: "run_pending_recent" });
+      const pendingOld = await createRun(prisma, ctx, { friendlyId: "run_pending_old" });
+
+      const base = {
+        taskIdentifier: "webhook.deliver",
+        runTags: ["booking:T"],
+        createdAt: new Date(Date.now() - 1 * DAY_MS),
+      };
+
+      await insertTaskRunV2Rows(clickhouse, [
+        { ...completed, ...base, status: "PENDING", updatedAt: new Date(Date.now() - 2 * DAY_MS) },
+        {
+          ...completed,
+          ...base,
+          status: "COMPLETED",
+          updatedAt: new Date(Date.now() - 1 * DAY_MS),
+        },
+        {
+          ...pendingRecent,
+          ...base,
+          status: "PENDING",
+          updatedAt: new Date(Date.now() - 1 * DAY_MS),
+        },
+        {
+          ...pendingOld,
+          ...base,
+          status: "PENDING",
+          createdAt: new Date(Date.now() - 60 * DAY_MS),
+          updatedAt: new Date(Date.now() - 60 * DAY_MS),
+        },
+      ]);
+
+      const repository = new RunsRepository({ prisma, clickhouse });
+
+      const { runIds } = await repository.listRunIds({
+        page: { size: 10 },
+        period: "365d",
+        organizationId: ctx.organizationId,
+        projectId: ctx.projectId,
+        environmentId: ctx.environmentId,
+        tasks: ["webhook.deliver"],
+        tags: ["booking:T"],
+        statuses: ["PENDING", "DELAYED"],
+      });
+
+      expect(runIds.sort()).toEqual([pendingOld.id, pendingRecent.id].sort());
+    }
+  );
+});