From 1a6f3bbe4ee48952e3ebfd36633ff8663ff2983c Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 5 Sep 2026 16:56:00 +0530 Subject: [PATCH 1/4] fix(openapi): reduce large import memory usage --- .changeset/openapi-large-catalog-memory.md | 7 + e2e/scenarios/openapi-large-catalog.test.ts | 140 +++++++++++++ e2e/vitest.config.ts | 1 + .../core/fumadb/src/adapters/drizzle/query.ts | 38 ++-- .../drizzle/upsert-many-generic.test.ts | 43 ++++ packages/core/sdk/src/executor.ts | 2 +- packages/core/sdk/src/plugin-storage.test.ts | 62 ++++++ packages/plugins/openapi/src/sdk/plugin.ts | 186 +++++++++--------- .../plugins/openapi/src/sdk/store.test.ts | 66 ++++++- packages/plugins/openapi/src/sdk/store.ts | 26 ++- 10 files changed, 448 insertions(+), 123 deletions(-) create mode 100644 .changeset/openapi-large-catalog-memory.md create mode 100644 e2e/scenarios/openapi-large-catalog.test.ts diff --git a/.changeset/openapi-large-catalog-memory.md b/.changeset/openapi-large-catalog-memory.md new file mode 100644 index 0000000000..1e171b63de --- /dev/null +++ b/.changeset/openapi-large-catalog-memory.md @@ -0,0 +1,7 @@ +--- +"@executor-js/sdk": patch +"@executor-js/plugin-openapi": patch +"@executor-js/fumadb": patch +--- + +Reduce large OpenAPI import memory by compiling operation bindings in chunks and filtering integration storage prefixes in SQL. Construct D1 upsert statements lazily in bounded native batches. diff --git a/e2e/scenarios/openapi-large-catalog.test.ts b/e2e/scenarios/openapi-large-catalog.test.ts new file mode 100644 index 0000000000..02ed0b03c7 --- /dev/null +++ b/e2e/scenarios/openapi-large-catalog.test.ts @@ -0,0 +1,140 @@ +import { randomUUID } from "node:crypto"; +import { assert, expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { AuthTemplateSlug, ConnectionName, IntegrationSlug } from "@executor-js/sdk/shared"; +import { scenario } from "../src/scenario"; +import { Api, Browser, Target } from "../src/services"; + +import { visit } from "../src/surfaces/browser"; + +const api = composePluginApi([openApiHttpPlugin()] as const); +const spec = (count: number) => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Large catalog", version: "1" }, + paths: Object.fromEntries( + Array.from({ length: count }, (_, i) => [ + `/items/${i}`, + { + get: { + operationId: `item${i}`, + responses: { + "200": { + description: "Item details", + content: { "application/json": { schema: { $ref: "#/components/schemas/Item" } } }, + }, + }, + }, + }, + ]), + ), + components: { + schemas: { + Item: { + type: "object", + properties: { + id: { type: "string" }, + description: { type: "string", description: "Item description. ".repeat(2000) }, + }, + }, + }, + }, + }); + +scenario( + "OpenAPI · large imports report complete counts and preserve unrelated catalogs", + {}, + Effect.gen(function* () { + const target = yield* Target; + const { client } = yield* Api; + const browser = yield* Browser; + const identity = yield* target.newIdentity(); + const http = yield* client(api, identity); + const slug = IntegrationSlug.make(`large-${randomUUID()}`); + const neighbor = IntegrationSlug.make(`neighbor-${randomUUID()}`); + yield* Effect.gen(function* () { + const add = (integration: string, count: number) => + http.openapi.addSpec({ + payload: { + slug: integration, + spec: { kind: "blob", value: spec(count) }, + baseUrl: "https://example.invalid", + authenticationTemplate: [], + }, + }); + expect((yield* add(neighbor, 1001)).toolCount).toBe(1001); + expect((yield* add(slug, 1201)).toolCount).toBe(1201); + yield* http.connections.create({ + payload: { + owner: "org", + integration: slug, + name: ConnectionName.make("main"), + template: AuthTemplateSlug.make("none"), + value: "catalog-fixture", + }, + }); + expect(yield* http.tools.list({ query: { integration: slug } })).toHaveLength(1201); + const updated = yield* http.openapi.updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: spec(1002) }, + }, + }); + expect(updated.toolCount).toBe(1002); + expect(updated.removedTools).toHaveLength(199); + expect(updated.addedTools).toEqual([]); + const failed = yield* http.openapi + .updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: "invalid" }, + }, + }) + .pipe(Effect.result); + expect(failed._tag).toBe("Failure"); + const intact = yield* http.openapi.updateSpec({ + params: { slug: neighbor }, + payload: { + spec: { kind: "blob", value: spec(1001) }, + }, + }); + expect(intact.toolCount).toBe(1001); + expect(intact.removedTools).toEqual([]); + const retry = yield* http.openapi.updateSpec({ + params: { slug }, + payload: { + spec: { kind: "blob", value: spec(1002) }, + }, + }); + expect(retry.toolCount).toBe(1002); + expect(retry.addedTools).toEqual([]); + expect(retry.removedTools).toEqual([]); + const catalog = yield* http.tools.list({ query: { integration: slug } }); + expect(catalog).toHaveLength(1002); + const first = catalog[0]; + assert(first); + const schema = yield* http.tools.schema({ + query: { address: first.address }, + }); + expect(JSON.stringify(schema)).toContain("description"); + yield* browser.session(identity, async ({ page, step }) => { + await step("Open the imported catalog after refresh", async () => { + await visit(page, `/integrations/${slug}?tab=tools`); + await page.getByPlaceholder("Filter 1002 tools…").waitFor(); + }); + }); + }).pipe( + Effect.ensuring( + Effect.forEach( + [slug, neighbor], + (slug) => http.openapi.removeSpec({ params: { slug } }).pipe(Effect.orDie), + { + discard: true, + }, + ), + ), + ); + }), +); diff --git a/e2e/vitest.config.ts b/e2e/vitest.config.ts index 2ed8ba01bd..880cbcafff 100644 --- a/e2e/vitest.config.ts +++ b/e2e/vitest.config.ts @@ -49,6 +49,7 @@ export default defineConfig({ include: [ "scenarios/browser-approval.test.ts", "scenarios/microsoft-graph-full.test.ts", + "scenarios/openapi-large-catalog.test.ts", "scenarios/toolkits-mcp.test.ts", "cloudflare/**/*.test.ts", ], diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index cadf7fce2e..ec4fed0cc7 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -28,6 +28,12 @@ type P_DBType = PostgreSQL.PgDatabase< >; const CREATE_MANY_BATCH_SIZE = 500; +// A D1 native batch is one Workers RPC. Large schemas can make a few hundred +// prepared statements exceed the platform's 32 MiB serialized-argument cap, +// even though every individual statement fits its bound-parameter limit. +// Bound statement construction and RPC group size. Each group commits +// independently; this count does not bound the size of an individual row. +const D1_NATIVE_BATCH_STATEMENT_LIMIT = 50; // A multi-row write binds (rows * columns) parameters in one statement, and // engines cap bound parameters per statement (older SQLite: 999, Cloudflare @@ -603,8 +609,7 @@ export function fromDrizzle( ]), ); - const buildStatements = (handle: typeof db): unknown[] => { - const statements: unknown[] = []; + const buildStatements = function* (handle: typeof db): Generator { for (let i = 0; i < values.length; i += batchSize) { const batch = values.slice(i, i + batchSize); const insert = handle.insert(drizzleTable).values(batch) as unknown as { @@ -614,15 +619,12 @@ export function fromDrizzle( readonly where?: typeof where; }) => unknown; }; - statements.push( - insert.onConflictDoUpdate({ - target, - set, - ...(where === undefined ? {} : { where }), - }), - ); + yield insert.onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }); } - return statements; }; const executeStatements = async (handle: typeof db) => { for (const statement of buildStatements(handle)) { @@ -632,14 +634,22 @@ export function fromDrizzle( const statementCount = Math.ceil(values.length / batchSize); // D1 rejects interactive transactions but its native batch API executes - // prepared statements as one transaction. Drizzle exposes that API on - // the database handle, so keep parameter-bounded upserts atomic instead - // of auto-committing each statement independently. + // each prepared-statement group as one transaction. Drizzle exposes that + // API on the database handle, so keep bounded groups atomic instead of + // auto-committing each statement independently. const nativeBatch = db as unknown as { readonly batch?: (statements: readonly unknown[]) => Promise; }; if (!interactiveTransactions && statementCount > 1 && nativeBatch.batch) { - await nativeBatch.batch(buildStatements(db)); + let statements: unknown[] = []; + for (const statement of buildStatements(db)) { + statements.push(statement); + if (statements.length >= D1_NATIVE_BATCH_STATEMENT_LIMIT) { + await nativeBatch.batch(statements); + statements = []; + } + } + if (statements.length > 0) await nativeBatch.batch(statements); return; } diff --git a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts index 333d4a2f7d..82309bb8d2 100644 --- a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts +++ b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts @@ -1,4 +1,5 @@ import { expect, test } from "@effect/vitest"; +import { sqliteTable, text } from "drizzle-orm/sqlite-core"; import { column, idColumn, schema, table } from "../../schema"; import { fromDrizzle } from "./query"; @@ -118,3 +119,45 @@ test("generic bulk upsert rolls earlier rows back when a later row fails", async expect(committed).toEqual([]); expect(events).toContain("transaction:rollback"); }); + +test("D1 bulk upsert bounds native batch RPC statement counts", async () => { + const drizzleRows = sqliteTable("generic_rows", { + id: text("id").primaryKey(), + value: text("value").notNull(), + }); + const batchSizes: number[] = []; + let constructed = 0; + let completed = 0; + let maxPending = 0; + const db = { + _: { fullSchema: { rows: drizzleRows } }, + insert: () => ({ + values: (values: readonly Record[]) => ({ + onConflictDoUpdate: () => { + constructed++; + maxPending = Math.max(maxPending, constructed - completed); + return { values }; + }, + }), + }), + batch: async (statements: readonly unknown[]) => { + batchSizes.push(statements.length); + completed += statements.length; + }, + }; + const orm = fromDrizzle(v1, db, "sqlite", false, 2); + const rows = v1.tables.rows; + + await orm.internal.upsertMany?.(rows, { + target: [rows.columns.id], + update: [rows.columns.value], + values: Array.from({ length: 121 }, (_, index) => ({ + id: `r${index}`, + value: `value-${index}`, + })), + }); + + expect(batchSizes).toEqual([50, 50, 21]); + expect(constructed).toBe(121); + expect(maxPending).toBe(50); +}); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 5d77abc207..f18925c206 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -2389,7 +2389,7 @@ const makePluginStorageFacade = (input: { list: (storageInput) => Effect.gen(function* () { const rows = yield* input.core.findMany("plugin_storage", { - where: whereFor(storageInput.collection), + where: whereForPrefix(storageInput.collection, storageInput.keyPrefix), }); return sortByOwnerPrecedence(rows) .filter((row) => diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index e7ebc7dd39..80876e0c54 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/packages/core/sdk/src/plugin-storage.test.ts @@ -76,6 +76,8 @@ const executionHistoryPlugin = definePlugin(() => ({ owner, entries: keys.map((key) => ({ collection: toolCalls.name, key })), }), + list: (keyPrefix?: string) => + ctx.storage.pluginStorage.list({ collection: toolCalls.name, keyPrefix }), get: (key: string) => ctx.storage.toolCalls.get({ key }), getMany: (keys: readonly string[]) => ctx.storage.toolCalls.getMany({ keys }), getForOwner: (owner: Owner, key: string) => ctx.storage.toolCalls.getForOwner({ owner, key }), @@ -164,6 +166,66 @@ const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => { }; describe("plugin storage collections", () => { + it.effect("filters list prefixes in the database before materializing unrelated rows", () => + Effect.gen(function* () { + const config = makeTestConfig({ + backend: "sqlite", + plugins: [executionHistoryPlugin] as const, + }); + const reads: number[] = []; + const wrap = (db: FumaDb): FumaDb => + new Proxy(db, { + get(target, property, receiver) { + if (property === "withContext") { + const withContext = target.withContext; + return withContext === undefined + ? undefined + : (context: unknown) => wrap(withContext(context)); + } + if (property === "findMany") { + const findMany: FumaDb["findMany"] = async (table, options) => { + const rows = await target.findMany(table, options); + if (table === "plugin_storage") reads.push(rows.length); + return rows; + }; + return findMany; + } + return Reflect.get(target, property, receiver); + }, + }); + const executor = yield* Effect.acquireRelease( + createExecutor({ ...config, db: wrap(config.db) }), + (instance) => + instance + .close() + .pipe(Effect.orDie, Effect.ensuring(Effect.promise(() => config.testDb.close()))), + ); + const data = call({ + runId: "run", + toolId: "tool", + status: "ok", + startedAt: "2026-01-01T00:00:00Z", + }); + yield* executor.executionHistory.recordMany("org", [ + { key: "selected.one", data }, + ...Array.from({ length: 100 }, (_, i) => ({ key: `unrelated.${i}`, data })), + ]); + reads.length = 0; + expect((yield* executor.executionHistory.list("selected.")).map((row) => row.key)).toEqual([ + "selected.one", + ]); + expect(reads).toEqual([1]); + yield* executor.executionHistory.recordMany("org", [ + { key: "literal_%.one", data }, + { key: "literalXX.one", data }, + { key: "LITERAL_%.one", data }, + ]); + expect((yield* executor.executionHistory.list("literal_%.")).map((row) => row.key)).toEqual([ + "literal_%.one", + ]); + }), + ); + it.effect("queries declared indexes through the executor's SQLite FumaDB target", () => Effect.gen(function* () { const executor = yield* makeTestExecutor({ diff --git a/packages/plugins/openapi/src/sdk/plugin.ts b/packages/plugins/openapi/src/sdk/plugin.ts index b0c0558236..65f3d76c84 100644 --- a/packages/plugins/openapi/src/sdk/plugin.ts +++ b/packages/plugins/openapi/src/sdk/plugin.ts @@ -54,10 +54,10 @@ import { ApiKeyAuthTemplate, describeApiKeyAuthMethod } from "@executor-js/sdk/h import { checkHealthOpenApi, compileAndPersistOpenApiSpecStreaming, - compileOpenApiSpec, + buildDefsJson, + compileAndPersistOpenApiOperations, invokeOpenApiBackedTool, listHealthCheckCandidatesOpenApi, - openApiStoredOperationsFromCompiled, resolveOpenApiBackedAnnotations, resolveOpenApiBackedTools, validateOpenApiBackedToolArgs, @@ -784,9 +784,7 @@ export const openApiPlugin = definePlugin< // Resolve URL → text and parse BEFORE opening a transaction. Holding // `BEGIN` across a network fetch is the Hyperdrive deadlock path. const resolved = yield* resolveSpecForInput(config, httpClientLayer); - const compiled = resolved.keepPathItem - ? undefined - : yield* compileOpenApiSpec(resolved.specText); + const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); const adapter = yield* resolveSpecFormatAdapter( options?.specFormats ?? [], config.specFormat, @@ -906,21 +904,24 @@ export const openApiPlugin = definePlugin< // The content-addressed defs blob lets the serve path resolve the // shared `definitions` without re-parsing the spec. Same idempotent, // outside-the-transaction rationale as the spec blob. - if (compiled) { - yield* ctx.storage.putDefs(specHash, JSON.stringify(compiled.hoistedDefs)); + if (document) { + yield* ctx.storage.putDefs(specHash, buildDefsJson(document)); } - yield* ctx.transaction( + const persisted = yield* ctx.transaction( Effect.gen(function* () { yield* ctx.core.integrations.register({ slug, name: - config.name?.trim() || derivedIdentity?.name || compiled?.title || resolvedSlug, + config.name?.trim() || + derivedIdentity?.name || + document?.info.title || + resolvedSlug, description: config.description ?? derivedIdentity?.description ?? - compiled?.description ?? - compiled?.title ?? + document?.info.description ?? + document?.info.title ?? resolvedSlug, config: integrationConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, canRemove: true, @@ -929,30 +930,24 @@ export const openApiPlugin = definePlugin< if (config.healthCheck) { yield* ctx.core.integrations.setHealthCheck(slug, config.healthCheck); } - if (compiled) { - yield* ctx.storage.putOperations( - resolvedSlug, - openApiStoredOperationsFromCompiled(resolvedSlug, compiled), - ); - return compiled.definitions.length; + if (document) { + return yield* compileAndPersistOpenApiOperations({ + doc: document, + integration: resolvedSlug, + storage: ctx.storage, + }); } - const persisted = yield* compileAndPersistOpenApiSpecStreaming({ + return yield* compileAndPersistOpenApiSpecStreaming({ specText: resolved.specText, integration: resolvedSlug, storage: ctx.storage, specHash, keepPathItem: resolved.keepPathItem, }); - return persisted.toolCount; }), ); - const toolCount = compiled - ? compiled.definitions.length - : yield* ctx.storage - .listOperations(resolvedSlug) - .pipe(Effect.map((operations) => operations.length)); - return { slug, toolCount }; + return { slug, toolCount: persisted.toolCount }; }); // Update the spec IN PLACE: re-resolve (stored source URL / bundle, or a @@ -994,82 +989,81 @@ export const openApiPlugin = definePlugin< }); } - // Resolve + compile BEFORE the transaction (same Hyperdrive-deadlock - // rule as addSpec: never hold BEGIN across a network fetch). - const resolved = yield* resolveSpecForInput( - { - spec: specInput, - specFormat: current.specFormat, - specOverrides: nextOverrides, - headers: current.headers, - queryParams: current.queryParams, - baseUrl: current.baseUrl, - authenticationTemplate: current.authenticationTemplate, - }, - httpClientLayer, - ); - const compiled = resolved.keepPathItem - ? undefined - : yield* compileOpenApiSpec(resolved.specText); - - const previousOperations = yield* ctx.storage.listOperations(rawSlug); - const previousNames = new Set(previousOperations.map((op) => op.toolName)); + const { nextNames, previousNames } = yield* Effect.gen(function* () { + // Resolve + parse BEFORE the transaction (same Hyperdrive-deadlock + // rule as addSpec: never hold BEGIN across a network fetch). + const resolved = yield* resolveSpecForInput( + { + spec: specInput, + specFormat: current.specFormat, + specOverrides: nextOverrides, + headers: current.headers, + queryParams: current.queryParams, + baseUrl: current.baseUrl, + authenticationTemplate: current.authenticationTemplate, + }, + httpClientLayer, + ); + const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); - // The resolved spec text lives in the plugin blob store keyed by its - // content hash (`spec/`); the config carries only the hash. Put - // the blob outside the transaction - re-puts are idempotent and an - // aborted config update just leaves an unreferenced blob. - const specHash = yield* sha256Hex(resolved.specText); - const sourceSpecHash = - nextOverrides.length > 0 ? yield* sha256Hex(resolved.sourceSpecText) : undefined; - yield* ctx.storage.putSpec(specHash, resolved.specText); - if (sourceSpecHash) { - yield* ctx.storage.putSpec(sourceSpecHash, resolved.sourceSpecText); - } - if (compiled) { - yield* ctx.storage.putDefs(specHash, JSON.stringify(compiled.hoistedDefs)); - } + const previousNames = new Set( + (yield* ctx.storage.listOperations(rawSlug)).map((op) => op.toolName), + ); - const { - sourceSpecHash: _currentSourceSpecHash, - specOverrides: _currentSpecOverrides, - ...currentWithoutOverrides - } = current; - const nextConfig: OpenApiIntegrationConfig = { - ...currentWithoutOverrides, - specHash, - ...((resolved.specUrl ?? specInputToSpecUrl(specInput)) !== undefined - ? { specUrl: resolved.specUrl ?? specInputToSpecUrl(specInput) } - : {}), - ...(nextOverrides.length > 0 ? { specOverrides: nextOverrides, sourceSpecHash } : {}), - }; + // The resolved spec text lives in the plugin blob store keyed by its + // content hash (`spec/`); the config carries only the hash. Put + // the blob outside the transaction - re-puts are idempotent and an + // aborted config update just leaves an unreferenced blob. + const specHash = yield* sha256Hex(resolved.specText); + const sourceSpecHash = + nextOverrides.length > 0 ? yield* sha256Hex(resolved.sourceSpecText) : undefined; + yield* ctx.storage.putSpec(specHash, resolved.specText); + if (sourceSpecHash) { + yield* ctx.storage.putSpec(sourceSpecHash, resolved.sourceSpecText); + } + if (document) { + yield* ctx.storage.putDefs(specHash, buildDefsJson(document)); + } - yield* ctx.transaction( - Effect.gen(function* () { - yield* ctx.core.integrations.update(slug, { - config: nextConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, - }); - if (compiled) { - yield* ctx.storage.putOperations( - rawSlug, - openApiStoredOperationsFromCompiled(rawSlug, compiled), - ); - } else { - yield* compileAndPersistOpenApiSpecStreaming({ - specText: resolved.specText, - integration: rawSlug, - storage: ctx.storage, - specHash, - keepPathItem: resolved.keepPathItem, + const { + sourceSpecHash: _currentSourceSpecHash, + specOverrides: _currentSpecOverrides, + ...currentWithoutOverrides + } = current; + const nextConfig: OpenApiIntegrationConfig = { + ...currentWithoutOverrides, + specHash, + ...((resolved.specUrl ?? specInputToSpecUrl(specInput)) !== undefined + ? { specUrl: resolved.specUrl ?? specInputToSpecUrl(specInput) } + : {}), + ...(nextOverrides.length > 0 ? { specOverrides: nextOverrides, sourceSpecHash } : {}), + }; + + const persisted = yield* ctx.transaction( + Effect.gen(function* () { + yield* ctx.core.integrations.update(slug, { + config: nextConfig satisfies OpenApiIntegrationConfig as IntegrationConfig, }); - } - }), - ); + if (document) { + return yield* compileAndPersistOpenApiOperations({ + doc: document, + integration: rawSlug, + storage: ctx.storage, + }); + } else { + return yield* compileAndPersistOpenApiSpecStreaming({ + specText: resolved.specText, + integration: rawSlug, + storage: ctx.storage, + specHash, + keepPathItem: resolved.keepPathItem, + }); + } + }), + ); - const nextOperations = compiled - ? openApiStoredOperationsFromCompiled(rawSlug, compiled) - : yield* ctx.storage.listOperations(rawSlug); - const nextNames = new Set(nextOperations.map((op) => op.toolName)); + return { nextNames: new Set(persisted.toolNames), previousNames }; + }); // Rebuild each connection's tool rows from the new spec. Outside the // transaction: refresh opens its own, and a half-refreshed catalog diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index fb7021e5a2..22b8e7fa10 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option } from "effect"; +import { Effect, Option, Schema } from "effect"; import { Subject, @@ -13,11 +13,14 @@ import { import { makeDefaultOpenapiStore } from "./store"; import { OperationBinding } from "./types"; +const encodeBinding = Schema.encodeSync(OperationBinding); + describe("OpenAPI operation store", () => { it.effect("bounds operation storage keys while preserving tool-name lookup", () => Effect.gen(function* () { const rows = new Map(); const capturedKeys: string[] = []; + const listedPrefixes: (string | undefined)[] = []; const storageKey = (collection: string, key: string) => `${collection}\0${key}`; const unexpectedCollectionCall = () => Effect.die("Unexpected collection storage call"); const now = new Date(); @@ -105,14 +108,19 @@ describe("OpenAPI operation store", () => { }), ), ), - list: (input: { readonly collection: string; readonly keyPrefix?: string }) => - Effect.succeed( + list: (input: { + readonly collection: string; + readonly keyPrefix?: string; + }) => { + listedPrefixes.push(input.keyPrefix); + return Effect.succeed( [...rows.values()].filter( (row) => row.collection === input.collection && (input.keyPrefix === undefined || row.key.startsWith(input.keyPrefix)), ) as PluginStorageEntry[], - ), + ); + }, put: (input: { readonly owner: "org" | "user"; readonly collection: string; @@ -177,6 +185,40 @@ describe("OpenAPI operation store", () => { }, ]); + const binding = OperationBinding.make({ + method: "get", + servers: [], + pathTemplate: "/legacy", + parameters: [], + requestBody: Option.none(), + responseBody: Option.none(), + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "microsoft_graph.legacy", + data: { + integration: "microsoft_graph", + toolName: "legacy", + binding: encodeBinding(binding), + }, + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "microsoft_graph.other.legacy", + data: { + integration: "microsoft_graph.other", + toolName: "legacy", + binding: encodeBinding(binding), + }, + }); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "unrelated.invalid", + data: { integration: "unrelated", toolName: "invalid", binding: "invalid" }, + }); expect(capturedKeys).toHaveLength(1); expect(capturedKeys[0]!.length).toBeLessThanOrEqual(255); expect(capturedKeys[0]).not.toContain(toolName); @@ -184,6 +226,22 @@ describe("OpenAPI operation store", () => { const operation = yield* store.getOperation("microsoft_graph", toolName); expect(operation?.toolName).toBe(toolName); expect(operation?.binding.pathTemplate).toBe("/users/{userId}/messages"); + expect((yield* store.listOperations("microsoft_graph")).map((op) => op.toolName)).toEqual([ + toolName, + "legacy", + ]); + yield* store.removeOperations("microsoft_graph"); + expect(yield* store.listOperations("microsoft_graph")).toEqual([]); + expect((yield* store.getOperation("microsoft_graph.other", "legacy"))?.toolName).toBe( + "legacy", + ); + expect(rows.has(storageKey("operation", "unrelated.invalid"))).toBe(true); + expect(listedPrefixes).toEqual( + Array.from({ length: 4 }, () => [ + capturedKeys[0]!.slice(0, capturedKeys[0]!.lastIndexOf(".") + 1), + "microsoft_graph.", + ]).flat(), + ); }), ); }); diff --git a/packages/plugins/openapi/src/sdk/store.ts b/packages/plugins/openapi/src/sdk/store.ts index 548a521031..45bd9df57b 100644 --- a/packages/plugins/openapi/src/sdk/store.ts +++ b/packages/plugins/openapi/src/sdk/store.ts @@ -84,8 +84,11 @@ const stableKeyHash = (value: string): string => { return hash.toString(36).padStart(13, "0"); }; +const operationKeyPrefix = (integration: string): string => + `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.`; + const operationKey = (integration: string, toolName: string): string => - `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.${stableKeyHash(toolName)}`; + `${operationKeyPrefix(integration)}${stableKeyHash(toolName)}`; const legacyOperationKey = (integration: string, toolName: string): string => `${integration}.${toolName}`; @@ -148,13 +151,20 @@ export const makeDefaultOpenapiStore = ({ pluginStorage, blobs }: StorageDeps): }); const listRows = (integration: string) => - pluginStorage - .list({ collection: OPERATION_COLLECTION }) - .pipe( - Effect.map((rows: readonly PluginStorageEntry[]) => - rows.filter((row) => rowToOperation(row)?.integration === integration), - ), - ); + Effect.gen(function* () { + const current = yield* pluginStorage.list({ + collection: OPERATION_COLLECTION, + keyPrefix: operationKeyPrefix(integration), + }); + const legacy = yield* pluginStorage.list({ + collection: OPERATION_COLLECTION, + keyPrefix: `${integration}.`, + }); + return [...current, ...legacy].filter((row) => { + const decoded = decodeOperationStorage(row.data); + return Option.isSome(decoded) && decoded.value.integration === integration; + }); + }); const removeOperations = (integration: string) => Effect.gen(function* () { From 56566c7c8d438f13b56afe466b3652b541568dc6 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 5 Sep 2026 17:03:48 +0530 Subject: [PATCH 2/4] fix(openapi): reject pathless updates before storage changes --- e2e/scenarios/openapi-large-catalog.test.ts | 23 +++++++++++++++++++++ packages/plugins/openapi/src/sdk/plugin.ts | 10 +++++++++ 2 files changed, 33 insertions(+) diff --git a/e2e/scenarios/openapi-large-catalog.test.ts b/e2e/scenarios/openapi-large-catalog.test.ts index 02ed0b03c7..9f84a9d2ba 100644 --- a/e2e/scenarios/openapi-large-catalog.test.ts +++ b/e2e/scenarios/openapi-large-catalog.test.ts @@ -94,6 +94,29 @@ scenario( }) .pipe(Effect.result); expect(failed._tag).toBe("Failure"); + const missingPaths = yield* http.openapi + .updateSpec({ + params: { slug }, + payload: { + spec: { + kind: "blob", + value: JSON.stringify({ + openapi: "3.1.0", + info: { title: "No paths", version: "1" }, + }), + }, + }, + }) + .pipe(Effect.result); + expect(missingPaths._tag).toBe("Failure"); + yield* http.connections.refresh({ + params: { + owner: "org", + integration: slug, + name: ConnectionName.make("main"), + }, + }); + expect(yield* http.tools.list({ query: { integration: slug } })).toHaveLength(1002); const intact = yield* http.openapi.updateSpec({ params: { slug: neighbor }, payload: { diff --git a/packages/plugins/openapi/src/sdk/plugin.ts b/packages/plugins/openapi/src/sdk/plugin.ts index 65f3d76c84..6f9175982f 100644 --- a/packages/plugins/openapi/src/sdk/plugin.ts +++ b/packages/plugins/openapi/src/sdk/plugin.ts @@ -785,6 +785,11 @@ export const openApiPlugin = definePlugin< // `BEGIN` across a network fetch is the Hyperdrive deadlock path. const resolved = yield* resolveSpecForInput(config, httpClientLayer); const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); + if (document && !document.paths) { + return yield* new OpenApiExtractionError({ + message: "OpenAPI document has no paths defined", + }); + } const adapter = yield* resolveSpecFormatAdapter( options?.specFormats ?? [], config.specFormat, @@ -1005,6 +1010,11 @@ export const openApiPlugin = definePlugin< httpClientLayer, ); const document = resolved.keepPathItem ? undefined : yield* parse(resolved.specText); + if (document && !document.paths) { + return yield* new OpenApiExtractionError({ + message: "OpenAPI document has no paths defined", + }); + } const previousNames = new Set( (yield* ctx.storage.listOperations(rawSlug)).map((op) => op.toolName), From 6001383e6771116896198d6de825f2a9ad54d603 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 5 Sep 2026 17:13:03 +0530 Subject: [PATCH 3/4] fix(openapi): deduplicate legacy reads and preserve atomic upserts (greptile) --- .changeset/openapi-large-catalog-memory.md | 3 +- .../core/fumadb/src/adapters/drizzle/query.ts | 38 ++++++---------- .../drizzle/upsert-many-generic.test.ts | 43 ------------------- .../plugins/openapi/src/sdk/store.test.ts | 16 +++++++ packages/plugins/openapi/src/sdk/store.ts | 13 +++--- 5 files changed, 39 insertions(+), 74 deletions(-) diff --git a/.changeset/openapi-large-catalog-memory.md b/.changeset/openapi-large-catalog-memory.md index 1e171b63de..09ce7d772e 100644 --- a/.changeset/openapi-large-catalog-memory.md +++ b/.changeset/openapi-large-catalog-memory.md @@ -1,7 +1,6 @@ --- "@executor-js/sdk": patch "@executor-js/plugin-openapi": patch -"@executor-js/fumadb": patch --- -Reduce large OpenAPI import memory by compiling operation bindings in chunks and filtering integration storage prefixes in SQL. Construct D1 upsert statements lazily in bounded native batches. +Reduce large OpenAPI import memory by compiling operation bindings in chunks and filtering integration storage prefixes in SQL. diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts index ec4fed0cc7..cadf7fce2e 100644 --- a/packages/core/fumadb/src/adapters/drizzle/query.ts +++ b/packages/core/fumadb/src/adapters/drizzle/query.ts @@ -28,12 +28,6 @@ type P_DBType = PostgreSQL.PgDatabase< >; const CREATE_MANY_BATCH_SIZE = 500; -// A D1 native batch is one Workers RPC. Large schemas can make a few hundred -// prepared statements exceed the platform's 32 MiB serialized-argument cap, -// even though every individual statement fits its bound-parameter limit. -// Bound statement construction and RPC group size. Each group commits -// independently; this count does not bound the size of an individual row. -const D1_NATIVE_BATCH_STATEMENT_LIMIT = 50; // A multi-row write binds (rows * columns) parameters in one statement, and // engines cap bound parameters per statement (older SQLite: 999, Cloudflare @@ -609,7 +603,8 @@ export function fromDrizzle( ]), ); - const buildStatements = function* (handle: typeof db): Generator { + const buildStatements = (handle: typeof db): unknown[] => { + const statements: unknown[] = []; for (let i = 0; i < values.length; i += batchSize) { const batch = values.slice(i, i + batchSize); const insert = handle.insert(drizzleTable).values(batch) as unknown as { @@ -619,12 +614,15 @@ export function fromDrizzle( readonly where?: typeof where; }) => unknown; }; - yield insert.onConflictDoUpdate({ - target, - set, - ...(where === undefined ? {} : { where }), - }); + statements.push( + insert.onConflictDoUpdate({ + target, + set, + ...(where === undefined ? {} : { where }), + }), + ); } + return statements; }; const executeStatements = async (handle: typeof db) => { for (const statement of buildStatements(handle)) { @@ -634,22 +632,14 @@ export function fromDrizzle( const statementCount = Math.ceil(values.length / batchSize); // D1 rejects interactive transactions but its native batch API executes - // each prepared-statement group as one transaction. Drizzle exposes that - // API on the database handle, so keep bounded groups atomic instead of - // auto-committing each statement independently. + // prepared statements as one transaction. Drizzle exposes that API on + // the database handle, so keep parameter-bounded upserts atomic instead + // of auto-committing each statement independently. const nativeBatch = db as unknown as { readonly batch?: (statements: readonly unknown[]) => Promise; }; if (!interactiveTransactions && statementCount > 1 && nativeBatch.batch) { - let statements: unknown[] = []; - for (const statement of buildStatements(db)) { - statements.push(statement); - if (statements.length >= D1_NATIVE_BATCH_STATEMENT_LIMIT) { - await nativeBatch.batch(statements); - statements = []; - } - } - if (statements.length > 0) await nativeBatch.batch(statements); + await nativeBatch.batch(buildStatements(db)); return; } diff --git a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts index 82309bb8d2..333d4a2f7d 100644 --- a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts +++ b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts @@ -1,5 +1,4 @@ import { expect, test } from "@effect/vitest"; -import { sqliteTable, text } from "drizzle-orm/sqlite-core"; import { column, idColumn, schema, table } from "../../schema"; import { fromDrizzle } from "./query"; @@ -119,45 +118,3 @@ test("generic bulk upsert rolls earlier rows back when a later row fails", async expect(committed).toEqual([]); expect(events).toContain("transaction:rollback"); }); - -test("D1 bulk upsert bounds native batch RPC statement counts", async () => { - const drizzleRows = sqliteTable("generic_rows", { - id: text("id").primaryKey(), - value: text("value").notNull(), - }); - const batchSizes: number[] = []; - let constructed = 0; - let completed = 0; - let maxPending = 0; - const db = { - _: { fullSchema: { rows: drizzleRows } }, - insert: () => ({ - values: (values: readonly Record[]) => ({ - onConflictDoUpdate: () => { - constructed++; - maxPending = Math.max(maxPending, constructed - completed); - return { values }; - }, - }), - }), - batch: async (statements: readonly unknown[]) => { - batchSizes.push(statements.length); - completed += statements.length; - }, - }; - const orm = fromDrizzle(v1, db, "sqlite", false, 2); - const rows = v1.tables.rows; - - await orm.internal.upsertMany?.(rows, { - target: [rows.columns.id], - update: [rows.columns.value], - values: Array.from({ length: 121 }, (_, index) => ({ - id: `r${index}`, - value: `value-${index}`, - })), - }); - - expect(batchSizes).toEqual([50, 50, 21]); - expect(constructed).toBe(121); - expect(maxPending).toBe(50); -}); diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index 22b8e7fa10..6124df814f 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -242,6 +242,22 @@ describe("OpenAPI operation store", () => { "microsoft_graph.", ]).flat(), ); + yield* store.appendOperations("op", [{ integration: "op", toolName: "current", binding }]); + yield* pluginStorage.put({ + owner: "org", + collection: "operation", + key: "op.legacy", + data: { integration: "op", toolName: "legacy", binding: encodeBinding(binding) }, + }); + expect((yield* store.listOperations("op")).map((op) => op.toolName)).toEqual([ + "current", + "legacy", + ]); + yield* store.removeOperations("op"); + expect(yield* store.listOperations("op")).toEqual([]); + expect((yield* store.getOperation("microsoft_graph.other", "legacy"))?.toolName).toBe( + "legacy", + ); }), ); }); diff --git a/packages/plugins/openapi/src/sdk/store.ts b/packages/plugins/openapi/src/sdk/store.ts index 45bd9df57b..a63410c0db 100644 --- a/packages/plugins/openapi/src/sdk/store.ts +++ b/packages/plugins/openapi/src/sdk/store.ts @@ -152,18 +152,21 @@ export const makeDefaultOpenapiStore = ({ pluginStorage, blobs }: StorageDeps): const listRows = (integration: string) => Effect.gen(function* () { + const currentPrefix = operationKeyPrefix(integration); const current = yield* pluginStorage.list({ collection: OPERATION_COLLECTION, - keyPrefix: operationKeyPrefix(integration), + keyPrefix: currentPrefix, }); const legacy = yield* pluginStorage.list({ collection: OPERATION_COLLECTION, keyPrefix: `${integration}.`, }); - return [...current, ...legacy].filter((row) => { - const decoded = decodeOperationStorage(row.data); - return Option.isSome(decoded) && decoded.value.integration === integration; - }); + return [...current, ...legacy.filter((row) => !row.key.startsWith(currentPrefix))].filter( + (row) => { + const decoded = decodeOperationStorage(row.data); + return Option.isSome(decoded) && decoded.value.integration === integration; + }, + ); }); const removeOperations = (integration: string) => From 0fae0d2e4a7dab8b3e36469a35595b98f9d11766 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 5 Sep 2026 17:18:58 +0530 Subject: [PATCH 4/4] test(openapi): enforce bounded import and update writes (greptile) --- .../plugins/openapi/src/sdk/plugin.test.ts | 37 ++++++++++++++++++- 1 file changed, 36 insertions(+), 1 deletion(-) diff --git a/packages/plugins/openapi/src/sdk/plugin.test.ts b/packages/plugins/openapi/src/sdk/plugin.test.ts index 1617e8b555..a41182fad9 100644 --- a/packages/plugins/openapi/src/sdk/plugin.test.ts +++ b/packages/plugins/openapi/src/sdk/plugin.test.ts @@ -38,6 +38,7 @@ import { } from "@executor-js/sdk/testing"; import { openApiPlugin } from "./plugin"; +import { makeDefaultOpenapiStore } from "./store"; import { type AuthenticationInput } from "./types"; import { addOpenApiTestConnection, @@ -1125,7 +1126,26 @@ paths: "addSpec accepts Microsoft Graph-scale operation catalogs from one spec", () => Effect.gen(function* () { - const executor = yield* createExecutor(makeTestConfig({ plugins: testPlugins() })); + const batchSizes: number[] = []; + const executor = yield* createExecutor( + makeTestConfig({ + plugins: [ + openApiPlugin({ + storage: (deps) => + makeDefaultOpenapiStore({ + ...deps, + pluginStorage: { + ...deps.pluginStorage, + putMany: (input) => { + batchSizes.push(input.entries.length); + return deps.pluginStorage.putMany(input); + }, + }, + }), + }), + ] as const, + }), + ); const added = yield* executor.openapi.addSpec({ spec: { kind: "blob", value: microsoftGraphScaleSpecText() }, @@ -1134,6 +1154,21 @@ paths: }); expect(added.toolCount).toBe(MICROSOFT_GRAPH_V1_OPERATION_COUNT); + expect(batchSizes.length).toBeGreaterThan(1); + expect(Math.max(...batchSizes)).toBeLessThanOrEqual(500); + expect(batchSizes.reduce((sum, count) => sum + count, 0)).toBe( + MICROSOFT_GRAPH_V1_OPERATION_COUNT, + ); + batchSizes.length = 0; + const updated = yield* executor.openapi.updateSpec("microsoft_graph_scale", { + spec: { kind: "blob", value: microsoftGraphScaleSpecText() }, + }); + expect(updated.toolCount).toBe(MICROSOFT_GRAPH_V1_OPERATION_COUNT); + expect(batchSizes.length).toBeGreaterThan(1); + expect(Math.max(...batchSizes)).toBeLessThanOrEqual(500); + expect(batchSizes.reduce((sum, count) => sum + count, 0)).toBe( + MICROSOFT_GRAPH_V1_OPERATION_COUNT, + ); }), 30_000, );