From 8deea1f7ee2fd47a89de5608c61b59e56047557e Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:29:42 +0200 Subject: [PATCH 1/5] feat(query-engine): migrate spanHierarchy (55/61) Three defs: the recent-window probe, the unbounded fallback probe, and the pruned hierarchy read. All three carry cache: undefined on purpose. The handler keeps a single cachedDirect around the whole probe-then-read sequence, so the probe only fires on an outer cache miss. Caching them individually would run the probe on every request and cache a result nobody asked for -- and the probe exists precisely because trace_detail_spans is partitioned by toDate(Timestamp), where an unpruned hierarchy read is p95 ~8.8s against ~2.3s pruned. Adds a `withDeps` helper. cachedDirect takes an Effect with no requirements, while runQuery reads WarehouseQueryService and QueryEngineService from context; withDeps supplies the instances already bound at the top of the group so a registry query can run INSIDE a cache wrapper. Without it the only options were hoisting the probe out of the cache (losing the miss-only property) or threading services through every call site. Verified: apps/api typecheck; SQL baseline byte-identical; routes + warehouse suites 310 pass. --- apps/api/src/routes/v1/query-engine.http.ts | 90 +++++++------------ packages/query-engine/src/registry/queries.ts | 60 +++++++++++++ 2 files changed, 94 insertions(+), 56 deletions(-) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index c1098dd48..4dec71935 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -152,6 +152,16 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const queryEngine = yield* QueryEngineService const warehouse = yield* WarehouseQueryService + + // `cachedDirect` takes an Effect with no requirements, but `runQuery` reads + // its services from context. Supplying the ones already bound above lets a + // registry query run INSIDE a cache wrapper — needed by spanHierarchy, + // where the probe must only fire on an outer cache miss. + const withDeps = (effect: Effect.Effect) => + effect.pipe( + Effect.provideService(WarehouseQueryService, warehouse), + Effect.provideService(QueryEngineService, queryEngine), + ) const serviceOperationsRollupEnabled = yield* Config.boolean( "SERVICE_OPERATIONS_ROLLUP_ENABLED", ).pipe(Config.withDefault(false)) @@ -182,65 +192,33 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // shared link, AI link), resolve one via a cheap LIMIT-1 probe and // derive a ±1h window so the main query can prune. The probe itself // tries the recent window first (see PROBE_RECENT_WINDOW_MS). - Effect.gen(function* () { - let startTime = payload.startTime - let endTime = payload.endTime - if (startTime == null || endTime == null) { - const runProbe = (narrowByTime: boolean) => - mapExecError( - warehouse - .compiledQueryFirst( - tenant, - CH.compile( - CH.traceTimeProbeQuery({ - traceId: payload.traceId, - narrowByTime, - }), - narrowByTime - ? { - orgId: tenant.orgId, - startTime: formatWarehouseDateTime( - nowMs - PROBE_RECENT_WINDOW_MS, - ), - } - : { orgId: tenant.orgId }, - ), - { - profile: "discovery", - context: narrowByTime - ? "spanHierarchyProbeRecent" - : "spanHierarchyProbe", - }, - ) - .pipe(Effect.map(Option.getOrNull)), - "spanHierarchy probe failed", - ) - const probe = (yield* runProbe(true)) ?? (yield* runProbe(false)) - if (probe?.timestamp != null) { - const window = partitionWindowAround(probe.timestamp) - startTime = window.startTime - endTime = window.endTime + withDeps( + Effect.gen(function* () { + let startTime = payload.startTime + let endTime = payload.endTime + if (startTime == null || endTime == null) { + const probe = + (yield* runQueryFirst(Queries.spanHierarchyProbeRecent, tenant, { + traceId: payload.traceId, + startTime: formatWarehouseDateTime( + nowMs - PROBE_RECENT_WINDOW_MS, + ), + })) ?? + (yield* runQueryFirst(Queries.spanHierarchyProbe, tenant, payload)) + if (probe?.timestamp != null) { + const window = partitionWindowAround(probe.timestamp) + startTime = window.startTime + endTime = window.endTime + } } - } - const narrowByTime = startTime != null && endTime != null - const compiled = CH.compile( - CH.spanHierarchyQuery({ + return yield* runQuery(Queries.spanHierarchy, tenant, { traceId: payload.traceId, spanId: payload.spanId, - narrowByTime, - }), - narrowByTime - ? { orgId: tenant.orgId, startTime, endTime } - : { orgId: tenant.orgId }, - ) - return yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "spanHierarchy", - }), - "spanHierarchy query failed", - ) - }), + startTime, + endTime, + }) + }), + ), traceCacheTtlSeconds(payload.endTime, nowMs), ) const typedRows = rows.map((row) => ({ diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index 51aefe969..a49236480 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -749,3 +749,63 @@ export const listPodsCount = defineQuery({ { rowSchema: CH.ListPodsSummaryOutputSchema }, ), }) + +// --- spanHierarchy: probe, then the pruned hierarchy read ----------------- +// +// `trace_detail_spans` is partitioned by toDate(Timestamp); without a time +// predicate the hierarchy query seeks every daily partition (~30) — p95 ~8.8s +// vs ~2.3s when pruned to one. When the caller has no timestamp (direct URL, +// shared link, AI link) a cheap LIMIT-1 probe resolves one. +// +// All three carry `cache: undefined` ON PURPOSE. The handler wraps the whole +// probe-then-read sequence in a single `cachedDirect`, so the probe only fires +// on an outer cache miss. Caching them individually would run the probe on +// every request and cache a result nobody asked for. + +/** Probe restricted to the recent window — tries ~2 daily partitions first. */ +export const spanHierarchyProbeRecent = defineQuery({ + id: "spanHierarchyProbeRecent", + profile: "discovery", + cache: undefined, + compile: (payload: { readonly traceId: string; readonly startTime: string }, orgId: string) => + CH.compile(CH.traceTimeProbeQuery({ traceId: payload.traceId, narrowByTime: true }), { + orgId, + startTime: payload.startTime, + }), +}) + +/** Unbounded fallback probe: every partition, only when the recent one missed. */ +export const spanHierarchyProbe = defineQuery({ + id: "spanHierarchyProbe", + profile: "discovery", + cache: undefined, + compile: (payload: { readonly traceId: string }, orgId: string) => + CH.compile(CH.traceTimeProbeQuery({ traceId: payload.traceId, narrowByTime: false }), { + orgId, + }), +}) + +export const spanHierarchy = defineQuery({ + id: "spanHierarchy", + profile: "list", + cache: undefined, + compile: ( + payload: { + readonly traceId: string + readonly spanId?: string | undefined + readonly startTime?: string | undefined + readonly endTime?: string | undefined + }, + orgId: string, + ) => { + const narrowByTime = payload.startTime != null && payload.endTime != null + return CH.compile( + CH.spanHierarchyQuery({ + traceId: payload.traceId, + spanId: payload.spanId, + narrowByTime, + }), + narrowByTime ? { orgId, startTime: payload.startTime, endTime: payload.endTime } : { orgId }, + ) + }, +}) From 301a4ad669317b14200bbf2ef5c3d836c579db4c Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:32:03 +0200 Subject: [PATCH 2/5] refactor(query-engine): curry runner deps instead of laundering context withDeps was an anti-pattern. It re-provided WarehouseQueryService and QueryEngineService into an Effect that read them from context, using the very instances already bound as values three lines above -- laundering a requirement purely to satisfy cachedDirect's R = never. makeQueryRunners takes those services as values once per handler group and returns runQuery/runQueryFirst closed over them. Every call site is unchanged (`runQuery(def, tenant, payload)`), but the effects now carry R = never, so a registry query composes inside a cache wrapper with no ceremony. spanHierarchy's probe-inside-cachedDirect works for the right reason rather than by re-injection. Verified: apps/api typecheck; SQL baseline byte-identical; routes + warehouse suites 310 pass. --- apps/api/src/routes/query-helpers.ts | 48 ++++++ apps/api/src/routes/query-runner.ts | 168 ++++++++++---------- apps/api/src/routes/v1/query-engine.http.ts | 59 +++---- 3 files changed, 159 insertions(+), 116 deletions(-) diff --git a/apps/api/src/routes/query-helpers.ts b/apps/api/src/routes/query-helpers.ts index c5278bde1..409075a40 100644 --- a/apps/api/src/routes/query-helpers.ts +++ b/apps/api/src/routes/query-helpers.ts @@ -1,6 +1,7 @@ import * as Integrations from "@maple/query-engine-integrations" import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine" import type { + HostInfraTimeseriesRequest, NodeInfraTimeseriesRequest, PodInfraTimeseriesRequest, WorkloadInfraTimeseriesRequest, @@ -111,3 +112,50 @@ export const workloadMetricSpec = (metric: WorkloadInfraTimeseriesRequest["metri } } } + +/** + * Metric name, grouping key, unit and query-family flag for a host metric. + * + * Shared like the pod/node/workload specs: the registry needs `metricName` and + * `isNetwork` to build the query, the handler needs `unit` and + * `groupByAttributeKey` for its response. + */ +export const hostMetricSpec = (metric: HostInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu": + return { + metricName: "system.cpu.utilization", + groupByAttributeKey: "state", + unit: "percent" as const, + isNetwork: false, + } + case "memory": + return { + metricName: "system.memory.utilization", + groupByAttributeKey: "state", + unit: "percent" as const, + isNetwork: false, + } + case "filesystem": + return { + metricName: "system.filesystem.utilization", + groupByAttributeKey: "mountpoint", + unit: "percent" as const, + isNetwork: false, + } + case "load15": + return { + metricName: "system.cpu.load_average.15m", + groupByAttributeKey: undefined, + unit: "load" as const, + isNetwork: false, + } + case "network": + return { + metricName: "system.network.io", + groupByAttributeKey: "direction", + unit: "bytes_per_second" as const, + isNetwork: true, + } + } +} diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts index db7e788dc..a9c8170ae 100644 --- a/apps/api/src/routes/query-runner.ts +++ b/apps/api/src/routes/query-runner.ts @@ -2,8 +2,8 @@ import type { QueryDef } from "@maple/query-engine/registry" import type { QueryEngineDirectError } from "@maple/query-engine/runtime" import { Clock, Effect, Option } from "effect" import type { TenantContext } from "@/services/auth/AuthService" -import { QueryEngineService } from "@/services/warehouse/QueryEngineService" -import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" +import type { QueryEngineServiceShape } from "@/services/warehouse/QueryEngineService" +import type { WarehouseQueryServiceShape } from "@/services/warehouse/WarehouseQueryService" /** * Execute registry-declared warehouse queries. @@ -15,72 +15,81 @@ import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryServic * * Handlers keep their own row-to-response mapping. What they lose is the * plumbing, not the presentation. + * + * The services are taken as VALUES, not read from context, and the runners are + * built once per handler group. That is load-bearing rather than stylistic: + * `QueryEngineService.cachedDirect` accepts an `Effect` with no requirements, so + * a context-reading runner could never be composed inside a cache wrapper — + * which `spanHierarchy` needs, since its probe must only fire on a cache miss. + * Currying here keeps every call site written as `runQuery(def, tenant, payload)` + * while the effects it returns carry `R = never`. */ - -/** - * Settings may be static or payload-dependent. Resolved to a spread so an - * absent/undefined result omits the key entirely rather than passing an - * explicit `settings: undefined`, which would read as "clear the profile - * defaults" downstream. - */ -const resolveSettings = (def: QueryDef, payload: Payload) => { - const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings - return settings === undefined ? {} : { settings } +export interface QueryRunnerDeps { + readonly warehouse: WarehouseQueryServiceShape + readonly queryEngine: QueryEngineServiceShape } -/** - * Shared tail: annotate failures with the query id, then apply the declared - * cache policy (or don't). `cachedDirect` wraps the whole execution so a hit - * skips the warehouse entirely, and it takes the raw payload as key input — - * it snaps timestamps and sorts set-valued keys itself, which is why the - * payload goes in unnormalized. - */ -const withPolicy = ( - def: QueryDef, - tenant: TenantContext, - payload: Payload, - execute: Effect.Effect, -) => - Effect.gen(function* () { - const queryEngine = yield* QueryEngineService - // Only read the clock when a def actually needs it — a static policy must - // not pay for, or depend on, a Clock read. - const cache = - typeof def.cache === "function" - ? def.cache(payload, yield* Clock.currentTimeMillis) - : def.cache - const labelled = execute.pipe( - // Same annotation the old inline `mapExecError` produced, with the label - // derived from `def.id` rather than a hand-written string that could - // disagree with the span context beside it. - Effect.tapError(() => - Effect.annotateCurrentSpan({ - "maple.query_engine.failed_step": `${def.id} query failed`, - }), - ), - ) - if (cache === undefined) { - return yield* labelled - } - return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, cache) - }) +export const makeQueryRunners = ({ warehouse, queryEngine }: QueryRunnerDeps) => { + /** + * Settings may be static or payload-dependent. Resolved to a spread so an + * absent result omits the key entirely rather than passing an explicit + * `settings: undefined`, which would read as "clear the profile defaults". + */ + const resolveSettings = (def: QueryDef, payload: Payload) => { + const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings + return settings === undefined ? {} : { settings } + } -/** - * Run a `QueryDef` that returns many rows. - * - * Rows-vs-first-row is a call-site concern rather than a field on the def: the - * same compiled query legitimately supports both, and `compiledQueryFirst` - * takes the identical `CompiledQuery`. Putting it in the def would only let a - * caller disagree with it. - */ -export const runQuery = ( - def: QueryDef, - tenant: TenantContext, - payload: Payload, -) => - Effect.gen(function* () { - const warehouse = yield* WarehouseQueryService - return yield* withPolicy( + /** + * Annotate failures with the query id, then apply the declared cache policy + * (or don't). `cachedDirect` wraps the whole execution so a hit skips the + * warehouse entirely, and it takes the raw payload as key input — it snaps + * timestamps and sorts set-valued keys itself, which is why the payload goes + * in unnormalized. + */ + const withPolicy = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + execute: Effect.Effect, + ) => + Effect.gen(function* () { + // Only read the clock when a def actually needs it — a static policy + // must not pay for, or depend on, a Clock read. + const cache = + typeof def.cache === "function" + ? def.cache(payload, yield* Clock.currentTimeMillis) + : def.cache + const labelled = execute.pipe( + // Same annotation the old inline `mapExecError` produced, with the + // label derived from `def.id` rather than a hand-written string that + // could disagree with the span context beside it. + Effect.tapError(() => + Effect.annotateCurrentSpan({ + "maple.query_engine.failed_step": `${def.id} query failed`, + }), + ), + ) + if (cache === undefined) { + return yield* labelled + } + return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, cache) + }) + + /** + * Run a `QueryDef` that returns many rows. + * + * Rows-vs-first-row is a call-site concern rather than a field on the def: + * the same compiled query legitimately supports both, and + * `compiledQueryFirst` takes the identical `CompiledQuery`. Putting it in the + * def would only let a caller disagree with it. + */ + const runQuery = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + ) => + withPolicy( def, tenant, payload, @@ -90,22 +99,19 @@ export const runQuery = ( context: def.id, }), ) - }) -/** - * Run a `QueryDef` that returns at most one row, as `Row | null`. - * - * Null rather than `Option` because every current caller immediately does - * `Option.getOrNull` to build a nullable response field. - */ -export const runQueryFirst = ( - def: QueryDef, - tenant: TenantContext, - payload: Payload, -) => - Effect.gen(function* () { - const warehouse = yield* WarehouseQueryService - return yield* withPolicy( + /** + * Run a `QueryDef` that returns at most one row, as `Row | null`. + * + * Null rather than `Option` because every current caller immediately does + * `Option.getOrNull` to build a nullable response field. + */ + const runQueryFirst = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + ) => + withPolicy( def, tenant, payload, @@ -117,4 +123,6 @@ export const runQueryFirst = ( }) .pipe(Effect.map(Option.getOrNull)), ) - }) + + return { runQuery, runQueryFirst } as const +} diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 4dec71935..353909083 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -91,7 +91,7 @@ import { workloadMetricSpec, } from "@/routes/query-helpers" import { Queries } from "@/routes/queries" -import { runQuery, runQueryFirst } from "@/routes/query-runner" +import { makeQueryRunners } from "@/routes/query-runner" import type { ExecutionTenant, WarehouseSqlError } from "@maple/query-engine/execution" import { buildBreakdownQuerySpec, buildTimeseriesQuerySpec } from "@maple/query-engine/query-builder" import * as Integrations from "@maple/query-engine-integrations" @@ -152,16 +152,8 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const queryEngine = yield* QueryEngineService const warehouse = yield* WarehouseQueryService + const { runQuery, runQueryFirst } = makeQueryRunners({ warehouse, queryEngine }) - // `cachedDirect` takes an Effect with no requirements, but `runQuery` reads - // its services from context. Supplying the ones already bound above lets a - // registry query run INSIDE a cache wrapper — needed by spanHierarchy, - // where the probe must only fire on an outer cache miss. - const withDeps = (effect: Effect.Effect) => - effect.pipe( - Effect.provideService(WarehouseQueryService, warehouse), - Effect.provideService(QueryEngineService, queryEngine), - ) const serviceOperationsRollupEnabled = yield* Config.boolean( "SERVICE_OPERATIONS_ROLLUP_ENABLED", ).pipe(Config.withDefault(false)) @@ -192,33 +184,28 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // shared link, AI link), resolve one via a cheap LIMIT-1 probe and // derive a ±1h window so the main query can prune. The probe itself // tries the recent window first (see PROBE_RECENT_WINDOW_MS). - withDeps( - Effect.gen(function* () { - let startTime = payload.startTime - let endTime = payload.endTime - if (startTime == null || endTime == null) { - const probe = - (yield* runQueryFirst(Queries.spanHierarchyProbeRecent, tenant, { - traceId: payload.traceId, - startTime: formatWarehouseDateTime( - nowMs - PROBE_RECENT_WINDOW_MS, - ), - })) ?? - (yield* runQueryFirst(Queries.spanHierarchyProbe, tenant, payload)) - if (probe?.timestamp != null) { - const window = partitionWindowAround(probe.timestamp) - startTime = window.startTime - endTime = window.endTime - } + Effect.gen(function* () { + let startTime = payload.startTime + let endTime = payload.endTime + if (startTime == null || endTime == null) { + const probe = + (yield* runQueryFirst(Queries.spanHierarchyProbeRecent, tenant, { + traceId: payload.traceId, + startTime: formatWarehouseDateTime(nowMs - PROBE_RECENT_WINDOW_MS), + })) ?? (yield* runQueryFirst(Queries.spanHierarchyProbe, tenant, payload)) + if (probe?.timestamp != null) { + const window = partitionWindowAround(probe.timestamp) + startTime = window.startTime + endTime = window.endTime } - return yield* runQuery(Queries.spanHierarchy, tenant, { - traceId: payload.traceId, - spanId: payload.spanId, - startTime, - endTime, - }) - }), - ), + } + return yield* runQuery(Queries.spanHierarchy, tenant, { + traceId: payload.traceId, + spanId: payload.spanId, + startTime, + endTime, + }) + }), traceCacheTtlSeconds(payload.endTime, nowMs), ) const typedRows = rows.map((row) => ({ From d5987c4243a649f789005abe8cb5ea2630df886b Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:34:12 +0200 Subject: [PATCH 3/5] feat(query-engine): migrate hostInfraTimeseries (56/61) Network reads a counter family, everything else a gauge family, so they are two defs rather than one def with a branch -- the row shapes differ and the handler maps them differently. Both keep the id "hostInfraTimeseries", which is what their spans already report; renaming would break continuity of existing telemetry for no gain. hostMetricSpec joins the pod/node/workload specs in query-helpers: the def needs metricName and groupByAttributeKey, the handler needs unit and isNetwork. Verified: apps/api typecheck; SQL baseline byte-identical. Local test suites not re-run for this commit -- CI covers them. --- apps/api/src/routes/queries.ts | 47 ++++++++++++ apps/api/src/routes/v1/query-engine.http.ts | 85 ++------------------- 2 files changed, 53 insertions(+), 79 deletions(-) diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts index c24e311b1..a84de8193 100644 --- a/apps/api/src/routes/queries.ts +++ b/apps/api/src/routes/queries.ts @@ -3,6 +3,7 @@ import * as Integrations from "@maple/query-engine-integrations" import { defineQuery } from "@maple/query-engine/registry" import { Queries as Core } from "@maple/query-engine/registry" import type { + HostInfraTimeseriesRequest, CloudflareInfraZoneFacetsRequest, CloudflareInfraZoneDetailRequest, ServicePlanetScaleStatsRequest, @@ -24,6 +25,7 @@ import type { WorkloadInfraTimeseriesRequest, } from "@maple/domain/http" import { + hostMetricSpec, nodeMetricSpec, partitionWindowAround, podMetricSpec, @@ -469,6 +471,49 @@ const cloudflareInfraZoneFacets = defineQuery({ }), }) +// --- hostInfraTimeseries: two query families behind one endpoint ---------- +// +// Network reads a counter family, everything else a gauge family, so they are +// separate defs rather than one def with a branch — the row shapes differ and +// the handler maps them differently. Both keep the id "hostInfraTimeseries", +// which is what their spans already report; renaming would break continuity of +// existing telemetry for no gain. + +const hostInfraNetworkTimeseries = defineQuery({ + id: "hostInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: HostInfraTimeseriesRequest, orgId: string) => + CH.compile(CH.hostNetworkTimeseriesQuery({ hostName: payload.hostName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }), +}) + +const hostInfraGaugeTimeseries = defineQuery({ + id: "hostInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: HostInfraTimeseriesRequest, orgId: string) => { + const spec = hostMetricSpec(payload.metric) + return CH.compile( + CH.hostGaugeTimeseriesQuery({ + hostName: payload.hostName, + metricName: spec.metricName, + groupByAttributeKey: spec.groupByAttributeKey, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ) + }, +}) + export const Queries = { ...Core, @@ -651,4 +696,6 @@ export const Queries = { planetscaleServiceConnections, planetscaleServiceStorage, cloudflareInfraZoneFacets, + hostInfraNetworkTimeseries, + hostInfraGaugeTimeseries, } as const diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 353909083..6258bab74 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -84,6 +84,7 @@ import { } from "@maple/query-engine" import { LOGS_BODY_SEARCH_SETTINGS } from "@maple/query-engine/profiles" import { + hostMetricSpec, nodeMetricSpec, partitionWindowAround, podMetricSpec, @@ -1501,66 +1502,12 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const tenant = yield* CurrentTenant.Context const bucketSeconds = payload.bucketSeconds ?? 60 - const spec = (() => { - switch (payload.metric) { - case "cpu": - return { - metricName: "system.cpu.utilization", - groupByAttributeKey: "state", - unit: "percent" as const, - isNetwork: false, - } - case "memory": - return { - metricName: "system.memory.utilization", - groupByAttributeKey: "state", - unit: "percent" as const, - isNetwork: false, - } - case "filesystem": - return { - metricName: "system.filesystem.utilization", - groupByAttributeKey: "mountpoint", - unit: "percent" as const, - isNetwork: false, - } - case "load15": - return { - metricName: "system.cpu.load_average.15m", - groupByAttributeKey: undefined, - unit: "load" as const, - isNetwork: false, - } - case "network": - return { - metricName: "system.network.io", - groupByAttributeKey: "direction", - unit: "bytes_per_second" as const, - isNetwork: true, - } - } - })() + const spec = hostMetricSpec(payload.metric) if (spec.isNetwork) { - const compiled = CH.compile( - CH.hostNetworkTimeseriesQuery({ hostName: payload.hostName }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "hostInfraTimeseries", - }), - "hostInfraTimeseries (network) query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.hostInfraNetworkTimeseries, tenant, payload) return new HostInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.sumValue) || 0, @@ -1570,29 +1517,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", }) } - const compiled = CH.compile( - CH.hostGaugeTimeseriesQuery({ - hostName: payload.hostName, - metricName: spec.metricName, - groupByAttributeKey: spec.groupByAttributeKey, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "hostInfraTimeseries", - }), - "hostInfraTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.hostInfraGaugeTimeseries, tenant, payload) return new HostInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, From 9e95ccccc8dee50529f7c65d56ad360a1f7ec02d Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:40:25 +0200 Subject: [PATCH 4/5] feat(query-engine): migrate the last two handlers (58/58) cloudflareInfraZoneBreakdown and serviceOperations, the two that were left because their control flow is not query construction. ZoneBreakdown runs three queries in parallel and then a fourth whose grouping keys come from the first's ranked output. `topKeys` therefore rides in the PAYLOAD rather than being derived inside compile -- a def has no way to see a previous query's result. The caller still skips the fourth entirely when the key list is empty. Its coverage sub-query stays deliberately unfiltered: it answers "what did the poller collect here", which is how the UI distinguishes "not collected yet" from "no traffic". serviceOperations has a rollup form and a raw form of each of its two queries. The CHOICE stays in the handler because it is policy: a feature flag selects the rollup, and a typed isMissingServiceOperationsRollup error falls back to raw at runtime, flipping a flag the timeseries query then honors too. The defs own compile, profile and context; the handler owns the fallback. Rollup/raw pairs share an id, matching the context their spans already report -- the fallback is recorded separately as query.rollup.fallback. Also removes dead code the earlier passes left behind: an unused `params` object in each of the two handlers above, and a whole conditional CH.compile in planetscaleInfraTimeseries that its def had already superseded. query-engine.http.ts now contains ZERO CH.compile calls and is down from 3275 to 1774 lines. The three remaining handlers -- execute, executeQueryBuilder and executeRawSql -- are the QuerySpec and raw-SQL surfaces and should not become QueryDefs. Verified: both packages typecheck. Test suites not run locally by request; CI covers them, including the SQL baseline and the ClickHouse DESCRIBE sweep. --- apps/api/src/routes/queries.ts | 89 +++++++ apps/api/src/routes/v1/query-engine.http.ts | 226 ++++-------------- packages/query-engine/src/registry/queries.ts | 94 ++++++++ 3 files changed, 223 insertions(+), 186 deletions(-) diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts index a84de8193..080bcd70f 100644 --- a/apps/api/src/routes/queries.ts +++ b/apps/api/src/routes/queries.ts @@ -3,6 +3,7 @@ import * as Integrations from "@maple/query-engine-integrations" import { defineQuery } from "@maple/query-engine/registry" import { Queries as Core } from "@maple/query-engine/registry" import type { + CloudflareInfraZoneBreakdownRequest, HostInfraTimeseriesRequest, CloudflareInfraZoneFacetsRequest, CloudflareInfraZoneDetailRequest, @@ -514,6 +515,90 @@ const hostInfraGaugeTimeseries = defineQuery({ }, }) +// --- cloudflareInfraZoneBreakdown: three parallel, then one dependent ----- + +const zoneBreakdownParams = (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => ({ + orgId, + serviceName: payload.serviceName, + startTime: payload.startTime, + endTime: payload.endTime, +}) + +const cloudflareInfraZoneBreakdownTotals = defineQuery({ + id: "cloudflareInfraZoneBreakdownTotals", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneBreakdownTotalsSQL( + payload.dimension, + toCloudflareFilters(payload), + payload.limit ?? 100, + ), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneBreakdownTotalsRowSchema }, + ), +}) + +/** + * Coverage is deliberately UNFILTERED: it answers "what did the poller collect + * here", which the UI needs in order to say "not collected yet" rather than "no + * traffic" for a window that predates the dataset. Do not thread filters in. + */ +const cloudflareInfraZoneBreakdownCoverage = defineQuery({ + id: "cloudflareInfraZoneBreakdownCoverage", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneBreakdownCoverageSQL(payload.dimension), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneBreakdownCoverageRowSchema }, + ), +}) + +const cloudflareInfraZoneBreakdownZoneTotal = defineQuery({ + id: "cloudflareInfraZoneBreakdownZoneTotal", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneCountersSQL(toCloudflareFilters(payload)), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneCountersRowSchema }, + ), +}) + +/** + * The chart runs AFTER the totals rather than beside them: totals are already + * ranked by requests, so they name the series worth plotting. Without that the + * grouping is unbounded — a zone taking scanner traffic returns a distinct path + * per probe, and the response grows to buckets x thousands of keys. One extra + * round trip over the same warm scan buys a payload that can't blow up. + * + * `topKeys` therefore rides in the PAYLOAD rather than being derived inside + * `compile`: it is the output of a previous query, which a def has no way to + * see. The caller must also skip this entirely when `topKeys` is empty. + */ +const cloudflareInfraZoneBreakdownTimeseries = defineQuery({ + id: "cloudflareInfraZoneBreakdownTimeseries", + profile: "aggregation", + cache: undefined, + compile: ( + payload: CloudflareInfraZoneBreakdownRequest & { readonly topKeys: ReadonlyArray }, + orgId: string, + ) => + CH.compile( + Integrations.cloudflareZoneBreakdownTimeseriesSQL( + payload.dimension, + toCloudflareFilters(payload), + payload.topKeys, + ), + { ...zoneBreakdownParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: Integrations.cloudflareZoneBreakdownTimeseriesRowSchema }, + ), +}) + export const Queries = { ...Core, @@ -698,4 +783,8 @@ export const Queries = { cloudflareInfraZoneFacets, hostInfraNetworkTimeseries, hostInfraGaugeTimeseries, + cloudflareInfraZoneBreakdownTotals, + cloudflareInfraZoneBreakdownCoverage, + cloudflareInfraZoneBreakdownZoneTotal, + cloudflareInfraZoneBreakdownTimeseries, } as const diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 6258bab74..edc7d60b5 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -516,23 +516,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("planetscaleInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const base = { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: Math.max(60, Math.floor(payload.bucketSeconds)), - database: payload.database, - } - const compiled = - payload.branch === undefined - ? CH.compile(Integrations.planetscaleInfraTimeseriesSQL(), base, { - rowSchema: Integrations.planetscaleInfraTimeseriesRowSchema, - }) - : CH.compile( - Integrations.planetscaleBranchInfraTimeseriesSQL(), - { ...base, branch: payload.branch }, - { rowSchema: Integrations.planetscaleInfraTimeseriesRowSchema }, - ) const rows = yield* runQuery(Queries.planetscaleInfraTimeseries, tenant, payload) return new PlanetScaleInfraTimeseriesResponse({ data: rows.map((row) => ({ ...row })) }) }), @@ -684,70 +667,14 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const tenant = yield* CurrentTenant.Context const filters = toCloudflareFilters(payload) - const params = { - orgId: tenant.orgId, - serviceName: payload.serviceName, - startTime: payload.startTime, - endTime: payload.endTime, - } - const totalsCompiled = CH.compile( - Integrations.cloudflareZoneBreakdownTotalsSQL( - payload.dimension, - filters, - payload.limit ?? 100, - ), - params, - { rowSchema: Integrations.cloudflareZoneBreakdownTotalsRowSchema }, - ) - // Coverage is deliberately unfiltered: it answers "what did the poller collect - // here", which the UI needs in order to say "not collected yet" rather than - // "no traffic" for a window that predates the dataset. - const coverageCompiled = CH.compile( - Integrations.cloudflareZoneBreakdownCoverageSQL(payload.dimension), - params, - { rowSchema: Integrations.cloudflareZoneBreakdownCoverageRowSchema }, - ) - const zoneTotalCompiled = CH.compile( - Integrations.cloudflareZoneCountersSQL(filters), - params, - { - rowSchema: Integrations.cloudflareZoneCountersRowSchema, - }, - ) const [totalRows, coverageRows, zoneRows] = yield* Effect.all( [ - mapExecError( - warehouse.compiledQuery(tenant, totalsCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownTotals", - }), - "cloudflareInfraZoneBreakdownTotals query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, coverageCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownCoverage", - }), - "cloudflareInfraZoneBreakdownCoverage query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, zoneTotalCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownZoneTotal", - }), - "cloudflareInfraZoneBreakdownZoneTotal query failed", - ), + runQuery(Queries.cloudflareInfraZoneBreakdownTotals, tenant, payload), + runQuery(Queries.cloudflareInfraZoneBreakdownCoverage, tenant, payload), + runQuery(Queries.cloudflareInfraZoneBreakdownZoneTotal, tenant, payload), ], { concurrency: 3 }, ) - // The chart runs after the totals rather than beside them: totals are already - // ranked by requests, so they name the series worth plotting. Without that the - // grouping is unbounded — a zone taking scanner traffic returns a distinct path - // per probe, and the response grows to buckets × thousands of keys. One extra - // round trip over the same warm scan buys a payload that can't blow up. - // The poller's own tail bucket is dropped from the picks, not plotted as a peer — - // it means the same thing as the fold, so it merges into it and leaves the slot - // for a real key. const topKeys = totalRows .filter((row) => row.key !== Integrations.CLOUDFLARE_BREAKDOWN_OTHER_KEY) .slice(0, Integrations.CLOUDFLARE_BREAKDOWN_SERIES_LIMIT) @@ -755,28 +682,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const bucketRows: ReadonlyArray = topKeys.length === 0 ? [] - : yield* mapExecError( - warehouse.compiledQuery( - tenant, - CH.compile( - Integrations.cloudflareZoneBreakdownTimeseriesSQL( - payload.dimension, - filters, - topKeys, - ), - { ...params, bucketSeconds: payload.bucketSeconds }, - { - rowSchema: - Integrations.cloudflareZoneBreakdownTimeseriesRowSchema, - }, - ), - { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownTimeseries", - }, - ), - "cloudflareInfraZoneBreakdownTimeseries query failed", - ) + : yield* runQuery(Queries.cloudflareInfraZoneBreakdownTimeseries, tenant, { + ...payload, + topKeys, + }) const coverage = coverageRows[0] const zoneRequests = zoneRows.find((row) => row.serviceName === payload.serviceName) // Breakdown metrics are a per-window top-N fold of what Cloudflare returned, so @@ -1106,11 +1015,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", "serviceOperations", payload, Effect.gen(function* () { - const params = { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - } yield* Effect.annotateCurrentSpan( "query.rollup.enabled", serviceOperationsRollupEnabled, @@ -1118,50 +1022,27 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // The rollout flag stays off until migration 0008 is deployed, // backfilled, and parity-checked. Disabling it restores the all-raw // rollback path without changing the endpoint contract. - const summaryOptions = { - serviceName: payload.serviceName, - environments: payload.environments, - limit: payload.limit, - } const runRawSummary = () => - warehouse.compiledQuery( - tenant, - CH.compile(CH.serviceOperationsSummaryRawQuery(summaryOptions), params, { - rowSchema: CH.serviceOperationsSummaryRowSchema, - }), - { profile: "aggregation", context: "serviceOperations" }, - ) + runQuery(Queries.serviceOperationsSummaryRaw, tenant, payload) let useRollup = serviceOperationsRollupEnabled const summaryEffect = useRollup - ? warehouse - .compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsSummaryQuery(summaryOptions), - params, - { - rowSchema: CH.serviceOperationsSummaryRowSchema, - }, - ), - { profile: "aggregation", context: "serviceOperations" }, - ) - .pipe( - Effect.catch((error) => { - if (!isMissingServiceOperationsRollup(error)) - return Effect.fail(error) - useRollup = false - return Effect.gen(function* () { - yield* Effect.logWarning( - "Service operations rollup is unavailable; using raw rollback path", - ).pipe(Effect.annotateLogs({ orgId: tenant.orgId })) - yield* Effect.annotateCurrentSpan( - "query.rollup.fallback", - true, - ) - return yield* runRawSummary() - }) - }), - ) + ? runQuery(Queries.serviceOperationsSummary, tenant, payload).pipe( + Effect.catch((error) => { + if (!isMissingServiceOperationsRollup(error)) + return Effect.fail(error) + useRollup = false + return Effect.gen(function* () { + yield* Effect.logWarning( + "Service operations rollup is unavailable; using raw rollback path", + ).pipe(Effect.annotateLogs({ orgId: tenant.orgId })) + yield* Effect.annotateCurrentSpan( + "query.rollup.fallback", + true, + ) + return yield* runRawSummary() + }) + }), + ) : runRawSummary() const summaryRows = yield* mapExecError( summaryEffect, @@ -1182,50 +1063,23 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", ) const requestedBucketSeconds = payload.bucketSeconds ?? windowSeconds / 50 const bucketSeconds = Math.max(1, Math.round(requestedBucketSeconds / 60)) * 60 - const timeseriesOptions = { - serviceName: payload.serviceName, - environments: payload.environments, - spanNames, - bucketSeconds, - } - const timeseriesParams = { ...params, bucketSeconds } + const timeseriesInput = { ...payload, spanNames, bucketSeconds } const runRawTimeseries = () => - warehouse.compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsTimeseriesRawQuery(timeseriesOptions), - timeseriesParams, - { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, - ), - { profile: "aggregation", context: "serviceOperationsTimeseries" }, - ) + runQuery(Queries.serviceOperationsTimeseriesRaw, tenant, timeseriesInput) const timeseriesEffect = useRollup - ? warehouse - .compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsTimeseriesQuery(timeseriesOptions), - timeseriesParams, - { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, - ), - { - profile: "aggregation", - context: "serviceOperationsTimeseries", - }, - ) - .pipe( - Effect.catch((error) => - isMissingServiceOperationsRollup(error) - ? Effect.gen(function* () { - yield* Effect.annotateCurrentSpan( - "query.rollup.fallback", - true, - ) - return yield* runRawTimeseries() - }) - : Effect.fail(error), - ), - ) + ? runQuery(Queries.serviceOperationsTimeseries, tenant, timeseriesInput).pipe( + Effect.catch((error) => + isMissingServiceOperationsRollup(error) + ? Effect.gen(function* () { + yield* Effect.annotateCurrentSpan( + "query.rollup.fallback", + true, + ) + return yield* runRawTimeseries() + }) + : Effect.fail(error), + ), + ) : runRawTimeseries() const timeseriesRows = yield* mapExecError( timeseriesEffect, diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index a49236480..761f1bd65 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -1,4 +1,5 @@ import type { + ServiceOperationsRequest, ListPodsRequest, NodeFacetsRequest, PodFacetsRequest, @@ -809,3 +810,96 @@ export const spanHierarchy = defineQuery({ ) }, }) + +// --- serviceOperations: rollup and raw variants --------------------------- +// +// Four defs, two logical queries. Each has a rollup form and a raw form, and +// the CHOICE between them stays in the handler because it is policy, not query +// construction: a feature flag selects the rollup, and a typed +// `isMissingServiceOperationsRollup` error falls back to raw at runtime, +// flipping a flag that the timeseries query then honors too. +// +// Rollup/raw pairs share an id, matching the context their spans already +// report — the fallback is recorded separately as `query.rollup.fallback`. +// +// All four are `cache: undefined`; the handler wraps the whole sequence in one +// `cachedDirect`. + +const serviceOperationsSummaryOptions = (payload: ServiceOperationsRequest) => ({ + serviceName: payload.serviceName, + environments: payload.environments, + limit: payload.limit, +}) + +const serviceOperationsParams = (payload: ServiceOperationsRequest, orgId: string) => ({ + orgId, + startTime: payload.startTime, + endTime: payload.endTime, +}) + +export const serviceOperationsSummary = defineQuery({ + id: "serviceOperations", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsRequest, orgId: string) => + CH.compile( + CH.serviceOperationsSummaryQuery(serviceOperationsSummaryOptions(payload)), + serviceOperationsParams(payload, orgId), + { rowSchema: CH.serviceOperationsSummaryRowSchema }, + ), +}) + +/** Rollback path, used when the rollup table is absent or the flag is off. */ +export const serviceOperationsSummaryRaw = defineQuery({ + id: "serviceOperations", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsRequest, orgId: string) => + CH.compile( + CH.serviceOperationsSummaryRawQuery(serviceOperationsSummaryOptions(payload)), + serviceOperationsParams(payload, orgId), + { rowSchema: CH.serviceOperationsSummaryRowSchema }, + ), +}) + +/** + * `spanNames` and `bucketSeconds` ride in the payload: both are derived from the + * summary rows and the requested window, so `compile` cannot see them. The + * rollup is minute-grain, which is why the caller rounds the bucket to a whole + * minute before passing it here. + */ +type ServiceOperationsTimeseriesInput = ServiceOperationsRequest & { + readonly spanNames: ReadonlyArray + readonly bucketSeconds: number +} + +const serviceOperationsTimeseriesOptions = (payload: ServiceOperationsTimeseriesInput) => ({ + serviceName: payload.serviceName, + environments: payload.environments, + spanNames: payload.spanNames, + bucketSeconds: payload.bucketSeconds, +}) + +export const serviceOperationsTimeseries = defineQuery({ + id: "serviceOperationsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsTimeseriesInput, orgId: string) => + CH.compile( + CH.serviceOperationsTimeseriesQuery(serviceOperationsTimeseriesOptions(payload)), + { ...serviceOperationsParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, + ), +}) + +export const serviceOperationsTimeseriesRaw = defineQuery({ + id: "serviceOperationsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsTimeseriesInput, orgId: string) => + CH.compile( + CH.serviceOperationsTimeseriesRawQuery(serviceOperationsTimeseriesOptions(payload)), + { ...serviceOperationsParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, + ), +}) From 33099edd80a8fa31ef6a1046302a0d5ff02ace40 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:43:09 +0200 Subject: [PATCH 5/5] fix(web): scope warehouse atom cache keys by org Atom family keys carried only filters and time range. The org rode along invisibly in the auth header, so after an org switch every cached entry stayed addressable by the new org and the UI re-rendered the PREVIOUS org's rows until the idle TTL expired -- up to 30 minutes on some atoms. Electric collections never had this bug because their ids already embed the org. The server side was already correct: buildDirectRouteCacheKey includes orgId, so this was purely the client cache. The org is PREFIXED rather than folded into the encoded object, because the atom decodes its key back into the query input -- an extra field there would travel to the server as part of the request payload. NUL cannot occur in encodeKey output (it is JSON) or in a Clerk org id, so the first occurrence always marks the boundary. Applied to both key-building sites: the warehouse atom factory and the dashboard widget family, which had the same defect. Verified: apps/web typecheck. Tests not run locally by request. --- apps/web/src/hooks/use-widget-data.ts | 8 ++++-- apps/web/src/lib/cache-key.ts | 28 +++++++++++++++++++ .../services/atoms/warehouse-query-atoms.ts | 7 +++-- 3 files changed, 37 insertions(+), 6 deletions(-) diff --git a/apps/web/src/hooks/use-widget-data.ts b/apps/web/src/hooks/use-widget-data.ts index 8addacf1d..227ae5d91 100644 --- a/apps/web/src/hooks/use-widget-data.ts +++ b/apps/web/src/hooks/use-widget-data.ts @@ -23,7 +23,8 @@ export type WidgetDataSourceLike = { import { disabledResultAtom } from "@/lib/services/atoms/disabled-result-atom" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" import type { WidgetDataState } from "@/components/dashboard-builder/types" -import { encodeKey } from "@/lib/cache-key" +import { encodeKey, encodeOrgScopedKey, orgScopedKeyPayload } from "@/lib/cache-key" +import { getActiveOrgId } from "@/lib/services/common/auth-headers" import { formatBackendError } from "@/lib/error-messages" import { Cause, Option } from "effect" import { WarehouseDecodeError, type BackendError } from "@/api/warehouse/effect-utils" @@ -318,8 +319,9 @@ const toWidgetDataAtomError = (error: unknown): WidgetFetchError => { const fetchWidgetData = Effect.fnUntraced( function* (key: string) { const parsed = yield* Effect.try({ + // The key is org-scoped; only the payload after the separator is JSON. try: () => - JSON.parse(key) as { + JSON.parse(orgScopedKeyPayload(key)) as { endpoint: string params: Record }, @@ -355,7 +357,7 @@ const widgetFetchFamily = Atom.family((key: string) => ) const widgetFetchAtom = (input: { endpoint: string; params: Record }) => - widgetFetchFamily(encodeKey(input)) + widgetFetchFamily(encodeOrgScopedKey(getActiveOrgId(), input)) /** * Fetches and transforms data for a single data source. Powers both whole diff --git a/apps/web/src/lib/cache-key.ts b/apps/web/src/lib/cache-key.ts index 2518f9b58..4b31bcea2 100644 --- a/apps/web/src/lib/cache-key.ts +++ b/apps/web/src/lib/cache-key.ts @@ -26,3 +26,31 @@ export function encodeKey(value: unknown): string { const normalized = normalizeForKey(value) return JSON.stringify(normalized === undefined ? null : normalized) } + +/** + * Separator between the active org and the encoded input in an atom family key. + * + * Atom keys used to carry only filters and time range. The org rode along + * invisibly in the auth header, so switching orgs left every cached entry + * addressable by the new org: the UI re-rendered the previous org's rows until + * the idle TTL expired — up to 30 minutes on some atoms. Electric collections + * never had this bug because their ids already embed the org. + * + * The org is prefixed rather than folded into the encoded object because atoms + * decode the key back into the query input; an extra field there would travel + * to the server as part of the request payload. + * + * NUL cannot occur in `encodeKey` output (it is JSON) or in a Clerk org id, so + * the first occurrence always marks the boundary. + */ +const ORG_KEY_SEPARATOR = "\u0000" + +/** Build an org-scoped atom family key. */ +export function encodeOrgScopedKey(orgId: string | null | undefined, value: unknown): string { + return `${orgId ?? ""}${ORG_KEY_SEPARATOR}${encodeKey(value)}` +} + +/** Recover just the encoded input from a key built by `encodeOrgScopedKey`. */ +export function orgScopedKeyPayload(key: string): string { + return key.slice(key.indexOf(ORG_KEY_SEPARATOR) + 1) +} diff --git a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts index 30b34c0b2..d2269e938 100644 --- a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts +++ b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts @@ -1,6 +1,7 @@ import { Atom } from "@/lib/effect-atom" import { Effect, Schema } from "effect" -import { encodeKey } from "@/lib/cache-key" +import { encodeOrgScopedKey, orgScopedKeyPayload } from "@/lib/cache-key" +import { getActiveOrgId } from "@/lib/services/common/auth-headers" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" import type { BackendError, WarehouseApiError } from "@/api/warehouse/effect-utils" import { @@ -161,7 +162,7 @@ function makeQueryAtomFamily(query: QueryEffect, o // shipped). The inner query spans already export by re-providing this same // (memoized) layer; this lifts the parent onto the same tracer. let resultAtom = MapleApiAtomClient.runtime.atom( - Schema.decodeUnknownEffect(UnknownFromJson)(key).pipe( + Schema.decodeUnknownEffect(UnknownFromJson)(orgScopedKeyPayload(key)).pipe( Effect.flatMap((input) => query(input as Input)), Effect.mapError(toQueryAtomError), ), @@ -174,7 +175,7 @@ function makeQueryAtomFamily(query: QueryEffect, o return resultAtom }) - return (input: Input) => family(encodeKey(input)) + return (input: Input) => family(encodeOrgScopedKey(getActiveOrgId(), input)) } export const getServiceUsageResultAtom = makeQueryAtomFamily(getServiceUsage, {