From 071b970cef87a8b42a20c5e31c03998d717cecd6 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 05:55:00 -0700 Subject: [PATCH 1/5] Scope the tools read catalog refresh to the read filter The stale-catalog scan ran unfiltered and refreshed connections sequentially, so a read for one integration paid for every stale connection in the workspace. Scope it to the read filter, narrow the TTL and config-revision triggers to the integrations they can fire for, project only the columns it reads, and refresh concurrently. A plugin defect can no longer fail the read. --- packages/core/sdk/src/core-schema.ts | 10 + packages/core/sdk/src/executor.ts | 281 ++++++++++++++++++--------- 2 files changed, 204 insertions(+), 87 deletions(-) diff --git a/packages/core/sdk/src/core-schema.ts b/packages/core/sdk/src/core-schema.ts index b03adf5dfc..baa71ee003 100644 --- a/packages/core/sdk/src/core-schema.ts +++ b/packages/core/sdk/src/core-schema.ts @@ -454,6 +454,16 @@ export const TOOL_INVOCATION_COLUMNS = [ "created_at", "updated_at", ] as const satisfies readonly (keyof ToolRow)[]; +/** The connection columns the tools read's catalog-refresh scan projects: the + * address `produceConnectionTools` needs plus the stamp its trigger check + * reads. Credentials, OAuth state and health JSON stay unread — the scan runs + * on every `tools.list`, and in steady state it must match no rows at all. */ +export const CONNECTION_CATALOG_SCAN_COLUMNS = [ + "owner", + "integration", + "name", + "tools_synced_at", +] as const satisfies readonly (keyof ConnectionRow)[]; export type DefinitionRow = FumaRow; export type ToolPolicyRow = FumaRow; export type ArtifactRow = FumaRow; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 56cb2e2997..9a6050294c 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -31,6 +31,7 @@ import { HealthCheckResult, HealthCheckSpec } from "./health-check"; import type { HealthCheckCandidate } from "./health-check"; import { ARTIFACT_SUMMARY_COLUMNS, + CONNECTION_CATALOG_SCAN_COLUMNS, coreSchema, isToolPolicyAction, TOOL_INVOCATION_COLUMNS, @@ -677,6 +678,22 @@ export interface ExecutorConfig [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( - (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); + const refreshCatalogsForRead = ( + filter: ToolListFilter | undefined, + ): Effect.Effect => + 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 NO_CATALOG_REFRESH; + const integrations = yield* core.findMany("integration", {}); + if (integrations.length === 0) return NO_CATALOG_REFRESH; + const integrationBySlug = new Map(integrations.map((row) => [row.slug, row] as const)); + + // Both age-based triggers are per-integration, so they scope to the + // integrations they can actually fire for rather than to every row old + // enough to qualify. Folding them into one global `staleBefore` made an + // OpenAPI connection older than the TTL a candidate whenever ANY loaded + // plugin listed a remote catalog, so the scan fetched rows that the + // per-row checks below then discarded. + const remoteSlugs = integrations + .filter((row) => runtimes.get(row.plugin_id)?.plugin.remoteToolCatalog === true) + .map((row) => row.slug); + const revisedSlugs = integrations + .filter((row) => row.config_revised_at != null) + .map((row) => row.slug); + const cutoff = + toolsSyncTtlMs == null || remoteSlugs.length === 0 ? null : Date.now() - toolsSyncTtlMs; + // One bound across all revised integrations: the per-row check + // re-verifies against each row's OWN revision, so over-selecting here + // costs a discarded row, never a spurious re-list. + const latestRevision = integrations.reduce( + (max, row) => + row.config_revised_at == null + ? max + : Math.max(max ?? Number(row.config_revised_at), Number(row.config_revised_at)), + null, + ); - const connections = yield* core.findMany("connection", { - where: (b: AnyCb) => - staleBefore === null - ? b.isNull("tools_synced_at") - : b.or(b.isNull("tools_synced_at"), b("tools_synced_at", "<", staleBefore)), - }); - 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; - const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime; - const expired = - cutoff !== null && - runtime.plugin.remoteToolCatalog === true && - syncedAt !== null && - syncedAt < cutoff; - if (!staleMarked && !configRevised && !expired) continue; + // Scoped to the same three fields the tool query below narrows by, so a + // read for one integration can never pay for another's handshakes. + // `filter.query` is deliberately absent: it is a post-read substring + // match over names, not a partition. + // + // An empty slug list would compile to `IN ()`, which is a syntax error + // on the SQL backends, so each arm folds to `false` instead (precedent: + // `blob.getMany`). The NULL-stamp arm always stands, so the OR can + // never degenerate to a bare `false`. + const connections = yield* core.findMany("connection", { + where: (b: AnyCb) => + b.and( + filter?.integration === undefined + ? true + : b("integration", "=", String(filter.integration)), + filter?.owner === undefined ? true : b("owner", "=", filter.owner), + filter?.connection === undefined ? true : b("name", "=", String(filter.connection)), + b.or( + b.isNull("tools_synced_at"), + revisedSlugs.length === 0 || latestRevision === null + ? false + : b.and( + b("integration", "in", revisedSlugs), + b("tools_synced_at", "<", latestRevision), + ), + remoteSlugs.length === 0 || cutoff === null + ? false + : b.and(b("integration", "in", remoteSlugs), b("tools_synced_at", "<", cutoff)), + ), + ), + select: CONNECTION_CATALOG_SCAN_COLUMNS, + }); - yield* produceConnectionTools( - integrationRow, - { - owner: connection.owner as Owner, - integration: IntegrationSlug.make(connection.integration), - name: ConnectionName.make(connection.name), - }, - "background", - ).pipe( - Effect.catch(() => Effect.succeed([] as readonly Tool[])), - Effect.withSpan("executor.tools.sync_stale", { - attributes: { - "executor.integration": connection.integration, - "executor.connection": connection.name, + const due: { + readonly integrationRow: IntegrationRow; + readonly ref: ConnectionRef; + readonly trigger: CatalogRefreshTrigger; + }[] = []; + 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; + const configRevised = revisedTime !== null && (syncedAt ?? 0) < revisedTime; + const expired = + cutoff !== null && + runtime.plugin.remoteToolCatalog === true && + syncedAt !== null && + syncedAt < cutoff; + if (!staleMarked && !configRevised && !expired) continue; + + due.push({ + integrationRow, + ref: { + owner: connection.owner as Owner, + integration: IntegrationSlug.make(connection.integration), + name: ConnectionName.make(connection.name), }, - }), + trigger: staleMarked ? "stale_marked" : configRevised ? "config_revised" : "expired", + }); + } + + // Bounded fan-out: these are independent upstream handshakes, and a + // sequential loop made a read cost the SUM of every candidate's + // round trip. The bound keeps a wide refresh from opening one socket + // per connection. + const outcomes = yield* Effect.forEach( + due, + ({ integrationRow, ref, trigger }) => + produceConnectionTools(integrationRow, ref, "background").pipe( + Effect.andThen(Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "ok" })), + Effect.as(true), + // catchCause, not catch: a plugin DEFECT is not in the error + // channel, and a tools READ must not be failable by one + // connection's misbehaving refresh. The stale-but-working + // catalog stays in place and the next read retries. + Effect.catchCause((cause) => + Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "fail" }).pipe( + Effect.andThen( + Effect.logDebug("Tool catalog refresh failed").pipe( + Effect.annotateLogs("cause", cause), + ), + ), + Effect.as(false), + ), + ), + Effect.withSpan("executor.tools.sync", { + attributes: { + "executor.integration": String(ref.integration), + "executor.connection": String(ref.name), + "executor.tools.sync.trigger": trigger, + }, + }), + ), + { concurrency: 4 }, ); - } - }); + + return { + candidates: connections.length, + synced: outcomes.filter((ok) => ok).length, + }; + }); const toolsList = (filter?: ToolListFilter): Effect.Effect => Effect.gen(function* () { - yield* syncStaleConnectionTools; + const refresh = yield* refreshCatalogsForRead(filter); // Projected: the list surface is metadata (address, description, // annotations) — loading every tool's input/output schema JSON made // an unbounded list scale with schema bytes, not tool count. @@ -3725,8 +3811,29 @@ export const createExecutor = Date: Sun, 16 Aug 2026 05:55:08 -0700 Subject: [PATCH 2/5] Push the search namespace into the tools read Empty-query enumeration filtered tools by exact integration slug after listing everything. Pass it as the read filter instead, so the read refreshes only that integration. Ranked search stays unscoped: its namespace match is token-prefix. --- packages/core/execution/src/tool-invoker.ts | 45 ++++++++++++++------- 1 file changed, 31 insertions(+), 14 deletions(-) diff --git a/packages/core/execution/src/tool-invoker.ts b/packages/core/execution/src/tool-invoker.ts index 4da9251767..594a550d9f 100644 --- a/packages/core/execution/src/tool-invoker.ts +++ b/packages/core/execution/src/tool-invoker.ts @@ -13,6 +13,7 @@ import { authToolFailure, isUserActionableError, isToolResult, + IntegrationSlug, ToolResult, ToolAddress, parseToolAddress, @@ -665,13 +666,17 @@ export const searchTools = Effect.fn("executor.tools.search")(function* ( }); const emptyQuery = normalizeSearchText(query).length === 0; - const hasNamespace = - options?.namespace !== undefined && normalizeSearchText(options.namespace).length > 0; + // The exact integration slug an empty-query enumeration reads under; null + // when the caller named no usable namespace. + const scope = + options?.namespace !== undefined && normalizeSearchText(options.namespace).length > 0 + ? IntegrationSlug.make(options.namespace.trim()) + : null; // An empty query with no namespace stays empty: it carries neither a // ranking signal nor a scope, and listing the whole workspace "by default" // is exactly the arbitrary dump the ranked search refuses to be. - if (emptyQuery && !hasNamespace) { + if (emptyQuery && scope === null) { return { items: [], total: 0, @@ -680,15 +685,26 @@ export const searchTools = Effect.fn("executor.tools.search")(function* ( } satisfies PagedResult; } - const all = yield* executor.tools.list({ includeAnnotations: false }).pipe( - Effect.mapError( - (cause) => - new ExecutionToolError({ - message: "Failed to list tools for search", - cause, - }), - ), - ); + // Enumeration's scope is an exact integration slug (see below), so it is the + // read's filter, not a post-read predicate: pushing it down lets the list + // narrow its own catalog refresh to that one integration instead of + // re-listing every stale connection in the workspace. Ranked search is + // deliberately NOT scoped here — `matchesNamespace` is token-prefix, so a + // scoped read would drop the prefix-sibling integrations it means to match. + const all = yield* executor.tools + .list({ + includeAnnotations: false, + ...(emptyQuery && scope !== null ? { integration: scope } : {}), + }) + .pipe( + Effect.mapError( + (cause) => + new ExecutionToolError({ + message: "Failed to list tools for search", + cause, + }), + ), + ); const searchable = all.map(toSearchableTool); // An empty query WITH a namespace is enumeration, not search: there is no @@ -698,10 +714,11 @@ export const searchTools = Effect.fn("executor.tools.search")(function* ( // sweep in prefix-sibling integrations (namespace "google" matching // google_gmail and google_sheets), which would silently break the census // guarantee: `total` here must reconcile against - // `executor.integrations.list`'s per-integration toolCount. + // `executor.integrations.list`'s per-integration toolCount. That exact match + // is now the `integration` filter on the read above, so `searchable` is + // already the namespace's catalog and re-filtering it here would be a no-op. const ranked: readonly ToolDiscoveryResult[] = emptyQuery ? searchable - .filter((tool) => tool.integration === options?.namespace?.trim()) .sort((left, right) => left.path.localeCompare(right.path)) .map((tool) => ({ path: tool.path, From 15137174f72a3ea0796d27c7c724ea7754a75883 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 05:55:08 -0700 Subject: [PATCH 3/5] Test that a tools read refreshes only its own filter scope --- .../core/sdk/src/tools-sync-scope.test.ts | 220 ++++++++++++++++++ 1 file changed, 220 insertions(+) create mode 100644 packages/core/sdk/src/tools-sync-scope.test.ts diff --git a/packages/core/sdk/src/tools-sync-scope.test.ts b/packages/core/sdk/src/tools-sync-scope.test.ts new file mode 100644 index 0000000000..81abb2421c --- /dev/null +++ b/packages/core/sdk/src/tools-sync-scope.test.ts @@ -0,0 +1,220 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { createExecutor } from "./executor"; +import type { FumaDb } from "./fuma-runtime"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + ProviderItemId, + ProviderKey, + ToolName, +} from "./ids"; +import { definePlugin } from "./plugin"; +import type { CredentialProvider } from "./provider"; +import { makeTestConfig } from "./testing"; + +// A tools READ refreshes stale catalogs before it answers, and each refresh is +// a live upstream handshake. These cases pin the two properties that keep that +// affordable: the refresh sees only connections inside the read's own filter, +// and in steady state it sees nothing at all. + +const INTEG_A = IntegrationSlug.make("alpha"); +const INTEG_B = IntegrationSlug.make("beta"); +const TEMPLATE = AuthTemplateSlug.make("apiKey"); + +const memoryProvider = (): CredentialProvider => { + const store = new Map(); + return { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + has: (id) => Effect.sync(() => store.has(String(id))), + list: () => + Effect.sync(() => + Array.from(store.keys()).map((key) => ({ id: ProviderItemId.make(key), name: key })), + ), + }; +}; + +/** A plugin owning two integrations that records every `resolveTools` it is + * asked for, as `/`. The recorded list IS the cost + * a read pays upstream, so the assertions below are about its contents. */ +const makeCountingPlugin = () => { + const resolved: string[] = []; + let dying = false; + const plugin = definePlugin(() => ({ + id: "counting" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: (input) => { + const slug = String(input.connection.integration); + resolved.push(`${slug}/${String(input.connection.name)}`); + return dying + ? // oxlint-disable-next-line executor/no-error-constructor -- boundary: a plugin defect IS a raw throw from third-party code, and reproducing it faithfully is the point of this case + Effect.die(new Error("resolveTools blew up")) + : Effect.succeed({ + tools: [{ name: ToolName.make(`${slug}_deploy`), description: "deploy" }], + }); + }, + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + Effect.all([ + ctx.core.integrations.register({ slug: INTEG_A, description: "Alpha", config: {} }), + ctx.core.integrations.register({ slug: INTEG_B, description: "Beta", config: {} }), + ]), + }), + }))(); + return { plugin, resolved, startDying: () => void (dying = true) }; +}; + +/** Wrap a query object so every `connection` read reports its row count. The + * refresh's scan is the query this change narrows, and "returned zero rows" + * is the only assertion that separates a narrowed scan from one that + * over-fetches and then discards. Forwards `withContext` so the executor's + * owner-policy binding keeps using the wrapper (as FumaDB requires). */ +const observeConnectionReads = (db: FumaDb) => { + const connectionRowCounts: number[] = []; + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop, receiver) { + const value: unknown = Reflect.get(target, prop, receiver); + if (prop === "withContext" && typeof value === "function") { + const withContext = value as (context: unknown) => FumaDb; + return (context: unknown) => wrap(withContext(context)); + } + if (prop !== "findMany" || typeof value !== "function") return value; + const findMany = value as (table: string, options?: unknown) => Promise; + return async (table: string, options?: unknown) => { + const rows = await findMany(table, options); + if (table === "connection") connectionRowCounts.push(rows.length); + return rows; + }; + }, + }); + return { db: wrap(db), connectionRowCounts }; +}; + +const makeHarness = Effect.fnUntraced(function* () { + const counting = makeCountingPlugin(); + const base = makeTestConfig({ plugins: [counting.plugin] as const }); + const observed = observeConnectionReads(base.db); + const config = { ...base, db: observed.db }; + const executor = yield* createExecutor(config); + yield* executor.counting.seed(); + return { + ...counting, + executor, + connectionRowCounts: observed.connectionRowCounts, + connect: (integration: IntegrationSlug, name: string) => + executor.connections.create({ + owner: "org", + name: ConnectionName.make(name), + integration, + template: TEMPLATE, + value: "secret-token", + }), + /** Clear every connection's catalog stamp — the `stale-marked` trigger, as + * `connections.markToolsStale` leaves it. */ + markEveryCatalogStale: () => + Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b.isNotNull("tools_synced_at"), + set: { tools_synced_at: null }, + }), + ), + }; +}); + +describe("tools read catalog refresh scope", () => { + it.effect("refreshes only the integration the read filters to", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.connect(INTEG_B, "main"); + yield* harness.markEveryCatalogStale(); + harness.resolved.length = 0; + + const alpha = yield* harness.executor.tools.list({ integration: INTEG_A }); + + expect(harness.resolved).toEqual(["alpha/main"]); + expect(alpha.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + + const beta = yield* harness.executor.tools.list({ integration: INTEG_B }); + + expect(harness.resolved).toEqual(["alpha/main", "beta/main"]); + expect(beta.map((tool) => String(tool.name))).toEqual(["beta_deploy"]); + }), + ), + ); + + it.effect("scans no connection rows once every catalog is fresh", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.connect(INTEG_B, "main"); + harness.resolved.length = 0; + harness.connectionRowCounts.length = 0; + + const tools = yield* harness.executor.tools.list(); + + expect(tools.map((tool) => String(tool.name)).sort()).toEqual([ + "alpha_deploy", + "beta_deploy", + ]); + // One scan, and it matched nothing: the steady-state read pays an + // indexed lookup and no upstream call at all. + expect(harness.connectionRowCounts).toEqual([0]); + expect(harness.resolved).toEqual([]); + }), + ), + ); + + it.effect("narrows the refresh by owner and connection too", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.connect(INTEG_A, "second"); + yield* harness.connect(INTEG_B, "main"); + yield* harness.markEveryCatalogStale(); + harness.resolved.length = 0; + harness.connectionRowCounts.length = 0; + + const tools = yield* harness.executor.tools.list({ + integration: INTEG_A, + owner: "org", + connection: ConnectionName.make("second"), + }); + + expect(harness.resolved).toEqual(["alpha/second"]); + expect(harness.connectionRowCounts).toEqual([1]); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + }), + ), + ); + + it.effect("survives a plugin whose resolveTools dies with a defect", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.markEveryCatalogStale(); + harness.startDying(); + harness.resolved.length = 0; + + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + + expect(harness.resolved).toEqual(["alpha/main"]); + // The refresh died; the read still answers from the stale-but-working + // catalog rather than failing or returning an empty list. + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + }), + ), + ); +}); From 53af9170857786d90cd9d0e7cbfb6fe819a8d09f Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 05:55:08 -0700 Subject: [PATCH 4/5] Document the tools.list spans and add a changeset --- .changeset/tools-list-scoped-catalog-refresh.md | 10 ++++++++++ .claude/skills/prod-telemetry/SKILL.md | 12 ++++++++++++ 2 files changed, 22 insertions(+) create mode 100644 .changeset/tools-list-scoped-catalog-refresh.md diff --git a/.changeset/tools-list-scoped-catalog-refresh.md b/.changeset/tools-list-scoped-catalog-refresh.md new file mode 100644 index 0000000000..f6f256ea29 --- /dev/null +++ b/.changeset/tools-list-scoped-catalog-refresh.md @@ -0,0 +1,10 @@ +--- +"@executor-js/sdk": patch +"@executor-js/execution": patch +--- + +**Fix: a filtered tools read no longer waits on unrelated integrations' catalog refreshes** + +`tools.list` refreshes stale tool catalogs before it answers, and each refresh is a live upstream handshake. That scan ignored the read's own filter, so `GET /api/tools?integration=railway` re-listed every stale connection in the workspace — thirteen sequential handshakes for integrations the caller had not asked for, none of them railway. The scan also applied the remote-catalog freshness TTL globally, so any connection older than the TTL was fetched in full and then discarded by a per-row check. + +The refresh now scopes to the same `integration`/`owner`/`connection` filter the read uses, narrows the TTL and config-revision triggers to the integrations they can actually fire for, projects only the four columns it reads, and runs the remaining refreshes concurrently instead of one at a time. A plugin defect raised during a refresh can no longer fail the read. Tool search's empty-query enumeration pushes its namespace into the read for the same reason. diff --git a/.claude/skills/prod-telemetry/SKILL.md b/.claude/skills/prod-telemetry/SKILL.md index 47f79bbc75..f8c7f7e055 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -37,6 +37,18 @@ join the same traces via traceparent). `mcp.tool.integration`, same outcome attrs. - `plugin.openapi.invoke` — `plugin.openapi.method` / `path_template` / `base_url`, and since PR #992 `http.status_code`. +- `executor.tools.list` — `executor.tools.filter.integration` / `.owner` / + `.connection` (present only when the read was filtered), + `executor.tools.result_count`, and the catalog-refresh counters + `executor.tools.sync.candidates` (rows the stale scan returned) / + `executor.tools.sync.synced`. A slow tools read is almost always + `candidates` > 0: subtract the child span durations to confirm. +- `executor.tools.sync` (child of the above, one per refreshed connection; + older spans carry the previous name `executor.tools.sync_stale` and no + trigger/outcome attrs) — `executor.integration`, + `executor.connection`, `executor.tools.sync.trigger` + (`stale_marked`/`config_revised`/`expired`) and + `executor.tools.sync.outcome` (`ok`/`fail`). - `mcp.request` (outer) — `mcp.auth.organization_id`, `mcp.auth.account_id`, `mcp.tool.name`, CF edge fields (`cf.country`…), MCP client fingerprint (`mcp.client.name`…). From 1c658ba53749992899c45b0f94721418b0fa2303 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 06:28:26 -0700 Subject: [PATCH 5/5] Serialize tool catalog persists and sanitize refresh failure logs The read refresh fans out to 4 concurrent produceConnectionTools calls, but fumadb runs a SQLite transaction as raw BEGIN/COMMIT on the shared connection. The second BEGIN fails with "cannot start a transaction within a transaction" and the losing persist drops its whole rebuild: five connections refreshed at once, four rebuilds lost. Hold a one-permit semaphore around the persist only, so the upstream resolveTools handshakes keep overlapping. Log refresh failures as a warning with enumerable fields instead of a debug line carrying the raw cause, and report a failed count on the tools.list span. --- .claude/skills/prod-telemetry/SKILL.md | 13 +- packages/core/sdk/src/executor.ts | 148 ++++++-- .../core/sdk/src/tools-sync-scope.test.ts | 333 +++++++++++++----- 3 files changed, 367 insertions(+), 127 deletions(-) diff --git a/.claude/skills/prod-telemetry/SKILL.md b/.claude/skills/prod-telemetry/SKILL.md index f8c7f7e055..8437b5f568 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -41,14 +41,21 @@ join the same traces via traceparent). `.connection` (present only when the read was filtered), `executor.tools.result_count`, and the catalog-refresh counters `executor.tools.sync.candidates` (rows the stale scan returned) / - `executor.tools.sync.synced`. A slow tools read is almost always - `candidates` > 0: subtract the child span durations to confirm. + `executor.tools.sync.synced` / `executor.tools.sync.failed`. A slow tools + read is almost always `candidates` > 0: subtract the child span durations to + confirm. A connection stuck permanently stale shows up as `failed` > 0 on + every read for its scope — the refresh is best-effort and never fails the + read, so this counter and the `executor tool catalog refresh failed` warning + are the only signals it emits. - `executor.tools.sync` (child of the above, one per refreshed connection; older spans carry the previous name `executor.tools.sync_stale` and no trigger/outcome attrs) — `executor.integration`, `executor.connection`, `executor.tools.sync.trigger` (`stale_marked`/`config_revised`/`expired`) and - `executor.tools.sync.outcome` (`ok`/`fail`). + `executor.tools.sync.outcome` (`ok`/`fail`). The matching warning log carries + `integration`, `connection`, `trigger` and `errorTags` only: a refresh runs + on a live credential, so no cause or upstream message is logged and Axiom + will not have one to search. - `mcp.request` (outer) — `mcp.auth.organization_id`, `mcp.auth.account_id`, `mcp.tool.name`, CF edge fields (`cf.country`…), MCP client fingerprint (`mcp.client.name`…). diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 9a6050294c..e587f8c38e 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,4 +1,4 @@ -import { Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect"; +import { Cause, Effect, Inspectable, Layer, Option, Predicate, Schema, Semaphore } from "effect"; import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; import { fumadb } from "@executor-js/fumadb"; import { memoryAdapter } from "@executor-js/fumadb/adapters/memory"; @@ -686,13 +686,30 @@ type CatalogRefreshTrigger = "stale_marked" | "config_revised" | "expired"; /** How much work a read's catalog refresh found and did, for the enclosing * `executor.tools.list` span. `candidates` is what the stale scan returned — - * the number the scan's filter scoping exists to keep at zero. */ + * the number the scan's filter scoping exists to keep at zero. `failed` + * counts refreshes that were swallowed to keep the read answerable, so a + * connection stuck permanently stale is visible as a rate rather than only + * as an absent `synced`. */ interface CatalogRefreshSummary { readonly candidates: number; readonly synced: number; + readonly failed: number; } -const NO_CATALOG_REFRESH: CatalogRefreshSummary = { candidates: 0, synced: 0 }; +const NO_CATALOG_REFRESH: CatalogRefreshSummary = { candidates: 0, synced: 0, failed: 0 }; + +/** The error tags in a swallowed refresh cause, as a stable comma-joined set. + * Tags only: this cause can carry an upstream HTTP request/response, so no + * message, field or rendered cause from it may reach a log line. A defect + * has no tag to report — its shape is arbitrary third-party throw data — + * and reports as `Die`. */ +const causeErrorTags = (cause: Cause.Cause): string => + Array.from( + new Set( + // oxlint-disable-next-line executor/no-manual-tag-check -- boundary: projects the tag out for telemetry rather than branching on it; catchTag/isTagged answer "is it THIS tag", and the log line needs whichever tag it actually was + cause.reasons.map((reason) => (Cause.isFailReason(reason) ? reason.error._tag : reason._tag)), + ), + ).join(","); // --------------------------------------------------------------------------- // collectTables — return the executor-owned Fuma table set. Plugins persist @@ -1646,6 +1663,40 @@ export const createExecutor = (effect: Effect.Effect) => fuma.transaction(effect); + // INVARIANT: tool-catalog persists must not interleave. + // + // fumadb implements a SQLite transaction as raw BEGIN/COMMIT/ROLLBACK on + // the shared connection and hands the callback that same handle + // (`fumadb/adapters/drizzle/query.ts`), so concurrent persists do not get + // one transaction each. SQLite has no nested transaction: the second BEGIN + // fails with "cannot start a transaction within a transaction", and + // because that throw happens before the adapter's try block, the losing + // persist never runs its callback and never rolls back. It simply loses + // its entire rebuild while the connection keeps serving the catalog it + // already had — a silent no-op, since the caller's error is swallowed by + // the best-effort refresh above. Measured: five connections refreshed + // concurrently, four rebuilds dropped. + // + // Postgres checks out a connection per transaction and does not need this; + // the permit is cheap there and the invariant is not worth making + // dialect-conditional. D1 (`interactiveTransactions: false`) issues no + // BEGIN at all, so its statements auto-commit and the permit is what keeps + // one rebuild's delete-then-insert from splitting around another's. + // + // Held around the persist ONLY. `produceConnectionTools` runs its upstream + // `resolveTools` handshake (seconds) outside the permit, so the read + // refresh's concurrent fan-out keeps every bit of its overlap. + // + // Scope is this executor instance, which covers every persist a single + // read fans out. It does NOT cover two executors sharing one SQLite + // handle — self-host mints an executor per request over a long-lived + // handle — because the permit would have to be keyed by connection in + // module state to reach that far. Serializing a shared connection's + // transactions belongs to whoever owns the connection: the fix for the + // cross-request case is a mutex in the fumadb sqlite adapter, not here. + const catalogPersistLock = yield* Semaphore.make(1); + const withCatalogPersistLock = catalogPersistLock.withPermits(1); + // Populated once, never mutated after startup. const staticTools = new Map(); const runtimes = new Map(); @@ -2560,20 +2611,29 @@ export const createExecutor = core.updateMany("connection", { where: connectionWhere, set: syncedSet(row), }); + // Locked: the preserve branches stamp without a surrounding + // transaction, and a bare UPDATE issued while another fiber holds an + // open BEGIN on the shared SQLite connection is enrolled in that + // fiber's transaction — and lost with it if it rolls back. const stampSyncedWithHealth = (reason: string) => - core.updateMany("connection", { - where: connectionWhere, - set: { - tools_synced_at: Date.now(), - last_health: toolSyncHealth(reason), - updated_at: new Date(), - }, - }); + withCatalogPersistLock( + core.updateMany("connection", { + where: connectionWhere, + set: { + tools_synced_at: Date.now(), + last_health: toolSyncHealth(reason), + updated_at: new Date(), + }, + }), + ); // Defense in depth (and cleanup for rows created before the create-time // guard, or emptied by an external edit): a credentialed non-OAuth @@ -2590,24 +2650,28 @@ export const createExecutor = @@ -3736,12 +3802,21 @@ export const createExecutor = Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "fail" }).pipe( Effect.andThen( - Effect.logDebug("Tool catalog refresh failed").pipe( - Effect.annotateLogs("cause", cause), - ), + Effect.logWarning("executor tool catalog refresh failed", { + integration: String(ref.integration), + connection: String(ref.name), + trigger, + errorTags: causeErrorTags(cause), + }), ), Effect.as(false), ), @@ -3757,9 +3832,11 @@ export const createExecutor = ok).length; return { candidates: connections.length, - synced: outcomes.filter((ok) => ok).length, + synced, + failed: outcomes.length - synced, }; }); @@ -3815,6 +3892,7 @@ export const createExecutor = { - const store = new Map(); - return { - key: ProviderKey.make("memory"), - writable: true, - get: (id) => Effect.sync(() => store.get(String(id)) ?? null), - set: (id, value) => Effect.sync(() => void store.set(String(id), value)), - has: (id) => Effect.sync(() => store.has(String(id))), - list: () => - Effect.sync(() => - Array.from(store.keys()).map((key) => ({ id: ProviderItemId.make(key), name: key })), - ), - }; -}; - /** A plugin owning two integrations that records every `resolveTools` it is * asked for, as `/`. The recorded list IS the cost - * a read pays upstream, so the assertions below are about its contents. */ + * a read pays upstream, so most assertions below are about its contents. + * + * `holdResolvesUntil` turns "the handshakes overlapped" from a race the test + * hopes for into one it establishes: every resolve blocks on a latch that + * only opens once N of them are in flight together. Left disarmed during + * setup, where connections are created one at a time and a latch waiting for + * a second resolve would simply hang. */ const makeCountingPlugin = () => { const resolved: string[] = []; let dying = false; + let toolSuffix = "deploy"; + let holdUntil: number | null = null; + let inFlight = 0; + let peakInFlight = 0; + const gate = Latch.makeUnsafe(true); + const plugin = definePlugin(() => ({ id: "counting" as const, - credentialProviders: [memoryProvider()], storage: () => ({}), - resolveTools: (input) => { - const slug = String(input.connection.integration); - resolved.push(`${slug}/${String(input.connection.name)}`); - return dying - ? // oxlint-disable-next-line executor/no-error-constructor -- boundary: a plugin defect IS a raw throw from third-party code, and reproducing it faithfully is the point of this case - Effect.die(new Error("resolveTools blew up")) - : Effect.succeed({ - tools: [{ name: ToolName.make(`${slug}_deploy`), description: "deploy" }], - }); - }, + resolveTools: (input) => + Effect.gen(function* () { + const slug = String(input.connection.integration); + resolved.push(`${slug}/${String(input.connection.name)}`); + inFlight += 1; + peakInFlight = Math.max(peakInFlight, inFlight); + if (holdUntil !== null && inFlight >= holdUntil) gate.openUnsafe(); + yield* gate.await; + inFlight -= 1; + if (dying) return yield* Effect.die("resolveTools blew up"); + return { + tools: [{ name: ToolName.make(`${slug}_${toolSuffix}`), description: toolSuffix }], + }; + }), invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), extension: (ctx) => ({ seed: () => @@ -66,68 +60,156 @@ const makeCountingPlugin = () => { ctx.core.integrations.register({ slug: INTEG_A, description: "Alpha", config: {} }), ctx.core.integrations.register({ slug: INTEG_B, description: "Beta", config: {} }), ]), + /** The `config_revised` trigger as a plugin actually fires it. The public + * `integrations.update` accepts only name/description; `config` — and so + * the `config_revised_at` stamp — is on the plugin-facing surface. */ + revise: (slug: IntegrationSlug) => + ctx.core.integrations.update(slug, { config: { revision: 2 } }), }), }))(); - return { plugin, resolved, startDying: () => void (dying = true) }; + + return { + plugin, + resolved, + startDying: () => void (dying = true), + /** Change what the next resolve returns, so a rebuild that never landed is + * distinguishable from one that did. Without this every refresh rewrites + * the identical row and a persist that silently failed still looks + * correct — the catalog it left behind is the one it would have written. */ + renameToolsTo: (suffix: string) => void (toolSuffix = suffix), + holdResolvesUntil: (count: number) => { + holdUntil = count; + gate.closeUnsafe(); + }, + peakConcurrentResolves: () => peakInFlight, + }; }; -/** Wrap a query object so every `connection` read reports its row count. The - * refresh's scan is the query this change narrows, and "returned zero rows" - * is the only assertion that separates a narrowed scan from one that - * over-fetches and then discards. Forwards `withContext` so the executor's - * owner-policy binding keeps using the wrapper (as FumaDB requires). */ +/** Wrap a query object so every `connection` read reports its row count AND its + * column projection, and so overlapping transactions are visible. + * + * The refresh's scan is the query this change narrows: "returned zero rows" is + * what separates a narrowed scan from one that over-fetches and then discards, + * and the recorded `select` is what stops the projection from being deleted + * while the row-count assertions stay green. + * + * `transaction` is wrapped too — the wrapped orm is forwarded into the run + * callback so reads issued inside a transaction are counted, and the open + * count is what proves catalog persists do not interleave. `internal` is + * deliberately out of scope: it is the test harness's own lazy-open escape + * hatch, so wrapping it would count harness traffic rather than the + * executor's. `withContext` is forwarded so the executor's owner-policy + * binding keeps using the wrapper (as FumaDB requires). */ const observeConnectionReads = (db: FumaDb) => { - const connectionRowCounts: number[] = []; + const connectionScans: { readonly rows: number; readonly select: unknown }[] = []; + let openTransactions = 0; + let peakOpenTransactions = 0; + const wrap = (inner: FumaDb): FumaDb => new Proxy(inner, { get(target, prop, receiver) { const value: unknown = Reflect.get(target, prop, receiver); - if (prop === "withContext" && typeof value === "function") { + if (typeof value !== "function") return value; + + if (prop === "withContext") { const withContext = value as (context: unknown) => FumaDb; - return (context: unknown) => wrap(withContext(context)); + return (context: unknown) => wrap(Reflect.apply(withContext, inner, [context])); } - if (prop !== "findMany" || typeof value !== "function") return value; - const findMany = value as (table: string, options?: unknown) => Promise; - return async (table: string, options?: unknown) => { - const rows = await findMany(table, options); - if (table === "connection") connectionRowCounts.push(rows.length); + + if (prop === "transaction") { + const transaction = value as (run: (orm: FumaDb) => Promise) => Promise; + return (run: (orm: FumaDb) => Promise) => { + openTransactions += 1; + peakOpenTransactions = Math.max(peakOpenTransactions, openTransactions); + return Reflect.apply(transaction, inner, [(orm: FumaDb) => run(wrap(orm))]).finally( + () => { + openTransactions -= 1; + }, + ); + }; + } + + if (prop !== "findMany") return value; + const findMany = value as ( + table: string, + options?: { readonly select?: unknown }, + ) => Promise; + return async (table: string, options?: { readonly select?: unknown }) => { + const rows = await Reflect.apply(findMany, inner, [table, options]); + if (table === "connection") { + connectionScans.push({ rows: rows.length, select: options?.select }); + } return rows; }; }, }); - return { db: wrap(db), connectionRowCounts }; -}; -const makeHarness = Effect.fnUntraced(function* () { - const counting = makeCountingPlugin(); - const base = makeTestConfig({ plugins: [counting.plugin] as const }); - const observed = observeConnectionReads(base.db); - const config = { ...base, db: observed.db }; - const executor = yield* createExecutor(config); - yield* executor.counting.seed(); return { - ...counting, - executor, - connectionRowCounts: observed.connectionRowCounts, - connect: (integration: IntegrationSlug, name: string) => - executor.connections.create({ - owner: "org", - name: ConnectionName.make(name), - integration, - template: TEMPLATE, - value: "secret-token", - }), - /** Clear every connection's catalog stamp — the `stale-marked` trigger, as - * `connections.markToolsStale` leaves it. */ - markEveryCatalogStale: () => - Effect.promise(() => - config.db.updateMany("connection", { - where: (b) => b.isNotNull("tools_synced_at"), - set: { tools_synced_at: null }, - }), - ), + db: wrap(db), + connectionScans, + peakOpenTransactions: () => peakOpenTransactions, }; -}); +}; + +/** The projection the catalog-refresh scan is expected to ask for, as a plain + * array so a structural compare against the recorded `select` reads cleanly. */ +const SCAN_PROJECTION = [...CONNECTION_CATALOG_SCAN_COLUMNS]; + +const makeHarness = () => + Effect.acquireRelease( + Effect.gen(function* () { + const counting = makeCountingPlugin(); + const config = makeTestConfig({ + plugins: [memoryCredentialsPlugin(), counting.plugin] as const, + }); + const observed = observeConnectionReads(config.db); + const executor = yield* createExecutor({ ...config, db: observed.db }); + yield* executor.counting.seed(); + return { + ...counting, + executor, + testDb: config.testDb, + connectionScans: observed.connectionScans, + peakOpenTransactions: observed.peakOpenTransactions, + connect: (integration: IntegrationSlug, name: string) => + executor.connections.create({ + owner: "org", + name: ConnectionName.make(name), + integration, + template: TEMPLATE, + value: "secret-token", + }), + /** Clear every connection's catalog stamp — the `stale_marked` trigger, + * as `connections.markToolsStale` leaves it. */ + markEveryCatalogStale: () => + Effect.promise(() => + observed.db.updateMany("connection", { + where: (b) => b.isNotNull("tools_synced_at"), + set: { tools_synced_at: null }, + }), + ), + /** Push every catalog stamp into the past. The `config_revised` trigger + * compares `tools_synced_at` against a stamp taken later in the same + * test; on an in-memory database both can land in one millisecond and + * the comparison is strict, so the ordering is made explicit here + * rather than left to the clock. */ + backdateEveryCatalog: () => + Effect.promise(() => + observed.db.updateMany("connection", { + where: (b) => b.isNotNull("tools_synced_at"), + set: { tools_synced_at: Date.now() - 60_000 }, + }), + ), + }; + }), + ({ executor, testDb }) => + executor + .close() + .pipe( + Effect.ignore, + Effect.andThen(Effect.promise(() => testDb.close()).pipe(Effect.ignore)), + ), + ); describe("tools read catalog refresh scope", () => { it.effect("refreshes only the integration the read filters to", () => @@ -159,7 +241,7 @@ describe("tools read catalog refresh scope", () => { yield* harness.connect(INTEG_A, "main"); yield* harness.connect(INTEG_B, "main"); harness.resolved.length = 0; - harness.connectionRowCounts.length = 0; + harness.connectionScans.length = 0; const tools = yield* harness.executor.tools.list(); @@ -169,7 +251,7 @@ describe("tools read catalog refresh scope", () => { ]); // One scan, and it matched nothing: the steady-state read pays an // indexed lookup and no upstream call at all. - expect(harness.connectionRowCounts).toEqual([0]); + expect(harness.connectionScans).toEqual([{ rows: 0, select: SCAN_PROJECTION }]); expect(harness.resolved).toEqual([]); }), ), @@ -184,7 +266,7 @@ describe("tools read catalog refresh scope", () => { yield* harness.connect(INTEG_B, "main"); yield* harness.markEveryCatalogStale(); harness.resolved.length = 0; - harness.connectionRowCounts.length = 0; + harness.connectionScans.length = 0; const tools = yield* harness.executor.tools.list({ integration: INTEG_A, @@ -193,12 +275,40 @@ describe("tools read catalog refresh scope", () => { }); expect(harness.resolved).toEqual(["alpha/second"]); - expect(harness.connectionRowCounts).toEqual([1]); + // One row, and only the four columns the refresh actually reads: the + // scan runs on every read, so the projection is part of its cost. + expect(harness.connectionScans).toEqual([{ rows: 1, select: SCAN_PROJECTION }]); expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); }), ), ); + it.effect("re-resolves only the integration whose config was revised", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.connect(INTEG_B, "main"); + yield* harness.backdateEveryCatalog(); + yield* harness.executor.counting.revise(INTEG_A); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + + expect(harness.resolved).toEqual(["alpha/main"]); + + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_B }); + + // Beta's catalog is older than alpha's revision stamp, and under one + // global `staleBefore` that alone made it a candidate. The trigger is + // per-integration: only the revised integration re-lists. + expect(harness.resolved).toEqual([]); + }), + ), + ); + it.effect("survives a plugin whose resolveTools dies with a defect", () => Effect.scoped( Effect.gen(function* () { @@ -214,6 +324,51 @@ describe("tools read catalog refresh scope", () => { // The refresh died; the read still answers from the stale-but-working // catalog rather than failing or returning an empty list. expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + + const retried = yield* harness.executor.tools.list({ integration: INTEG_A }); + + // A failed refresh stamps nothing, so the connection stays stale and + // the NEXT read attempts it again — best-effort means retried, not + // abandoned. + expect(harness.resolved).toEqual(["alpha/main", "alpha/main"]); + expect(retried.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + }), + ), + ); + + it.effect("persists every catalog when one read refreshes many connections", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + const names = ["one", "two", "three", "four", "five"]; + for (const name of names) yield* harness.connect(INTEG_A, name); + // Make the rebuild observably different from what `connect` already + // persisted, so a refresh whose write never landed is visible. + harness.renameToolsTo("redeploy"); + yield* harness.markEveryCatalogStale(); + harness.resolved.length = 0; + // Arm the overlap latch only now: `connect` resolves one connection at + // a time, and a latch waiting for a second in-flight resolve would + // never open during setup. + harness.holdResolvesUntil(2); + + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + + expect(harness.resolved.slice().sort()).toEqual(names.map((n) => `alpha/${n}`).sort()); + // Every connection's rebuild landed. fumadb runs a SQLite transaction + // as raw BEGIN/COMMIT on the shared connection, so a second persist + // entering while the first is open hits "cannot start a transaction + // within a transaction" and loses its whole rebuild, leaving that + // connection on the catalog it had before. Without the persist permit + // this assertion reports four of the five still on `alpha_deploy`. + expect( + tools.map((tool) => `${String(tool.connection)}/${String(tool.name)}`).sort(), + ).toEqual(names.map((n) => `${n}/alpha_redeploy`).sort()); + + // The two halves of the invariant: discovery overlapped, persists did + // not. + expect(harness.peakConcurrentResolves()).toBeGreaterThan(1); + expect(harness.peakOpenTransactions()).toBe(1); }), ), );