Skip to content
Open
6 changes: 6 additions & 0 deletions .server-changes/runs-list-read-isolation.md
Original file line number Diff line number Diff line change
@@ -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.
9 changes: 9 additions & 0 deletions apps/webapp/app/env.server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ export class CreateBulkActionPresenter extends BasePresenter {

const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
organizationId,
"standard"
"runsList"
);
const runsRepository = new RunsRepository({
clickhouse,
Expand Down
60 changes: 58 additions & 2 deletions apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(),
});
}

Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<T>(
queryBuilder: ClickhouseQueryBuilder<T>,
options: FilterRunsOptions
Expand All @@ -426,28 +443,26 @@ 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.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
Expand All @@ -467,51 +482,55 @@ 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],
});
Comment thread
ericallam marked this conversation as resolved.
}

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) {
Expand All @@ -520,11 +539,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,
});
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down Expand Up @@ -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 },
Expand Down
4 changes: 2 additions & 2 deletions apps/webapp/app/v3/services/bulk/BulkActionV2.server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -275,7 +275,7 @@ export class BulkActionService extends BaseService {

const clickhouse = await clickhouseFactory.getClickhouseForOrganization(
group.project.organizationId,
"standard"
"runsList"
);
const runsRepository = new RunsRepository({
clickhouse,
Expand Down
66 changes: 66 additions & 0 deletions apps/webapp/test/runsListClickhouseSettings.test.ts
Original file line number Diff line number Diff line change
@@ -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/);
}
);
});
Loading