From 2ef7161a4ee2de2d0da8fd681dfda0122d393deb Mon Sep 17 00:00:00 2001 From: utpal singh Date: Sun, 11 Oct 2026 06:48:02 +0530 Subject: [PATCH] fix: reconcile cache capacity recovery with current v2 Preserve limit diagnostics, retention clamping and combined catalog reads. Retain transactional eviction and lease-safe rollback. Bound optional catalog metadata by the complete manifest size. --- .changeset/cache-capacity-recovery.md | 6 + e2e/test-plan.ts | 60 +++ e2e/tests/app-cache-capacity.spec.ts | 401 ++++++++++++++++++ packages/app-cache/src/contracts/cache.ts | 6 +- packages/app-cache/src/index.ts | 7 +- packages/app-cache/src/sqlite.ts | 54 ++- packages/apps/package.json | 2 +- .../apps/src/implementation/catalog-cache.ts | 84 +++- 8 files changed, 593 insertions(+), 27 deletions(-) create mode 100644 .changeset/cache-capacity-recovery.md create mode 100644 e2e/tests/app-cache-capacity.spec.ts diff --git a/.changeset/cache-capacity-recovery.md b/.changeset/cache-capacity-recovery.md new file mode 100644 index 0000000000..fb31b3bb96 --- /dev/null +++ b/.changeset/cache-capacity-recovery.md @@ -0,0 +1,6 @@ +--- +"@executor-js/app-cache": patch +"apps": patch +--- + +Include the safe reason in CacheError messages and preserve it when reading through the host transport. Reclaim old, unleased cache entries at aggregate capacity without evicting the current write, and reload catalogs whose cached parts have been evicted. diff --git a/e2e/test-plan.ts b/e2e/test-plan.ts index 83b6c91ca2..951addc593 100644 --- a/e2e/test-plan.ts +++ b/e2e/test-plan.ts @@ -806,6 +806,66 @@ export const scenarios = plan({ local: na("Hosted profile API fixture; Node adapter exercised by self-host"), }, }, + appCacheEntryCapacity: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App cache reclaims old entries at the entry count limit", + targets: { + "self-host": scheduled, + cloud: sourceStorageCloud, + local: na("Hosted API fixture; the cache implementation is shared with Local"), + }, + }, + appCacheCatalogCapacity: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App catalogs recover when pressure evicts parts of a retained manifest", + targets: { + "self-host": scheduled, + cloud: scheduled, + local: na("Hosted API fixture; the catalog cache implementation is shared with Local"), + }, + }, + appCacheManifestCapacity: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App catalogs omit optional names when the full manifest exceeds the entry limit", + targets: { + "self-host": scheduled, + cloud: scheduled, + local: na("Hosted API fixture; the catalog cache implementation is shared with Local"), + }, + }, + appCacheProtectedCapacity: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App cache rolls back pressure when every retained value has a live loader", + targets: { + "self-host": scheduled, + cloud: scheduled, + local: na("Hosted API fixture; the cache implementation is shared with Local"), + }, + }, + appCacheCapacity: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App cache reclaims old values at capacity and keeps valid calls working", + targets: { + "self-host": scheduled, + cloud: scheduled, + local: na("Hosted API fixture; the cache implementation is shared with Local"), + }, + }, + appCacheReasons: { + fixtures: "actors", + file: "app-cache-capacity.spec.ts", + title: "App cache errors expose safe reasons to authored callers", + targets: { + "self-host": scheduled, + cloud: scheduled, + local: na("Hosted API fixture; the cache implementation is shared with Local"), + }, + }, appCacheEvaluation: { fixtures: "actors", file: "app-cache-evaluation.spec.ts", diff --git a/e2e/tests/app-cache-capacity.spec.ts b/e2e/tests/app-cache-capacity.spec.ts new file mode 100644 index 0000000000..f643c6f6e5 --- /dev/null +++ b/e2e/tests/app-cache-capacity.spec.ts @@ -0,0 +1,401 @@ +/** Cache pressure and diagnostics through an authored app's real host transport. */ +import { expect, layer } from "@effect/vitest"; +import { Deferred, Effect, Fiber, Layer, Schema } from "effect"; +import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; +import { HttpRouter, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/http"; +import { createServer } from "node:http"; +import { randomUUID } from "node:crypto"; +import { Actors } from "../support/actors.ts"; +import { Api, body } from "../support/api.ts"; +import { App } from "../support/contracts.ts"; +import { saveAndDeploy } from "../support/app-authoring.ts"; +import { HostedLive, withHostedCase } from "../support/case.ts"; +import { appsManifest, mcpSdkVersion, withApps } from "../support/apps-release.ts"; +import { scenarios } from "../test-plan.ts"; + +const fixture = ({ upstream, hold }: { readonly upstream?: string; readonly hold?: string } = {}) => + Effect.gen(function* () { + const actors = yield* Actors; + const api = yield* Api; + const prefix = `/api/organizations/${actors.organization.id}/apps`; + const files = [ + { + path: "index.ts", + content: `import { CacheError, defineApp, query, object, string, number, boolean, router } from "apps"; +${upstream === undefined ? "" : 'import { mcpRouter } from "apps/mcp";'} +export default defineApp({ accounts: {} }, async ({ cache }) => ({ tools: router({ + ${upstream === undefined ? "" : `upstream: await mcpRouter({ url: ${JSON.stringify(upstream)}, cache }),`} + diagnostic: query({ input: object({ read: number() }) }, async (_, { read }) => { + try { + if (read === 1) await cache.readMany(Array.from({ length: 129 }, (_, i) => i), string()); + else if (read === 2) await cache.write([{ key: "SYNTHETIC_PRIVATE_CACHE_KEY" + "x".repeat(9000), value: true }], "1 minute"); + else if (read === 3) await cache.write(Array.from({ length: 4 }, (_, i) => ({ key: "batch-" + i, value: "x".repeat(1999990) })), "1 minute"); + else if (read === 4) await cache.write([{ key: "atomic", value: "kept" }], -1); + else if (read === 5) await cache.write(Array.from({ length: 129 }, (_, i) => ({ key: "count-" + i, value: true })), "1 minute"); + else await cache.write([{ key: "atomic", value: "kept" }, { key: "SYNTHETIC_PRIVATE_CACHE_KEY", value: "SYNTHETIC_PRIVATE_CACHE_VALUE" + "x".repeat(2000000) }], "1 minute"); + return null; + } catch (error) { + if (!(error instanceof CacheError)) throw error; + return { reason: error.reason, message: error.message, limit: error.limit ?? null, atomic: await cache.read("atomic", string()) === undefined }; + } + }), + held: query({ input: object({}) }, async () => cache.get({ key: "held", schema: string(), freshFor: "1 minute", load: async ({ cache, signal }) => { + await cache.write([{ key: "held-started", value: true }], "1 day"); + while (!await cache.read("held-release", boolean())) { + if (signal.aborted) throw new Error("Cancelled"); + await new Promise(resolve => setTimeout(resolve, 25)); + } + return "held"; + } })), + heldStarted: query({ input: object({}) }, async () => await cache.read("held-started", boolean()) ?? false), + heldRelease: query({ input: object({}) }, async () => { await cache.write([{ key: "held-release", value: true }], "1 minute"); return true; }), + ${ + hold === undefined + ? "" + : `holdValues: query({ input: object({}) }, async () => { + await Promise.all(Array.from({ length: 67 }, (_, i) => cache.revalidate({ key: "pressure-" + i, schema: string(), freshFor: "1 day", load: async ({ signal }) => { + const response = await fetch(${JSON.stringify(hold)}, { signal }); + if (!response.ok) throw new Error("Loader hold failed"); + return "refreshed"; + } }).then(() => true))); + return true; + }),` + } + blockedWrite: query({ input: object({}) }, async () => { + try { + await cache.write([{ key: "pressure-66", value: "r".repeat(1900000) }, { key: "blocked-write", value: "x".repeat(1900000) }], "1 day"); + return null; + } catch (error) { + if (!(error instanceof CacheError)) throw error; + return { reason: error.reason, message: error.message, limit: error.limit ?? null, absent: await cache.read("blocked-write", string()) === undefined, kept: (await cache.read("pressure-66", string()))?.startsWith("x") ?? false }; + } + }), + countFill: query({ input: object({ start: number(), count: number() }) }, async (_, { start, count }) => { + for (let i = 0; i < count; i += 128) await cache.write(Array.from({ length: Math.min(128, count - i) }, (_, j) => ({ key: "entry-" + (start + i + j), value: "kept" })), "1 day"); + return true; + }), + fill: query({ input: object({ batch: number() }) }, async (_, { batch }) => { + await cache.write(Array.from({ length: 4 }, (_, i) => ({ key: "pressure-" + (batch * 4 + i), value: "x".repeat(1900000) })), "1 day"); + return true; + }), + size: query({ input: object({ key: string() }) }, async (_, { key }) => (await cache.read(key, string()))?.length ?? null), + replace: query({ input: object({ key: string(), size: number() }) }, async (_, { key, size }) => { + await cache.write([{ key, value: "x".repeat(size) }], "1 day"); + return true; + }), + load: query({ input: object({}) }, async () => cache.get({ key: "loaded", schema: string(), freshFor: "1 minute", load: async () => "loaded" })), +}) }));`, + }, + upstream === undefined + ? appsManifest + : { + path: "package.json", + content: JSON.stringify({ + dependencies: withApps({ "@modelcontextprotocol/sdk": mcpSdkVersion }), + }), + }, + ]; + const deployed = yield* api.request(actors.owner, "POST", `${prefix}/deploy`, { + name: `Cache capacity ${randomUUID().slice(0, 8)}`, + files, + }); + expect(deployed.status).toBe(200); + const app = (yield* body(App, deployed)).id; + const path = `${prefix}/${app}`; + yield* Effect.addFinalizer(() => api.request(actors.owner, "DELETE", path).pipe(Effect.orDie)); + const request = (tool: string, input: Schema.Json = {}) => + api.request(actors.owner, "POST", `${path}/tools/call`, { tool, kind: "query", input }); + const call = (tool: string, input: Schema.Json = {}) => + Effect.gen(function* () { + const response = yield* request(tool, input); + expect(response.status, `${tool} ${JSON.stringify(input)}`).toBe(200); + return yield* body(Schema.Json, response); + }); + const redeploy = () => + saveAndDeploy(actors.owner, path, { + files: files.map((file) => + file.path === "index.ts" + ? { + ...file, + content: `${file.content}\n// A distinct build with the same cache keys.\n`, + } + : file, + ), + }); + return { + request, + call, + redeploy, + index: () => api.request(actors.owner, "GET", `${path}/tools/index`), + }; + }); + +const loaderServer = Effect.gen(function* () { + const released = yield* Deferred.make(); + let started = 0; + const handler = Effect.gen(function* () { + started++; + yield* Deferred.await(released); + return HttpServerResponse.empty(); + }); + const services = yield* Layer.build( + HttpRouter.serve(HttpRouter.add("GET", "/hold", handler), { + disableLogger: true, + disableListenLog: true, + }).pipe(Layer.provideMerge(NodeHttpServer.layer(createServer, { host: "127.0.0.1", port: 0 }))), + ); + yield* Effect.addFinalizer(() => Deferred.succeed(released, undefined)); + const server = yield* HttpServer.HttpServer.pipe(Effect.provideContext(services)); + if (!("port" in server.address)) return yield* Effect.die("Expected a TCP listener"); + return { + url: `http://127.0.0.1:${server.address.port}/hold`, + started: () => started, + release: Deferred.succeed(released, undefined), + }; +}); + +const catalogServer = (largeHeader = false) => + Effect.gen(function* () { + let lists = 0; + const Rpc = Schema.Struct({ id: Schema.optional(Schema.Number), method: Schema.String }); + const handler = Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const rpc = yield* request.json.pipe(Effect.flatMap(Schema.decodeUnknownEffect(Rpc))); + if (rpc.id === undefined) return HttpServerResponse.empty({ status: 202 }); + let result: Schema.Json = {}; + switch (rpc.method) { + case "initialize": + result = { + protocolVersion: "2025-03-26", + capabilities: { tools: {} }, + serverInfo: { name: "pressure", version: "1" }, + ...(largeHeader ? { instructions: "x".repeat(1_995_000) } : {}), + }; + break; + case "tools/list": + lists++; + // Full tool parts and pages exceed one reclamation batch, so pressure can remove + // older parts while leaving the later manifest and schema-free summaries intact. + result = { + tools: Array.from({ length: largeHeader ? 1000 : 12 }, (_, i) => ({ + name: `fixture_${i}`, + inputSchema: { + type: "object", + properties: {}, + ...(largeHeader ? {} : { description: "x".repeat(900_000) }), + }, + annotations: { readOnlyHint: true }, + })), + }; + break; + case "tools/call": + result = { content: [{ type: "text", text: "pressure-receipt" }] }; + break; + } + return HttpServerResponse.jsonUnsafe({ jsonrpc: "2.0", id: rpc.id, result }); + }).pipe(Effect.orDie); + const services = yield* Layer.build( + HttpRouter.serve( + Layer.mergeAll( + HttpRouter.add("POST", "/mcp", handler), + HttpRouter.add("GET", "/mcp", Effect.succeed(HttpServerResponse.empty({ status: 405 }))), + HttpRouter.add( + "DELETE", + "/mcp", + Effect.succeed(HttpServerResponse.empty({ status: 204 })), + ), + ), + { disableLogger: true, disableListenLog: true }, + ).pipe( + Layer.provideMerge(NodeHttpServer.layer(createServer, { host: "127.0.0.1", port: 0 })), + ), + ); + const server = yield* HttpServer.HttpServer.pipe(Effect.provideContext(services)); + if (!("port" in server.address)) return yield* Effect.die("Expected a TCP listener"); + return { url: `http://127.0.0.1:${server.address.port}/mcp`, lists: () => lists }; + }); + +layer(HostedLive, { excludeTestServices: true })("App caching", (it) => { + it.effect(scenarios.appCacheReasons.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + const { call } = yield* fixture(); + for (const read of [0, 1, 2, 3, 4, 5]) { + const diagnostic = yield* call("diagnostic", { read }); + const reason = read === 4 ? "invalid" : "capacity"; + const limit = [ + "entryBytes", + "batchEntries", + "keyBytes", + "batchBytes", + null, + "batchEntries", + ][read]; + expect(diagnostic).toEqual({ + reason, + message: `CacheError: ${reason}`, + limit, + atomic: true, + }); + expect(JSON.stringify(diagnostic)).not.toContain("SYNTHETIC_PRIVATE_CACHE"); + } + }), + ), + ); + + it.effect(scenarios.appCacheEntryCapacity.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + const { call, redeploy } = yield* fixture(); + expect(yield* call("countFill", { start: 0, count: 100_000 })).toBe(true); + expect(yield* call("size", { key: "entry-0" })).toBe(4); + expect(yield* call("countFill", { start: 100_000, count: 128 })).toBe(true); + expect(yield* call("size", { key: "entry-0" })).toBeNull(); + expect(yield* call("size", { key: "entry-100127" })).toBe(4); + expect(yield* call("load")).toBe("loaded"); + expect(yield* call("countFill", { start: 100_128, count: 128 })).toBe(true); + expect(yield* call("load")).toBe("loaded"); + const rebuilt = yield* redeploy(); + expect(rebuilt.status, JSON.stringify(rebuilt.body)).toBe(200); + // Builds share the store's limit, but never each other's values. Old-build rows + // must be eligible victims when a new namespace reaches aggregate capacity. + expect(yield* call("size", { key: "entry-100127" })).toBeNull(); + expect(yield* call("countFill", { start: 0, count: 512 })).toBe(true); + expect(yield* call("size", { key: "entry-0" })).toBe(4); + expect(yield* call("size", { key: "entry-511" })).toBe(4); + expect(yield* call("load")).toBe("loaded"); + }), + ), + ); + + it.effect(scenarios.appCacheCatalogCapacity.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + const server = yield* catalogServer(); + const { call, index } = yield* fixture({ upstream: server.url }); + const indexed = yield* index(); + expect(indexed.status, JSON.stringify(indexed.body)).toBe(200); + expect(server.lists()).toBe(1); + for (let batch = 0; batch < 14; batch++) expect(yield* call("fill", { batch })).toBe(true); + // Summaries remain readable, but the first tool's full metadata was older and evicted. + expect((yield* index()).status).toBe(200); + expect(server.lists()).toBe(1); + expect(JSON.stringify(yield* call("upstream.fixture_0"))).toContain("pressure-receipt"); + expect(server.lists()).toBeGreaterThan(1); + expect((yield* index()).status).toBe(200); + expect(JSON.stringify(yield* call("upstream.fixture_0"))).toContain("pressure-receipt"); + }), + ), + ); + + it.effect(scenarios.appCacheManifestCapacity.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + // The instructions fit the entry limit alone. The 1000 short names fit their + // own cap, but adding them would push the complete manifest over 2 MB. + const names = Array.from({ length: 1000 }, (_, i) => `fixture_${i}`); + expect(new TextEncoder().encode(JSON.stringify(names)).byteLength).toBeLessThan(500_000); + expect( + new TextEncoder().encode(JSON.stringify({ instructions: "x".repeat(1_995_000), names })) + .byteLength, + ).toBeGreaterThan(2_000_000); + const server = yield* catalogServer(true); + const { call, request, index } = yield* fixture({ upstream: server.url }); + const indexed = yield* index(); + expect(indexed.status, JSON.stringify(indexed.body)).toBe(200); + expect(server.lists()).toBe(1); + for (const name of ["fixture_0", "fixture_999"]) + expect(JSON.stringify(yield* call(`upstream.${name}`))).toContain("pressure-receipt"); + // With no names index, retained pages must still reject unknown tools without + // rediscovering the catalog or looping through recovery. + for (const round of [1, 2]) { + const unknown = yield* request("upstream.missing_tool").pipe(Effect.timeout("5 seconds")); + expect(unknown.status, `unknown lookup ${round}`).toBe(404); + expect(server.lists()).toBe(1); + } + expect(JSON.stringify(yield* call("upstream.fixture_0"))).toContain("pressure-receipt"); + expect(server.lists()).toBe(1); + }), + ), + ); + + it.effect(scenarios.appCacheProtectedCapacity.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + const server = yield* loaderServer; + const { call, request } = yield* fixture({ hold: server.url }); + for (let batch = 0; batch < 16; batch++) expect(yield* call("fill", { batch })).toBe(true); + for (let key = 64; key < 67; key++) + expect(yield* call("replace", { key: `pressure-${key}`, size: 1_900_000 })).toBe(true); + const holder = yield* request("holdValues").pipe(Effect.forkScoped); + yield* Effect.sync(server.started).pipe( + Effect.repeat({ until: (count) => count === 67 }), + Effect.timeout("10 seconds"), + ); + // Every retained value now has a live lease. A valid batch cannot fit, and its + // replacement must restore both the old value and the loader's publication fence. + expect(yield* call("blockedWrite")).toEqual({ + reason: "capacity", + message: "CacheError: capacity", + limit: "totalBytes", + absent: true, + kept: true, + }); + expect(yield* call("size", { key: "pressure-0" })).toBe(1_900_000); + yield* server.release; + const published = yield* Fiber.join(holder).pipe(Effect.timeout("10 seconds")); + expect(published.status).toBe(200); + expect(yield* body(Schema.Json, published)).toBe(true); + expect(yield* call("size", { key: "pressure-66" })).toBe(9); + expect(yield* call("replace", { key: "blocked-write", size: 1_900_000 })).toBe(true); + expect(yield* call("size", { key: "blocked-write" })).toBe(1_900_000); + expect(yield* call("load")).toBe("loaded"); + }), + ), + ); + + it.effect(scenarios.appCacheCapacity.title, (context) => + withHostedCase( + context, + Effect.gen(function* () { + const { call, request } = yield* fixture(); + const held = yield* request("held").pipe(Effect.forkScoped); + yield* call("heldStarted").pipe( + Effect.repeat({ until: (value) => value === true }), + Effect.timeout("10 seconds"), + ); + // Each batch is below 8 MB and each entry below 2 MB. Retention is still a day away. + for (let batch = 0; batch < 16; batch++) expect(yield* call("fill", { batch })).toBe(true); + expect(yield* call("replace", { key: "pressure-63", size: 1 })).toBe(true); + expect(yield* call("fill", { batch: 16 })).toBe(true); + // Replacements count their byte delta, not the whole new value. + expect(yield* call("size", { key: "pressure-0" })).toBe(1_900_000); + expect(yield* call("replace", { key: "pressure-63", size: 1_900_000 })).toBe(true); + expect(yield* call("size", { key: "pressure-0" })).toBeNull(); + expect(yield* call("size", { key: "pressure-63" })).toBe(1_900_000); + expect(yield* call("heldRelease")).toBe(true); + const published = yield* Fiber.join(held).pipe(Effect.timeout("10 seconds")); + expect(published.status).toBe(200); + expect(yield* body(Schema.Json, published)).toBe("held"); + expect(yield* call("held")).toBe("held"); + expect(yield* call("load")).toBe("loaded"); + for (let batch = 17; batch < 20; batch++) { + expect(yield* call("fill", { batch })).toBe(true); + expect(yield* call("size", { key: `pressure-${batch * 4 + 3}` })).toBe(1_900_000); + expect(yield* call("load")).toBe("loaded"); + } + expect(yield* call("diagnostic", { read: 0 })).toEqual({ + reason: "capacity", + message: "CacheError: capacity", + limit: "entryBytes", + atomic: true, + }); + }), + ), + ); +}); diff --git a/packages/app-cache/src/contracts/cache.ts b/packages/app-cache/src/contracts/cache.ts index 95193d2a75..f78aa4a46c 100644 --- a/packages/app-cache/src/contracts/cache.ts +++ b/packages/app-cache/src/contracts/cache.ts @@ -15,7 +15,11 @@ export class CacheError extends Schema.TaggedError()("CacheError", { "totalEntries", ]), ), -}) {} +}) { + override get message() { + return `CacheError: ${this.reason}`; + } +} /** Bounded defaults shared by loaders and persistent adapters. */ export const cacheLimits = { diff --git a/packages/app-cache/src/index.ts b/packages/app-cache/src/index.ts index 48d8586791..2f6a890ad1 100644 --- a/packages/app-cache/src/index.ts +++ b/packages/app-cache/src/index.ts @@ -77,8 +77,11 @@ export const makeCache = ( const keyOf = (key: Schema.Json) => cacheKey([scope, key]); const readKeys = (keys: readonly string[]) => transport({ operation: "read", keys }).pipe( - Effect.flatMap(Schema.decodeUnknownEffect(Schema.Array(Schema.NullOr(CacheEntry)))), - Effect.mapError(() => new CacheError({ reason: "storage" })), + Effect.flatMap((reply) => + Schema.decodeUnknownEffect(Schema.Array(Schema.NullOr(CacheEntry)))(reply).pipe( + Effect.mapError(() => new CacheError({ reason: "storage" })), + ), + ), ); const durations = (freshFor: Duration.Input, staleFor: Duration.Input = 0) => Effect.try({ diff --git a/packages/app-cache/src/sqlite.ts b/packages/app-cache/src/sqlite.ts index 993776db83..d56b38ff6e 100644 --- a/packages/app-cache/src/sqlite.ts +++ b/packages/app-cache/src/sqlite.ts @@ -38,6 +38,9 @@ const initialize = (storage: CacheSqlStorage) => bytes INTEGER NOT NULL, touched REAL NOT NULL, PRIMARY KEY(namespace, key) )`); storage.sql.exec("CREATE INDEX IF NOT EXISTS executor_cache_expiry ON executor_cache(stale)"); + storage.sql.exec( + "CREATE INDEX IF NOT EXISTS executor_cache_oldest ON executor_cache(touched, namespace, key)", + ); storage.sql.exec(`CREATE TABLE IF NOT EXISTS executor_cache_usage ( id INTEGER PRIMARY KEY CHECK (id = 0), bytes INTEGER NOT NULL, count INTEGER NOT NULL )`); @@ -118,13 +121,50 @@ export const sqliteCache = (storage: CacheSqlStorage) => { now, ); }; - const bound = () => { - const totals = Schema.decodeUnknownSync(Usage)( + const usage = () => + Schema.decodeUnknownSync(Usage)( storage.sql.exec("SELECT bytes, count FROM executor_cache_usage WHERE id = 0").one(), ); - if (totals.bytes > cacheLimits.totalBytes) + const bound = (protectedKeys: readonly string[]) => { + const totals = usage(); + if (totals.bytes <= cacheLimits.totalBytes && totals.count <= cacheLimits.totalEntries) + return; + // Reclaim in write-age order, not LRU: reads do not change touched. Leave room for + // another batch so pressure does not scan victims on every small write. Live leases + // and every entry this command installs are ineligible, even after publish clears its lease. + const bytes = + totals.bytes > cacheLimits.totalBytes + ? totals.bytes - cacheLimits.totalBytes + cacheLimits.batchBytes + : 0; + const count = + totals.count > cacheLimits.totalEntries + ? totals.count - cacheLimits.totalEntries + cacheLimits.batchEntries + : 0; + storage.sql.exec( + `WITH victims AS ( + SELECT rowid, bytes, + sum(bytes) OVER (ORDER BY touched, namespace, key) - bytes AS prior_bytes, + row_number() OVER (ORDER BY touched, namespace, key) - 1 AS prior_count + FROM executor_cache + WHERE lease_until <= ? AND NOT (namespace = ? AND key IN (SELECT value FROM json_each(?))) + ORDER BY touched, namespace, key LIMIT ? + ) + DELETE FROM executor_cache WHERE rowid IN ( + SELECT rowid FROM victims WHERE prior_bytes < ? OR prior_count < ? + )`, + now, + namespace, + JSON.stringify(protectedKeys), + cacheLimits.totalEntries + cacheLimits.batchEntries, + bytes, + count, + ); + const remaining = usage(); + // No safe victim can make this command fit. Roll back its writes and reclamation; + // never remove a live fence or claim that an unretained direct write succeeded. + if (remaining.bytes > cacheLimits.totalBytes) throw new CacheError({ reason: "capacity", limit: "totalBytes" }); - if (totals.count > cacheLimits.totalEntries) + if (remaining.count > cacheLimits.totalEntries) throw new CacheError({ reason: "capacity", limit: "totalEntries" }); }; const claim = (key: string) => { @@ -140,7 +180,7 @@ export const sqliteCache = (storage: CacheSqlStorage) => { now + cacheLimits.leaseMs, now, ); - bound(); + bound([key]); return lease; }; switch (command.operation) { @@ -196,7 +236,7 @@ export const sqliteCache = (storage: CacheSqlStorage) => { const row = read(command.key); if (row?.lease !== command.lease || row.lease_until <= now) return false; write(command.key, command.entry); - bound(); + bound([command.key]); return true; } case "write": { @@ -208,7 +248,7 @@ export const sqliteCache = (storage: CacheSqlStorage) => { ) throw new CacheError({ reason: "capacity", limit: "batchBytes" }); for (const { key, entry } of command.entries) write(key, entry); - bound(); + bound(command.entries.map(({ key }) => key)); return null; } case "invalidate": diff --git a/packages/apps/package.json b/packages/apps/package.json index 963e082731..6bd43a4c00 100644 --- a/packages/apps/package.json +++ b/packages/apps/package.json @@ -1,6 +1,6 @@ { "name": "apps", - "version": "0.0.1-beta.79", + "version": "0.0.1-beta.80", "private": true, "description": "TypeScript apps with typed tools, accounts, storage, and UI.", "type": "module", diff --git a/packages/apps/src/implementation/catalog-cache.ts b/packages/apps/src/implementation/catalog-cache.ts index a1e69797fc..66b085093d 100644 --- a/packages/apps/src/implementation/catalog-cache.ts +++ b/packages/apps/src/implementation/catalog-cache.ts @@ -1,5 +1,6 @@ /** Revisioned catalog storage shared by remote protocols. Executables are never persisted. */ import { CacheError } from "@executor-js/app-cache"; +import { cacheLimits } from "@executor-js/app-cache/contracts"; import { Deferred, Duration, Effect, Exit, Schema } from "effect"; import type { AccountCredential, AppCache, CacheLoadContext } from "../contracts/cache.ts"; import { JsonObject, type JsonValue } from "../contracts/schema.ts"; @@ -52,6 +53,12 @@ export const catalogScope = ( }); }; +/** A disposable part disappeared; transport and decoding failures are not cache misses. */ +class MissingCatalogPart extends Schema.TaggedError()( + "MissingCatalogPart", + {}, +) {} + const schema = (decoder: Schema.Decoder) => wrap(decoder, false); /** @@ -78,6 +85,8 @@ const Manifest = Schema.Struct({ revision: Schema.String, pages: Schema.Int, summaries: Schema.Int, + /** Names distinguish an evicted tool part from an unknown name without another read. */ + names: Schema.optionalKey(Schema.Array(Schema.String)), header: Schema.optionalKey(JsonObject), }); @@ -149,7 +158,7 @@ export const catalogCache = ( ); const values: B[] = []; for (const page of batches.flat()) { - if (page === undefined) return yield* new CacheError({ reason: "unavailable" }); + if (page === undefined) return yield* new MissingCatalogPart(); values.push(...page); } return values; @@ -157,13 +166,14 @@ export const catalogCache = ( /** The catalog a refresh replaces, from the scope that refresh writes. */ const kept = (cache: AppCache) => fromPromise(method(cache, "read"), "cache")(key, schema(Manifest)).pipe( - Effect.map((manifest) => + Effect.flatMap((manifest) => manifest === undefined - ? undefined - : { - header: manifest.header, - tools: pages(cache, manifest.revision, "page", manifest.pages, options.schema), - }, + ? Effect.succeed(undefined) + : pages(cache, manifest.revision, "page", manifest.pages, options.schema).pipe( + Effect.map((tools) => ({ header: manifest.header, tools: Effect.succeed(tools) })), + // A source must not confirm an incomplete cached revision as unchanged. + Effect.catchIf(Schema.is(MissingCatalogPart), () => Effect.succeed(undefined)), + ), ), ); const refresh = (context: CacheLoadContext) => @@ -249,13 +259,21 @@ export const catalogCache = ( yield* pageOut(); yield* summaryOut(); yield* flush(); - // The cache publishes this manifest only after all parts, under its fenced loader lease. - return { + const names = tools.map(({ name }) => name); + const manifest = { revision, pages, summaries, ...(header === undefined ? {} : { header: yield* json(header) }), }; + const candidate = { ...manifest, names }; + // Optional names must fit both their own cap and the complete manifest's entry bound. + // Catalogs without names still resolve missing parts through their retained pages. + const named = + new TextEncoder().encode(JSON.stringify(names)).byteLength <= 500_000 && + new TextEncoder().encode(JSON.stringify(candidate)).byteLength <= cacheLimits.entryBytes; + // The cache publishes this manifest only after all parts, under its fenced loader lease. + return named ? candidate : manifest; }); const getOptions = { key, @@ -331,11 +349,25 @@ export const catalogCache = ( if (cache === undefined) yield* local; else yield* fromPromise(method(cache, "revalidate"), "cache")(getOptions); } + // One uncached load completes this invocation. Invalidate the incomplete manifest so the + // next invocation rebuilds it and hosts forget declarations evaluated from its missing parts. + const recover = () => + local.pipe( + Effect.tap(() => + cache === undefined + ? Effect.void + : fromPromise(method(cache, "invalidate"), "cache")(key), + ), + ); const metadata = () => Effect.gen(function* () { if (cache === undefined) return (yield* local).tools; const manifest = yield* current(); - return yield* pages(cache, manifest.revision, "page", manifest.pages, options.schema); + return yield* pages(cache, manifest.revision, "page", manifest.pages, options.schema).pipe( + Effect.catchIf(Schema.is(MissingCatalogPart), () => + recover().pipe(Effect.map(({ tools }) => tools)), + ), + ); }); return { @@ -356,6 +388,12 @@ export const catalogCache = ( "summary", manifest.summaries, options.summary.schema, + ).pipe( + Effect.catchIf(Schema.is(MissingCatalogPart), () => + recover().pipe( + Effect.map(({ tools }) => tools.map((tool) => options.summary.of(tool))), + ), + ), ); }), resolve: (name: string) => @@ -367,12 +405,26 @@ export const catalogCache = ( part(manifest.revision, "tool", name), schema(options.schema), ); - if (found === undefined) return separately; - if (found.value === undefined) return Effect.succeed(undefined); - // A part this schema cannot read is read again on its own, as without the hint, - // and fails there if it is still unreadable. - return Schema.decodeUnknownEffect(options.schema)(found.value).pipe( - Effect.catch(() => separately), + if (manifest.names !== undefined && !manifest.names.includes(name)) + return Effect.succeed(undefined); + const tool = + found === undefined + ? separately + : found.value === undefined + ? Effect.succeed(undefined) + : // An unreadable hinted part is read separately, and fails there if still unreadable. + Schema.decodeUnknownEffect(options.schema)(found.value).pipe( + Effect.catch(() => separately), + ); + return tool.pipe( + Effect.flatMap((value) => + value === undefined + ? // Retained pages distinguish an evicted part from an unknown name. + metadata().pipe( + Effect.map((tools) => tools.find((tool) => tool.name === name)), + ) + : Effect.succeed(value), + ), ); }), ),