From e05f569dbdeb52dc4eb5c52bf398bf2750d32bac Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 8 Oct 2026 16:42:27 +0200 Subject: [PATCH 1/3] perf(api): share organization region reads across isolates ensureServedHere called Clerk on every fresh isolate, about 170 ms of every cold request. Region answers now go through the Workers cache as well as the isolate memo, so a new isolate in the same data center reuses the answer instead of calling Clerk. Each entry keeps the time Clerk was read and every tier measures age from it, so the 60 s (chosen) and 5 s (still open) bounds hold end to end. Only a yes is stored and Clerk failures are never stored. chooseRegion drops the cached answer after it writes the region. --- .../org/OrganizationRegionService.test.ts | 164 ++++++++++++ .../services/org/OrganizationRegionService.ts | 253 +++++++++++++----- .../src/services/org/OrganizationService.ts | 8 +- 3 files changed, 350 insertions(+), 75 deletions(-) create mode 100644 packages/backend/src/services/org/OrganizationRegionService.test.ts diff --git a/packages/backend/src/services/org/OrganizationRegionService.test.ts b/packages/backend/src/services/org/OrganizationRegionService.test.ts new file mode 100644 index 0000000000..c342b88ffc --- /dev/null +++ b/packages/backend/src/services/org/OrganizationRegionService.test.ts @@ -0,0 +1,164 @@ +import { assert, describe, it } from "@effect/vitest" +import { + type EdgeCacheBackend, + EdgeCacheService, + makeEdgeCacheService, + makeMemoryBackend, +} from "@maple/cache" +import { OrganizationWrongRegionError, OrgId } from "@maple/domain/http" +import type { MapleRegion } from "@maple/domain/organization-regions" +import { Effect, Exit, Option, Ref, Schema } from "effect" +import { TestClock } from "effect/testing" +import { ClerkRequestError } from "@maple/backend/services/auth/clerk-request" +import { + type DirectoryOrganization, + makeOrganizationRegionService, + type ReadDirectoryOrganization, +} from "./OrganizationRegionService" + +const orgId = Schema.decodeUnknownSync(OrgId)("org_region_cache") + +const chosen = (region: MapleRegion): DirectoryOrganization => ({ + publicMetadata: { regions: [region] }, + createdAt: 0, +}) +/** Created at the test clock's start with no region yet, so still inside the choice window. */ +const open: DirectoryOrganization = { publicMetadata: {}, createdAt: 0 } + +/** A stand-in Clerk whose answer can change, counting reads. */ +const makeDirectory = Effect.fnUntraced(function* (initial: Option.Option) { + const answer = yield* Ref.make(initial) + const reads = yield* Ref.make(0) + const read = (_: OrgId) => + Ref.update(reads, (n) => n + 1).pipe( + Effect.andThen(Ref.get(answer)), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail( + new ClerkRequestError({ + operation: "Clerk.organizations.getOrganization", + message: "Clerk is down", + cause: undefined, + }), + ), + onSome: Effect.succeed, + }), + ), + ) + return { answer, reads, read } +}) + +/** A fresh isolate: its own memory, the data center's shared cache. */ +const isolate = (backend: EdgeCacheBackend, directory: { readonly read: ReadDirectoryOrganization }) => + makeOrganizationRegionService("us", Option.some(directory.read)).pipe( + Effect.provideService(EdgeCacheService, makeEdgeCacheService(backend)), + ) + +const isWrongRegion = (exit: Exit.Exit): boolean => + Exit.isFailure(exit) && + Option.exists(Exit.findErrorOption(exit), (error) => error instanceof OrganizationWrongRegionError) + +describe("OrganizationRegionService", () => { + it.effect("a miss reads Clerk once, then the isolate answers", () => + Effect.gen(function* () { + const directory = yield* makeDirectory(Option.some(chosen("us"))) + const regions = yield* isolate(makeMemoryBackend(), directory) + yield* regions.ensureServedHere(orgId) + yield* regions.ensureServedHere(orgId) + assert.strictEqual(yield* Ref.get(directory.reads), 1) + }), + ) + + it.effect("a fresh isolate reuses another isolate's answer from the shared cache", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(chosen("us"))) + yield* (yield* isolate(backend, directory)).ensureServedHere(orgId) + yield* TestClock.adjust("30 seconds") + yield* (yield* isolate(backend, directory)).ensureServedHere(orgId) + assert.strictEqual(yield* Ref.get(directory.reads), 1) + }), + ) + + it.effect("a chosen region expires 60 s after Clerk was read, in every tier", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(chosen("us"))) + yield* (yield* isolate(backend, directory)).ensureServedHere(orgId) + yield* TestClock.adjust("30 seconds") + const second = yield* isolate(backend, directory) + yield* second.ensureServedHere(orgId) + // The second isolate got the answer at 30 s, but it was read at 0 s. + yield* TestClock.adjust("30 seconds") + yield* second.ensureServedHere(orgId) + assert.strictEqual(yield* Ref.get(directory.reads), 2) + }), + ) + + it.effect("an open region expires after 5 s, so a choice of EU lands quickly", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(open)) + const regions = yield* isolate(backend, directory) + yield* regions.ensureServedHere(orgId) + yield* Ref.set(directory.answer, Option.some(chosen("eu"))) + yield* TestClock.adjust("4 seconds") + yield* regions.ensureServedHere(orgId) + yield* TestClock.adjust("1 second") + assert.isTrue( + isWrongRegion( + yield* Effect.exit((yield* isolate(backend, directory)).ensureServedHere(orgId)), + ), + ) + assert.isTrue(isWrongRegion(yield* Effect.exit(regions.ensureServedHere(orgId)))) + }), + ) + + it.effect("forgetting drops the answer from both tiers", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(open)) + const regions = yield* isolate(backend, directory) + yield* regions.ensureServedHere(orgId) + yield* Ref.set(directory.answer, Option.some(chosen("eu"))) + yield* regions.forget(orgId) + assert.isTrue(isWrongRegion(yield* Effect.exit(regions.ensureServedHere(orgId)))) + assert.isTrue( + isWrongRegion( + yield* Effect.exit((yield* isolate(backend, directory)).ensureServedHere(orgId)), + ), + ) + }), + ) + + it.effect("a Clerk failure serves the request and is not cached as a region", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.none()) + const regions = yield* isolate(backend, directory) + yield* regions.ensureServedHere(orgId) + assert.deepStrictEqual(yield* regions.servedHere(orgId), Option.none()) + yield* Ref.set(directory.answer, Option.some(chosen("eu"))) + assert.isTrue(isWrongRegion(yield* Effect.exit(regions.ensureServedHere(orgId)))) + assert.isTrue( + isWrongRegion( + yield* Effect.exit((yield* isolate(backend, directory)).ensureServedHere(orgId)), + ), + ) + }), + ) + + it.effect("a refusal is not cached, so a region just chosen here is served at once", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(chosen("eu"))) + const regions = yield* isolate(backend, directory) + assert.isTrue(isWrongRegion(yield* Effect.exit(regions.ensureServedHere(orgId)))) + yield* Ref.set(directory.answer, Option.some(chosen("us"))) + yield* regions.ensureServedHere(orgId) + yield* (yield* isolate(backend, directory)).ensureServedHere(orgId) + assert.strictEqual(yield* Ref.get(directory.reads), 2) + }), + ) +}) diff --git a/packages/backend/src/services/org/OrganizationRegionService.ts b/packages/backend/src/services/org/OrganizationRegionService.ts index 903a923ccf..7ce72222a9 100644 --- a/packages/backend/src/services/org/OrganizationRegionService.ts +++ b/packages/backend/src/services/org/OrganizationRegionService.ts @@ -9,6 +9,7 @@ * not what keeps data in a region. That is why a Clerk read that fails lets the request through. */ import { createClerkClient } from "@clerk/backend" +import { EdgeCacheService } from "@maple/cache" import { OrganizationWrongRegionError, type OrgId } from "@maple/domain/http" import { type MapleRegion, @@ -18,9 +19,9 @@ import { organizationRegionsFrom, organizationServedIn, } from "@maple/domain/organization-regions" -import { Clock, Context, Effect, Layer, Option, Redacted } from "effect" +import { Clock, Context, Effect, HashMap, Layer, Option, Redacted, Ref, Schema } from "effect" import { Env } from "@maple/backend/platform/Env" -import { clerkRequest } from "@maple/backend/services/auth/clerk-request" +import { type ClerkRequestError, clerkRequest } from "@maple/backend/services/auth/clerk-request" /** * A region that can no longer change: chosen, or past the choice window (`REGION_CHOICE_WINDOW_MS`, @@ -33,6 +34,48 @@ const REGIONS_TTL_MS = 60_000 */ const OPEN_REGION_TTL_MS = 5_000 +/** + * The Workers cache tier, shared by every isolate in a data center. A fresh isolate reads it in a + * few ms instead of paying a ~180 ms Clerk call. Same TTLs as the isolate tier. + */ +const REGIONS_CACHE_BUCKET = "org-regions" + +/** One Clerk answer. Every tier measures its age from `readAtMs`, so reuse never adds up across tiers. */ +const RegionRead = Schema.Struct({ + metadata: Schema.Unknown, + createdAtMs: Schema.Number, + readAtMs: Schema.Number, +}) +type RegionRead = typeof RegionRead.Type + +/** Keeps a read that refuses this instance out of the shared tier. Recovered where it is raised. */ +class RegionReadNotShared extends Schema.TaggedError()( + "@maple/backend/services/org/RegionReadNotShared", + { message: Schema.String, read: RegionRead }, +) {} + +const regionTtlMs = (entry: RegionRead, nowMs: number): number => + organizationRegionOpen(entry.metadata, entry.createdAtMs, nowMs) ? OPEN_REGION_TTL_MS : REGIONS_TTL_MS + +/** Where a region answer came from, for the `cache.layer` span attribute. */ +interface RegionAnswer { + readonly entry: RegionRead + readonly layer: "isolate" | "edge" | "origin" +} + +const isFresh = (entry: RegionRead, nowMs: number): boolean => + nowMs - entry.readAtMs < regionTtlMs(entry, nowMs) + +/** The organization fields a region decision needs, as the directory returns them. */ +export interface DirectoryOrganization { + readonly publicMetadata: unknown + readonly createdAt: number +} + +export type ReadDirectoryOrganization = ( + orgId: OrgId, +) => Effect.Effect + export interface OrganizationRegionServiceApi { /** The instance's own region. */ readonly region: MapleRegion @@ -44,88 +87,149 @@ export interface OrganizationRegionServiceApi { * rather than be served on a guess. */ readonly servedHere: (orgId: OrgId) => Effect.Effect> + /** Drops what this instance remembers about the organization, after its region was written. */ + readonly forget: (orgId: OrgId) => Effect.Effect } +/** + * Builds the service over a directory read, or none for a single instance without Clerk. + * Exported so tests can stand in for Clerk. + */ +export const makeOrganizationRegionService = Effect.fnUntraced(function* ( + region: MapleRegion, + readOrganization: Option.Option, +) { + const edgeCache = yield* EdgeCacheService + const remembered = yield* Ref.make(HashMap.empty()) + + const readDirectory = Effect.fnUntraced(function* (read: ReadDirectoryOrganization, orgId: OrgId) { + const readAtMs = yield* Clock.currentTimeMillis + const organization = yield* read(orgId) + return { metadata: organization.publicMetadata, createdAtMs: organization.createdAt, readAtMs } + }) + + // Only a yes is shared. An organization whose region was just chosen in onboarding arrives + // here straight away, and a cached no would refuse it. + const readShared = (read: ReadDirectoryOrganization, orgId: OrgId, nowMs: number) => + edgeCache + .getOrCompute( + { + bucket: REGIONS_CACHE_BUCKET, + key: orgId, + schema: RegionRead, + ttlSeconds: (entry) => Math.ceil(regionTtlMs(entry, entry.readAtMs) / 1000), + }, + readDirectory(read, orgId).pipe( + Effect.filterOrFail( + (entry) => organizationServedIn(entry.metadata, region), + (entry) => + new RegionReadNotShared({ + message: "Organization is not served here", + read: entry, + }), + ), + ), + ) + .pipe( + Effect.flatMap(({ value, hit }): Effect.Effect => { + if (!hit) return Effect.succeed({ entry: value, layer: "origin" }) + if (isFresh(value, nowMs)) return Effect.succeed({ entry: value, layer: "edge" }) + // Written once Clerk answered, so the entry can outlive its TTL by that call. + return Effect.map(readDirectory(read, orgId), (entry) => ({ entry, layer: "origin" })) + }), + Effect.catchTag( + "@maple/backend/services/org/RegionReadNotShared", + (notShared): Effect.Effect => + Effect.succeed({ entry: notShared.read, layer: "origin" }), + ), + ) + + const read = Effect.fn("OrganizationRegionService.read")(function* (orgId: OrgId) { + if (Option.isNone(readOrganization)) return Option.none() + const nowMs = yield* Clock.currentTimeMillis + const local = Option.filter(HashMap.get(yield* Ref.get(remembered), orgId), (entry) => + isFresh(entry, nowMs), + ) + if (Option.isSome(local)) { + yield* Effect.annotateCurrentSpan({ "cache.hit": true, "cache.layer": "isolate" }) + return Option.some(local.value.metadata) + } + return yield* readShared(readOrganization.value, orgId, nowMs).pipe( + Effect.tap(({ entry, layer }) => + Effect.andThen( + Effect.annotateCurrentSpan({ "cache.hit": layer === "edge", "cache.layer": layer }), + organizationServedIn(entry.metadata, region) + ? Ref.update(remembered, HashMap.set(orgId, entry)) + : Effect.void, + ), + ), + Effect.map(({ entry }) => Option.some(entry.metadata)), + Effect.catch((error) => + Effect.logWarning("Could not read organization regions; serving the request").pipe( + Effect.annotateLogs({ orgId, error: error.message }), + Effect.as(Option.none()), + ), + ), + ) + }) + + const ensureServedHere: OrganizationRegionServiceApi["ensureServedHere"] = Effect.fn( + "OrganizationRegionService.ensureServedHere", + )(function* (orgId) { + const metadata = yield* read(orgId) + if (Option.isNone(metadata)) return + if (organizationRegionsFrom(metadata.value).includes(region)) return + const orgRegion = organizationHomeRegion(metadata.value) + yield* Effect.annotateCurrentSpan({ "maple.org_region": orgRegion, "maple.region": region }) + return yield* new OrganizationWrongRegionError({ + message: `This organization lives in the ${MAPLE_REGION_LABELS[orgRegion].short} region.`, + orgId, + orgRegion, + region, + }) + }) + + const servedHere: OrganizationRegionServiceApi["servedHere"] = Effect.fn( + "OrganizationRegionService.servedHere", + )(function* (orgId) { + // No directory means one instance, which serves everything. + if (Option.isNone(readOrganization)) return Option.some(true) + return Option.map(yield* read(orgId), (metadata) => organizationHomeRegion(metadata) === region) + }) + + const forget: OrganizationRegionServiceApi["forget"] = Effect.fn("OrganizationRegionService.forget")( + function* (orgId) { + yield* Ref.update(remembered, HashMap.remove(orgId)) + yield* edgeCache.invalidate({ bucket: REGIONS_CACHE_BUCKET, key: orgId }) + }, + ) + + return { region, ensureServedHere, servedHere, forget } satisfies OrganizationRegionServiceApi +}) + export class OrganizationRegionService extends Context.Service< OrganizationRegionService, OrganizationRegionServiceApi >()("@maple/backend/services/org/OrganizationRegionService", { make: Effect.gen(function* () { const env = yield* Env - const region = env.MAPLE_REGION const clerk = env.MAPLE_AUTH_MODE.toLowerCase() === "clerk" - ? Option.match(env.CLERK_SECRET_KEY, { - onNone: () => undefined, - onSome: (secretKey) => createClerkClient({ secretKey: Redacted.value(secretKey) }), - }) - : undefined - - const cache = new Map< - OrgId, - { readonly metadata: unknown; readonly createdAtMs: number; readonly atMs: number } - >() - - const read = Effect.fn("OrganizationRegionService.read")(function* (orgId: OrgId) { - if (clerk === undefined) return Option.none() - const nowMs = yield* Clock.currentTimeMillis - const cached = cache.get(orgId) - const ttlMs = - cached !== undefined && !organizationRegionOpen(cached.metadata, cached.createdAtMs, nowMs) - ? REGIONS_TTL_MS - : OPEN_REGION_TTL_MS - if (cached !== undefined && nowMs - cached.atMs < ttlMs) { - return Option.some(cached.metadata) - } - return yield* clerkRequest("Clerk.organizations.getOrganization", { orgId }, () => - clerk.organizations.getOrganization({ organizationId: orgId }), - ).pipe( - Effect.map((organization) => { - // Only a yes is cached. An organization whose region was just chosen in onboarding - // arrives here straight away, and a cached no would refuse it for a minute. - if (organizationServedIn(organization.publicMetadata, region)) { - cache.set(orgId, { - metadata: organization.publicMetadata, - createdAtMs: organization.createdAt, - atMs: nowMs, - }) - } - return Option.some(organization.publicMetadata) - }), - Effect.catch((error) => - Effect.logWarning("Could not read organization regions; serving the request").pipe( - Effect.annotateLogs({ orgId, error: error.message }), - Effect.as(Option.none()), - ), - ), - ) - }) - - const ensureServedHere: OrganizationRegionServiceApi["ensureServedHere"] = Effect.fn( - "OrganizationRegionService.ensureServedHere", - )(function* (orgId) { - const metadata = yield* read(orgId) - if (Option.isNone(metadata)) return - if (organizationRegionsFrom(metadata.value).includes(region)) return - const orgRegion = organizationHomeRegion(metadata.value) - yield* Effect.annotateCurrentSpan({ "maple.org_region": orgRegion, "maple.region": region }) - return yield* new OrganizationWrongRegionError({ - message: `This organization lives in the ${MAPLE_REGION_LABELS[orgRegion].short} region.`, - orgId, - orgRegion, - region, - }) - }) - - const servedHere: OrganizationRegionServiceApi["servedHere"] = Effect.fn( - "OrganizationRegionService.servedHere", - )(function* (orgId) { - // No directory means one instance, which serves everything. - if (clerk === undefined) return Option.some(true) - return Option.map(yield* read(orgId), (metadata) => organizationHomeRegion(metadata) === region) - }) - - return { region, ensureServedHere, servedHere } satisfies OrganizationRegionServiceApi + ? Option.map(env.CLERK_SECRET_KEY, (secretKey) => + createClerkClient({ secretKey: Redacted.value(secretKey) }), + ) + : Option.none() + return yield* makeOrganizationRegionService( + env.MAPLE_REGION, + Option.map( + clerk, + (client): ReadDirectoryOrganization => + (orgId) => + clerkRequest("Clerk.organizations.getOrganization", { orgId }, () => + client.organizations.getOrganization({ organizationId: orgId }), + ), + ), + ) }), }) { static readonly layer = Layer.effect(this, this.make) @@ -135,5 +239,6 @@ export class OrganizationRegionService extends Context.Service< region: "us", ensureServedHere: () => Effect.void, servedHere: () => Effect.succeed(Option.some(true)), + forget: () => Effect.void, }) } diff --git a/packages/backend/src/services/org/OrganizationService.ts b/packages/backend/src/services/org/OrganizationService.ts index 89cd7bc8f5..03cb0a824d 100644 --- a/packages/backend/src/services/org/OrganizationService.ts +++ b/packages/backend/src/services/org/OrganizationService.ts @@ -72,6 +72,7 @@ import { Env } from "@maple/backend/platform/Env" import { clerkRequest } from "@maple/backend/services/auth/clerk-request" import { AutumnClient } from "@maple/backend/services/billing/autumn-http" import { responseHasPlanHistory } from "@maple/backend/services/billing/autumn-client" +import { OrganizationRegionService } from "@maple/backend/services/org/OrganizationRegionService" const ROOT_ROLE = Schema.decodeSync(RoleName)("root") const ORG_ADMIN_ROLE = Schema.decodeSync(RoleName)("org:admin") @@ -258,6 +259,7 @@ export class OrganizationService extends Context.Service, @@ -427,6 +429,8 @@ export class OrganizationService extends Context.Service toProviderError(error.cause))) + // A yes this instance cached before the write may now be wrong. + yield* regions.forget(orgId) // Clerk has no conditional write, so two admins choosing at once both succeed and the // later write wins. Answering with a fresh read rather than the request sends both to // the region that stuck in all but a same-instant race. The write already landed, so a @@ -466,7 +470,9 @@ export class OrganizationService extends Context.Service this.use((service) => service.retrieve(orgId)) From 1270c6bc0c1ac146ff14331e9860d71fba6770ef Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 8 Oct 2026 16:55:54 +0200 Subject: [PATCH 2/3] chore(api): measure the first region cache read on a fresh isolate Temporary. Lifts the read deadline for the org-regions bucket so the preview records how long the first Workers cache read in an isolate really takes. --- packages/backend/src/services/org/OrganizationRegionService.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/backend/src/services/org/OrganizationRegionService.ts b/packages/backend/src/services/org/OrganizationRegionService.ts index 7ce72222a9..5a4e2c171c 100644 --- a/packages/backend/src/services/org/OrganizationRegionService.ts +++ b/packages/backend/src/services/org/OrganizationRegionService.ts @@ -118,6 +118,7 @@ export const makeOrganizationRegionService = Effect.fnUntraced(function* ( key: orgId, schema: RegionRead, ttlSeconds: (entry) => Math.ceil(regionTtlMs(entry, entry.readAtMs) / 1000), + readTimeoutMs: 1_000, }, readDirectory(read, orgId).pipe( Effect.filterOrFail( From d9c2d99f1ca3d49fdcc176a94f91481be58113a0 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Thu, 8 Oct 2026 17:09:42 +0200 Subject: [PATCH 3/3] perf(api): race the shared region read against Clerk on fresh isolates The preview showed that a fresh isolate's first Workers cache read takes 90 to 340 ms (p10 to p90), against 13 ms warm. A 40 ms deadline threw every cold hit away, and a long one let slow reads cost more than Clerk. The shared read now runs alone for 40 ms. If it has not answered, Clerk is asked as well and the first answer wins. A late hit still beats a slower Clerk call, and a hung read can no longer cost more than Clerk. EdgeCacheService gains a per-call read deadline on rawGetDetailed and a rawDelete for raw-keyed entries. --- lib/cache/src/edge-cache.ts | 17 ++- .../auth/MembershipRevocationService.test.ts | 1 + .../org/OrganizationRegionService.test.ts | 34 ++++- .../services/org/OrganizationRegionService.ts | 123 ++++++++++++------ .../warehouse/QueryEngineCacheHitRate.test.ts | 1 + .../QueryEngineEvaluateCache.test.ts | 1 + 6 files changed, 134 insertions(+), 43 deletions(-) diff --git a/lib/cache/src/edge-cache.ts b/lib/cache/src/edge-cache.ts index 99bca9c524..11f18d885d 100644 --- a/lib/cache/src/edge-cache.ts +++ b/lib/cache/src/edge-cache.ts @@ -101,9 +101,11 @@ export interface EdgeCacheServiceApi { * failure is logged and swallowed (the entry simply expires via its TTL). */ readonly invalidate: (options: EdgeCacheInvalidateOptions) => Effect.Effect + /** `readTimeoutMs` overrides the service's read deadline, as on `getOrCompute`. */ readonly rawGetDetailed: ( bucket: string, key: string, + options?: { readonly readTimeoutMs?: number }, ) => Effect.Effect, EdgeCacheIOError> readonly rawGet: (bucket: string, key: string) => Effect.Effect, EdgeCacheIOError> readonly rawPut: ( @@ -112,6 +114,8 @@ export interface EdgeCacheServiceApi { value: unknown, ttlSeconds: number, ) => Effect.Effect + /** Evict a `rawPut` entry by the same raw key. */ + readonly rawDelete: (bucket: string, key: string) => Effect.Effect } const sha256Hex = async (input: string): Promise => { @@ -449,7 +453,11 @@ export const makeEdgeCacheService = ( ) }) - const rawGetDetailed = Effect.fn("EdgeCache.rawGetDetailed")(function* (bucket: string, key: string) { + const rawGetDetailed = Effect.fn("EdgeCache.rawGetDetailed")(function* ( + bucket: string, + key: string, + options?: { readonly readTimeoutMs?: number }, + ) { yield* Effect.annotateCurrentSpan({ "cache.bucket": bucket, "cache.hit": false, @@ -461,7 +469,7 @@ export const makeEdgeCacheService = ( const skipRead = shouldSkipRead(bucket, nowMs) const read = skipRead ? { value: undefined, timedOut: false as const } - : yield* readBackend(bucket, key, nowMs, boundedReadTimeoutMs).pipe( + : yield* readBackend(bucket, key, nowMs, resolveReadTimeoutMs(options?.readTimeoutMs)).pipe( Effect.mapError(toIOError(bucket, key)), ) if (!skipRead) recordReadOutcome(bucket, read.timedOut, nowMs) @@ -500,12 +508,17 @@ export const makeEdgeCacheService = ( ) }) + const rawDelete = Effect.fn("EdgeCache.rawDelete")(function* (bucket: string, key: string) { + return yield* backend.delete(bucket, key).pipe(Effect.mapError(toIOError(bucket, key))) + }) + return { getOrCompute, invalidate, rawGetDetailed, rawGet, rawPut, + rawDelete, } satisfies EdgeCacheServiceApi } diff --git a/packages/backend/src/services/auth/MembershipRevocationService.test.ts b/packages/backend/src/services/auth/MembershipRevocationService.test.ts index acadec16ec..4a3833a1c3 100644 --- a/packages/backend/src/services/auth/MembershipRevocationService.test.ts +++ b/packages/backend/src/services/auth/MembershipRevocationService.test.ts @@ -43,6 +43,7 @@ const recordingEdgeCache = () => { rawGetDetailed: () => Effect.succeed({ status: "miss", value: Option.none(), readMs: 0 }), rawGet: () => Effect.succeed(Option.none()), rawPut: () => Effect.void, + rawDelete: () => Effect.void, } const layer = Layer.succeed(EdgeCacheService, api) return { invalidated, layer } diff --git a/packages/backend/src/services/org/OrganizationRegionService.test.ts b/packages/backend/src/services/org/OrganizationRegionService.test.ts index c342b88ffc..746f293554 100644 --- a/packages/backend/src/services/org/OrganizationRegionService.test.ts +++ b/packages/backend/src/services/org/OrganizationRegionService.test.ts @@ -7,7 +7,7 @@ import { } from "@maple/cache" import { OrganizationWrongRegionError, OrgId } from "@maple/domain/http" import type { MapleRegion } from "@maple/domain/organization-regions" -import { Effect, Exit, Option, Ref, Schema } from "effect" +import { Effect, Exit, Fiber, Option, Ref, Schema } from "effect" import { TestClock } from "effect/testing" import { ClerkRequestError } from "@maple/backend/services/auth/clerk-request" import { @@ -161,4 +161,36 @@ describe("OrganizationRegionService", () => { assert.strictEqual(yield* Ref.get(directory.reads), 2) }), ) + + it.effect("a shared read that hangs hands over to Clerk after 40 ms", () => + Effect.gen(function* () { + const hung: EdgeCacheBackend = { ...makeMemoryBackend(), get: () => Effect.never } + const directory = yield* makeDirectory(Option.some(chosen("eu"))) + const pending = yield* Effect.forkChild( + Effect.exit((yield* isolate(hung, directory)).ensureServedHere(orgId)), + ) + yield* TestClock.adjust("40 millis") + assert.isTrue(isWrongRegion(yield* Fiber.join(pending))) + }), + ) + + it.effect("a slow shared answer still wins over a slower Clerk", () => + Effect.gen(function* () { + const backend = makeMemoryBackend() + const directory = yield* makeDirectory(Option.some(chosen("us"))) + yield* (yield* isolate(backend, directory)).ensureServedHere(orgId) + // Clerk now says EU and takes 500 ms, so only the shared answer can serve the request. + yield* Ref.set(directory.answer, Option.some(chosen("eu"))) + const slowClerk = { read: (id: OrgId) => Effect.delay(directory.read(id), "500 millis") } + const slowCache: EdgeCacheBackend = { + ...backend, + get: (...args) => Effect.delay(backend.get(...args), "100 millis"), + } + const pending = yield* Effect.forkChild( + Effect.exit((yield* isolate(slowCache, slowClerk)).ensureServedHere(orgId)), + ) + yield* TestClock.adjust("1 second") + assert.isTrue(Exit.isSuccess(yield* Fiber.join(pending))) + }), + ) }) diff --git a/packages/backend/src/services/org/OrganizationRegionService.ts b/packages/backend/src/services/org/OrganizationRegionService.ts index 5a4e2c171c..5a4641a1bf 100644 --- a/packages/backend/src/services/org/OrganizationRegionService.ts +++ b/packages/backend/src/services/org/OrganizationRegionService.ts @@ -19,7 +19,7 @@ import { organizationRegionsFrom, organizationServedIn, } from "@maple/domain/organization-regions" -import { Clock, Context, Effect, HashMap, Layer, Option, Redacted, Ref, Schema } from "effect" +import { Clock, Context, Effect, Fiber, HashMap, Layer, Option, Redacted, Ref, Schema } from "effect" import { Env } from "@maple/backend/platform/Env" import { type ClerkRequestError, clerkRequest } from "@maple/backend/services/auth/clerk-request" @@ -35,10 +35,17 @@ const REGIONS_TTL_MS = 60_000 const OPEN_REGION_TTL_MS = 5_000 /** - * The Workers cache tier, shared by every isolate in a data center. A fresh isolate reads it in a - * few ms instead of paying a ~180 ms Clerk call. Same TTLs as the isolate tier. + * The Workers cache tier, shared by every isolate in a data center, so a fresh isolate can skip + * the Clerk call. Holds only answers that let this instance serve. Same TTLs as the isolate tier. */ const REGIONS_CACHE_BUCKET = "org-regions" +/** + * How long the shared read runs alone before Clerk is asked too. Warm reads answer in ~13 ms. A + * fresh isolate's first read took 90 to 340 ms (p10 to p90) on a preview, so there both race. + */ +const CLERK_HEDGE_AFTER_MS = 40 +/** Frees a hung shared read. Clerk is already answering by then. */ +const SHARED_READ_DEADLINE_MS = 1_000 /** One Clerk answer. Every tier measures its age from `readAtMs`, so reuse never adds up across tiers. */ const RegionRead = Schema.Struct({ @@ -48,10 +55,12 @@ const RegionRead = Schema.Struct({ }) type RegionRead = typeof RegionRead.Type -/** Keeps a read that refuses this instance out of the shared tier. Recovered where it is raised. */ -class RegionReadNotShared extends Schema.TaggedError()( - "@maple/backend/services/org/RegionReadNotShared", - { message: Schema.String, read: RegionRead }, +const decodeRegionRead = Schema.decodeUnknownOption(RegionRead) + +/** The shared tier had no usable answer, so the race against Clerk goes to Clerk. */ +class RegionNotShared extends Schema.TaggedError()( + "@maple/backend/services/org/RegionNotShared", + { message: Schema.String }, ) {} const regionTtlMs = (entry: RegionRead, nowMs: number): number => @@ -63,6 +72,8 @@ interface RegionAnswer { readonly layer: "isolate" | "edge" | "origin" } +const answer = (entry: RegionRead, layer: RegionAnswer["layer"]): RegionAnswer => ({ entry, layer }) + const isFresh = (entry: RegionRead, nowMs: number): boolean => nowMs - entry.readAtMs < regionTtlMs(entry, nowMs) @@ -108,43 +119,67 @@ export const makeOrganizationRegionService = Effect.fnUntraced(function* ( return { metadata: organization.publicMetadata, createdAtMs: organization.createdAt, readAtMs } }) - // Only a yes is shared. An organization whose region was just chosen in onboarding arrives - // here straight away, and a cached no would refuse it. - const readShared = (read: ReadDirectoryOrganization, orgId: OrgId, nowMs: number) => + const readEdge = (orgId: OrgId, nowMs: number) => edgeCache - .getOrCompute( - { - bucket: REGIONS_CACHE_BUCKET, - key: orgId, - schema: RegionRead, - ttlSeconds: (entry) => Math.ceil(regionTtlMs(entry, entry.readAtMs) / 1000), - readTimeoutMs: 1_000, - }, - readDirectory(read, orgId).pipe( - Effect.filterOrFail( - (entry) => organizationServedIn(entry.metadata, region), - (entry) => - new RegionReadNotShared({ - message: "Organization is not served here", - read: entry, - }), - ), - ), - ) + .rawGetDetailed(REGIONS_CACHE_BUCKET, orgId, { readTimeoutMs: SHARED_READ_DEADLINE_MS }) .pipe( - Effect.flatMap(({ value, hit }): Effect.Effect => { - if (!hit) return Effect.succeed({ entry: value, layer: "origin" }) - if (isFresh(value, nowMs)) return Effect.succeed({ entry: value, layer: "edge" }) - // Written once Clerk answered, so the entry can outlive its TTL by that call. - return Effect.map(readDirectory(read, orgId), (entry) => ({ entry, layer: "origin" })) - }), - Effect.catchTag( - "@maple/backend/services/org/RegionReadNotShared", - (notShared): Effect.Effect => - Effect.succeed({ entry: notShared.read, layer: "origin" }), + Effect.map(({ value }) => + Option.filter(Option.flatMap(value, decodeRegionRead), (entry) => isFresh(entry, nowMs)), ), + // A failed read is a miss. Clerk still answers. + Effect.orElseSucceed(() => Option.none()), ) + // Only a yes is shared. An organization whose region was just chosen in onboarding arrives + // here straight away, and a cached no would refuse it. + const readOrigin = (read: ReadDirectoryOrganization, orgId: OrgId) => + readDirectory(read, orgId).pipe( + Effect.tap((entry) => + organizationServedIn(entry.metadata, region) + ? edgeCache + .rawPut( + REGIONS_CACHE_BUCKET, + orgId, + entry, + Math.ceil(regionTtlMs(entry, entry.readAtMs) / 1000), + ) + .pipe( + Effect.catch((error) => + Effect.logWarning("Could not share the organization's regions").pipe( + Effect.annotateLogs({ orgId, error: error.message }), + ), + ), + ) + : Effect.void, + ), + Effect.map((entry) => answer(entry, "origin")), + ) + + const readShared = Effect.fnUntraced(function* ( + read: ReadDirectoryOrganization, + orgId: OrgId, + nowMs: number, + ) { + const shared = yield* Effect.forkChild(readEdge(orgId, nowMs)) + const early = yield* Fiber.join(shared).pipe(Effect.timeoutOption(CLERK_HEDGE_AFTER_MS)) + if (Option.isSome(early)) { + return Option.isSome(early.value) + ? answer(early.value.value, "edge") + : yield* readOrigin(read, orgId) + } + // Usually a fresh isolate's first read. Ask Clerk as well and take whichever answers first. + yield* Effect.annotateCurrentSpan("cache.hedged", true) + const lateHit = Fiber.join(shared).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.fail(new RegionNotShared({ message: "No shared region answer" })), + onSome: (entry) => Effect.succeed(answer(entry, "edge")), + }), + ), + ) + return yield* Effect.race(lateHit, readOrigin(read, orgId)) + }) + const read = Effect.fn("OrganizationRegionService.read")(function* (orgId: OrgId) { if (Option.isNone(readOrganization)) return Option.none() const nowMs = yield* Clock.currentTimeMillis @@ -201,7 +236,15 @@ export const makeOrganizationRegionService = Effect.fnUntraced(function* ( const forget: OrganizationRegionServiceApi["forget"] = Effect.fn("OrganizationRegionService.forget")( function* (orgId) { yield* Ref.update(remembered, HashMap.remove(orgId)) - yield* edgeCache.invalidate({ bucket: REGIONS_CACHE_BUCKET, key: orgId }) + yield* edgeCache + .rawDelete(REGIONS_CACHE_BUCKET, orgId) + .pipe( + Effect.catch((error) => + Effect.logWarning( + "Could not drop the shared organization regions; they expire on their TTL", + ).pipe(Effect.annotateLogs({ orgId, error: error.message })), + ), + ) }, ) diff --git a/packages/backend/src/services/warehouse/QueryEngineCacheHitRate.test.ts b/packages/backend/src/services/warehouse/QueryEngineCacheHitRate.test.ts index a0528b4c2f..ed901b7d6d 100644 --- a/packages/backend/src/services/warehouse/QueryEngineCacheHitRate.test.ts +++ b/packages/backend/src/services/warehouse/QueryEngineCacheHitRate.test.ts @@ -68,6 +68,7 @@ const makeObservedEdgeCache = () => { }, rawGet: inner.rawGet, rawPut: inner.rawPut, + rawDelete: inner.rawDelete, } return { layer: Layer.succeed(EdgeCacheService, observed), diff --git a/packages/backend/src/services/warehouse/QueryEngineEvaluateCache.test.ts b/packages/backend/src/services/warehouse/QueryEngineEvaluateCache.test.ts index d92395bc0b..e44b6052e3 100644 --- a/packages/backend/src/services/warehouse/QueryEngineEvaluateCache.test.ts +++ b/packages/backend/src/services/warehouse/QueryEngineEvaluateCache.test.ts @@ -312,6 +312,7 @@ const makeRecordingEdge = ( rawGetDetailed: () => Effect.succeed({ status: "miss" as const, value: Option.none(), readMs: 0 }), rawGet: () => Effect.succeed(Option.none()), rawPut: () => Effect.void, + rawDelete: () => Effect.void, }) describe("QueryEngineService.cachedDirect TTL", () => {