diff --git a/.server-changes/runs-empty-state-clickhouse.md b/.server-changes/runs-empty-state-clickhouse.md new file mode 100644 index 00000000000..29a5d102b29 --- /dev/null +++ b/.server-changes/runs-empty-state-clickhouse.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Reduce primary database load on the runs page by serving its empty-state check from ClickHouse instead of Postgres. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index cabe0159a6d..36c2302a5c8 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -330,6 +330,13 @@ const EnvironmentSchema = z .string() .default(process.env.REDIS_TLS_DISABLED ?? "false"), TASK_META_CACHE_CURRENT_ENV_TTL_SECONDS: z.coerce.number().default(86400), + + // Runs-list empty-state check: how far back the ClickHouse "does this env have any run" + // probe looks. Bounds the prove-absence partition scan. 0 = unbounded ("any run ever"). + RUN_LIST_HAS_RUNS_LOOKBACK_DAYS: z.coerce.number().default(30), + // SWR TTLs for the empty-state has-runs cache (memory + Redis). + RUN_LIST_HAS_RUNS_CACHE_FRESH_MS: z.coerce.number().default(86_400_000), + RUN_LIST_HAS_RUNS_CACHE_STALE_MS: z.coerce.number().default(604_800_000), TASK_META_CACHE_BY_WORKER_TTL_SECONDS: z.coerce.number().default(2592000), REALTIME_STREAMS_REDIS_HOST: z diff --git a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts index b030f5a3f12..fe6499b8e10 100644 --- a/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts @@ -11,11 +11,47 @@ import { timeFilters } from "~/components/runs/v3/SharedFilters"; import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server"; import { getTaskIdentifiers } from "~/models/task.server"; import { RunsRepository } from "~/services/runsRepository/runsRepository.server"; +import { env } from "~/env.server"; +import { + createCache, + createLRUMemoryStore, + DefaultStatefulContext, + Namespace, +} from "@internal/cache"; +import { RedisCacheStore } from "~/services/unkey/redisCacheStore.server"; +import { singleton } from "~/utils/singleton"; import { regionForDisplay } from "~/runEngine/concerns/workerQueueSplit.server"; import { machinePresetFromRun } from "~/v3/machinePresets.server"; import { ServiceValidationError } from "~/v3/services/baseService.server"; import { isCancellableRunStatus, isFinalRunStatus, isPendingRunStatus } from "~/v3/taskStatus"; +// Positive-only cache: only envs known to have runs are stored (empty envs are re-checked), +// so "has runs" is monotonic and the TTL can be very long. Tiered memory + Redis. +const runsExistCache = singleton("runsExistCache", () => { + const ctx = new DefaultStatefulContext(); + const memory = createLRUMemoryStore(5000, "runs-has-runs-cache"); + const redis = new RedisCacheStore({ + name: "runs-has-runs", + connection: { + keyPrefix: "tr:cache:runs-has-runs", + port: env.CACHE_REDIS_PORT, + host: env.CACHE_REDIS_HOST, + username: env.CACHE_REDIS_USERNAME, + password: env.CACHE_REDIS_PASSWORD, + tlsDisabled: env.CACHE_REDIS_TLS_DISABLED === "true", + clusterMode: env.CACHE_REDIS_CLUSTER_MODE_ENABLED === "1", + }, + }); + + return createCache({ + hasRuns: new Namespace(ctx, { + stores: [memory, redis], + fresh: env.RUN_LIST_HAS_RUNS_CACHE_FRESH_MS, + stale: env.RUN_LIST_HAS_RUNS_CACHE_STALE_MS, + }), + }); +}); + export type RunListOptions = { userId?: string; projectId: string; @@ -42,6 +78,8 @@ export type RunListOptions = { direction?: Direction; cursor?: string; pageSize?: number; + // Run the empty-state "has any run ever" probe. Only the runs list consumes it. + includeHasAnyRuns?: boolean; }; const DEFAULT_PAGE_SIZE = 25; @@ -65,37 +103,31 @@ export class NextRunListPresenter { } ) {} - // Existence probe for the empty-state. run-ops (TaskRun) read. - // Split off / single-DB: one plain findFirst on `replica` (passthrough). - // Split on: true if a row exists in the NEW run-ops DB OR the LEGACY run-ops - // read replica only (never the legacy writer). New first to avoid touching - // legacy whenever the new DB already answers. - async #anyRunExistsInEnv(environmentId: string): Promise { - const splitEnabled = this.readThroughDeps?.splitEnabled ?? false; + // Empty-state existence probe, served from ClickHouse (same connection as the runs + // list) so it no longer scans TaskRun in Postgres. SWR-cached to spare ClickHouse; + // RUN_LIST_HAS_RUNS_LOOKBACK_DAYS bounds the prove-absence partition scan. + async #anyRunExistsInEnv( + runsRepository: RunsRepository, + organizationId: string, + projectId: string, + environmentId: string + ): Promise { + const lookbackDays = env.RUN_LIST_HAS_RUNS_LOOKBACK_DAYS; + const createdAtLowerBoundMs = + lookbackDays > 0 ? Date.now() - lookbackDays * 24 * 60 * 60 * 1000 : undefined; - if (!splitEnabled) { - const firstRun = await this.replica.taskRun.findFirst({ - where: { runtimeEnvironmentId: environmentId }, - select: { id: true }, + const result = await runsExistCache.hasRuns.swr(environmentId, async () => { + const exists = await runsRepository.runExistsInEnvironment({ + organizationId, + projectId, + environmentId, + createdAtLowerBoundMs, }); - return firstRun !== null; - } - - const newClient = this.readThroughDeps?.newClient ?? this.replica; - const newRun = await newClient.taskRun.findFirst({ - where: { runtimeEnvironmentId: environmentId }, - select: { id: true }, + // undefined (not false) so swr does NOT cache the empty result — re-check until a run exists. + return exists ? true : undefined; }); - if (newRun) { - return true; - } - const legacyReplica = this.readThroughDeps?.legacyReplica ?? this.replica; - const legacyRun = await legacyReplica.taskRun.findFirst({ - where: { runtimeEnvironmentId: environmentId }, - select: { id: true }, - }); - return legacyRun !== null; + return result.val ?? false; } public async call( @@ -125,6 +157,7 @@ export class NextRunListPresenter { direction = "forward", cursor, pageSize = DEFAULT_PAGE_SIZE, + includeHasAnyRuns = false, }: RunListOptions ) { //get the time values from the raw values (including a default period) @@ -252,8 +285,13 @@ export class NextRunListPresenter { let hasAnyRuns = runs.length > 0; - if (!hasAnyRuns) { - hasAnyRuns = await this.#anyRunExistsInEnv(environmentId); + if (!hasAnyRuns && includeHasAnyRuns) { + hasAnyRuns = await this.#anyRunExistsInEnv( + runsRepository, + organizationId, + projectId, + environmentId + ); } return { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx index f00a4548167..fbcbbfeac63 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx @@ -111,6 +111,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { userId, projectId: project.id, ...filters, + includeHasAnyRuns: true, }); // Only persist rootOnly when no tasks are filtered. While a task filter is active, diff --git a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts index f57e37275c3..2aa9d5f3a68 100644 --- a/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/clickhouseRunsRepository.server.ts @@ -35,6 +35,38 @@ export class ClickHouseRunsRepository implements IRunsRepository { return "clickhouse"; } + async runExistsInEnvironment(options: { + organizationId: string; + projectId: string; + environmentId: string; + createdAtLowerBoundMs?: number; + }): Promise { + const queryBuilder = this.options.clickhouse.taskRuns.existsQueryBuilder(); + + queryBuilder + .where("organization_id = {organizationId: String}", { + organizationId: options.organizationId, + }) + .where("project_id = {projectId: String}", { projectId: options.projectId }) + .where("environment_id = {environmentId: String}", { environmentId: options.environmentId }); + + if (typeof options.createdAtLowerBoundMs === "number") { + queryBuilder.where("created_at >= fromUnixTimestamp64Milli({createdAtLowerBound: Int64})", { + createdAtLowerBound: options.createdAtLowerBoundMs, + }); + } + + queryBuilder.limit(1); + + const [queryError, result] = await queryBuilder.execute(); + + if (queryError) { + throw queryError; + } + + return (result?.length ?? 0) > 0; + } + /** * Runs the keyset-paginated query and returns `{ runId, createdAt }` rows * (one extra beyond `page.size` to signal "has more"). The ordering is always diff --git a/apps/webapp/app/services/runsRepository/runsRepository.server.ts b/apps/webapp/app/services/runsRepository/runsRepository.server.ts index fd095150ac0..a349c5bf534 100644 --- a/apps/webapp/app/services/runsRepository/runsRepository.server.ts +++ b/apps/webapp/app/services/runsRepository/runsRepository.server.ts @@ -176,6 +176,12 @@ export interface IRunsRepository { }>; countRuns(options: RunListInputOptions): Promise; listTags(options: TagListOptions): Promise; + runExistsInEnvironment(options: { + organizationId: string; + projectId: string; + environmentId: string; + createdAtLowerBoundMs?: number; + }): Promise; } export class RunsRepository implements IRunsRepository { @@ -189,6 +195,26 @@ export class RunsRepository implements IRunsRepository { return "runsRepository"; } + async runExistsInEnvironment(options: { + organizationId: string; + projectId: string; + environmentId: string; + createdAtLowerBoundMs?: number; + }): Promise { + return startActiveSpan( + "runsRepository.runExistsInEnvironment", + async () => this.clickHouseRunsRepository.runExistsInEnvironment(options), + { + attributes: { + "repository.name": "clickhouse", + organizationId: options.organizationId, + projectId: options.projectId, + environmentId: options.environmentId, + }, + } + ); + } + async listRunIds(options: ListRunsOptions): Promise { return startActiveSpan( "runsRepository.listRunIds", diff --git a/apps/webapp/test/nextRunListPresenter.readthrough.test.ts b/apps/webapp/test/nextRunListPresenter.readthrough.test.ts index 41daf9834de..189ab51c6e3 100644 --- a/apps/webapp/test/nextRunListPresenter.readthrough.test.ts +++ b/apps/webapp/test/nextRunListPresenter.readthrough.test.ts @@ -144,220 +144,12 @@ async function createRun( }); } -/** - * Wraps a real prisma handle in a Proxy whose `taskRun.findFirst` throws if invoked. Used to - * prove the empty-state probe never touches the legacy replica when the new DB already answers. - * All other access forwards to the real client (so FK parents etc. still resolve). - */ -function throwingFindFirst(prisma: PrismaClient, label: string): PrismaClient { - return new Proxy(prisma, { - get(target, prop) { - if (prop === "taskRun") { - return new Proxy((target as any).taskRun, { - get(trTarget, trProp) { - if (trProp === "findFirst") { - return async () => { - throw new Error(`${label}.taskRun.findFirst must not be invoked`); - }; - } - return (trTarget as any)[trProp]; - }, - }); - } - return (target as any)[prop]; - }, - }) as unknown as PrismaClient; -} - -const callOptions = (ctx: SeedContext, overrides?: { to?: number }) => ({ +const callOptions = (ctx: SeedContext) => ({ projectId: ctx.projectId, pageSize: 10, - ...overrides, }); -// `to` one hour in the past. The CH page filters `created_at <= to`, so a just-created run is -// deterministically excluded regardless of replication timing — the empty-state tests otherwise -// raced on the run not having replicated yet (held locally, failed on CI). The PG existence probe -// has no time filter, so it still finds the row and `hasAnyRuns` stays true. -const emptyPageWindow = (): { to: number } => ({ to: Date.now() - 60 * 60 * 1000 }); - -describe("NextRunListPresenter dual-DB empty-state probe + routed hydrate (legacy + new Postgres)", () => { - // no-false-empty. Runs ONLY on legacy, none on new. Empty CH page -> listRuns returns []. - // splitEnabled true. The probe misses NEW, falls through to the legacy replica and finds the - // row, so the dashboard must NOT show "no runs". - replicationContainerTest( - "no-false-empty: runs only on the legacy replica still report hasAnyRuns true", - async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => { - const { clickhouse } = await setupClickhouseReplication({ - prisma, - databaseUrl: postgresContainer.getConnectionUri(), - clickhouseUrl: clickhouseContainer.getConnectionUrl(), - redisOptions, - }); - - const { url: newUrl } = await createPostgresContainer(network, { - imageTag: "docker.io/postgres:17", - }); - const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } }); - legacyReplicaHolder.client = prisma; - - try { - const ctx = await seedParents(prisma, "nofalseempty"); - await mirrorParents(prismaNew, ctx, "nofalseempty"); - - // Run lives ONLY on the legacy DB. We seed it to legacy and never wait for CH replication, - // so within the page window the CH id-set page is empty and listRuns returns []. The - // empty-state probe (NEW miss -> legacy hit) is what proves hasAnyRuns stays true. - await createRun(prisma, ctx, { friendlyId: "run_legacyOnly" }); - - const presenter = new NextRunListPresenter(prisma, clickhouse, { - newClient: prismaNew, - legacyReplica: prisma, - splitEnabled: true, - }); - - const result = await presenter.call( - ctx.organizationId, - ctx.environmentId, - callOptions(ctx, emptyPageWindow()) - ); - - // CH id-set is empty within the page window, but the legacy probe finds the row. - expect(result.runs).toHaveLength(0); - expect(result.hasAnyRuns).toBe(true); - } finally { - await prismaNew.$disconnect(); - } - } - ); - - // new-DB short-circuit. A run on NEW, legacy replica wrapped so its taskRun.findFirst - // throws. Empty CH page. The probe answers from NEW and must NEVER fall through to legacy. The - // post-migration straggler is the same shape: present on NEW, absent from LEGACY, legacy never - // invoked. - replicationContainerTest( - "new-DB short-circuit: hasAnyRuns answered from the new DB without touching the legacy replica", - async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => { - const { clickhouse } = await setupClickhouseReplication({ - prisma, - databaseUrl: postgresContainer.getConnectionUri(), - clickhouseUrl: clickhouseContainer.getConnectionUrl(), - redisOptions, - }); - - const { url: newUrl } = await createPostgresContainer(network, { - imageTag: "docker.io/postgres:17", - }); - const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } }); - legacyReplicaHolder.client = prisma; - - try { - const ctx = await seedParents(prisma, "newshortcircuit"); - await mirrorParents(prismaNew, ctx, "newshortcircuit"); - - // The (migrated/straggler) run lives ONLY on NEW. - await createRun(prismaNew, ctx, { friendlyId: "run_newOnly" }); - - const legacySpy = throwingFindFirst(prisma, "legacyReplica"); - - const presenter = new NextRunListPresenter(prisma, clickhouse, { - newClient: prismaNew, - legacyReplica: legacySpy, - splitEnabled: true, - }); - - // If the legacy spy were invoked, this would throw — the test passing IS the proof. - const result = await presenter.call( - ctx.organizationId, - ctx.environmentId, - callOptions(ctx) - ); - - expect(result.runs).toHaveLength(0); - expect(result.hasAnyRuns).toBe(true); - } finally { - await prismaNew.$disconnect(); - } - } - ); - - // genuinely empty. Nothing on either DB. Empty CH page. splitEnabled true. Both - // probes run and return null -> the true empty state is preserved. - replicationContainerTest( - "genuinely empty: both DBs empty reports hasAnyRuns false", - async ({ clickhouseContainer, redisOptions, postgresContainer, prisma, network }) => { - const { clickhouse } = await setupClickhouseReplication({ - prisma, - databaseUrl: postgresContainer.getConnectionUri(), - clickhouseUrl: clickhouseContainer.getConnectionUrl(), - redisOptions, - }); - - const { url: newUrl } = await createPostgresContainer(network, { - imageTag: "docker.io/postgres:17", - }); - const prismaNew = new PrismaClient({ datasources: { db: { url: newUrl } } }); - legacyReplicaHolder.client = prisma; - - try { - const ctx = await seedParents(prisma, "trulyempty"); - await mirrorParents(prismaNew, ctx, "trulyempty"); - - const presenter = new NextRunListPresenter(prisma, clickhouse, { - newClient: prismaNew, - legacyReplica: prisma, - splitEnabled: true, - }); - - const result = await presenter.call( - ctx.organizationId, - ctx.environmentId, - callOptions(ctx) - ); - - expect(result.runs).toHaveLength(0); - expect(result.hasAnyRuns).toBe(false); - } finally { - await prismaNew.$disconnect(); - } - } - ); - - // passthrough single-DB (two-arg ctor). One `prisma`, seed a run, empty CH page. - // splitEnabled defaults false -> exactly one plain findFirst against the single handle; the - // split branch (new/legacy) is structurally never entered (no second handle is injected). - // Also covers "served from the replica only" — the ctor exposes no legacy-writer field, so a - // no-primary-read guarantee is structural. - replicationContainerTest( - "passthrough single-DB: two-arg ctor finds the run via the single handle", - async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => { - const { clickhouse } = await setupClickhouseReplication({ - prisma, - databaseUrl: postgresContainer.getConnectionUri(), - clickhouseUrl: clickhouseContainer.getConnectionUrl(), - redisOptions, - }); - - legacyReplicaHolder.client = prisma; - - const ctx = await seedParents(prisma, "passthrough"); - await createRun(prisma, ctx, { friendlyId: "run_passthrough" }); - - // Two-arg ctor: no readThroughDeps -> RunsRepository.readThrough is undefined - // (passthrough) and the probe is one plain `this.replica.taskRun.findFirst`. - const presenter = new NextRunListPresenter(prisma, clickhouse); - - const result = await presenter.call( - ctx.organizationId, - ctx.environmentId, - callOptions(ctx, emptyPageWindow()) - ); - - expect(result.runs).toHaveLength(0); - expect(result.hasAnyRuns).toBe(true); - } - ); - +describe("NextRunListPresenter routed hydrate (legacy + new Postgres)", () => { // list hydrate flows through the routed store: split, non-empty CH id-set whose rows are // split across NEW + the legacy replica. result.runs must be the union, id-desc ordered. This // proves the deps are threaded so the routed store is actually used. diff --git a/internal-packages/clickhouse/src/index.ts b/internal-packages/clickhouse/src/index.ts index 0b252a98f67..bc8b6e662c8 100644 --- a/internal-packages/clickhouse/src/index.ts +++ b/internal-packages/clickhouse/src/index.ts @@ -15,6 +15,7 @@ import { getTaskRunsCountQueryBuilder, getTaskRunTagsQueryBuilder, getPendingVersionIdsQueryBuilder, + getTaskRunExistsQueryBuilder, } from "./taskRuns.js"; import { getSpanDetailsQueryBuilder, @@ -228,6 +229,7 @@ export class ClickHouse { insertPayloadsCompactArrays: insertRawTaskRunPayloadsCompactArrays(this.writer), queryBuilder: getTaskRunsQueryBuilder(this.reader), countQueryBuilder: getTaskRunsCountQueryBuilder(this.reader), + existsQueryBuilder: getTaskRunExistsQueryBuilder(this.reader, { max_execution_time: 10 }), tagQueryBuilder: getTaskRunTagsQueryBuilder(this.reader), pendingVersionIdsQueryBuilder: getPendingVersionIdsQueryBuilder(this.reader), getTaskActivity: getTaskActivityQueryBuilder(this.reader), diff --git a/internal-packages/clickhouse/src/taskRuns.ts b/internal-packages/clickhouse/src/taskRuns.ts index 517afb70c5c..a0c5f8c4f9b 100644 --- a/internal-packages/clickhouse/src/taskRuns.ts +++ b/internal-packages/clickhouse/src/taskRuns.ts @@ -461,6 +461,23 @@ export function getTaskRunsCountQueryBuilder(ch: ClickhouseReader, settings?: Cl }); } +export const TaskRunExistsQueryResult = z.object({ + run_exists: z.number().int(), +}); + +export type TaskRunExistsQueryResult = z.infer; + +// Empty-state existence probe. No FINAL (a stale/dup row still answers "a run exists"). +// Callers must filter organization_id + project_id + environment_id (the sort-key prefix). +export function getTaskRunExistsQueryBuilder(ch: ClickhouseReader, settings?: ClickHouseSettings) { + return ch.queryBuilder({ + name: "getTaskRunExists", + baseQuery: "SELECT 1 AS run_exists FROM trigger_dev.task_runs_v2", + schema: TaskRunExistsQueryResult, + settings, + }); +} + export const TaskRunTagsQueryResult = z.object({ tag: z.string(), });