From 97bf4e90f9f2a874a98477b1aa9e67aad96e5e7d Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 17:06:20 +0100 Subject: [PATCH 1/7] feat(webapp): isolate the runs list ClickHouse read pool Give the runs-list ClickHouse pool server-side query protection (max_execution_time, thread and memory caps, a per-user concurrency breaker, readonly) so one tenant expensive query cannot saturate the shared read service, and cap the runs list created_at lower bound to a bounded window so an unbounded filter cannot scan every partition. Billing and bulk count reads move to the read pool, off the ingestion writer. Count queries are never date-capped so billing keeps counting runs of any age. --- .server-changes/runs-list-read-isolation.md | 6 ++ apps/webapp/app/env.server.ts | 16 +++++ .../v3/CreateBulkActionPresenter.server.ts | 2 +- .../v3/NextRunListPresenter.server.ts | 1 + .../clickhouse/clickhouseFactory.server.ts | 55 +++++++++++++++- .../clickhouseRunsRepository.server.ts | 23 ++++++- .../runsRepository/runsRepository.server.ts | 8 +++ .../billingLimitQueuedRuns.server.ts | 4 +- .../v3/services/bulk/BulkActionV2.server.ts | 4 +- .../test/runsListClickhouseSettings.test.ts | 66 +++++++++++++++++++ .../test/runsListCreatedAtClamp.test.ts | 61 +++++++++++++++++ 11 files changed, 237 insertions(+), 9 deletions(-) create mode 100644 .server-changes/runs-list-read-isolation.md create mode 100644 apps/webapp/test/runsListClickhouseSettings.test.ts create mode 100644 apps/webapp/test/runsListCreatedAtClamp.test.ts diff --git a/.server-changes/runs-list-read-isolation.md b/.server-changes/runs-list-read-isolation.md new file mode 100644 index 00000000000..07f2ff346d8 --- /dev/null +++ b/.server-changes/runs-list-read-isolation.md @@ -0,0 +1,6 @@ +--- +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. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index c9179306124..5f03d5a6368 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -2233,6 +2233,22 @@ 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().default(30_000), + RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().default(35), + RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().optional(), + RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().optional(), + 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 diff --git a/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts b/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts index a04368b4255..6ddd6250b58 100644 --- a/apps/webapp/app/presenters/v3/CreateBulkActionPresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts index 52836fad293..2ed3a2f5580 100644 --- a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts @@ -256,6 +256,7 @@ 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, diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index 5a2b3b86eee..9818bb363b5 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/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,40 @@ function initializeRealtimeClickhouseClient(): ClickHouse { }); } +/** + * Server-side query protection for the runs-list read pool. Safe as client-level settings ONLY + * because this pool is read-only (no inserts); a client-level `max_execution_time` on a mixed + * read+write pool would also kill slow inserts. `readonly=2` enforces read-only while still + * allowing these settings to apply (`readonly=1` rejects them). `max_concurrent_queries_for_user` + * is a per-ClickHouse-user (`default`) fail-fast circuit breaker, not per-tenant isolation. + */ +function getRunsListClickhouseSettings(): ClickHouseSettings { + const settings: ClickHouseSettings = { + max_execution_time: env.RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME, + timeout_before_checking_execution_speed: 0, + }; + + if (env.RUNS_LIST_CLICKHOUSE_READONLY !== "0") { + settings.readonly = env.RUNS_LIST_CLICKHOUSE_READONLY; + } + if (env.RUNS_LIST_CLICKHOUSE_MAX_THREADS !== undefined) { + settings.max_threads = env.RUNS_LIST_CLICKHOUSE_MAX_THREADS; + } + if (env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE !== undefined) { + settings.max_memory_usage = env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE.toString(); + } + if (env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE_FOR_USER !== undefined) { + settings.max_memory_usage_for_user = + env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE_FOR_USER.toString(); + } + if (env.RUNS_LIST_CLICKHOUSE_MAX_CONCURRENT_QUERIES_FOR_USER !== undefined) { + settings.max_concurrent_queries_for_user = + env.RUNS_LIST_CLICKHOUSE_MAX_CONCURRENT_QUERIES_FOR_USER; + } + + return settings; +} + /** Runs list reads — dashboard + API (`RUNS_LIST_CLICKHOUSE_URL`); * falls back to the default client if unset. */ const defaultRunsListClickhouseClient = singleton( @@ -319,6 +353,8 @@ function initializeRunsListClickhouseClient(): ClickHouse { request: env.RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST === "1", }, maxOpenConnections: env.RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS, + requestTimeoutMs: env.RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS, + clickhouseSettings: getRunsListClickhouseSettings(), }); } @@ -550,10 +586,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: env.RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS, + clickhouseSettings: getRunsListClickhouseSettings(), + }); case "standard": case "query": case "admin": - case "runsList": return new ClickHouse({ url: parsed.toString(), name, diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index 2e911e5e958..dcabe4eef43 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -127,7 +127,8 @@ 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; @@ -335,6 +336,12 @@ 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( @@ -411,9 +418,15 @@ 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. + */ function applyRunFiltersToQueryBuilder( queryBuilder: ClickhouseQueryBuilder, - options: FilterRunsOptions + options: FilterRunsOptions, + maxCreatedAtAgeMs?: number ) { queryBuilder .where("organization_id = {organizationId: String}", { @@ -426,6 +439,12 @@ function applyRunFiltersToQueryBuilder( 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 }); } diff --git a/apps/webapp/app/services/runsRepository/runsRepository.server.ts b/apps/webapp/app/services/runsRepository/runsRepository.server.ts index 0b1049125dd..b8d8f2c6d4e 100644 --- a/apps/webapp/app/services/runsRepository/runsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/runsRepository.server.ts @@ -33,6 +33,14 @@ 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[]]); diff --git a/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts b/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts index 066af4be101..6eff2f9336f 100644 --- a/apps/webapp/app/v3/services/billingLimit/billingLimitQueuedRuns.server.ts +++ b/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 { 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 }, diff --git a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts b/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts index d3cd77b143c..b19f6e4bc69 100644 --- a/apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts +++ b/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, diff --git a/apps/webapp/test/runsListClickhouseSettings.test.ts b/apps/webapp/test/runsListClickhouseSettings.test.ts new file mode 100644 index 00000000000..fc066b60369 --- /dev/null +++ b/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/); + } + ); +}); diff --git a/apps/webapp/test/runsListCreatedAtClamp.test.ts b/apps/webapp/test/runsListCreatedAtClamp.test.ts new file mode 100644 index 00000000000..c1eaf5c4b21 --- /dev/null +++ b/apps/webapp/test/runsListCreatedAtClamp.test.ts @@ -0,0 +1,61 @@ +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 created_at clamp", () => { + containerTest( + "listRuns is capped to the window; countRuns and the unclamped instance are not", + async ({ clickhouseContainer, prisma }) => { + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + name: "created-at-clamp-test", + }); + + const ctx = await seedParents(prisma, "clamp"); + + const recent = await createRun(prisma, ctx, { friendlyId: "run_recent", status: "PENDING" }); + const old = await createRun(prisma, ctx, { friendlyId: "run_old", status: "PENDING" }); + + await insertTaskRunV2Rows(clickhouse, [ + { ...recent, createdAt: new Date(Date.now() - 1 * DAY_MS) }, + { ...old, createdAt: new Date(Date.now() - 60 * DAY_MS) }, + ]); + + const listArgs = { + page: { size: 10 }, + period: "365d", + organizationId: ctx.organizationId, + projectId: ctx.projectId, + environmentId: ctx.environmentId, + }; + const countArgs = { + period: "365d", + organizationId: ctx.organizationId, + projectId: ctx.projectId, + environmentId: ctx.environmentId, + }; + + const clamped = new RunsRepository({ prisma, clickhouse, maxCreatedAtAgeMs: 30 * DAY_MS }); + const clampedList = await clamped.listRuns(listArgs); + expect(clampedList.runs.map((r) => r.friendlyId)).toEqual(["run_recent"]); + + const unclamped = new RunsRepository({ prisma, clickhouse }); + const unclampedList = await unclamped.listRuns(listArgs); + expect(unclampedList.runs.map((r) => r.friendlyId).sort()).toEqual(["run_old", "run_recent"]); + + expect(await clamped.countRuns(countArgs)).toBe(2); + } + ); +}); From d2bbe1bcdf849d122671fcb0ecf84b1896ed3bbb Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 18:00:05 +0100 Subject: [PATCH 2/7] 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. --- .server-changes/runs-list-read-isolation.md | 2 +- apps/webapp/app/env.server.ts | 9 -- .../v3/NextRunListPresenter.server.ts | 1 - .../clickhouseRunsRepository.server.ts | 93 +++++++++---------- .../runsRepository/runsRepository.server.ts | 8 -- .../test/runsListCreatedAtClamp.test.ts | 61 ------------ apps/webapp/test/runsListQueryShape.test.ts | 77 +++++++++++++++ .../clickhouse/src/client/queryBuilder.ts | 26 ++++++ 8 files changed, 147 insertions(+), 130 deletions(-) delete mode 100644 apps/webapp/test/runsListCreatedAtClamp.test.ts create mode 100644 apps/webapp/test/runsListQueryShape.test.ts diff --git a/.server-changes/runs-list-read-isolation.md b/.server-changes/runs-list-read-isolation.md index 07f2ff346d8..5c411635735 100644 --- a/.server-changes/runs-list-read-isolation.md +++ b/.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. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 5f03d5a6368..9ee69e9339d 100644 --- a/apps/webapp/app/env.server.ts +++ b/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 diff --git a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts index 2ed3a2f5580..52836fad293 100644 --- a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts +++ b/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, diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index dcabe4eef43..7d44ae01966 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/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( queryBuilder: ClickhouseQueryBuilder, - options: FilterRunsOptions, - maxCreatedAtAgeMs?: number + options: FilterRunsOptions ) { queryBuilder .where("organization_id = {organizationId: String}", { @@ -439,36 +436,10 @@ function applyRunFiltersToQueryBuilder( 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( 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( 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, }); } diff --git a/apps/webapp/app/services/runsRepository/runsRepository.server.ts b/apps/webapp/app/services/runsRepository/runsRepository.server.ts index b8d8f2c6d4e..0b1049125dd 100644 --- a/apps/webapp/app/services/runsRepository/runsRepository.server.ts +++ b/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[]]); diff --git a/apps/webapp/test/runsListCreatedAtClamp.test.ts b/apps/webapp/test/runsListCreatedAtClamp.test.ts deleted file mode 100644 index c1eaf5c4b21..00000000000 --- a/apps/webapp/test/runsListCreatedAtClamp.test.ts +++ /dev/null @@ -1,61 +0,0 @@ -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 created_at clamp", () => { - containerTest( - "listRuns is capped to the window; countRuns and the unclamped instance are not", - async ({ clickhouseContainer, prisma }) => { - const clickhouse = new ClickHouse({ - url: clickhouseContainer.getConnectionUrl(), - name: "created-at-clamp-test", - }); - - const ctx = await seedParents(prisma, "clamp"); - - const recent = await createRun(prisma, ctx, { friendlyId: "run_recent", status: "PENDING" }); - const old = await createRun(prisma, ctx, { friendlyId: "run_old", status: "PENDING" }); - - await insertTaskRunV2Rows(clickhouse, [ - { ...recent, createdAt: new Date(Date.now() - 1 * DAY_MS) }, - { ...old, createdAt: new Date(Date.now() - 60 * DAY_MS) }, - ]); - - const listArgs = { - page: { size: 10 }, - period: "365d", - organizationId: ctx.organizationId, - projectId: ctx.projectId, - environmentId: ctx.environmentId, - }; - const countArgs = { - period: "365d", - organizationId: ctx.organizationId, - projectId: ctx.projectId, - environmentId: ctx.environmentId, - }; - - const clamped = new RunsRepository({ prisma, clickhouse, maxCreatedAtAgeMs: 30 * DAY_MS }); - const clampedList = await clamped.listRuns(listArgs); - expect(clampedList.runs.map((r) => r.friendlyId)).toEqual(["run_recent"]); - - const unclamped = new RunsRepository({ prisma, clickhouse }); - const unclampedList = await unclamped.listRuns(listArgs); - expect(unclampedList.runs.map((r) => r.friendlyId).sort()).toEqual(["run_old", "run_recent"]); - - expect(await clamped.countRuns(countArgs)).toBe(2); - } - ); -}); diff --git a/apps/webapp/test/runsListQueryShape.test.ts b/apps/webapp/test/runsListQueryShape.test.ts new file mode 100644 index 00000000000..e4340685b23 --- /dev/null +++ b/apps/webapp/test/runsListQueryShape.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()); + } + ); +}); diff --git a/internal-packages/clickhouse/src/client/queryBuilder.ts b/internal-packages/clickhouse/src/client/queryBuilder.ts index bcdc68089c9..7b4f172e931 100644 --- a/internal-packages/clickhouse/src/client/queryBuilder.ts +++ b/internal-packages/clickhouse/src/client/queryBuilder.ts @@ -12,6 +12,7 @@ export type WhereCondition = { export class ClickhouseQueryBuilder { private name: string; private baseQuery: string; + private prewhereClauses: string[] = []; private whereClauses: string[] = []; private havingClauses: string[] = []; private params: QueryParams = {}; @@ -42,6 +43,28 @@ export class ClickhouseQueryBuilder { return this; } + /** + * Adds a PREWHERE clause. On a `... FINAL` base query, only use this for columns that are + * immutable or additive across a run's versions (e.g. task_identifier, tags): PREWHERE filters + * rows before FINAL reconciles versions, so a mutable column (e.g. status) would keep a stale + * version and drop the winning one. It filters before materialising the wide columns, which is + * what bounds memory on `task_runs_v2 FINAL` scans. + */ + prewhere(clause: string, params?: QueryParams): this { + this.prewhereClauses.push(clause); + if (params) { + Object.assign(this.params, params); + } + return this; + } + + prewhereIf(condition: any, clause: string, params?: QueryParams): this { + if (condition) { + this.prewhere(clause, params); + } + return this; + } + where(clause: string, params?: QueryParams): this { this.whereClauses.push(clause); if (params) { @@ -117,6 +140,9 @@ export class ClickhouseQueryBuilder { build(): { query: string; params: QueryParams } { let query = this.baseQuery; + if (this.prewhereClauses.length > 0) { + query += " PREWHERE " + this.prewhereClauses.join(" AND "); + } if (this.whereClauses.length > 0) { query += " WHERE " + this.whereClauses.join(" AND "); } From 7f8db900ae866b64ce7d12a423b3631e45676fe3 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 22:17:21 +0100 Subject: [PATCH 3/7] fix(webapp): use only per-query caps on the runs-list ClickHouse pool Drops max_memory_usage_for_user and max_concurrent_queries_for_user. Those are per-ClickHouse-user limits, and every connection is the default user, so hitting the shared budget rejects whichever query arrives next rather than the one responsible, which would fail queries for uninvolved tenants. The per-query caps (max_execution_time, max_memory_usage, max_threads) bound a bad query to itself, and the server-level max_server_memory_usage protects the node. --- apps/webapp/app/env.server.ts | 2 -- .../clickhouse/clickhouseFactory.server.ts | 22 ++++++++----------- 2 files changed, 9 insertions(+), 15 deletions(-) diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 9ee69e9339d..12dababb94c 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -2237,8 +2237,6 @@ const EnvironmentSchema = z RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().default(35), RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().optional(), RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().optional(), - 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"), /** * Dedicated ClickHouse service for queue metrics: the ingestion consumer's inserts and every diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index 9818bb363b5..a9d69817e72 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -293,11 +293,15 @@ function initializeRealtimeClickhouseClient(): ClickHouse { } /** - * Server-side query protection for the runs-list read pool. Safe as client-level settings ONLY - * because this pool is read-only (no inserts); a client-level `max_execution_time` on a mixed - * read+write pool would also kill slow inserts. `readonly=2` enforces read-only while still - * allowing these settings to apply (`readonly=1` rejects them). `max_concurrent_queries_for_user` - * is a per-ClickHouse-user (`default`) fail-fast circuit breaker, not per-tenant isolation. + * 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). */ function getRunsListClickhouseSettings(): ClickHouseSettings { const settings: ClickHouseSettings = { @@ -314,14 +318,6 @@ function getRunsListClickhouseSettings(): ClickHouseSettings { if (env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE !== undefined) { settings.max_memory_usage = env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE.toString(); } - if (env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE_FOR_USER !== undefined) { - settings.max_memory_usage_for_user = - env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE_FOR_USER.toString(); - } - if (env.RUNS_LIST_CLICKHOUSE_MAX_CONCURRENT_QUERIES_FOR_USER !== undefined) { - settings.max_concurrent_queries_for_user = - env.RUNS_LIST_CLICKHOUSE_MAX_CONCURRENT_QUERIES_FOR_USER; - } return settings; } From 1592856b116abf6caca28d241bd9b1b4a729ddaf Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 22:22:21 +0100 Subject: [PATCH 4/7] fix(webapp): default the runs-list per-query caps on max_threads and max_memory_usage were opt-in env vars, so out of the box, or if deploy config lagged, the pool ran with no thread or per-query memory cap, which is the thread oversubscription that hurt throughput under load. Give both a conservative default (4 threads, 1 GiB) so the guardrails hold without depending on a deploy-time config step, following the logs and query read pools. --- apps/webapp/app/env.server.ts | 4 ++-- .../app/services/clickhouse/clickhouseFactory.server.ts | 8 ++------ 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 12dababb94c..5d1d10ae5d4 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -2235,8 +2235,8 @@ const EnvironmentSchema = z RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"), RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS: z.coerce.number().int().default(30_000), RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().default(35), - RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().optional(), - RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().optional(), + RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().default(4), + RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().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 diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index a9d69817e72..d8411d54074 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -307,17 +307,13 @@ 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; } - if (env.RUNS_LIST_CLICKHOUSE_MAX_THREADS !== undefined) { - settings.max_threads = env.RUNS_LIST_CLICKHOUSE_MAX_THREADS; - } - if (env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE !== undefined) { - settings.max_memory_usage = env.RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE.toString(); - } return settings; } From 3d7ef3be9f932085d32e2fe7313c70e65fcd0ddc Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 23:09:37 +0100 Subject: [PATCH 5/7] fix(webapp): harden runs-list pool timeout and cap validation Client request timeout now sits above the server max_execution_time (default 40s vs 35s, and the factory forces it to at least exec + 5s), so the server-side cap is what stops a slow query and the client stays connected to receive the error, instead of aborting first and leaving the query running. The numeric caps reject zero and negative values, since ClickHouse treats 0 as unlimited for max_execution_time and max_memory_usage, which would silently disable them. --- apps/webapp/app/env.server.ts | 12 ++++++++---- .../clickhouse/clickhouseFactory.server.ts | 17 +++++++++++++++-- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 5d1d10ae5d4..6bf7dafa618 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -2233,10 +2233,14 @@ 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().default(30_000), - RUNS_LIST_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().default(35), - RUNS_LIST_CLICKHOUSE_MAX_THREADS: z.coerce.number().int().default(4), - RUNS_LIST_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().default(1_073_741_824), + 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 diff --git a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index d8411d54074..a29664a4502 100644 --- a/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts +++ b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts @@ -303,6 +303,19 @@ function initializeRealtimeClickhouseClient(): ClickHouse { * 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, @@ -345,7 +358,7 @@ function initializeRunsListClickhouseClient(): ClickHouse { request: env.RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST === "1", }, maxOpenConnections: env.RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS, - requestTimeoutMs: env.RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS, + requestTimeoutMs: getRunsListRequestTimeoutMs(), clickhouseSettings: getRunsListClickhouseSettings(), }); } @@ -591,7 +604,7 @@ function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHou request: env.RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST === "1", }, maxOpenConnections: env.RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS, - requestTimeoutMs: env.RUNS_LIST_CLICKHOUSE_REQUEST_TIMEOUT_MS, + requestTimeoutMs: getRunsListRequestTimeoutMs(), clickhouseSettings: getRunsListClickhouseSettings(), }); case "standard": From 6a835fa1cfccb0aaf0f443a2f404fc950f1ce7bc Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 23:24:48 +0100 Subject: [PATCH 6/7] fix(webapp): keep the runs-list region filter in WHERE, not PREWHERE The region filter uses if(region != "", region, worker_queue): a run is region="" at trigger (so the expression yields worker_queue) and gets a real region at dequeue, so the expression flips from one non-empty value to another across a run versions. Under PREWHERE that is evaluated before FINAL reconciles versions, so it could keep a stale pre-dequeue version and drop the winner, returning runs whose current region no longer matches (and listRunIds drives bulk actions). Moved it back to WHERE (post-FINAL) with a regression test. --- .../clickhouseRunsRepository.server.ts | 23 ++++++----- apps/webapp/test/runsListQueryShape.test.ts | 39 +++++++++++++++++++ 2 files changed, 52 insertions(+), 10 deletions(-) diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index 7d44ae01966..451e54e78c8 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -416,10 +416,13 @@ export class ClickHouseRunsRepository implements IRunsRepository { * * 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. + * which is what bounds memory on these scans. A filter stays in WHERE (post-FINAL) when its truth + * value can flip across a run's versions, since PREWHERE could then keep a stale version and drop + * the winner: `status` (lifecycle-mutable), and `regions` (its `if(region != '', region, + * worker_queue)` expression yields the worker_queue before dequeue and the region after, two + * different non-empty values). 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( queryBuilder: ClickhouseQueryBuilder, @@ -440,6 +443,12 @@ function applyRunFiltersToQueryBuilder( queryBuilder.where("status IN {statuses: Array(String)}", { statuses: options.statuses }); } + if (options.regions && options.regions.length > 0) { + queryBuilder.where("if(region != '', region, worker_queue) IN {regions: Array(String)}", { + regions: options.regions, + }); + } + // Period is a number of milliseconds duration if (options.period) { queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", { @@ -508,12 +517,6 @@ function applyRunFiltersToQueryBuilder( queryBuilder.prewhere("queue IN {queues: Array(String)}", { queues: options.queues }); } - if (options.regions && options.regions.length > 0) { - queryBuilder.prewhere("if(region != '', region, worker_queue) IN {regions: Array(String)}", { - regions: options.regions, - }); - } - if (options.machines && options.machines.length > 0) { queryBuilder.prewhere("machine_preset IN {machines: Array(String)}", { machines: options.machines, diff --git a/apps/webapp/test/runsListQueryShape.test.ts b/apps/webapp/test/runsListQueryShape.test.ts index e4340685b23..b2e2f13003a 100644 --- a/apps/webapp/test/runsListQueryShape.test.ts +++ b/apps/webapp/test/runsListQueryShape.test.ts @@ -74,4 +74,43 @@ describe("runs list query shape (PREWHERE routing under FINAL)", () => { expect(runIds.sort()).toEqual([pendingOld.id, pendingRecent.id].sort()); } ); + + containerTest( + "region filter uses the post-FINAL effective region, not a pre-dequeue worker_queue version", + async ({ clickhouseContainer, prisma }) => { + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + name: "query-shape-region-test", + }); + + const ctx = await seedParents(prisma, "region"); + const run = await createRun(prisma, ctx, { friendlyId: "run_region" }); + + const shared = { + taskIdentifier: "webhook.deliver", + workerQueue: "wq-legacy", + createdAt: new Date(Date.now() - 1 * DAY_MS), + }; + + await insertTaskRunV2Rows(clickhouse, [ + { ...run, ...shared, region: "", updatedAt: new Date(Date.now() - 2 * DAY_MS) }, + { ...run, ...shared, region: "us-east-1", updatedAt: new Date(Date.now() - 1 * DAY_MS) }, + ]); + + const repository = new RunsRepository({ prisma, clickhouse }); + const listArgs = { + page: { size: 10 } as const, + period: "365d", + organizationId: ctx.organizationId, + projectId: ctx.projectId, + environmentId: ctx.environmentId, + }; + + const byRegion = await repository.listRunIds({ ...listArgs, regions: ["us-east-1"] }); + expect(byRegion.runIds).toEqual([run.id]); + + const byWorkerQueue = await repository.listRunIds({ ...listArgs, regions: ["wq-legacy"] }); + expect(byWorkerQueue.runIds).toEqual([]); + } + ); }); From fcd9a8dc3c675e5b1a6807d7f18973f2f3a3ceff Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 24 Aug 2026 23:39:26 +0100 Subject: [PATCH 7/7] fix(webapp): move error_fingerprint and machine_preset filters to WHERE Both reflect execution/outcome and change across a run versions, so they are unsafe in PREWHERE (evaluated before FINAL): error_fingerprint is derived from status per snapshot and is cleared when a run recovers to a non-error status, and machine_preset can escalate to a larger machine on an out-of-memory retry. In either case an earlier version matches the filter while the winning version does not, so PREWHERE could keep the stale version and drop the winner. Only trigger-time identity columns and append-only arrays stay in PREWHERE. --- .../clickhouseRunsRepository.server.ts | 46 ++++++++++--------- 1 file changed, 25 insertions(+), 21 deletions(-) diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index 451e54e78c8..f81f6cac705 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -414,15 +414,19 @@ export class ClickHouseRunsRepository implements IRunsRepository { /** * 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. A filter stays in WHERE (post-FINAL) when its truth - * value can flip across a run's versions, since PREWHERE could then keep a stale version and drop - * the winner: `status` (lifecycle-mutable), and `regions` (its `if(region != '', region, - * worker_queue)` expression yields the worker_queue before dequeue and the region after, two - * different non-empty values). 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. + * 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 trigger-time identity columns + * that never change (task_identifier, task_version, schedule_id, is_test, root_run_id, batch_id, + * friendly_id, queue, task_kind) and for append-only arrays under `hasAny`/`hasAll` (tags, + * bulk_action_group_ids), so 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 reflect execution/outcome and change as a run runs stay in WHERE (post-FINAL): + * `status`, `machine_preset` (can escalate on OOM retry), `error_fingerprint` (set/cleared with + * status), and the `regions` expression (`if(region != '', region, worker_queue)` yields the + * worker_queue before dequeue and the region after). 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( queryBuilder: ClickhouseQueryBuilder, @@ -449,6 +453,18 @@ function applyRunFiltersToQueryBuilder( }); } + if (options.machines && options.machines.length > 0) { + queryBuilder.where("machine_preset IN {machines: Array(String)}", { + machines: options.machines, + }); + } + + if (options.errorId) { + queryBuilder.where("error_fingerprint = {errorFingerprint: String}", { + errorFingerprint: ErrorId.toId(options.errorId), + }); + } + // Period is a number of milliseconds duration if (options.period) { queryBuilder.where("created_at >= fromUnixTimestamp64Milli({period: Int64})", { @@ -517,18 +533,6 @@ function applyRunFiltersToQueryBuilder( queryBuilder.prewhere("queue IN {queues: Array(String)}", { queues: options.queues }); } - if (options.machines && options.machines.length > 0) { - queryBuilder.prewhere("machine_preset IN {machines: Array(String)}", { - machines: options.machines, - }); - } - - if (options.errorId) { - queryBuilder.prewhere("error_fingerprint = {errorFingerprint: String}", { - errorFingerprint: ErrorId.toId(options.errorId), - }); - } - if (options.taskKinds && options.taskKinds.length > 0) { const includesStandard = options.taskKinds.includes("STANDARD"); // Include empty string when filtering for STANDARD (default value for pre-existing runs)