diff --git a/apps/api/alchemy.run.ts b/apps/api/alchemy.run.ts index 57ff12665..35cfaaadd 100644 --- a/apps/api/alchemy.run.ts +++ b/apps/api/alchemy.run.ts @@ -99,13 +99,6 @@ 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 @@ -203,7 +196,6 @@ 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 deleted file mode 100644 index fc1d9ef00..000000000 --- a/apps/api/src/platform/CacheBackendLive.test.ts +++ /dev/null @@ -1,101 +0,0 @@ -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 8131399ea..6a996d6c7 100644 --- a/apps/api/src/platform/CacheBackendLive.ts +++ b/apps/api/src/platform/CacheBackendLive.ts @@ -1,6 +1,5 @@ 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" @@ -46,100 +45,16 @@ 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`. + * Workers KV was trialled here as a second backend (#387, 2026-08-10) on the + * theory that a KV `get` is a cancellable subrequest and so cheaper to abandon + * than an uncancellable `cache.match()`. Prod measurement refuted it and it was + * removed — see the note above `resolveCachedSettings` in + * `OrgClickHouseSettingsService.ts` for the numbers. Don't reach for it again + * without re-reading them. */ -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 @@ -155,4 +70,4 @@ export const CacheBackendLive = Layer.effect( } return CacheBackend.of(makeWorkersBackend(cache)) }), -).pipe(Layer.provide(Layer.mergeAll(WorkersCache.layer, WorkerEnvironment.layer))) +).pipe(Layer.provide(WorkersCache.layer)) diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts index 8e42d02c8..e2d446493 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.test.ts @@ -7,13 +7,6 @@ import { OrgId, RoleName, } from "@maple/domain/http" -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" @@ -24,10 +17,8 @@ import { cleanupTestDbs, createTestDb, executeSql, type TestDb } from "@/platfor import { type ClickHouseExecConfig, execClickHouse, - invalidateOrgRuntimeConfig, invalidateOrgRuntimeConfigMemo, isRetryableUpstream, - ORG_CH_CONFIG_BUCKET, OrgClickHouseSettingsService, shouldHealSchemaVersion, validateClickHouseCredentialTransport, @@ -286,16 +277,9 @@ describe("resolveRuntimeConfig caching", () => { }), ) - // EdgeCacheService is merged into the RUN context (not just the build) so the - // call-time `Effect.serviceOption(EdgeCacheService)` inside resolveRuntimeConfig - // resolves it — mirroring prod, where MainLive provides it at top level. const buildLayer = (testDb: TestDb) => { const envLive = Env.layer.pipe(Layer.provide(configLive)) - const edgeCacheLive = EdgeCacheService.layer.pipe(Layer.provide(MemoryCacheBackendLive)) - const orgSettingsLive = OrgClickHouseSettingsService.layer.pipe( - Layer.provide(Layer.mergeAll(envLive, testDb.layer)), - ) - return Layer.mergeAll(orgSettingsLive, edgeCacheLive) + return OrgClickHouseSettingsService.layer.pipe(Layer.provide(Layer.mergeAll(envLive, testDb.layer))) } const seedRow = (db: TestDb, orgId: string, chUrl: string) => @@ -374,19 +358,10 @@ describe("resolveRuntimeConfig caching", () => { const SOFT_TTL_MS = 300_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 }), - ) + // (did the caller block? did the forked refresh land?) — it bypasses the + // service, so it never busts the memo, which is what keeps the memo the thing + // under test. /** * Resolve until the forked background refresh has landed. @@ -428,8 +403,6 @@ 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) @@ -456,8 +429,6 @@ 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. @@ -485,8 +456,6 @@ describe("resolveRuntimeConfig caching", () => { ]), ) - 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 @@ -497,11 +466,10 @@ describe("resolveRuntimeConfig caching", () => { }).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", () => { + // The memo is the only tier, so a memo bust IS a complete invalidation. This + // was not true while a durable tier sat underneath it (#387, removed) — the + // read fell through to a shared entry that outlived every isolate. + it.effect("a memo bust is a complete invalidation", () => { const testDb = createTestDb(cacheTrackedDbs) const orgId = "org_ch_memo_module_bust" return Effect.gen(function* () { @@ -517,32 +485,18 @@ describe("resolveRuntimeConfig caching", () => { ) 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") + const afterBust = yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) + expect(expectSome(afterBust).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", () => { + // A cold isolate has no memo entry and must read Postgres. This is the cost + // two separate shared-cache attempts tried and failed to remove (see the note + // above `resolveCachedSettings`); the fix is to warm the config once before a + // fan-out, not to add a third tier. + it.effect("a cold isolate reads through to Postgres", () => { 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)), - ) - + const orgId = "org_ch_cold_isolate" return Effect.gen(function* () { yield* Effect.promise(() => seedRow(testDb, orgId, "https://a.example")) yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) @@ -554,40 +508,11 @@ describe("resolveRuntimeConfig caching", () => { ]), ) - // Cold isolate: memo gone, nothing else should be answering. + // What a fresh isolate looks like: no in-process state at all. 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))) }) @@ -623,7 +548,7 @@ describe("resolveRuntimeConfig caching", () => { "https://c.example", ]), ) - yield* evictDurable(orgId) + yield* TestClock.adjust(SOFT_TTL_MS + 1_000) yield* OrgClickHouseSettingsService.resolveRuntimeConfig(asOrgId(orgId)) expect(yield* resolveUntilUrl(orgId, "https://c.example")).toBe("https://c.example") diff --git a/apps/api/src/services/org/OrgClickHouseSettingsService.ts b/apps/api/src/services/org/OrgClickHouseSettingsService.ts index 0ac1c2020..3f361cfba 100644 --- a/apps/api/src/services/org/OrgClickHouseSettingsService.ts +++ b/apps/api/src/services/org/OrgClickHouseSettingsService.ts @@ -30,9 +30,6 @@ 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, @@ -167,28 +164,6 @@ export const invalidateOrgRuntimeConfigMemo = (orgId: string): void => { 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 @@ -207,45 +182,6 @@ 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, @@ -965,7 +901,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 - yield* invalidateOrgRuntimeConfig(orgId) + invalidateOrgRuntimeConfigMemo(orgId) return hadOverride }) @@ -1372,49 +1308,6 @@ export class OrgClickHouseSettingsService extends Context.Service< 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, @@ -1453,7 +1346,7 @@ export class OrgClickHouseSettingsService extends Context.Service< if (startedAt !== undefined && nowMs - startedAt < REFRESH_MARKER_STALE_MS) return false refreshInFlight.set(orgId, nowMs) - const work = lookupCachedSettings(orgId).pipe( + const work = readSettingsFromPostgres(orgId).pipe( Effect.flatMap((value) => Clock.currentTimeMillis.pipe( Effect.map((writeNowMs) => storeCachedSettings(orgId, value, writeNowMs)), @@ -1499,6 +1392,26 @@ export class OrgClickHouseSettingsService extends Context.Service< // abandoned read kept holding one of the Worker's six connection slots and // the fallback Postgres read queued behind it. The layer meant to avoid a // ~26ms round-trip was manufacturing a ~2.5s one 82% of the time. + // + // A SECOND attempt (#387) put the same tier on Workers KV instead, on the + // theory that a KV `get` is a cancellable subrequest and so cheap to + // abandon. Measured over 24h on the live deploy it was worse, and removed: + // KV reads that COMPLETE take 92ms (vs 6ms on the Cache API) and 79% of them + // still hit their deadline. It also could not reach the cost it was aimed + // at. 94% of the Postgres fallback is `apps/alerting` (4,646 resolutions/day + // at p50 573ms, ~44min of blocked wall time) which has no KV binding and so + // could never use the tier; `apps/api`, which had it, falls back at p50 24ms + // — cheaper than a KV read. + // + // The reason both attempts failed is that this is not a cold-isolate miss. + // Grouping alerting's Postgres resolutions by trace: 1,033 traces do zero, + // while 106 traces do 22 EACH (half of all of them). It is an in-request + // fan-out where every branch misses the memo because none has finished + // writing it yet. No shared cache can fix concurrent siblings — it just + // turns N Postgres reads into N cache reads contending for the same six + // connection slots. The fix is to resolve the config once BEFORE the + // fan-out, the way `warehouse.warmRoute` already does in + // `routes/v1/query-engine.http.ts`. const resolveCachedSettings = Effect.fn("OrgClickHouseSettingsService.resolveCachedSettings")( function* (orgId: OrgId) { const nowMs = yield* Clock.currentTimeMillis @@ -1530,7 +1443,7 @@ export class OrgClickHouseSettingsService extends Context.Service< "clickhouse.config.source": "postgres", "clickhouse.config.memoHit": false, }) - const cached = yield* lookupCachedSettings(orgId) + const cached = yield* readSettingsFromPostgres(orgId) storeCachedSettings(orgId, cached, nowMs) return cached }, diff --git a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts index c2d584d32..e41176aaa 100644 --- a/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts +++ b/apps/api/src/workflows/ClickHouseSchemaApplyWorkflow.run.ts @@ -29,34 +29,23 @@ import { type DesiredTable, } from "@maple/domain/clickhouse" import { eq } from "drizzle-orm" -import { Effect, Layer } from "effect" -import { EdgeCacheService } from "@maple/cache" -import { CacheBackendLive } from "@/platform/CacheBackendLive" +import { Effect } from "effect" import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { makeTracedPgConnection, type TracedPgConnection } from "@/platform/pg-execute" -import { invalidateOrgRuntimeConfig } from "@/services/org/OrgClickHouseSettingsService" +import { invalidateOrgRuntimeConfigMemo } from "@/services/org/OrgClickHouseSettingsService" /** - * Bust the org's cached runtime config after this workflow writes to - * `org_clickhouse_settings`. + * Bust this isolate's cached runtime config after the workflow writes to + * `org_clickhouse_settings` (it stamps `schema_version`, part of the cached + * projection). * - * 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. + * Isolate-local, and that is the whole story now that the durable tier is gone: + * the workflow runs in its own isolate, so API isolates converge on the new + * `schema_version` at the memo's soft TTL (`ORG_CH_CONFIG_MEMO_TTL_MS`, 300s) + * exactly as they do for every other writer. This call keeps the invariant + * "every writer of the row busts its own memo" true at every write site. */ -const bustRuntimeConfigCache = (orgId: string): Promise => - Effect.runPromise( - invalidateOrgRuntimeConfig(orgId).pipe( - Effect.provide(EdgeCacheService.layer.pipe(Layer.provide(CacheBackendLive))), - ), - ).catch(() => undefined) +const bustRuntimeConfigCache = (orgId: string): void => invalidateOrgRuntimeConfigMemo(orgId) /** * This workflow runs outside the worker's layer graph, so it owns its telemetry @@ -496,7 +485,7 @@ async function runWithDb( }) .where(eq(orgClickHouseSettings.orgId, orgId)), ) - await bustRuntimeConfigCache(orgId) + bustRuntimeConfigCache(orgId) }) for (const migration of clickHouseMigrations) { @@ -655,7 +644,7 @@ async function runWithDb( .set({ syncStatus: "error", lastSyncError: message, updatedAt: new Date(finishedAt) }) .where(eq(orgClickHouseSettings.orgId, orgId)), ).catch(() => undefined) - await bustRuntimeConfigCache(orgId) + bustRuntimeConfigCache(orgId) throw error } } diff --git a/apps/api/wrangler.jsonc b/apps/api/wrangler.jsonc index e721c241a..9c67db46a 100644 --- a/apps/api/wrangler.jsonc +++ b/apps/api/wrangler.jsonc @@ -65,13 +65,6 @@ "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 77dc67445..4cfbbdb24 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" | "workers-kv" | "memory" + readonly name: "workers-cache" | "memory" readonly get: (bucket: string, hash: string, nowMs: number) => Promise readonly put: ( bucket: string, @@ -28,7 +28,7 @@ export interface EdgeCacheBackend { } /** - * Injected edge-cache storage backend (Workers KV in prod, in-memory in + * Injected edge-cache storage backend (`caches.default` in prod, in-memory in * tests/dev). * * The tag string still names the old home. Tags are identity, not diff --git a/lib/cache/src/edge-cache.ts b/lib/cache/src/edge-cache.ts index dddf2d195..c391b5f63 100644 --- a/lib/cache/src/edge-cache.ts +++ b/lib/cache/src/edge-cache.ts @@ -66,17 +66,6 @@ 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, @@ -446,7 +435,6 @@ export const makeEdgeCacheService = ( }) return { - backendName: backend.name, getOrCompute, invalidate, rawGetDetailed,