feat(webapp): isolate the runs list ClickHouse read pool (#4763)

## Summary

Improves the performance and reliability of the runs list and the
`runs.list` API, especially for large projects and filtered views.

## What changed

- **Filtered runs-list queries use `PREWHERE`.** Immutable and
additive-only filters (tags, task identifier, version, queue, region,
machine, and the rest) are applied in `PREWHERE` on the `task_runs_v2
FINAL` scan, so ClickHouse filters, and uses the tags skip index, before
it reconciles versions and materialises the wide columns. Same results,
far less memory per query. `status` stays in `WHERE`: it changes across
a run's versions, so filtering it before `FINAL` could return stale
rows.
- **The runs-list ClickHouse pool gets per-query guardrails**, all
env-configurable: a `max_execution_time` paired with the client request
timeout, a per-query `max_memory_usage`, a `max_threads` cap, and
`readonly`. Each bounds a single query to itself, so a heavy query can't
affect other queries, and they are safe as pool-level settings only
because this pool is read-only.
- **Billing and bulk count reads move to the read pool**, off the write
pool.

Defaults are conservative for self-hosters; production values are set
via env.

Author: ericallam

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

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+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

@@ -2246,6 +2246,15 @@ const EnvironmentSchema = z
       .enum(["log", "error", "warn", "info", "debug"])
       .default("info"),
     RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
+    RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS: z.coerce.number().int().positive().default(40_000),
+    RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().positive().default(35),
+    RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().positive().default(4),
+    RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(1_073_741_824),
+    RUNS_LIST_CLICKHOUSE_READONLY: z.enum(["0", "1", "2"]).default("2"),
     /**
      * 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/CreateBulkActionPresenter.server.ts

@@ -26,7 +26,7 @@ export class CreateBulkActionPresenter extends BasePresenter {
 
     const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
       organizationId,
-      "standard"
+      "runsList"
     );
     const runsRepository = new RunsRepository({
       clickhouse,

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -1,4 +1,4 @@
-import { ClickHouse } from "@internal/clickhouse";
+import { ClickHouse, type ClickHouseSettings } from "@internal/clickhouse";
 import { createHash } from "crypto";
 import { ClickhouseEventRepository } from "~/v3/eventRepository/clickhouseEventRepository.server";
 import { env } from "~/env.server";
@@ -292,6 +292,45 @@ function initializeRealtimeClickhouseClient(): ClickHouse {
   });
 }
 
+/**
+ * Server-side query protection for the runs-list read pool. Every setting here is PER-QUERY, so a
+ * pathological query only ever kills itself: a slow one hits `max_execution_time`, a memory-hungry
+ * one hits `max_memory_usage`, a thread-hungry one hits `max_threads`. Per-USER limits
+ * (`max_*_for_user`) are deliberately NOT used: everything connects as `default`, so a per-user cap
+ * would reject whichever query arrives once the shared budget is hit, punishing innocent tenants
+ * for a noisy one. The node itself is protected by the server-level `max_server_memory_usage`.
+ * Safe as client-level settings ONLY because this pool is read-only; on a mixed read+write pool a
+ * client-level `max_execution_time` would also kill slow inserts. `readonly=2` enforces read-only
+ * while still allowing these settings to apply (`readonly=1` rejects them).
+ */
+/**
+ * Client request timeout for the runs-list pool, forced above the server-side `max_execution_time`
+ * so the server cap is what stops a slow query and the client stays connected to receive that
+ * error. If the client timed out first, it would abort while ClickHouse kept executing, which is
+ * the abandoned-query behaviour this pool is trying to prevent.
+ */
+function getRunsListRequestTimeoutMs() {
+  return Math.max(
+    env.RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS,
+    (env.RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME + 5) * 1000
+  );
+}
+
+function getRunsListClickhouseSettings(): ClickHouseSettings {
+  const settings: ClickHouseSettings = {
+    max_execution_time: env.RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME,
+    timeout_before_checking_execution_speed: 0,
+    max_threads: env.RUNS_LIST_CLICKHOUSE_MAX_THREADS,
+    max_memory_usage: env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE.toString(),
+  };
+
+  if (env.RUNS_LIST_CLICKHOUSE_READONLY !== "0") {
+    settings.readonly = env.RUNS_LIST_CLICKHOUSE_READONLY;
+  }
+
+  return settings;
+}
+
 /** Runs list reads — dashboard + API (`RUNS_LIST_CLICKHOUSE_URL`);
  *  falls back to the default client if unset. */
 const defaultRunsListClickhouseClient = singleton(
@@ -319,6 +358,8 @@ function initializeRunsListClickhouseClient(): ClickHouse {
       request: env.RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST === "1",
     },
     maxOpenConnections: env.RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+    requestTimeoutMs: getRunsListRequestTimeoutMs(),
+    clickhouseSettings: getRunsListClickhouseSettings(),
   });
 }
 
@@ -550,10 +591,25 @@ function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHou
         },
         maxOpenConnections: env.REALTIME_BACKEND_NATIVE_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
       });
+    case "runsList":
+      return new ClickHouse({
+        url: parsed.toString(),
+        name,
+        keepAlive: {
+          enabled: env.RUNS_LIST_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+          idleSocketTtl: env.RUNS_LIST_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+        },
+        logLevel: env.RUNS_LIST_CLICKHOUSE_LOG_LEVEL,
+        compression: {
+          request: env.RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+        },
+        maxOpenConnections: env.RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+        requestTimeoutMs: getRunsListRequestTimeoutMs(),
+        clickhouseSettings: getRunsListClickhouseSettings(),
+      });
     case "standard":
     case "query":
     case "admin":
-    case "runsList":
       return new ClickHouse({
         url: parsed.toString(),
         name,

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

@@ -411,6 +411,22 @@ export class ClickHouseRunsRepository implements IRunsRepository {
   }
 }
 
+/**
+ * Builds the shared filter clauses for the runs list against `task_runs_v2 FINAL`.
+ *
+ * A filter may go in PREWHERE only if its truth value can never flip true->false across a run's
+ * versions, because PREWHERE is evaluated before FINAL reconciles versions and would otherwise keep
+ * a stale matching version and drop the winning one. That holds for columns that are only ever set
+ * once and never change: trigger-time identity columns (task_identifier, task_version, schedule_id,
+ * is_test, root_run_id, batch_id, friendly_id, queue, task_kind), append-only arrays under
+ * `hasAny`/`hasAll` (tags, bulk_action_group_ids), `region` (set once at dequeue, `''` -> value,
+ * never changes), and `error_fingerprint` (empty until a terminal error status, then fixed). Those
+ * go in PREWHERE to filter (and, for tags, use the skip index) before FINAL and before materialising
+ * the wide columns, which is what bounds memory on these scans. Columns that change as a run runs
+ * stay in WHERE (post-FINAL): `status` (lifecycle) and `machine_preset` (escalates on OOM retry).
+ * The `(organization_id, project_id, environment_id)` primary-key prefix and the `created_at` range
+ * also stay in WHERE so they keep driving primary-key and partition pruning.
+ */
 function applyRunFiltersToQueryBuilder<T>(
   queryBuilder: ClickhouseQueryBuilder<T>,
   options: FilterRunsOptions
@@ -426,28 +442,14 @@ function applyRunFiltersToQueryBuilder<T>(
       environmentId: options.environmentId,
     });
 
-  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 });
+  if (options.machines && options.machines.length > 0) {
+    queryBuilder.where("machine_preset IN {machines: Array(String)}", {
+      machines: options.machines,
+    });
   }
 
   // Period is a number of milliseconds duration
@@ -467,49 +469,63 @@ 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)}", {
-      regions: options.regions,
-    });
-  }
-
-  if (options.machines && options.machines.length > 0) {
-    queryBuilder.where("machine_preset IN {machines: Array(String)}", {
-      machines: options.machines,
-    });
+    queryBuilder.prewhere("region IN {regions: Array(String)}", { regions: options.regions });
   }
 
   if (options.errorId) {
-    queryBuilder.where("error_fingerprint = {errorFingerprint: String}", {
+    queryBuilder.prewhere("error_fingerprint = {errorFingerprint: String}", {
       errorFingerprint: ErrorId.toId(options.errorId),
     });
   }
@@ -520,11 +536,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/v3/services/billingLimit/billingLimitQueuedRuns.server.ts

@@ -39,7 +39,7 @@ export async function getBillableEnvironmentsForBillingLimit(
 export async function createBillingLimitRunsRepository(organizationId: string) {
   const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
     organizationId,
-    "standard"
+    "runsList"
   );
 
   return new RunsRepository({
@@ -95,7 +95,7 @@ export async function countBillableQueuedRunsForOrganization(
 ): Promise<number> {
   const client =
     clickhouse ??
-    (await clickhouseFactory.getClickhouseForOrganization(organizationId, "standard"));
+    (await clickhouseFactory.getClickhouseForOrganization(organizationId, "runsList"));
 
   const queryBuilder = client.taskRuns.countQueryBuilder({
     settings: { max_execution_time: BILLING_LIMIT_QUEUED_COUNT_MAX_EXECUTION_S },

apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts

@@ -115,7 +115,7 @@ export class BulkActionService extends BaseService {
     // Count the runs that will be affected by the bulk action
     const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
       organizationId,
-      "standard"
+      "runsList"
     );
     const runsRepository = new RunsRepository({
       clickhouse,
@@ -275,7 +275,7 @@ export class BulkActionService extends BaseService {
 
     const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
       group.project.organizationId,
-      "standard"
+      "runsList"
     );
     const runsRepository = new RunsRepository({
       clickhouse,

apps/webapp/test/runsListClickhouseSettings.test.ts

@@ -0,0 +1,66 @@
+import { ClickHouse } from "@internal/clickhouse";
+import { clickhouseTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import { z } from "zod";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+describe("runs-list ClickHouse protection settings", () => {
+  clickhouseTest(
+    "server-side max_execution_time kills a slow read, and readonly=2 does not block the caps",
+    async ({ clickhouseContainer }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-list-settings-test",
+        requestTimeoutMs: 30_000,
+        clickhouseSettings: {
+          max_execution_time: 1,
+          timeout_before_checking_execution_speed: 0,
+          max_threads: 2,
+          readonly: "2",
+        },
+      });
+
+      const slow = clickhouse.reader.query({
+        name: "slow-read",
+        query: "SELECT sum(number) AS total FROM numbers(1000000000000)",
+        schema: z.object({ total: z.number() }),
+      });
+      const [slowError] = await slow({});
+
+      expect(slowError).not.toBeNull();
+      expect(slowError?.message.toLowerCase()).toMatch(/timeout|exceeded/);
+
+      const fast = clickhouse.reader.query({
+        name: "fast-read",
+        query: "SELECT 1 AS one",
+        schema: z.object({ one: z.number() }),
+      });
+      const [fastError, rows] = await fast({});
+
+      expect(fastError).toBeNull();
+      expect(rows).toEqual([{ one: 1 }]);
+    }
+  );
+
+  clickhouseTest(
+    "readonly=2 rejects writes while permitting reads",
+    async ({ clickhouseContainer }) => {
+      const clickhouse = new ClickHouse({
+        url: clickhouseContainer.getConnectionUrl(),
+        name: "runs-list-readonly-test",
+        clickhouseSettings: { readonly: "2" },
+      });
+
+      const write = clickhouse.reader.query({
+        name: "write-under-readonly",
+        query: "CREATE TABLE trigger_dev.runs_list_readonly_probe (id UInt8) ENGINE = Memory",
+        schema: z.object({}),
+      });
+      const [writeError] = await write({});
+
+      expect(writeError).not.toBeNull();
+      expect(writeError?.message.toLowerCase()).toMatch(/readonly|read-only|read only/);
+    }
+  );
+});