From 0dbcd3008ea7a13555de3f68ae25ca8ec2229c4c Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 22 Jun 2026 11:57:21 +0530 Subject: [PATCH 1/4] feat(sdk): expose the cache KeyValueStore on the Executor instance --- packages/core/sdk/src/executor.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 8bb7b30848..6708f4f477 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -361,6 +361,13 @@ export type Executor = { * machinery (e.g. execution observers) can attribute work to an owner * without re-threading identity through every call site. */ readonly owner: OwnerBinding; + + /** The host-provided key/value cache primitive (Cloudflare KV in prod, an + * in-memory fallback otherwise). Plugins reach durable cache storage through + * this without re-threading a binding — wrap it with + * `KeyValueStore.toSchemaStore` for typed entries (mirrors the internal + * `tools.list` / schema-view caches). */ + readonly cache: KeyValueStore.KeyValueStore; } & PluginExtensions; export interface ExecutorDb { @@ -4215,6 +4222,7 @@ export const createExecutor = => value as Executor; From 553a5ec881ec55167a127f165f1bad6b9e80380f Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 22 Jun 2026 11:57:21 +0530 Subject: [PATCH 2/4] perf(semantic-search): snapshot the tool manifest to KV per run --- .../semantic-search-manifest-snapshot.md | 10 ++ .../src/sdk/manifest-snapshot.ts | 39 ++++++ .../src/sdk/tool-search-index.ts | 115 +++++++++++++++--- 3 files changed, 144 insertions(+), 20 deletions(-) create mode 100644 .changeset/semantic-search-manifest-snapshot.md create mode 100644 packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts diff --git a/.changeset/semantic-search-manifest-snapshot.md b/.changeset/semantic-search-manifest-snapshot.md new file mode 100644 index 0000000000..6bbf6cb21a --- /dev/null +++ b/.changeset/semantic-search-manifest-snapshot.md @@ -0,0 +1,10 @@ +--- +"@executor-js/sdk": minor +"@executor-js/plugin-semantic-search": minor +--- + +Snapshot the tool manifest per reindex run instead of re-reading it from D1 on every scan page. + +`createExecutor` now exposes its cache `KeyValueStore` on the executor instance as `executor.cache`, so plugins can reach durable cache storage (Cloudflare KV in production) without re-threading a binding — wrap it with `KeyValueStore.toSchemaStore` for typed entries, the same way the internal `tools.list` / schema-view caches do. + +The semantic-search scan phase used to call `tools.manifest()` — a full `tool_schema_manifest` table read — once per partition page, so an unchanged catalog cost `O(N² / pageLimit)` cross-region reads. Now `create` reads the manifest once, partitions it, and writes one snapshot per partition to `executor.cache`; `scan` reads only its partition's snapshot (KV-only, with no D1 fallback on the hot path — a miss fails the message so the queue retries). Snapshots are deleted when the run completes. This collapses the scan's manifest reads from `O(N² / pageLimit)` to a single pass and cuts per-message memory from the whole manifest to one partition's slice. diff --git a/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts b/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts new file mode 100644 index 0000000000..b2a5087a79 --- /dev/null +++ b/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts @@ -0,0 +1,39 @@ +import { ToolSchemaManifest } from "@executor-js/sdk/core"; +import { Schema } from "effect"; + +// --------------------------------------------------------------------------- +// Per-run tool-manifest snapshot. +// +// The scan phase used to re-read the entire `tool_schema_manifest` table from D1 +// on every page (12 partitions × ~N/limit pages), so an unchanged catalog cost +// O(N²/pageLimit) cross-region reads. Instead, `create` (and `reconcile`, for +// resume) reads the manifest once, partitions it, and writes one snapshot per +// partition to the executor's KV cache. `scan` then reads only its partition's +// snapshot — KV-only, no D1 fallback on the hot path; a miss fails the message +// so the queue retries (the initial scan fan-out is delayed once to let the +// write propagate). Snapshots are deleted when the run reaches a terminal state. +// --------------------------------------------------------------------------- + +export const MANIFEST_SNAPSHOT_PREFIX = "index-manifest/"; +export const MANIFEST_SNAPSHOT_VERSION = "v1"; + +/** A manifest paired with its ordinal in the full (pre-partition) manifest list. + * The ordinal is preserved so partitioned jobs keep the same stable ordering + * they had when the scan filtered the full list in-process. */ +const ManifestSnapshotItem = Schema.Struct({ + ordinal: Schema.Number, + manifest: ToolSchemaManifest, +}); +export type ManifestSnapshotItem = typeof ManifestSnapshotItem.Type; + +/** One partition's slice of the run's manifest snapshot. Sharded per partition so + * a scan message loads only its ~N/partitions rows, not the whole catalog, and + * no single KV value approaches the 25 MB cap. */ +export const ManifestSnapshotEntry = Schema.Struct({ + version: Schema.Literal(MANIFEST_SNAPSHOT_VERSION), + items: Schema.Array(ManifestSnapshotItem), +}); +export type ManifestSnapshotEntry = typeof ManifestSnapshotEntry.Type; + +export const manifestSnapshotKey = (runId: string, partition: number): string => + `${MANIFEST_SNAPSHOT_PREFIX}${MANIFEST_SNAPSHOT_VERSION}/${runId}/${partition}`; diff --git a/packages/plugins/semantic-search/src/sdk/tool-search-index.ts b/packages/plugins/semantic-search/src/sdk/tool-search-index.ts index 47f6f5bfe6..0e2e1d115e 100644 --- a/packages/plugins/semantic-search/src/sdk/tool-search-index.ts +++ b/packages/plugins/semantic-search/src/sdk/tool-search-index.ts @@ -3,9 +3,11 @@ import type { Owner, PluginBlobStore, PluginStorageCollectionFacade, + ToolSchemaManifest, } from "@executor-js/sdk/core"; import { sha256Hex } from "@executor-js/sdk/core"; -import { Context, Effect, Predicate, Schema } from "effect"; +import { Context, Effect, Option, Predicate, Schema } from "effect"; +import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore"; import { type Chunker, type ToolChunk, ToolDocumentInput } from "./chunker"; import { @@ -28,6 +30,12 @@ import { } from "./documents"; import type { ToolEmbedder } from "./embedder"; import { SemanticSearchError } from "./errors"; +import { + ManifestSnapshotEntry, + type ManifestSnapshotItem, + MANIFEST_SNAPSHOT_VERSION, + manifestSnapshotKey, +} from "./manifest-snapshot"; import { cyrb53 } from "./fingerprint"; import type { VectorInput, VectorStore } from "./store"; import type { FtsDocumentInput, FtsLexicalStore } from "./store-fts"; @@ -906,6 +914,84 @@ export const reconcile = ( }; }); +const manifestSnapshotStore = (executor: Executor) => + KeyValueStore.toSchemaStore(executor.cache, ManifestSnapshotEntry); + +/** Partition the run's full manifest list (preserving each entry's ordinal in the + * full list) and write one snapshot per partition to the executor cache. Written + * once at run creation so the scan phase reads KV instead of re-querying the whole + * `tool_schema_manifest` table on every page. Fails loudly: the scan path has no + * D1 fallback, so a run must not start without its snapshot. */ +const writeManifestSnapshot = ( + executor: Executor, + runId: string, + partitionCount: number, + manifests: readonly ToolSchemaManifest[], +): Effect.Effect => + Effect.gen(function* () { + const store = manifestSnapshotStore(executor); + const byPartition: ManifestSnapshotItem[][] = Array.from({ length: partitionCount }, () => []); + manifests.forEach((manifest, ordinal) => { + byPartition[partitionForPath(manifest.path, partitionCount)]?.push({ ordinal, manifest }); + }); + yield* Effect.forEach( + byPartition, + (items, partition) => + store.set(manifestSnapshotKey(runId, partition), { + version: MANIFEST_SNAPSHOT_VERSION, + items, + }), + { concurrency: INDEX_STORAGE_CONCURRENCY, discard: true }, + ).pipe( + Effect.mapError( + (cause) => + new SemanticSearchError({ message: "Failed to write tool manifest snapshot.", cause }), + ), + ); + }); + +/** Read a partition's manifest snapshot. KV-only by design: a miss fails the scan + * so the queue retries (the initial scan fan-out is delayed once to let the write + * propagate) — there is deliberately no D1 fallback on this hot path. */ +const readManifestSnapshot = ( + executor: Executor, + runId: string, + partition: number, +): Effect.Effect => + manifestSnapshotStore(executor) + .get(manifestSnapshotKey(runId, partition)) + .pipe( + Effect.mapError( + (cause) => + new SemanticSearchError({ message: "Failed to read tool manifest snapshot.", cause }), + ), + Effect.flatMap( + Option.match({ + onNone: () => + Effect.fail( + new SemanticSearchError({ + message: `Tool manifest snapshot for run "${runId}" partition ${partition} is missing.`, + }), + ), + onSome: (entry) => Effect.succeed(entry.items), + }), + ), + ); + +/** Best-effort delete of every partition snapshot once a run reaches a terminal + * state. (Stalled / terminally-failed runs that never complete leave their + * snapshots until overwritten — a small, bounded leak; KV has no native TTL.) */ +const deleteManifestSnapshot = ( + executor: Executor, + runId: string, + partitionCount: number, +): Effect.Effect => + Effect.forEach( + Array.from({ length: partitionCount }, (_, partition) => partition), + (partition) => executor.cache.remove(manifestSnapshotKey(runId, partition)), + { concurrency: INDEX_STORAGE_CONCURRENCY, discard: true }, + ).pipe(Effect.ignore); + export const create = ( input: IndexStores & ToolSearchIndex.CreateInput, ): Effect.Effect => @@ -914,6 +1000,8 @@ export const create = ( const manifests = yield* listToolManifests(input.executor, { maxTools: input.maxTools }); const createdAt = nowIso(); + yield* writeManifestSnapshot(input.executor, input.runId, partitionCount, manifests); + yield* input.runs .put({ owner: input.owner, @@ -947,26 +1035,12 @@ export const scan = ( input: IndexStores & ToolSearchIndex.ScanInput, ): Effect.Effect => Effect.gen(function* () { - const run = yield* input.runs - .getForOwner({ owner: input.owner, key: input.runId }) - .pipe( - Effect.mapError( - (cause) => new SemanticSearchError({ message: "Failed to load index run.", cause }), - ), - ); - if (run === null) { - return yield* new SemanticSearchError({ - message: `Index run "${input.runId}" does not exist.`, - }); - } + const partitionManifests = yield* readManifestSnapshot( + input.executor, + input.runId, + input.partition, + ); - const manifests = yield* listToolManifests(input.executor, { maxTools: input.maxTools }); - const partitionManifests = manifests - .map((manifest, ordinal) => ({ manifest, ordinal })) - .filter( - ({ manifest }) => - partitionForPath(manifest.path, run.data.partitionCount) === input.partition, - ); const existing = yield* input.jobs .count({ where: { runId: input.runId, partition: input.partition } }) .pipe( @@ -1631,6 +1705,7 @@ export const complete = ( new SemanticSearchError({ message: "Failed to mark index run completed.", cause }), ), ); + yield* deleteManifestSnapshot(input.executor, input.runId, existing.data.partitionCount); } return { runId: input.runId, removed: result.removed }; }); From 0070378a138d11da1831886cfcdd0d90e2679c56 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 22 Jun 2026 11:57:21 +0530 Subject: [PATCH 3/4] test(semantic-search): cover manifest snapshot read, miss, and cleanup --- .../src/sdk/tool-search-index.test.ts | 160 +++++++++++++++++- 1 file changed, 157 insertions(+), 3 deletions(-) diff --git a/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts b/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts index b09e0a2c77..7428c8ab05 100644 --- a/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts +++ b/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts @@ -9,6 +9,7 @@ import type { } from "@executor-js/sdk/core"; import { makeInMemoryBlobStore, pluginBlobStore } from "@executor-js/sdk/core"; import { Effect } from "effect"; +import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore"; import { makeFacetChunker } from "./chunker"; import { @@ -22,7 +23,18 @@ import { toolFingerprints, } from "./collections"; import type { ToolEmbedder } from "./embedder"; -import { chunk, commit, create, embed, fail, reconcile, scan, status } from "./tool-search-index"; +import { SemanticSearchError } from "./errors"; +import { + chunk, + commit, + complete, + create, + embed, + fail, + reconcile, + scan, + status, +} from "./tool-search-index"; import type { VectorInput, VectorStore } from "./store"; const owner: Owner = "org" as Owner; @@ -31,6 +43,27 @@ const namespace = "test-ns"; const makeBlobs = () => pluginBlobStore(makeInMemoryBlobStore(), { org: "test-org", user: null }, "semanticSearch"); +/** In-memory cache primitive for the executor mock — durable within a single test + * so a `create` snapshot write is visible to the `scan` reads that follow. */ +const makeMemoryCache = (): Executor["cache"] => { + const rows = new Map(); + return KeyValueStore.makeStringOnly({ + get: (key) => Effect.sync(() => rows.get(key)), + set: (key, value) => + Effect.sync(() => { + rows.set(key, value); + }), + remove: (key) => + Effect.sync(() => { + rows.delete(key); + }), + clear: Effect.sync(() => { + rows.clear(); + }), + size: Effect.sync(() => rows.size), + }); +}; + const manifestForTool = ( tool: Tool, fingerprint = `fingerprint:${String(tool.address)}`, @@ -64,7 +97,7 @@ const makeExecutor = ( connection: "default" as never, pluginId: "test", }; - const executor: Pick = { + const executor: Pick = { tools: { list: () => Effect.succeed([tool]), manifest: () => Effect.succeed([manifestForTool(tool)]), @@ -77,6 +110,7 @@ const makeExecutor = ( }); }, }, + cache: makeMemoryCache(), }; return executor as Executor; }; @@ -511,12 +545,13 @@ describe("ToolSearchIndex", () => { pluginId: "test", }, ]; - const executor: Pick = { + const executor: Pick = { tools: { list: () => Effect.succeed(tools), manifest: () => Effect.succeed(tools.map((tool) => manifestForTool(tool))), schema: () => Effect.succeed(null), }, + cache: makeMemoryCache(), }; const runs = makeCollection(indexRuns.name); const jobs = makeCollection(indexJobs.name); @@ -852,3 +887,122 @@ describe("ToolSearchIndex", () => { }), ); }); + +describe("ToolSearchIndex manifest snapshot", () => { + const makeTools = (count: number): Tool[] => + Array.from({ length: count }, (_, i) => ({ + address: `tools.github.repos.get${i}` as never, + name: `repos.get${i}` as never, + integration: "github" as never, + description: `Get repository ${i}`, + owner, + connection: "default" as never, + pluginId: "test", + })); + + /** Executor mock that counts `tools.manifest()` calls so a test can prove the + * scan reads the KV snapshot rather than re-querying the manifest per page. */ + const makeCountingExecutor = (tools: Tool[]) => { + const counters = { manifest: 0 }; + const executor: Pick = { + tools: { + list: () => Effect.succeed(tools), + manifest: () => { + counters.manifest++; + return Effect.succeed(tools.map((tool) => manifestForTool(tool))); + }, + schema: () => Effect.succeed(null), + }, + cache: makeMemoryCache(), + }; + return { executor: executor as Executor, counters }; + }; + + const makeBase = (executor: Executor) => ({ + namespace, + executor, + runs: makeCollection(indexRuns.name), + jobs: makeCollection(indexJobs.name), + chunks: makeCollection(indexChunks.name), + fingerprints: makeCollection(toolFingerprints.name), + blobs: makeBlobs(), + owner, + }); + + it.effect("reads the manifest once at create and pages scans from the KV snapshot", () => + Effect.gen(function* () { + const { executor, counters } = makeCountingExecutor(makeTools(5)); + const base = makeBase(executor); + + yield* create({ ...base, runId: "run-snap", partitionCount: 1 }); + expect(counters.manifest).toBe(1); + + let pages = 0; + let hasMore = true; + while (hasMore) { + const result = yield* scan({ ...base, runId: "run-snap", partition: 0, limit: 2 }); + hasMore = result.hasMore; + pages++; + } + + expect(pages).toBe(3); // ceil(5 / 2) pages, each from the snapshot + expect(counters.manifest).toBe(1); // never re-read during scan + expect(base.jobs.data.size).toBe(5); // every tool scanned exactly once + }), + ); + + it.effect("fails the scan when the run snapshot is missing (no D1 fallback)", () => + Effect.gen(function* () { + const { executor, counters } = makeCountingExecutor(makeTools(2)); + const base = makeBase(executor); + + // No `create`, so nothing was written to the KV snapshot. + const error = yield* Effect.flip( + scan({ ...base, runId: "run-missing", partition: 0, limit: 10 }), + ); + + expect(error).toBeInstanceOf(SemanticSearchError); // KV miss → fail, not a live read + expect(counters.manifest).toBe(0); // the scan never falls back to a live manifest read + }), + ); + + it.effect("clears the partition snapshots when the run completes", () => + Effect.gen(function* () { + const { executor } = makeCountingExecutor(makeTools(3)); + const base = makeBase(executor); + + yield* create({ ...base, runId: "run-clear", partitionCount: 1 }); + let hasMore = true; + while (hasMore) { + const result = yield* scan({ ...base, runId: "run-clear", partition: 0, limit: 10 }); + hasMore = result.hasMore; + } + + const store: VectorStore = { + maxTopK: 100, + query: () => Effect.succeed([]), + upsert: () => Effect.void, + deleteByIds: () => Effect.void, + }; + const embedder: ToolEmbedder = { + model: "test", + dimensions: 3, + embedDocuments: (texts) => Effect.succeed(texts.map(() => [0.1, 0.2, 0.3])), + embedQuery: () => Effect.succeed([0.1, 0.2, 0.3]), + }; + yield* complete({ + ...base, + store, + embedder, + chunker: makeFacetChunker(), + runId: "run-clear", + }); + + // The snapshot is gone, so a post-completion scan can no longer read it. + const error = yield* Effect.flip( + scan({ ...base, runId: "run-clear", partition: 0, limit: 10 }), + ); + expect(error).toBeInstanceOf(SemanticSearchError); // snapshot cleared on completion + }), + ); +}); From f7685ded7b4767aee9d6911ecbd8fe1ca7920f90 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 22 Jun 2026 13:15:56 +0530 Subject: [PATCH 4/4] fix(semantic-search): correct snapshot reconcile comment and clean up on terminal fail (greptile) --- .../src/sdk/manifest-snapshot.ts | 17 +++++++++++------ .../src/sdk/tool-search-index.test.ts | 16 ++++++++++++++++ .../src/sdk/tool-search-index.ts | 8 +++++--- 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts b/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts index b2a5087a79..26efa6cbf4 100644 --- a/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts +++ b/packages/plugins/semantic-search/src/sdk/manifest-snapshot.ts @@ -6,12 +6,17 @@ import { Schema } from "effect"; // // The scan phase used to re-read the entire `tool_schema_manifest` table from D1 // on every page (12 partitions × ~N/limit pages), so an unchanged catalog cost -// O(N²/pageLimit) cross-region reads. Instead, `create` (and `reconcile`, for -// resume) reads the manifest once, partitions it, and writes one snapshot per -// partition to the executor's KV cache. `scan` then reads only its partition's -// snapshot — KV-only, no D1 fallback on the hot path; a miss fails the message -// so the queue retries (the initial scan fan-out is delayed once to let the -// write propagate). Snapshots are deleted when the run reaches a terminal state. +// O(N²/pageLimit) cross-region reads. Instead, `create` reads the manifest once, +// partitions it, and writes one snapshot per partition to the executor's KV cache. +// `scan` then reads only its partition's snapshot — KV-only, no D1 fallback on the +// hot path; a miss fails the message so the queue retries (the initial scan fan-out +// is delayed once to let the write propagate). Snapshots are deleted when the run +// reaches a terminal state (`complete` or terminal `fail`). +// +// Only `create` writes the snapshot — `reconcile` (resume) does not. The snapshot is +// durable (no TTL; removed only at terminal state), so a resumed stalled run still +// finds it. If a snapshot is ever genuinely absent at resume (e.g. a version-key bump +// or KV rollback), its scans DLQ and the recovery path is a fresh reindex, not resume. // --------------------------------------------------------------------------- export const MANIFEST_SNAPSHOT_PREFIX = "index-manifest/"; diff --git a/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts b/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts index 7428c8ab05..1c1b77bb22 100644 --- a/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts +++ b/packages/plugins/semantic-search/src/sdk/tool-search-index.test.ts @@ -1005,4 +1005,20 @@ describe("ToolSearchIndex manifest snapshot", () => { expect(error).toBeInstanceOf(SemanticSearchError); // snapshot cleared on completion }), ); + + it.effect("clears the partition snapshots when the run terminally fails", () => + Effect.gen(function* () { + const { executor } = makeCountingExecutor(makeTools(3)); + const base = makeBase(executor); + + yield* create({ ...base, runId: "run-failclear", partitionCount: 1 }); + const failed = yield* fail({ ...base, runId: "run-failclear", error: "queue exhausted" }); + expect(failed.runFailed).toBe(true); // no pending paths/chunks → run is terminally failed + + const error = yield* Effect.flip( + scan({ ...base, runId: "run-failclear", partition: 0, limit: 10 }), + ); + expect(error).toBeInstanceOf(SemanticSearchError); // snapshot cleared on terminal failure + }), + ); }); diff --git a/packages/plugins/semantic-search/src/sdk/tool-search-index.ts b/packages/plugins/semantic-search/src/sdk/tool-search-index.ts index 0e2e1d115e..2a2b539fa7 100644 --- a/packages/plugins/semantic-search/src/sdk/tool-search-index.ts +++ b/packages/plugins/semantic-search/src/sdk/tool-search-index.ts @@ -687,7 +687,7 @@ const countJobsByStatus = ( ); export const fail = ( - input: IndexCollections & ToolSearchIndex.FailInput, + input: IndexStores & ToolSearchIndex.FailInput, ): Effect.Effect => Effect.gen(function* () { const updatedAt = nowIso(); @@ -834,6 +834,7 @@ export const fail = ( new SemanticSearchError({ message: "Failed to mark index run failed.", cause }), ), ); + yield* deleteManifestSnapshot(input.executor, input.runId, run.data.partitionCount); } return { @@ -979,8 +980,9 @@ const readManifestSnapshot = ( ); /** Best-effort delete of every partition snapshot once a run reaches a terminal - * state. (Stalled / terminally-failed runs that never complete leave their - * snapshots until overwritten — a small, bounded leak; KV has no native TTL.) */ + * state (`complete` or terminal `fail`). A stalled run that never reaches a + * terminal state leaves its snapshot until overwritten — a small, bounded leak; + * KV has no native TTL through this abstraction. */ const deleteManifestSnapshot = ( executor: Executor, runId: string,