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..fc1d9ef00 --- /dev/null +++ b/apps/api/src/platform/CacheBackendLive.test.ts @@ -0,0 +1,101 @@ +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; 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({ + ...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) + }) + + // ...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 5055372af..8131399ea 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,101 @@ 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, 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, nowMs) => { + const envelope = await kv.get(composite(bucket, hash), "json") + if (envelope === null || typeof envelope !== "object" || !("v" in envelope)) return undefined + 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, 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)) + }, + } +} + 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 +155,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..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" @@ -18,7 +24,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 +368,50 @@ 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 }), + ) + + /** + * 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) @@ -380,17 +428,17 @@ 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) 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))) }) @@ -408,6 +456,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 +467,130 @@ 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") + + expect(yield* resolveUntilUrl(orgId, "https://b.example")).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))) + }) + + // `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", () => { + 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,11 +623,10 @@ 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) - 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/src/services/org/OrgClickHouseSettingsService.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.ts index d391ed15f..0ac1c2020 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.ts @@ -30,6 +30,9 @@ import { import { orgClickHouseSchemaApplyRuns, orgClickHouseSettings } from "@maple/db" import { eq } from "drizzle-orm" import { WorkerEnvironment } from "@maple/effect-cloudflare/worker-environment" +import { type EdgeCacheBackend, EdgeCacheService } from "@maple/cache" + +type EdgeCacheBackendName = EdgeCacheBackend["name"] import { Clock, Context, @@ -102,17 +105,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 +149,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 +207,45 @@ 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 + +/** + * 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, @@ -875,10 +965,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 +1367,54 @@ 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) || !durableTierIsInvalidated(edgeCache.value.backendName)) { + 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/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/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, 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()( 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)), ) }