Scope the stale catalog sync to the requested integration; harden the narrowed-read and lost-claim tests; release fumadb

Author: RhysSullivan

.changeset/mcp-passthrough-mode.md

@@ -1,5 +1,6 @@
 ---
 "@executor-js/sdk": minor
+"@executor-js/fumadb": minor
 "@executor-js/plugin-openapi": patch
 "@executor-js/plugin-graphql": patch
 "@executor-js/plugin-mcp": patch

packages/core/sdk/src/executor.test.ts

@@ -54,6 +54,10 @@ const addr = (tool: string): ToolAddress => ToolAddress.make(`tools.${INTEG}.org
 // resolveTools (with shared $defs), and supports ctx.transaction rollback.
 // ---------------------------------------------------------------------------
 
+/** Toggled by a test so the demo plugin's next discovery differs from what
+ *  is persisted. Module-level because the plugin closure is created once. */
+const demoDiscoversExtra = { value: false };
+
 const demoPlugin = definePlugin(() => ({
   id: "demo" as const,
   credentialProviders: [memoryProvider()],
@@ -68,6 +72,11 @@ const demoPlugin = definePlugin(() => ({
   resolveTools: () =>
     Effect.succeed({
       tools: [
+        // A test may flip this to make one discovery differ from the last
+        // persisted catalog (see the lost-claim rebuild test).
+        ...(demoDiscoversExtra.value
+          ? [{ name: ToolName.make("extra"), description: "extra" }]
+          : []),
         {
           name: ToolName.make("inspect"),
           description: "inspect",
@@ -1020,13 +1029,15 @@ describe("createExecutor", () => {
           config.db.findMany("tool", { where: (b) => b("integration", "=", String(INTEG)) }),
         );
 
-        // "A" rebuilds; the proxy plays "B" between A's claim and A's unit.
+        // "A" rebuilds — and discovers something NEW (`extra`) that the
+        // persisted catalog does not have, so a caller handed A's discovery
+        // instead of the persisted rows is distinguishable. The proxy plays
+        // "B" between A's claim and A's unit.
+        demoDiscoversExtra.value = true;
         raceState.armed = true;
-        const reported = yield* executor.connections.refresh({
-          owner: "org",
-          integration: INTEG,
-          name: CONN,
-        });
+        const reported = yield* executor.connections
+          .refresh({ owner: "org", integration: INTEG, name: CONN })
+          .pipe(Effect.ensuring(Effect.sync(() => void (demoDiscoversExtra.value = false))));
         expect(raceState.armed, "B interleaved").toBe(false);
         expect(raceState.applied, "A's fenced unit was discarded").toBe(false);
 
@@ -1040,9 +1051,14 @@ describe("createExecutor", () => {
         );
         // And A reported the persisted rows, not the ones it discovered and
         // failed to write — so its caller cannot disagree with the next list.
+        // A discovered `extra`; persisted has no `extra`; the report must not.
         expect(
-          reported.map((tool) => String(tool.address)).sort(),
+          reported.map((tool) => String(tool.name)).sort(),
           "refresh reports what is persisted",
+        ).toEqual(["inspect", "run"]);
+        expect(
+          reported.map((tool) => String(tool.address)).sort(),
+          "refresh reports the persisted addresses",
         ).toEqual(
           rowsAfter
             .map((row) => `tools.${row.integration}.${row.owner}.${row.connection}.${row.name}`)

packages/core/sdk/src/executor.ts

@@ -5555,123 +5555,137 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     //    is older than the freshness TTL.
     // Best-effort: a failed rebuild leaves the stale-but-working catalog in
     // place and retries on the next read.
-    const syncStaleConnectionTools = Effect.gen(function* () {
-      // The platform view can never persist a rebuilt catalog (writes are
-      // denied at the storage boundary), so attempting the sync would only
-      // fire upstream `resolveTools` calls whose results are thrown away —
-      // network side effects on a read-only credential. Skip it entirely:
-      // read-only-ness of the platform read path is a stated invariant here,
-      // not an accident of the best-effort catch below.
-      if (config.platformView === true) return;
-      const integrations = yield* core.findMany("integration", {});
-      if (integrations.length === 0) return;
-      const integrationBySlug = new Map(integrations.map((row) => [row.slug, row] as const));
-      // The TTL only matters when a loaded plugin actually lists a live remote
-      // catalog; otherwise skip it so age alone never widens the stale query.
-      const anyRemoteCatalog = Array.from(runtimes.values()).some(
-        (runtime) => runtime.plugin.remoteToolCatalog === true,
-      );
-      const cutoff =
-        toolsSyncTtlMs == null || !anyRemoteCatalog ? null : Date.now() - toolsSyncTtlMs;
-
-      // Bound the scan to potentially-stale rows: stale-marked (NULL stamp) or
-      // synced before the latest instant any trigger could fire at (the TTL
-      // cutoff / the newest config revision). Per-row trigger checks below
-      // re-verify against each row's own integration; in steady state this
-      // query returns nothing and the read pays one indexed lookup.
-      const latestRevision = integrations.reduce<number | null>(
-        (max, row) =>
-          row.config_revised_at == null
-            ? max
-            : Math.max(max ?? Number(row.config_revised_at), Number(row.config_revised_at)),
-        null,
-      );
-      const staleBefore =
-        cutoff === null && latestRevision === null
-          ? null
-          : Math.max(cutoff ?? Number.MIN_SAFE_INTEGER, latestRevision ?? Number.MIN_SAFE_INTEGER);
-
-      // A connection with no catalog manifest (built before the manifest
-      // existed) is stale too: `tools.describeAll` refuses it until a rebuild
-      // stamps one, so the read that first sees it is the read that fixes it.
-      const connections = yield* core.findMany("connection", {
-        where: (b: AnyCb) =>
-          staleBefore === null
-            ? b.or(b.isNull("tools_synced_at"), b.isNull("tools_manifest"))
-            : b.or(
-                b.isNull("tools_synced_at"),
-                b.isNull("tools_manifest"),
-                b("tools_synced_at", "<", staleBefore),
-              ),
-      });
-      // Each rebuild is an independent upstream listing, so they run together
-      // rather than one after another: a host with many stale remote-catalog
-      // connections otherwise pays the sum of every server's latency on the
-      // read that trips the TTL. Only the listings overlap — `persistCatalog`
-      // keeps the catalog writes in a single-file queue, so this fan-out never
-      // opens two transactions on a one-connection database.
-      const rebuilds: Effect.Effect<readonly Tool[]>[] = [];
-      for (const connection of connections) {
-        const integrationRow = integrationBySlug.get(connection.integration);
-        if (!integrationRow) continue;
-        const runtime = runtimes.get(integrationRow.plugin_id);
-        // Only re-produce catalogs this executor can actually re-list —
-        // rebuilding under an unloaded plugin would clear a working catalog.
-        // (A loaded plugin without `resolveTools` still flows through:
-        // `produceConnectionTools` runs its clear-and-stamp cleanup path.)
-        if (!runtime) continue;
-
-        const syncedAt =
-          connection.tools_synced_at == null ? null : Number(connection.tools_synced_at);
-        const revisedTime =
-          integrationRow.config_revised_at == null
+    // `scope` narrows the sync to one integration's connections. A read that
+    // only asked for that integration (`describeAll({ integration })`) must not
+    // scan, dial, or wait on every other stale connection in the workspace.
+    const syncStaleConnectionToolsScoped = (scope?: { readonly integration: IntegrationSlug }) =>
+      Effect.gen(function* () {
+        // The platform view can never persist a rebuilt catalog (writes are
+        // denied at the storage boundary), so attempting the sync would only
+        // fire upstream `resolveTools` calls whose results are thrown away —
+        // network side effects on a read-only credential. Skip it entirely:
+        // read-only-ness of the platform read path is a stated invariant here,
+        // not an accident of the best-effort catch below.
+        if (config.platformView === true) return;
+        const integrations = yield* core.findMany("integration", {
+          where: (b: AnyCb) =>
+            scope === undefined ? true : b("slug", "=", String(scope.integration)),
+        });
+        if (integrations.length === 0) return;
+        const integrationBySlug = new Map(integrations.map((row) => [row.slug, row] as const));
+        // The TTL only matters when a loaded plugin actually lists a live remote
+        // catalog; otherwise skip it so age alone never widens the stale query.
+        const anyRemoteCatalog = Array.from(runtimes.values()).some(
+          (runtime) => runtime.plugin.remoteToolCatalog === true,
+        );
+        const cutoff =
+          toolsSyncTtlMs == null || !anyRemoteCatalog ? null : Date.now() - toolsSyncTtlMs;
+
+        // Bound the scan to potentially-stale rows: stale-marked (NULL stamp) or
+        // synced before the latest instant any trigger could fire at (the TTL
+        // cutoff / the newest config revision). Per-row trigger checks below
+        // re-verify against each row's own integration; in steady state this
+        // query returns nothing and the read pays one indexed lookup.
+        const latestRevision = integrations.reduce<number | null>(
+          (max, row) =>
+            row.config_revised_at == null
+              ? max
+              : Math.max(max ?? Number(row.config_revised_at), Number(row.config_revised_at)),
+          null,
+        );
+        const staleBefore =
+          cutoff === null && latestRevision === null
             ? null
-            : Number(integrationRow.config_revised_at);
-
-        const staleMarked = syncedAt === null || connection.tools_manifest == null;
-        const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime;
-        const expired =
-          cutoff !== null &&
-          runtime.plugin.remoteToolCatalog === true &&
-          syncedAt !== null &&
-          syncedAt < cutoff;
-        if (!staleMarked && !configRevised && !expired) continue;
+            : Math.max(
+                cutoff ?? Number.MIN_SAFE_INTEGER,
+                latestRevision ?? Number.MIN_SAFE_INTEGER,
+              );
 
-        rebuilds.push(
-          produceConnectionTools(
-            integrationRow,
-            {
-              owner: connection.owner as Owner,
-              integration: IntegrationSlug.make(connection.integration),
-              name: ConnectionName.make(connection.name),
-            },
-            "background",
-          ).pipe(
-            // Best-effort, but never silent: the read still succeeds on the
-            // stale-but-working catalog and the peer rebuilds still finish,
-            // while the operator gets the connection that failed and why.
-            // Without this a connection whose upstream is permanently broken
-            // re-fails on every read and leaves no trace anywhere.
-            Effect.catch((error) =>
-              Effect.logWarning("executor stale tool sync failed", {
-                integration: connection.integration,
-                connection: connection.name,
-                error: describeSyncFailure(error),
-              }).pipe(Effect.as([] as readonly Tool[])),
+        // A connection with no catalog manifest (built before the manifest
+        // existed) is stale too: `tools.describeAll` refuses it until a rebuild
+        // stamps one, so the read that first sees it is the read that fixes it.
+        const connections = yield* core.findMany("connection", {
+          where: (b: AnyCb) =>
+            b.and(
+              scope === undefined ? true : b("integration", "=", String(scope.integration)),
+              staleBefore === null
+                ? b.or(b.isNull("tools_synced_at"), b.isNull("tools_manifest"))
+                : b.or(
+                    b.isNull("tools_synced_at"),
+                    b.isNull("tools_manifest"),
+                    b("tools_synced_at", "<", staleBefore),
+                  ),
             ),
-            Effect.withSpan("executor.tools.sync_stale", {
-              attributes: {
-                "executor.integration": connection.integration,
-                "executor.connection": connection.name,
+        });
+        // Each rebuild is an independent upstream listing, so they run together
+        // rather than one after another: a host with many stale remote-catalog
+        // connections otherwise pays the sum of every server's latency on the
+        // read that trips the TTL. Only the listings overlap — `persistCatalog`
+        // keeps the catalog writes in a single-file queue, so this fan-out never
+        // opens two transactions on a one-connection database.
+        const rebuilds: Effect.Effect<readonly Tool[]>[] = [];
+        for (const connection of connections) {
+          const integrationRow = integrationBySlug.get(connection.integration);
+          if (!integrationRow) continue;
+          const runtime = runtimes.get(integrationRow.plugin_id);
+          // Only re-produce catalogs this executor can actually re-list —
+          // rebuilding under an unloaded plugin would clear a working catalog.
+          // (A loaded plugin without `resolveTools` still flows through:
+          // `produceConnectionTools` runs its clear-and-stamp cleanup path.)
+          if (!runtime) continue;
+
+          const syncedAt =
+            connection.tools_synced_at == null ? null : Number(connection.tools_synced_at);
+          const revisedTime =
+            integrationRow.config_revised_at == null
+              ? null
+              : Number(integrationRow.config_revised_at);
+
+          const staleMarked = syncedAt === null || connection.tools_manifest == null;
+          const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime;
+          const expired =
+            cutoff !== null &&
+            runtime.plugin.remoteToolCatalog === true &&
+            syncedAt !== null &&
+            syncedAt < cutoff;
+          if (!staleMarked && !configRevised && !expired) continue;
+
+          rebuilds.push(
+            produceConnectionTools(
+              integrationRow,
+              {
+                owner: connection.owner as Owner,
+                integration: IntegrationSlug.make(connection.integration),
+                name: ConnectionName.make(connection.name),
               },
-            }),
-          ),
-        );
-      }
-      yield* Effect.all(rebuilds, {
-        concurrency: STALE_TOOLS_SYNC_CONCURRENCY,
+              "background",
+            ).pipe(
+              // Best-effort, but never silent: the read still succeeds on the
+              // stale-but-working catalog and the peer rebuilds still finish,
+              // while the operator gets the connection that failed and why.
+              // Without this a connection whose upstream is permanently broken
+              // re-fails on every read and leaves no trace anywhere.
+              Effect.catch((error) =>
+                Effect.logWarning("executor stale tool sync failed", {
+                  integration: connection.integration,
+                  connection: connection.name,
+                  error: describeSyncFailure(error),
+                }).pipe(Effect.as([] as readonly Tool[])),
+              ),
+              Effect.withSpan("executor.tools.sync_stale", {
+                attributes: {
+                  "executor.integration": connection.integration,
+                  "executor.connection": connection.name,
+                },
+              }),
+            ),
+          );
+        }
+        yield* Effect.all(rebuilds, {
+          concurrency: STALE_TOOLS_SYNC_CONCURRENCY,
+        });
       });
-    });
+    const syncStaleConnectionTools = syncStaleConnectionToolsScoped();
 
     // How long a tools read waits for the stale sync before answering from
     // the persisted rows (`ExecutorConfig.toolsSyncGraceMs`; `null` blocks
@@ -5689,10 +5703,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     // failures log and the stale-but-working catalog stays — so the fork
     // swallows its scan errors the same way each rebuild already swallows its
     // own.
-    const awaitStaleSyncWithinGrace = (graceMs: number) =>
+    const awaitStaleSyncWithinGrace = (
+      graceMs: number,
+      scope?: { readonly integration: IntegrationSlug },
+    ) =>
       Effect.gen(function* () {
         const fiber = yield* Effect.forkDetach(
-          syncStaleConnectionTools.pipe(
+          syncStaleConnectionToolsScoped(scope).pipe(
             Effect.catch((error) =>
               Effect.logWarning("executor stale tool sync scan failed", {
                 error: describeSyncFailure(error),
@@ -5909,10 +5926,15 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
       filter?: ToolListFilter,
     ): Effect.Effect<readonly ToolProjection[], StorageFailure> =>
       Effect.gen(function* () {
+        // The stale sync is scoped to the requested integration, so a
+        // narrowed read never scans, dials, or waits on unrelated
+        // connections.
+        const scope =
+          filter?.integration === undefined ? undefined : { integration: filter.integration };
         if (toolsSyncGraceMs === null) {
-          yield* syncStaleConnectionTools;
+          yield* syncStaleConnectionToolsScoped(scope);
         } else {
-          yield* awaitStaleSyncWithinGrace(toolsSyncGraceMs);
+          yield* awaitStaleSyncWithinGrace(toolsSyncGraceMs, scope);
         }
         const integrationWhere = (b: AnyCb) =>
           b.and(

packages/hosts/mcp/src/passthrough-tools.test.ts

@@ -595,25 +595,34 @@ describe("passthrough mode server", () => {
 
   it("narrows to the requested integrations and says which were not connected", async () => {
     const { engine } = makeRecordingEngine();
+    // The filter is pushed into the READ: the port is asked once per slug and
+    // never for the whole workspace. Recording the calls is what proves it —
+    // a stub that merely honoured the filter would also pass an unfiltered
+    // read followed by host-side filtering.
+    const reads: (string | undefined)[] = [];
     await withClient(
       {
         engine,
         mode: "passthrough",
         passthroughIntegrations: ["linear", "notion"],
-        // The filter is pushed into the READ: the port is asked per slug, so
-        // a real catalog never describes tools the session will not serve.
         tools: {
-          describeAll: (filter) =>
-            Effect.succeed(
+          describeAll: (filter) => {
+            reads.push(filter?.integration === undefined ? undefined : String(filter.integration));
+            return Effect.succeed(
               CATALOG.filter(
                 (tool) =>
                   filter?.integration === undefined || tool.integration === filter.integration,
               ),
-            ),
+            );
+          },
         },
       },
       async (client) => {
         const listed = await client.listTools();
+        expect(reads.sort(), "one read per requested slug, none unfiltered").toEqual([
+          "linear",
+          "notion",
+        ]);
         expect(listed.tools.map((tool) => tool.name)).toEqual(["linear__issueCreate"]);
         const instructions = client.getInstructions() ?? "";
         expect(instructions).toContain("1 integration tool");