diff --git a/.server-changes/runs-list-read-isolation.md b/.server-changes/runs-list-read-isolation.md new file mode 100644 index 00000000000..5c411635735 --- /dev/null +++ b/.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. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index c9179306124..6bf7dafa618 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -2233,6 +2233,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 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/services/clickhouse/clickhouseFactory.server.ts b/apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts index 5a2b3b86eee..a29664a4502 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,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, diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index 2e911e5e958..f81f6cac705 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -411,6 +411,23 @@ 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 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, options: FilterRunsOptions @@ -426,28 +443,26 @@ function applyRunFiltersToQueryBuilder( environmentId: options.environmentId, }); - if (options.tasks && options.tasks.length > 0) { - queryBuilder.where("task_identifier IN {tasks: Array(String)}", { tasks: options.tasks }); + if (options.statuses && options.statuses.length > 0) { + queryBuilder.where("status IN {statuses: Array(String)}", { statuses: options.statuses }); } - if (options.versions && options.versions.length > 0) { - queryBuilder.where("task_version IN {versions: Array(String)}", { - versions: options.versions, + if (options.regions && options.regions.length > 0) { + queryBuilder.where("if(region != '', region, worker_queue) IN {regions: Array(String)}", { + regions: options.regions, }); } - 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.machines && options.machines.length > 0) { + queryBuilder.where("machine_preset IN {machines: Array(String)}", { + machines: options.machines, + }); } - if (options.scheduleId) { - queryBuilder.where("schedule_id = {scheduleId: String}", { scheduleId: options.scheduleId }); + if (options.errorId) { + queryBuilder.where("error_fingerprint = {errorFingerprint: String}", { + errorFingerprint: ErrorId.toId(options.errorId), + }); } // Period is a number of milliseconds duration @@ -467,51 +482,55 @@ 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 }); - } - - 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, - }); - } - - if (options.errorId) { - queryBuilder.where("error_fingerprint = {errorFingerprint: String}", { - errorFingerprint: ErrorId.toId(options.errorId), - }); + queryBuilder.prewhere("queue IN {queues: Array(String)}", { queues: options.queues }); } if (options.taskKinds && options.taskKinds.length > 0) { @@ -520,11 +539,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/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/runsListQueryShape.test.ts b/apps/webapp/test/runsListQueryShape.test.ts new file mode 100644 index 00000000000..b2e2f13003a --- /dev/null +++ b/apps/webapp/test/runsListQueryShape.test.ts @@ -0,0 +1,116 @@ +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()); + } + ); + + 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([]); + } + ); +}); 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 "); }