From c3cc20c4bc679bdeed4e5a75f817b88a082109e3 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Mon, 10 Aug 2026 23:02:44 +0200 Subject: [PATCH 1/3] perf(api): serve the org warehouse config from a durable KV tier The SWR memo shipped in 445d3ee9c only helps an isolate that has already paid once, and Workers evict isolates constantly. Measured over the three days since it landed, `resolveCachedSettings` still fell through to Postgres 10-13k times/day at a p50 of 482-636ms each -- 1.4-2.6 hours of blocked wall time per day, on a read that sits in front of every widget query on a dashboard. That cost is the per-`.execute()` postgres.js dial, not the query; the "~26ms Postgres read" in the old comment was the edge-cache read p50, not this. Only a shared cache removes it, since the whole point is to help an isolate that has never seen the org. Put that tier on Workers KV rather than `caches.default`. The Cache API is not slow -- a completed `cache.match()` is 7ms p50 -- but it holds one of the Worker's six simultaneous-connection slots while awaiting headers and is not cancellable, so the read deadline abandons the wait but not the slot and `compute` queues behind a seventh connection. Still true on 2026-08-10 across the remaining buckets: 15% of reads time out, p95 16.4s, 1003s blocked in a day. A KV `get` is an ordinary subrequest, so abandoning one is closer to actually free. That last point is a hypothesis, not a measurement. `cache.backend` is on every getOrCompute span so the two can be compared; if KV contends for the same six slots it will show the same bimodal split, and dropping the EDGE_CACHE binding reverts to the Cache API with no code change. - CacheBackend gains a KV implementation, selected when EDGE_CACHE is bound. Swaps the backend for every bucket, so qe-direct / qe-execute / autumn-customer -- which show the same pathology -- get measured too. - The config read goes through EdgeCacheService between memo and Postgres, resolved with `Effect.serviceOption` so tests and unbound hosts degrade to a direct read. `readTimeoutMs` is 150ms, not the 40ms default, because that default assumes a cheap `compute` and here `compute` is the ~500ms dial. - Invalidation becomes load-bearing: a missed memo bust self-healed in minutes, a missed KV bust is an hour of the whole fleet reading a config that no longer exists. `invalidateOrgRuntimeConfig` drops both tiers and every writer calls it, including both schema-apply workflow writes (they stamp `schema_version`, which is in the cached projection). - Raise the memo hard ceiling 15min -> 6h. It is an isolate-lifetime backstop, not a staleness bound -- the soft TTL is -- and at 15min it made a bursty widget fan-out block at the head of every burst. KV returns null for a missing key while EdgeCacheBackend uses undefined for "not cached", and a cached null is a real value -- on this bucket the common one ("managed org, use Tinybird"). Values are stored wrapped as `{ v }`; a bare null would have made every managed org a permanent miss. The memory backend cannot catch that class of bug because it stores raw values, so CacheBackendLive.test.ts covers it directly. Scoped to apps/api. apps/alerting has no KV binding and still uses `caches.default`; sharing one namespace across two alchemy stacks needs an ID reference rather than a second create. --- apps/api/alchemy.run.ts | 8 ++ .../api/src/platform/CacheBackendLive.test.ts | 77 ++++++++++ apps/api/src/platform/CacheBackendLive.ts | 81 ++++++++++- .../org/OrgClickHouseSettingsService.test.ts | 115 ++++++++++++++- .../org/OrgClickHouseSettingsService.ts | 132 ++++++++++++++++-- .../ClickHouseSchemaApplyWorkflow.run.ts | 29 +++- apps/api/wrangler.jsonc | 7 + lib/cache/src/cache-backend.ts | 2 +- 8 files changed, 432 insertions(+), 19 deletions(-) create mode 100644 apps/api/src/platform/CacheBackendLive.test.ts diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index b6d528a13..e0a571fa7 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -99,6 +99,13 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => title: resolveWorkerName("mcp-sessions", stage), }) + // EdgeCacheService backend. Binding present => KV; drop it to fall back to + // `caches.default` (see CacheBackendLive.ts for the measured reason KV is + // being trialled). + const edgeCache = yield* Cloudflare.KV.Namespace("EDGE_CACHE", { + title: resolveWorkerName("edge-cache", stage), + }) + // Long-running schema-apply: chunks heavy backfill migrations across durable // steps so they never hit the Worker request budget. Class is exported from // src/worker.ts. The first Workflow arg IS the physical workflow name; the @@ -196,6 +203,7 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) => AI: Cloudflare.AI.Gateway("maple-api-ai"), CHAT_SESSION: chatSession, MCP_SESSIONS: mcpSessions, + EDGE_CACHE: edgeCache, // Read side of the replay payload store; absent bindings degrade to // inline-only hydration (see platform/ReplayBlobStore.ts). REPLAY_BLOBS: replayBlobs, diff --git a/apps/api/src/platform/CacheBackendLive.test.ts b/apps/api/src/platform/CacheBackendLive.test.ts new file mode 100644 index 000000000..6c5ee3245 --- /dev/null +++ b/apps/api/src/platform/CacheBackendLive.test.ts @@ -0,0 +1,77 @@ +import { describe, expect, it } from "@effect/vitest" +import { makeKvBackend } from "./CacheBackendLive" + +/** + * Stand-in for a Workers KV binding with the two semantics that matter here: + * values are stored as strings and round-trip through JSON, and a missing key + * reads back as `null` — indistinguishable, without care, from a stored `null`. + */ +const makeFakeKv = () => { + const store = new Map() + return { + store, + get: async (key: string, _type: "json") => { + const raw = store.get(key) + return raw === undefined ? null : (JSON.parse(raw) as unknown) + }, + put: async (key: string, value: string, _options?: { expirationTtl?: number }) => { + store.set(key, value) + }, + delete: async (key: string) => { + store.delete(key) + }, + } +} + +describe("makeKvBackend", () => { + // The regression this guards: `EdgeCacheBackend` signals "not cached" with + // `undefined`, but a cached `null` is a real value — and on the org-config + // bucket it is the common one ("managed org, use Tinybird"). KV returns + // `null` for a missing key, so a bare `null` write would make every managed + // org a permanent miss and send it to Postgres forever. + it("round-trips a cached null as a hit, not a miss", async () => { + const kv = makeFakeKv() + const backend = makeKvBackend(kv) + + await backend.put("org-clickhouse-config", "hash", null, 3600, Date.now()) + + expect(await backend.get("org-clickhouse-config", "hash", Date.now())).toBeNull() + }) + + it("reports an absent key as undefined", async () => { + const backend = makeKvBackend(makeFakeKv()) + + expect(await backend.get("org-clickhouse-config", "never-written", Date.now())).toBeUndefined() + }) + + it("round-trips a struct value and drops it on delete", async () => { + const kv = makeFakeKv() + const backend = makeKvBackend(kv) + const value = { chUrl: "https://a.example", schemaVersion: "4" } + + await backend.put("org-clickhouse-config", "hash", value, 3600, Date.now()) + expect(await backend.get("org-clickhouse-config", "hash", Date.now())).toEqual(value) + + await backend.delete("org-clickhouse-config", "hash") + expect(await backend.get("org-clickhouse-config", "hash", Date.now())).toBeUndefined() + }) + + // KV rejects a TTL below 60s; the query-result buckets use shorter ones. + it("raises a sub-minute TTL to KV's 60s floor", async () => { + const puts: Array<{ expirationTtl?: number }> = [] + const kv = makeFakeKv() + const backend = makeKvBackend({ + ...kv, + put: async (key, value, options) => { + puts.push(options ?? {}) + await kv.put(key, value, options) + }, + }) + + await backend.put("qe-direct", "hash", { a: 1 }, 15, Date.now()) + await backend.put("org-clickhouse-config", "hash", { a: 1 }, 3600, Date.now()) + + expect(puts[0]?.expirationTtl).toBe(60) + expect(puts[1]?.expirationTtl).toBe(3600) + }) +}) diff --git a/apps/api/src/platform/CacheBackendLive.ts b/apps/api/src/platform/CacheBackendLive.ts index 5055372af..3b1dc4366 100644 --- a/apps/api/src/platform/CacheBackendLive.ts +++ b/apps/api/src/platform/CacheBackendLive.ts @@ -1,5 +1,6 @@ import { Effect, Layer, Metric } from "effect" import { WorkersCache } from "@maple/effect-cloudflare/workers-cache" +import { WorkerEnvironment } from "@maple/effect-cloudflare/worker-environment" import { CacheBackend, type EdgeCacheBackend, makeMemoryBackend } from "@maple/cache" import * as QueryEngineMetrics from "@/observability/QueryEngineMetrics" @@ -44,9 +45,87 @@ const makeWorkersBackend = (cache: Cache): EdgeCacheBackend => ({ }, }) +/** + * Minimal structural view of a Workers KV binding — enough for the three + * `EdgeCacheBackend` operations. Duck-typed rather than importing + * `@cloudflare/workers-types`, matching how `worker.ts` reads `MCP_SESSIONS`. + */ +interface KvLike { + get(key: string, type: "json"): Promise + put(key: string, value: string, options?: { expirationTtl?: number }): Promise + delete(key: string): Promise +} + +const isKvLike = (candidate: unknown): candidate is KvLike => + typeof candidate === "object" && + candidate !== null && + "get" in candidate && + "put" in candidate && + "delete" in candidate + +// KV rejects an `expirationTtl` below 60s. +const KV_MIN_TTL_SECONDS = 60 + +/** + * Workers KV as the edge-cache backend, trialled against `caches.default`. + * + * The Cache API is not slow — measured on 2026-08-10, a completed `cache.match()` + * costs a p50 of 7ms. The problem is that 15% of reads never complete: a + * `cache.match()` holds one of the Worker's six simultaneous-connection slots + * while it waits for response headers, and it is **not cancellable**, so the + * 40ms deadline abandons the wait but not the slot — `compute` then opens a + * seventh connection and queues behind the read it just gave up on (p95 16.4s, + * 1003s of blocked wall time in one day). See `DEFAULT_EDGE_CACHE_READ_TIMEOUT_MS`. + * + * KV is worth trying because a `get` is an ordinary subrequest rather than an + * uncancellable cache read, so abandoning one is closer to actually free. That + * is the hypothesis, NOT a measured fact — if KV turns out to contend for the + * same six slots it will show the same bimodal split. `cache.backend` is on + * every `getOrCompute` span precisely so the two can be compared; check the + * `cache.read_status` distribution per backend before trusting this. + * + * `cacheTtl` is left at the KV default (60s). Raising it would buy more + * colo-local hits, but a colo that has the key cached would keep serving it for + * that long after an explicit `delete` — and invalidation-on-mutation is the + * property the org config cache is relying on. + * + * Values are stored wrapped as `{ v }`. `EdgeCacheBackend` uses `undefined` to + * mean "not cached", and a cached `null` is a legitimate value — for the org + * config bucket it is the COMMON one ("this org is managed, use Tinybird"), and + * caching it is the single biggest win there. KV's `get` returns `null` for a + * missing key, so storing a bare `null` would make every managed org a + * permanent miss. The envelope keeps absence and null distinguishable. + */ +export const makeKvBackend = (kv: KvLike): EdgeCacheBackend => { + const composite = (bucket: string, hash: string) => `${bucket}:${hash}` + + return { + name: "workers-kv", + get: async (bucket, hash) => { + const envelope = await kv.get(composite(bucket, hash), "json") + if (envelope === null || typeof envelope !== "object" || !("v" in envelope)) return undefined + return (envelope as { v: unknown }).v + }, + put: async (bucket, hash, value, ttlSeconds) => { + await kv.put(composite(bucket, hash), JSON.stringify({ v: value }), { + expirationTtl: Math.max(KV_MIN_TTL_SECONDS, Math.floor(ttlSeconds)), + }) + }, + delete: async (bucket, hash) => { + await kv.delete(composite(bucket, hash)) + }, + } +} + export const CacheBackendLive = Layer.effect( CacheBackend, Effect.gen(function* () { + const env = yield* WorkerEnvironment + const kv = env.EDGE_CACHE + if (isKvLike(kv)) { + return CacheBackend.of(makeKvBackend(kv)) + } + const cache = yield* WorkersCache if (!cache) { // The fallback is per-isolate, so nothing is shared across requests that @@ -62,4 +141,4 @@ export const CacheBackendLive = Layer.effect( } return CacheBackend.of(makeWorkersBackend(cache)) }), -).pipe(Layer.provide(WorkersCache.layer)) +).pipe(Layer.provide(Layer.mergeAll(WorkersCache.layer, WorkerEnvironment.layer))) diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts index cfb961987..331d8bda6 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts @@ -18,7 +18,10 @@ import { cleanupTestDbs, createTestDb, executeSql, type TestDb } from "@/platfor import { type ClickHouseExecConfig, execClickHouse, + invalidateOrgRuntimeConfig, + invalidateOrgRuntimeConfigMemo, isRetryableUpstream, + ORG_CH_CONFIG_BUCKET, OrgClickHouseSettingsService, shouldHealSchemaVersion, validateClickHouseCredentialTransport, @@ -359,11 +362,25 @@ describe("resolveRuntimeConfig caching", () => { }).pipe(Effect.provide(buildLayer(testDb))) }) - // The memo's soft TTL (5min) and hard ceiling (15min), mirrored from the + // The memo's soft TTL (5min) and hard ceiling (6h), mirrored from the // service. Tests drive TestClock across them rather than reaching into the // module's private state. const SOFT_TTL_MS = 300_000 - const HARD_TTL_MS = 900_000 + const HARD_TTL_MS = 21_600_000 + + // A raw UPDATE bypasses the service, so it is invisible to the DURABLE tier + // just as it is to the memo — the refresh behind a stale memo entry re-reads + // the cached row and sees nothing new. + // + // Tests below use a raw UPDATE as an instrument for observing MEMO mechanics + // (did the caller block? did the forked refresh land?), so they evict the + // durable entry to stand in for the invalidation a real write would have + // performed. Evicting only the durable tier — never the memo — is what keeps + // the memo the thing under test. + const evictDurable = (orgId: string) => + Effect.flatMap(EdgeCacheService, (cache) => + cache.invalidate({ bucket: ORG_CH_CONFIG_BUCKET, key: orgId }), + ) it.effect("past the soft TTL, serves the stale value and refreshes behind the request", () => { const testDb = createTestDb(cacheTrackedDbs) @@ -380,6 +397,8 @@ describe("resolveRuntimeConfig caching", () => { ]), ) + yield* evictDurable(orgId) + // Past the soft TTL but inside the hard ceiling: the caller must NOT // wait on Postgres, so it still sees the old URL. yield* TestClock.adjust(SOFT_TTL_MS + 1_000) @@ -408,6 +427,8 @@ describe("resolveRuntimeConfig caching", () => { ]), ) + yield* evictDurable(orgId) + // Nothing refreshed the entry in the meantime (an isolate whose requests // all ended before their refresh landed), so the ceiling forces a // blocking read rather than serving an unboundedly old value. @@ -417,6 +438,95 @@ describe("resolveRuntimeConfig caching", () => { }).pipe(Effect.provide(buildLayer(testDb))) }) + // The ceiling is an isolate-lifetime backstop, not a staleness bound. It used + // to be 15min, which made a bursty workload (idle isolate, then a widget + // fan-out) block on Postgres at the head of every burst. At 20min — well past + // the old ceiling — the caller must still be served from the memo. + it.effect("an idle stretch past the old 15min ceiling still serves without blocking", () => { + const testDb = createTestDb(cacheTrackedDbs) + const orgId = "org_ch_swr_ceiling_raised" + return Effect.gen(function* () { + yield* Effect.promise(() => seedRow(testDb, orgId, "https://a.example")) + yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + + yield* Effect.promise(() => + executeSql(testDb, "UPDATE org_clickhouse_settings SET ch_url = $2 WHERE org_id = $1", [ + orgId, + "https://b.example", + ]), + ) + + yield* evictDurable(orgId) + + yield* TestClock.adjust(1_200_000) + const stale = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + // The old URL proves nobody blocked on the read; the refresh this call + // forked lands next, so the burst behind it is already current. + expect(expectSome(stale).url).toBe("https://a.example") + + yield* TestClock.adjust(1) + const refreshed = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + expect(expectSome(refreshed).url).toBe("https://b.example") + }).pipe(Effect.provide(buildLayer(testDb))) + }) + + // The two tiers must be busted together. A memo-only bust used to be a + // complete invalidation; with a durable tier underneath it is not, and the + // failure is silent — the read falls through to a shared entry that outlives + // every isolate. `invalidateOrgRuntimeConfig` is the one writers must call. + it.effect("a memo-only bust falls through to the durable tier; the full bust does not", () => { + const testDb = createTestDb(cacheTrackedDbs) + const orgId = "org_ch_memo_module_bust" + return Effect.gen(function* () { + yield* Effect.promise(() => seedRow(testDb, orgId, "https://a.example")) + const before = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + expect(expectSome(before).url).toBe("https://a.example") + + yield* Effect.promise(() => + executeSql(testDb, "UPDATE org_clickhouse_settings SET ch_url = $2 WHERE org_id = $1", [ + orgId, + "https://b.example", + ]), + ) + + invalidateOrgRuntimeConfigMemo(orgId) + const afterMemoBust = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + // Still the old URL: the memo is gone, but the durable entry answered. + expect(expectSome(afterMemoBust).url).toBe("https://a.example") + + yield* invalidateOrgRuntimeConfig(orgId) + const afterFullBust = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + expect(expectSome(afterFullBust).url).toBe("https://b.example") + }).pipe(Effect.provide(buildLayer(testDb))) + }) + + // The reason the durable tier exists: a cold isolate has no memo entry at all, + // and in prod that path measured 10–13k resolutions/day at a p50 of ~500ms. + it.effect("a cold isolate is served by the durable tier, not Postgres", () => { + const testDb = createTestDb(cacheTrackedDbs) + const orgId = "org_ch_durable_cold" + return Effect.gen(function* () { + yield* Effect.promise(() => seedRow(testDb, orgId, "https://a.example")) + yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + + yield* Effect.promise(() => + executeSql(testDb, "UPDATE org_clickhouse_settings SET ch_url = $2 WHERE org_id = $1", [ + orgId, + "https://b.example", + ]), + ) + + // Drop only the memo — this is what a fresh isolate looks like: no + // in-process state, but the shared entry a sibling isolate wrote is there. + invalidateOrgRuntimeConfigMemo(orgId) + + const resolved = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + // The old URL proves the durable tier answered. Had it gone to Postgres + // it would have picked up the raw UPDATE — and paid the dial to do it. + expect(expectSome(resolved).url).toBe("https://a.example") + }).pipe(Effect.provide(buildLayer(testDb))) + }) + // Asserts the observable contract — none of the concurrent callers block, and // the in-flight marker is released so the entry can refresh again. It does // NOT count Postgres reads: the dedup marker is module-private, and a broken @@ -449,6 +559,7 @@ describe("resolveRuntimeConfig caching", () => { "https://c.example", ]), ) + yield* evictDurable(orgId) yield* TestClock.adjust(SOFT_TTL_MS + 1_000) yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) yield* TestClock.adjust(1) diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.ts index d391ed15f..2bfc75765 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.ts @@ -30,6 +30,7 @@ import { import { orgClickHouseSchemaApplyRuns, orgClickHouseSettings } from "@maple/db" import { eq } from "drizzle-orm" import { WorkerEnvironment } from "@maple/effect-cloudflare/worker-environment" +import { EdgeCacheService } from "@maple/cache" import { Clock, Context, @@ -102,17 +103,25 @@ type CachedSettingsRow = Pick< // otherwise -> block on Postgres (cold isolate, or a memo so idle // that no background refresh ever completed) // -// The hard ceiling exists because a background refresh is best-effort: it is -// forked into the triggering request's scope and interrupted if that request -// finishes first (see `refreshCachedSettings`). Without a ceiling, an isolate -// serving only short requests could serve one value forever. +// The SOFT TTL is what bounds staleness. Every write through this service busts +// the memo, but only in the isolate that served the write — other isolates +// converge by re-reading at the soft TTL. That degree of staleness is safe +// because the warehouse executor self-heals on `WarehouseAuthError`: it calls +// `invalidateRuntimeConfig` and retries once, so a credential rotation costs the +// first request one extra round-trip instead of costing the org every request +// until the entry ages out. // -// Staleness is safe to this degree because the warehouse executor self-heals on -// `WarehouseAuthError` — it calls `invalidateRuntimeConfig` and retries once, so -// a credential rotation costs the first request one extra round-trip instead of -// costing the org every request until the entry ages out. +// The HARD ceiling is not a staleness bound — it is an isolate-lifetime backstop. +// A background refresh is best-effort: it is forked into the triggering request's +// scope and interrupted if that request finishes first (see +// `refreshCachedSettings`). Without a ceiling, a pathological isolate serving +// only sub-refresh-length requests could serve one value forever. It is set well +// past a typical Workers isolate lifetime so that the blocking read happens once +// per cold isolate and never again — a bursty dashboard workload (idle isolate, +// then a widget fan-out) must not pay it on the first query of every burst, with +// the rest of the fan-out queued behind it. const ORG_CH_CONFIG_MEMO_TTL_MS = 300_000 -const ORG_CH_CONFIG_MEMO_HARD_MS = 900_000 +const ORG_CH_CONFIG_MEMO_HARD_MS = 21_600_000 interface RuntimeConfigMemoEntry { readonly value: CachedChSettings | null readonly freshUntil: number @@ -138,6 +147,46 @@ const runtimeConfigMemo = new Map() const refreshInFlight = new Map() const REFRESH_MARKER_STALE_MS = 10_000 +/** + * Drop an org's memoized runtime config, for writers that live OUTSIDE this + * service and so have no service instance to call — today the schema-apply + * workflow, which stamps `schema_version`/`sync_status` on the row directly. + * + * The two maps must always be cleared together: a refresh forked before a write + * must not land after it and restore the value that was just dropped. + * + * The workflow runs in its own isolate, so this clears that isolate's memo and + * not the API's — API isolates still converge at `ORG_CH_CONFIG_MEMO_TTL_MS`. + * It exists so the invariant "every writer of this row busts the memo" holds at + * every write site rather than at most of them. + */ +export const invalidateOrgRuntimeConfigMemo = (orgId: string): void => { + runtimeConfigMemo.delete(orgId) + refreshInFlight.delete(orgId) +} + +/** + * Drop BOTH cache tiers for an org: this isolate's memo and the shared durable + * entry. Every writer of `org_clickhouse_settings` must call it. + * + * The durable tier makes this non-optional in a way the memo never was. A memo + * is per-isolate and expires in minutes, so a missed bust self-healed; a KV + * entry is shared by every isolate in every colo and lives for + * `ORG_CH_CONFIG_EDGE_TTL_SECONDS`, so a missed bust is an hour of every request + * globally reading a config that no longer exists. + * + * `EdgeCacheService` is resolved with `Effect.serviceOption` at call time, so + * this is safe to run in a context that has no cache layer (tests, hosts without + * the binding) — it degrades to a memo-only bust. + */ +export const invalidateOrgRuntimeConfig = Effect.fnUntraced(function* (orgId: string) { + invalidateOrgRuntimeConfigMemo(orgId) + const edgeCache = yield* Effect.serviceOption(EdgeCacheService) + if (Option.isSome(edgeCache)) { + yield* edgeCache.value.invalidate({ bucket: ORG_CH_CONFIG_BUCKET, key: orgId }) + } +}) + /** * Projection of the settings row memoized by `resolveRuntimeConfig`. Holds the * ENCRYPTED password material (ciphertext/iv/tag) — never the plaintext — so @@ -156,6 +205,23 @@ const CachedChSettings = Schema.Struct({ }) type CachedChSettings = typeof CachedChSettings.Type +/** + * Codec for the durable (KV) tier. `null` — "this org is managed, use Tinybird" + * — is a cached value in its own right, not an absence, and it is the common + * case, so it must survive the JSON round-trip. + */ +const CachedChSettingsCodec = Schema.NullOr(CachedChSettings) + +/** + * Bucket + key shape for the durable tier. `invalidate` derives the storage hash + * from the same `{ bucket, key }`, so every write site must pass the org id + * unchanged. + */ +export const ORG_CH_CONFIG_BUCKET = "org-clickhouse-config" +/** Long, because eviction is driven by explicit invalidation at every write site. */ +const ORG_CH_CONFIG_EDGE_TTL_SECONDS = 3_600 +const ORG_CH_CONFIG_EDGE_READ_TIMEOUT_MS = 150 + const toCachedChSettings = (row: CachedSettingsRow): CachedChSettings => ({ schemaVersion: row.schemaVersion, chUrl: row.chUrl, @@ -875,10 +941,7 @@ export class OrgClickHouseSettingsService extends Context.Service< // changes no routing decision, so it does not count as a hit. const memoized = runtimeConfigMemo.get(orgId) const hadOverride = memoized !== undefined && memoized.value !== null - runtimeConfigMemo.delete(orgId) - // A refresh forked before this write must not land after it and restore - // the value we just dropped. - refreshInFlight.delete(orgId) + yield* invalidateOrgRuntimeConfig(orgId) return hadOverride }) @@ -1280,11 +1343,52 @@ export class OrgClickHouseSettingsService extends Context.Service< // The narrow Postgres read behind the memo. Returns the ENCRYPTED row // projection (or `null` for a managed org); decryption happens per-request // in `resolveRuntimeConfig`, so plaintext credentials never enter a cache. - const lookupCachedSettings = (orgId: OrgId) => + const readSettingsFromPostgres = (orgId: OrgId) => selectCachedRow(orgId).pipe( Effect.map((row) => (Option.isSome(row) ? toCachedChSettings(row.value) : null)), ) + /** + * The durable tier between the per-isolate memo and Postgres. + * + * The memo only ever helps an isolate that has already paid once, and + * Workers evict isolates constantly: measured over the three days after the + * SWR memo shipped, 10–13k resolutions/day still fell through to Postgres at + * a p50 of 482–636ms each — 1.4–2.6 HOURS of blocked wall time per day. That + * cost is the per-`.execute()` postgres.js dial, not the query. + * + * A shared cache is the only thing that removes it, since the whole point is + * to help an isolate that has never seen this org. It is deliberately on + * Workers KV rather than `caches.default` — see `makeKvBackend` in + * `CacheBackendLive.ts` for the measured reason, and treat this as an + * experiment until `cache.read_status` per `cache.backend` says otherwise. + * + * `readTimeoutMs` is well above the service default of 40ms because that + * default is tuned for a cheap `compute`. Here `compute` is a ~500ms dial, + * so waiting 150ms for a cache read that usually lands in single-digit ms is + * the better trade even when it occasionally loses. + * + * `Effect.serviceOption` keeps the dependency optional: tests and any host + * without the layer fall through to Postgres rather than failing to build. + */ + const lookupCachedSettings = Effect.fnUntraced(function* (orgId: OrgId) { + const edgeCache = yield* Effect.serviceOption(EdgeCacheService) + if (Option.isNone(edgeCache)) return yield* readSettingsFromPostgres(orgId) + + const result = yield* edgeCache.value.getOrCompute( + { + bucket: ORG_CH_CONFIG_BUCKET, + key: orgId, + ttlSeconds: ORG_CH_CONFIG_EDGE_TTL_SECONDS, + readTimeoutMs: ORG_CH_CONFIG_EDGE_READ_TIMEOUT_MS, + schema: CachedChSettingsCodec, + }, + readSettingsFromPostgres(orgId), + ) + yield* Effect.annotateCurrentSpan("clickhouse.config.edge_hit", result.hit) + return result.value + }) + const storeCachedSettings = (orgId: OrgId, value: CachedChSettings | null, nowMs: number) => { runtimeConfigMemo.set(orgId, { value, diff --git a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts index f36c8cf45..c2d584d32 100644 --- a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts +++ b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts @@ -29,9 +29,34 @@ import { type DesiredTable, } from "@maple/domain/clickhouse" import { eq } from "drizzle-orm" -import { Effect } from "effect" +import { Effect, Layer } from "effect" +import { EdgeCacheService } from "@maple/cache" +import { CacheBackendLive } from "@/platform/CacheBackendLive" import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { makeTracedPgConnection, type TracedPgConnection } from "@/platform/pg-execute" +import { invalidateOrgRuntimeConfig } from "@/services/org/OrgClickHouseSettingsService" + +/** + * Bust the org's cached runtime config after this workflow writes to + * `org_clickhouse_settings`. + * + * This matters more than it looks. The workflow stamps `schema_version`, which + * is part of the cached projection, and it runs in its OWN isolate — so clearing + * the module-scoped memo here reaches nothing the API is serving. What does + * reach the API is the durable KV entry, which is shared across every isolate + * and colo and lives an hour. Without this, a completed schema apply leaves the + * whole fleet reading a stale `schema_version` and reporting phantom + * `clickhouse.schemaDrift` until the entry expires. + * + * Best-effort: a failure here must never fail the apply, which has already + * committed its Postgres write. + */ +const bustRuntimeConfigCache = (orgId: string): Promise => + Effect.runPromise( + invalidateOrgRuntimeConfig(orgId).pipe( + Effect.provide(EdgeCacheService.layer.pipe(Layer.provide(CacheBackendLive))), + ), + ).catch(() => undefined) /** * This workflow runs outside the worker's layer graph, so it owns its telemetry @@ -471,6 +496,7 @@ async function runWithDb( }) .where(eq(orgClickHouseSettings.orgId, orgId)), ) + await bustRuntimeConfigCache(orgId) }) for (const migration of clickHouseMigrations) { @@ -629,6 +655,7 @@ async function runWithDb( .set({ syncStatus: "error", lastSyncError: message, updatedAt: new Date(finishedAt) }) .where(eq(orgClickHouseSettings.orgId, orgId)), ).catch(() => undefined) + await bustRuntimeConfigCache(orgId) throw error } } diff --git a/apps/api/wrangler.jsonc b/apps/api/wrangler.jsonc index 9c67db46a..e721c241a 100644 --- a/apps/api/wrangler.jsonc +++ b/apps/api/wrangler.jsonc @@ -65,6 +65,13 @@ "binding": "MCP_SESSIONS", "id": "00000000000000000000000000000000", }, + // Backend for EdgeCacheService. Present => KV is used; remove the binding + // and the service falls back to `caches.default`. See CacheBackendLive.ts + // for why KV is being trialled over the Cache API. + { + "binding": "EDGE_CACHE", + "id": "00000000000000000000000000000000", + }, ], // Cloudflare Email Service (Email Sending). `remote: true` routes sends through // the real Cloudflare sender even under local dev. (Mirrored in alchemy.run.ts diff --git a/lib/cache/src/cache-backend.ts b/lib/cache/src/cache-backend.ts index ea140d5f6..77dc67445 100644 --- a/lib/cache/src/cache-backend.ts +++ b/lib/cache/src/cache-backend.ts @@ -15,7 +15,7 @@ export interface EdgeCacheBackend { * `memory` fallback — which is silently selected whenever `caches` is * undefined, and makes every cross-request hit disappear. */ - readonly name: "workers-cache" | "memory" + readonly name: "workers-cache" | "workers-kv" | "memory" readonly get: (bucket: string, hash: string, nowMs: number) => Promise readonly put: ( bucket: string, From 863f4b1103542b488e0cf1d4ae416ae708c68708 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Mon, 10 Aug 2026 23:11:07 +0200 Subject: [PATCH 2/3] fix(api): export env from the cloudflare:workers test stub Two CI failures from the KV tier, both invisible locally because I only ran targeted files. `WorkerEnvironment` destructures `env` off the `cloudflare:workers` module, but the vitest stub only exported `DurableObject` and `WorkflowEntrypoint`. Every consumer so far layered its bindings in rather than reading them off the service, so nothing had dereferenced the result before; `CacheBackendLive` reading `env.EDGE_CACHE` is the first, and it crashed 10 alerts.http tests with "Cannot read properties of undefined". Export `env` from the stub so the double matches the real module, and make `WorkerEnvironment.layer` fall back to `{}` so a host without one hands out an empty binding record rather than `undefined` behind a non-nullable type. The SWR refresh assertions also went flaky. A forked refresh completes when real promises settle -- a PGlite read, plus a durable-tier read whose key hashing goes through `crypto.subtle` -- and none of that is driven by `TestClock`, so `adjust(1)` was never a guarantee. Adding the durable tier lengthened that chain enough for a slower CI machine to lose the race. Poll for the refreshed value with `resolveUntilUrl` instead, bounded so a refresh that never lands still fails. The stale assertion stays exactly where it was, so "nobody blocked" is still asserted on the nose. --- .../org/OrgClickHouseSettingsService.test.ts | 39 ++++++++++++++----- apps/api/test/stubs/cloudflare-workers.ts | 11 ++++-- .../src/worker-environment.ts | 7 +++- 3 files changed, 43 insertions(+), 14 deletions(-) diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts index 331d8bda6..e7521600d 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts @@ -382,6 +382,31 @@ describe("resolveRuntimeConfig caching", () => { cache.invalidate({ bucket: ORG_CH_CONFIG_BUCKET, key: orgId }), ) + /** + * Resolve until the forked background refresh has landed. + * + * A single `TestClock.adjust(1)` is NOT enough. The refresh completes when + * real promises settle — a PGlite read, plus a durable-tier read whose key + * hashing goes through `crypto.subtle` — and none of that is driven by the + * test clock. Asserting on one tick passed locally and failed in CI, where + * the machine is slower and contended. + * + * Polling keeps the contract honest: callers assert the STALE value first + * (proving nobody blocked), then use this to assert the refresh eventually + * lands. Bounded, so a refresh that never lands still fails the test. + */ + const resolveUntilUrl = Effect.fnUntraced(function* (orgId: string, expected: string) { + let last: string | undefined + for (let attempt = 0; attempt < 200; attempt++) { + const resolved = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + last = Option.isSome(resolved) ? resolved.value.url : undefined + if (last === expected) return last + yield* TestClock.adjust(1) + yield* Effect.yieldNow + } + return last + }) + it.effect("past the soft TTL, serves the stale value and refreshes behind the request", () => { const testDb = createTestDb(cacheTrackedDbs) const orgId = "org_ch_swr_stale" @@ -405,11 +430,9 @@ describe("resolveRuntimeConfig caching", () => { const stale = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) expect(expectSome(stale).url).toBe("https://a.example") - // The refresh forked by that call now lands, so the NEXT resolve sees the + // The refresh forked by that call lands, so a subsequent resolve sees the // new value without anyone having blocked on the read. - yield* TestClock.adjust(1) - const refreshed = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) - expect(expectSome(refreshed).url).toBe("https://b.example") + expect(yield* resolveUntilUrl(orgId, "https://b.example")).toBe("https://b.example") }).pipe(Effect.provide(buildLayer(testDb))) }) @@ -464,9 +487,7 @@ describe("resolveRuntimeConfig caching", () => { // forked lands next, so the burst behind it is already current. expect(expectSome(stale).url).toBe("https://a.example") - yield* TestClock.adjust(1) - const refreshed = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) - expect(expectSome(refreshed).url).toBe("https://b.example") + expect(yield* resolveUntilUrl(orgId, "https://b.example")).toBe("https://b.example") }).pipe(Effect.provide(buildLayer(testDb))) }) @@ -562,9 +583,7 @@ describe("resolveRuntimeConfig caching", () => { yield* evictDurable(orgId) yield* TestClock.adjust(SOFT_TTL_MS + 1_000) yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) - yield* TestClock.adjust(1) - const refreshed = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) - expect(expectSome(refreshed).url).toBe("https://c.example") + expect(yield* resolveUntilUrl(orgId, "https://c.example")).toBe("https://c.example") }).pipe(Effect.provide(buildLayer(testDb))) }) diff --git a/apps/api/test/stubs/cloudflare-workers.ts b/apps/api/test/stubs/cloudflare-workers.ts index 1c1d1775a..083be4417 100644 --- a/apps/api/test/stubs/cloudflare-workers.ts +++ b/apps/api/test/stubs/cloudflare-workers.ts @@ -1,7 +1,9 @@ // Stub for the `cloudflare:workers` virtual module so it can be imported in the -// node/vitest environment. Only `DurableObject` and `WorkflowEntrypoint` are -// needed — the modules in `@maple/effect-cloudflare` that statically import -// them are never exercised at runtime in unit tests (bindings are layered in). +// node/vitest environment. Alongside `DurableObject` and `WorkflowEntrypoint`, +// this must export `env`: the real module always does, and `WorkerEnvironment` +// destructures it. Omitting it made the service yield `undefined` rather than an +// empty binding record, so the first consumer to read a binding off it (rather +// than layering one in) crashed with "Cannot read properties of undefined". // // `DurableObject` keeps `ctx`/`env` because `ChatSession` uses both: `ctx.storage.sql` for the // event log and `ctx.waitUntil` to own its own turn. `test/chat/fake-do-state.ts` hands it a real @@ -16,3 +18,6 @@ export class DurableObject { } } export class WorkflowEntrypoint {} + +/** No bindings in node/vitest — tests layer in whatever they need. */ +export const env: Record = {} diff --git a/lib/effect-cloudflare/src/worker-environment.ts b/lib/effect-cloudflare/src/worker-environment.ts index 73eaf4ccb..a1d452cd3 100644 --- a/lib/effect-cloudflare/src/worker-environment.ts +++ b/lib/effect-cloudflare/src/worker-environment.ts @@ -24,7 +24,12 @@ export class WorkerEnvironment extends Context.Service = Layer.effect( this, - cloudflareWorkers.pipe(Effect.map(({ env }) => env as Record)), + // `?? {}` because the service's type promises a record and consumers read + // bindings straight off it. A host whose `cloudflare:workers` module has no + // `env` (the vitest stub, historically) would otherwise hand out + // `undefined` behind a non-nullable type, and the first consumer to + // dereference a binding crashes instead of seeing "binding absent". + cloudflareWorkers.pipe(Effect.map(({ env }) => (env ?? {}) as Record)), ) } From 78f124b4cb6c72e62c955a39606d337cd7174bb4 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Mon, 10 Aug 2026 23:16:16 +0200 Subject: [PATCH 3/3] fix(api): keep the durable config tier off stores nobody invalidates Two findings from review, both real. `apps/alerting` runs OrgClickHouseSettingsService too, but it has no EDGE_CACHE binding, so its CacheBackend is `caches.default` -- a store no `invalidateOrgRuntimeConfig` ever reaches, because every writer lives in apps/api and busts KV. An hour-long entry there would let alert evaluation keep using a warehouse config the customer already rotated or deleted: firing on stale data, or missing incidents outright. My PR description waved this away on the grounds that alerting never writes the config, which was the wrong question -- the hazard is that it reads from a store the writers cannot bust. Expose `backendName` on EdgeCacheServiceShape and use the durable tier only where invalidation lands: `workers-kv` (the binding apps/api owns, where the busts go) and `memory` (per-isolate, self-consistent by construction). A `workers-cache` host falls back to memo + Postgres -- slower, and right. The proper fix is binding the same namespace into the alerting stack, which already binds Hyperdrive by ID across stacks; until then the tier stays off there rather than silently serving stale routing. Second: KV refuses an `expirationTtl` under 60s and the backend clamped upward, but real buckets ask for less -- `qe-evaluate` and the integrations routes use 30s. The other two backends carry their own deadline (`Cache-Control: max-age`, `expiresAt`), so a 30s entry was being served for 60s. Carry the real deadline in the envelope as `exp` and expire on read; `expirationTtl` still goes out at the floor to bound storage. --- .../api/src/platform/CacheBackendLive.test.ts | 28 +++++++++++- apps/api/src/platform/CacheBackendLive.ts | 38 +++++++++++----- .../org/OrgClickHouseSettingsService.test.ts | 45 ++++++++++++++++++- .../org/OrgClickHouseSettingsService.ts | 30 ++++++++++++- lib/cache/src/edge-cache.ts | 20 ++++++++- 5 files changed, 143 insertions(+), 18 deletions(-) diff --git a/apps/api/src/platform/CacheBackendLive.test.ts b/apps/api/src/platform/CacheBackendLive.test.ts index 6c5ee3245..fc1d9ef00 100644 --- a/apps/api/src/platform/CacheBackendLive.test.ts +++ b/apps/api/src/platform/CacheBackendLive.test.ts @@ -56,8 +56,9 @@ describe("makeKvBackend", () => { expect(await backend.get("org-clickhouse-config", "hash", Date.now())).toBeUndefined() }) - // KV rejects a TTL below 60s; the query-result buckets use shorter ones. - it("raises a sub-minute TTL to KV's 60s floor", async () => { + // KV rejects a TTL below 60s; real buckets ask for less (`qe-evaluate` and the + // integrations routes use 30s). + it("raises a sub-minute TTL to KV's 60s storage floor", async () => { const puts: Array<{ expirationTtl?: number }> = [] const kv = makeFakeKv() const backend = makeKvBackend({ @@ -74,4 +75,27 @@ describe("makeKvBackend", () => { expect(puts[0]?.expirationTtl).toBe(60) expect(puts[1]?.expirationTtl).toBe(3600) }) + + // ...but the floor must not extend the entry's observable life. The other two + // backends carry their own deadline (Cache-Control max-age / expiresAt), so + // without an envelope deadline a 30s entry would be served for 60s. + it("expires a sub-minute entry at its real TTL, not at the 60s floor", async () => { + const backend = makeKvBackend(makeFakeKv()) + const writtenAt = 1_000_000 + + await backend.put("qe-evaluate", "hash", { a: 1 }, 30, writtenAt) + + expect(await backend.get("qe-evaluate", "hash", writtenAt + 29_000)).toEqual({ a: 1 }) + expect(await backend.get("qe-evaluate", "hash", writtenAt + 31_000)).toBeUndefined() + }) + + it("keeps a long entry alive to its full TTL", async () => { + const backend = makeKvBackend(makeFakeKv()) + const writtenAt = 1_000_000 + + await backend.put("org-clickhouse-config", "hash", null, 3600, writtenAt) + + expect(await backend.get("org-clickhouse-config", "hash", writtenAt + 3_000_000)).toBeNull() + expect(await backend.get("org-clickhouse-config", "hash", writtenAt + 3_601_000)).toBeUndefined() + }) }) diff --git a/apps/api/src/platform/CacheBackendLive.ts b/apps/api/src/platform/CacheBackendLive.ts index 3b1dc4366..8131399ea 100644 --- a/apps/api/src/platform/CacheBackendLive.ts +++ b/apps/api/src/platform/CacheBackendLive.ts @@ -89,27 +89,41 @@ const KV_MIN_TTL_SECONDS = 60 * that long after an explicit `delete` — and invalidation-on-mutation is the * property the org config cache is relying on. * - * Values are stored wrapped as `{ v }`. `EdgeCacheBackend` uses `undefined` to - * mean "not cached", and a cached `null` is a legitimate value — for the org - * config bucket it is the COMMON one ("this org is managed, use Tinybird"), and - * caching it is the single biggest win there. KV's `get` returns `null` for a - * missing key, so storing a bare `null` would make every managed org a - * permanent miss. The envelope keeps absence and null distinguishable. + * Values are stored wrapped as `{ v, exp }`, which the other two backends do not + * need: + * + * - `v` because `EdgeCacheBackend` uses `undefined` to mean "not cached" while a + * cached `null` is a legitimate value — for the org config bucket it is the + * COMMON one ("this org is managed, use Tinybird"), and caching it is the + * single biggest win there. KV's `get` returns `null` for a missing key, so + * storing a bare `null` would make every managed org a permanent miss. + * - `exp` because KV refuses an `expirationTtl` under 60s, and real buckets ask + * for less (`qe-evaluate` and the integrations routes use 30s). The other + * backends carry their own deadline (`Cache-Control: max-age`, `expiresAt`), + * so without this a 30s entry would be served for 60s. `expirationTtl` still + * goes out at the 60s floor to bound storage; `exp` is what expires the read. */ export const makeKvBackend = (kv: KvLike): EdgeCacheBackend => { const composite = (bucket: string, hash: string) => `${bucket}:${hash}` return { name: "workers-kv", - get: async (bucket, hash) => { + get: async (bucket, hash, nowMs) => { const envelope = await kv.get(composite(bucket, hash), "json") if (envelope === null || typeof envelope !== "object" || !("v" in envelope)) return undefined - return (envelope as { v: unknown }).v + const { v, exp } = envelope as { v: unknown; exp?: number } + // KV cannot expire an entry sooner than 60s, so short-TTL buckets carry + // their real deadline in the envelope and expire on read. + if (typeof exp === "number" && nowMs >= exp) return undefined + return v }, - put: async (bucket, hash, value, ttlSeconds) => { - await kv.put(composite(bucket, hash), JSON.stringify({ v: value }), { - expirationTtl: Math.max(KV_MIN_TTL_SECONDS, Math.floor(ttlSeconds)), - }) + put: async (bucket, hash, value, ttlSeconds, nowMs) => { + const ttl = Math.floor(ttlSeconds) + await kv.put( + composite(bucket, hash), + JSON.stringify({ v: value, exp: nowMs + ttl * 1_000 }), + { expirationTtl: Math.max(KV_MIN_TTL_SECONDS, ttl) }, + ) }, delete: async (bucket, hash) => { await kv.delete(composite(bucket, hash)) diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts index e7521600d..8e42d02c8 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts @@ -7,7 +7,13 @@ import { OrgId, RoleName, } from "@maple/domain/http" -import { EdgeCacheService, MemoryCacheBackendLive } from "@maple/cache" +import { + type EdgeCacheBackend, + EdgeCacheService, + makeEdgeCacheService, + makeMemoryBackend, + MemoryCacheBackendLive, +} from "@maple/cache" import { Cause, ConfigProvider, Effect, Exit, Layer, Option, Schema } from "effect" import { TestClock } from "effect/testing" import { FetchHttpClient } from "effect/unstable/http" @@ -521,6 +527,43 @@ describe("resolveRuntimeConfig caching", () => { }).pipe(Effect.provide(buildLayer(testDb))) }) + // `apps/alerting` runs this same service but has no EDGE_CACHE binding, so its + // CacheBackend is `caches.default` — a store that no `invalidateOrgRuntimeConfig` + // ever reaches, since the writers all live in apps/api and bust KV. An + // hour-long entry there would let alert evaluation keep using a warehouse + // config the customer already rotated or deleted. + it("skips the durable tier on a backend the writers' invalidation cannot reach", () => { + const testDb = createTestDb(cacheTrackedDbs) + const orgId = "org_ch_durable_uninvalidated" + const workersCacheBackend: EdgeCacheBackend = { ...makeMemoryBackend(), name: "workers-cache" } + const layer = Layer.mergeAll( + OrgClickHouseSettingsService.layer.pipe( + Layer.provide(Layer.mergeAll(Env.layer.pipe(Layer.provide(configLive)), testDb.layer)), + ), + Layer.succeed(EdgeCacheService, makeEdgeCacheService(workersCacheBackend)), + ) + + return Effect.gen(function* () { + yield* Effect.promise(() => seedRow(testDb, orgId, "https://a.example")) + yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + + yield* Effect.promise(() => + executeSql(testDb, "UPDATE org_clickhouse_settings SET ch_url = $2 WHERE org_id = $1", [ + orgId, + "https://b.example", + ]), + ) + + // Cold isolate: memo gone, nothing else should be answering. + invalidateOrgRuntimeConfigMemo(orgId) + + const resolved = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + // Reads through to Postgres. The KV-backed test above sees the OLD url + // here — that divergence is the whole point. + expect(expectSome(resolved).url).toBe("https://b.example") + }).pipe(Effect.provide(layer)) + }) + // The reason the durable tier exists: a cold isolate has no memo entry at all, // and in prod that path measured 10–13k resolutions/day at a p50 of ~500ms. it.effect("a cold isolate is served by the durable tier, not Postgres", () => { diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.ts index 2bfc75765..0ac1c2020 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.ts @@ -30,7 +30,9 @@ import { import { orgClickHouseSchemaApplyRuns, orgClickHouseSettings } from "@maple/db" import { eq } from "drizzle-orm" import { WorkerEnvironment } from "@maple/effect-cloudflare/worker-environment" -import { EdgeCacheService } from "@maple/cache" +import { type EdgeCacheBackend, EdgeCacheService } from "@maple/cache" + +type EdgeCacheBackendName = EdgeCacheBackend["name"] import { Clock, Context, @@ -222,6 +224,28 @@ export const ORG_CH_CONFIG_BUCKET = "org-clickhouse-config" const ORG_CH_CONFIG_EDGE_TTL_SECONDS = 3_600 const ORG_CH_CONFIG_EDGE_READ_TIMEOUT_MS = 150 +/** + * Whether this host's cache store is one the writers' invalidation actually + * reaches. An hour-long entry is only safe under that condition. + * + * The org config is mutated exclusively from `apps/api`, which owns the + * `EDGE_CACHE` KV binding — so `"workers-kv"` is the store those busts land in, + * and `"memory"` is per-isolate and therefore self-consistent by construction. + * + * `"workers-cache"` is neither. `apps/alerting` runs this same service against + * `caches.default` (it has no KV binding), which no `invalidateOrgRuntimeConfig` + * ever touches — so an hour-long entry there would let alert evaluation keep + * using a warehouse config the customer already rotated or deleted, firing on + * stale data or missing incidents. Falling back to the memo + Postgres is the + * correct behaviour for such a host: it is slower, and it is right. + * + * The proper fix is to bind the same KV namespace into the alerting stack (it + * already binds Hyperdrive by ID across stacks, so there is precedent); until + * then the tier stays off there rather than silently serving stale routing. + */ +const durableTierIsInvalidated = (backendName: EdgeCacheBackendName): boolean => + backendName !== "workers-cache" + const toCachedChSettings = (row: CachedSettingsRow): CachedChSettings => ({ schemaVersion: row.schemaVersion, chUrl: row.chUrl, @@ -1373,7 +1397,9 @@ export class OrgClickHouseSettingsService extends Context.Service< */ const lookupCachedSettings = Effect.fnUntraced(function* (orgId: OrgId) { const edgeCache = yield* Effect.serviceOption(EdgeCacheService) - if (Option.isNone(edgeCache)) return yield* readSettingsFromPostgres(orgId) + if (Option.isNone(edgeCache) || !durableTierIsInvalidated(edgeCache.value.backendName)) { + return yield* readSettingsFromPostgres(orgId) + } const result = yield* edgeCache.value.getOrCompute( { diff --git a/lib/cache/src/edge-cache.ts b/lib/cache/src/edge-cache.ts index 59c647916..dddf2d195 100644 --- a/lib/cache/src/edge-cache.ts +++ b/lib/cache/src/edge-cache.ts @@ -66,6 +66,17 @@ export interface EdgeCacheReadResult { } export interface EdgeCacheServiceShape { + /** + * Which store is behind this service, mirroring `cache.backend` on the spans. + * + * Exposed because it decides whether a bucket may be cached ACROSS isolates at + * all. `"memory"` is per-isolate, so a bust always reaches the reader. + * `"workers-kv"` is shared, and only the worker that owns the binding writes + * it. `"workers-cache"` is shared but colo-local and busted per-isolate, so a + * long-lived entry there can outlive an invalidation issued elsewhere — fine + * for query results keyed by their own inputs, wrong for mutable config. + */ + readonly backendName: EdgeCacheBackend["name"] readonly getOrCompute: ( options: EdgeCacheGetOrComputeOptions, compute: Effect.Effect, @@ -434,7 +445,14 @@ export const makeEdgeCacheService = ( }) }) - return { getOrCompute, invalidate, rawGetDetailed, rawGet, rawPut } satisfies EdgeCacheServiceShape + return { + backendName: backend.name, + getOrCompute, + invalidate, + rawGetDetailed, + rawGet, + rawPut, + } satisfies EdgeCacheServiceShape } export class EdgeCacheService extends Context.Service()(