`executor.tools.list.annotations` was spawning ~8 serial storage round-trips per request (p90 1.9s, ~85% of the total list span). Axiom traces showed 7.79 child `executor.storage.find_many` calls on average, up to 11, each ~200-300ms and stacked back-to-back. Two nested serial loops produced this: - executor.ts ran `resolveAnnotations` one (plugin_id, source_id) group at a time inside a `for..yield*` loop. - openapi / graphql / google-discovery plugins each ran `listOperationsBySource` / `getBindingsForSource` one scope at a time for shadowed sources. Both sets of calls are independent reads, so replace the serial loops with `Effect.forEach(..., { concurrency: "unbounded" })`. Expected annotations span to collapse from sum-of-children to max-of-children (~300-400ms), taking `executor.tools.list` p90 from ~2.2s to ~700-800ms.
Unified · Split
| @@ -1844,18 +1844,30 @@ export const createExecutor = < · 1843 unmodified lines | |||
| 1844 | else groups.set(key, [row]); | 1844 | else groups.set(key, [row]); |
| 1845 | } | 1845 | } |
| 1846 | 1846 | ||
| 1847 | for (const [key, groupRows] of groups) { | 1847 | // Each (plugin_id, source_id) group is an independent DB read, |
| 1848 | const [pluginId, sourceId] = key.split("\u0000") as [ | 1848 | // so fan them out concurrently. Yielding them serially stacks |
| 1849 | string, | 1849 | // ~200-300ms storage round-trips end-to-end and dominates the |
| 1850 | string, | 1850 | // `executor.tools.list.annotations` span. |
| 1851 | ]; | 1851 | const maps = yield* Effect.forEach( |
| 1852 | const runtime = runtimes.get(pluginId); | 1852 | [...groups], |
| 1853 | if (!runtime?.plugin.resolveAnnotations) continue; | 1853 | ([key, groupRows]) => |
| 1854 | const map = yield* runtime.plugin.resolveAnnotations({ | 1854 | Effect.gen(function* () { |
| 1855 | ctx: runtime.ctx, | 1855 | const [pluginId, sourceId] = key.split("\u0000") as [ |
| 1856 | sourceId, | 1856 | string, |
| 1857 | toolRows: groupRows, | 1857 | string, |
| 1858 | }); | 1858 | ]; |
| 1859 | const runtime = runtimes.get(pluginId); | ||
| 1860 | if (!runtime?.plugin.resolveAnnotations) return undefined; | ||
| 1861 | return yield* runtime.plugin.resolveAnnotations({ | ||
| 1862 | ctx: runtime.ctx, | ||
| 1863 | sourceId, | ||
| 1864 | toolRows: groupRows, | ||
| 1865 | }); | ||
| 1866 | }), | ||
| 1867 | { concurrency: "unbounded" }, | ||
| 1868 | ); | ||
| 1869 | for (const map of maps) { | ||
| 1870 | if (!map) continue; | ||
| 1859 | for (const [toolId, annotations] of Object.entries(map)) { | 1871 | for (const [toolId, annotations] of Object.entries(map)) { |
| 1860 | result.set(toolId, annotations); | 1872 | result.set(toolId, annotations); |
| 1861 | } | 1873 | } |
| @@ -592,11 +592,20 @@ export const googleDiscoveryPlugin = definePlugin(() => ({ · 591 unmodified lines | |||
| 592 | const typedCtx = ctx as PluginCtx<GoogleDiscoveryStore>; | 592 | const typedCtx = ctx as PluginCtx<GoogleDiscoveryStore>; |
| 593 | const scopes = new Set<string>(); | 593 | const scopes = new Set<string>(); |
| 594 | for (const row of toolRows) scopes.add(row.scope_id as string); | 594 | for (const row of toolRows) scopes.add(row.scope_id as string); |
| 595 | const byScope = new Map<string, ReadonlyMap<string, GoogleDiscoveryMethodBinding>>(); | 595 | // One getBindingsForSource per scope is independent storage |
| 596 | for (const scope of scopes) { | 596 | // work; run them in parallel so a shadowed source doesn't |
| 597 | const bindings = yield* typedCtx.storage.getBindingsForSource(sourceId, scope); | 597 | // serialise two ~200ms reads back-to-back in the caller's |
| 598 | byScope.set(scope, bindings); | 598 | // `executor.tools.list.annotations` span. |
| 599 | } | 599 | const entries = yield* Effect.forEach( |
| 600 | [...scopes], | ||
| 601 | (scope) => | ||
| 602 | Effect.gen(function* () { | ||
| 603 | const bindings = yield* typedCtx.storage.getBindingsForSource(sourceId, scope); | ||
| 604 | return [scope, bindings] as const; | ||
| 605 | }), | ||
| 606 | { concurrency: "unbounded" }, | ||
| 607 | ); | ||
| 608 | const byScope = new Map<string, ReadonlyMap<string, GoogleDiscoveryMethodBinding>>(entries); | ||
| 600 | const out: Record<string, ToolAnnotations> = {}; | 609 | const out: Record<string, ToolAnnotations> = {}; |
| 601 | for (const row of toolRows) { | 610 | for (const row of toolRows) { |
| 602 | const binding = byScope.get(row.scope_id as string)?.get(row.id); | 611 | const binding = byScope.get(row.scope_id as string)?.get(row.id); |
| @@ -534,16 +534,25 @@ export const graphqlPlugin = definePlugin( · 533 unmodified lines | |||
| 534 | for (const row of toolRows as readonly ToolRow[]) { | 534 | for (const row of toolRows as readonly ToolRow[]) { |
| 535 | scopes.add(row.scope_id as string); | 535 | scopes.add(row.scope_id as string); |
| 536 | } | 536 | } |
| 537 | const byScope = new Map<string, Map<string, OperationBinding>>(); | 537 | // One listOperationsBySource per scope is independent storage |
| 538 | for (const scope of scopes) { | 538 | // work; run them in parallel so a shadowed source doesn't |
| 539 | const ops = yield* ctx.storage.listOperationsBySource( | 539 | // serialise two ~200ms reads back-to-back in the caller's |
| 540 | sourceId, | 540 | // `executor.tools.list.annotations` span. |
| 541 | scope, | 541 | const entries = yield* Effect.forEach( |
| 542 | ); | 542 | [...scopes], |
| 543 | const byId = new Map<string, OperationBinding>(); | 543 | (scope) => |
| 544 | for (const op of ops) byId.set(op.toolId, op.binding); | 544 | Effect.gen(function* () { |
| 545 | byScope.set(scope, byId); | 545 | const ops = yield* ctx.storage.listOperationsBySource( |
| 546 | } | 546 | sourceId, |
| 547 | scope, | ||
| 548 | ); | ||
| 549 | const byId = new Map<string, OperationBinding>(); | ||
| 550 | for (const op of ops) byId.set(op.toolId, op.binding); | ||
| 551 | return [scope, byId] as const; | ||
| 552 | }), | ||
| 553 | { concurrency: "unbounded" }, | ||
| 554 | ); | ||
| 555 | const byScope = new Map<string, Map<string, OperationBinding>>(entries); | ||
| 547 | 556 | ||
| 548 | const out: Record<string, ToolAnnotations> = {}; | 557 | const out: Record<string, ToolAnnotations> = {}; |
| 549 | for (const row of toolRows as readonly ToolRow[]) { | 558 | for (const row of toolRows as readonly ToolRow[]) { |
| @@ -1017,13 +1017,22 @@ export const openApiPlugin = definePlugin( · 1016 unmodified lines | |||
| 1017 | for (const row of toolRows as readonly ToolRow[]) { | 1017 | for (const row of toolRows as readonly ToolRow[]) { |
| 1018 | scopes.add(row.scope_id as string); | 1018 | scopes.add(row.scope_id as string); |
| 1019 | } | 1019 | } |
| 1020 | const byScope = new Map<string, Map<string, OperationBinding>>(); | 1020 | // One listOperationsBySource per scope is independent storage |
| 1021 | for (const scope of scopes) { | 1021 | // work; run them in parallel so a shadowed source doesn't |
| 1022 | const ops = yield* ctx.storage.listOperationsBySource(sourceId, scope); | 1022 | // serialise two ~200ms reads back-to-back in the caller's |
| 1023 | const byId = new Map<string, OperationBinding>(); | 1023 | // `executor.tools.list.annotations` span. |
| 1024 | for (const op of ops) byId.set(op.toolId, op.binding); | 1024 | const entries = yield* Effect.forEach( |
| 1025 | byScope.set(scope, byId); | 1025 | [...scopes], |
| 1026 | } | 1026 | (scope) => |
| 1027 | Effect.gen(function* () { | ||
| 1028 | const ops = yield* ctx.storage.listOperationsBySource(sourceId, scope); | ||
| 1029 | const byId = new Map<string, OperationBinding>(); | ||
| 1030 | for (const op of ops) byId.set(op.toolId, op.binding); | ||
| 1031 | return [scope, byId] as const; | ||
| 1032 | }), | ||
| 1033 | { concurrency: "unbounded" }, | ||
| 1034 | ); | ||
| 1035 | const byScope = new Map<string, Map<string, OperationBinding>>(entries); | ||
| 1027 | 1036 | ||
| 1028 | const out: Record<string, ToolAnnotations> = {}; | 1037 | const out: Record<string, ToolAnnotations> = {}; |
| 1029 | for (const row of toolRows as readonly ToolRow[]) { | 1038 | for (const row of toolRows as readonly ToolRow[]) { |