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..8437b5f568 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -37,6 +37,25 @@ 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` / `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`). 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/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, 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..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"; @@ -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,39 @@ export interface ExecutorConfig(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 // through host-owned facades (`pluginStorage`, `blobs`) instead of contributing @@ -1629,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(); @@ -2543,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 @@ -2573,24 +2650,28 @@ export const createExecutor = @@ -3575,8 +3658,9 @@ export const createExecutor = [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. + // + // Warning, not debug: swallowing this is the whole point, so the + // log line is the only signal that a connection is serving a + // stale catalog. Enumerable fields only — a refresh runs on a + // live credential and its cause can embed the upstream request + // and response. + Effect.catchCause((cause) => + Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "fail" }).pipe( + Effect.andThen( + Effect.logWarning("executor tool catalog refresh failed", { + integration: String(ref.integration), + connection: String(ref.name), + trigger, + errorTags: causeErrorTags(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 }, ); - } - }); + + const synced = outcomes.filter((ok) => ok).length; + return { + candidates: connections.length, + synced, + failed: outcomes.length - synced, + }; + }); 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 +3888,30 @@ export const createExecutor = /`. The recorded list IS the cost + * 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, + storage: () => ({}), + 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: () => + Effect.all([ + 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), + /** 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 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 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 (typeof value !== "function") return value; + + if (prop === "withContext") { + const withContext = value as (context: unknown) => FumaDb; + return (context: unknown) => wrap(Reflect.apply(withContext, inner, [context])); + } + + 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), + 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", () => + 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.connectionScans.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.connectionScans).toEqual([{ rows: 0, select: SCAN_PROJECTION }]); + 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.connectionScans.length = 0; + + const tools = yield* harness.executor.tools.list({ + integration: INTEG_A, + owner: "org", + connection: ConnectionName.make("second"), + }); + + expect(harness.resolved).toEqual(["alpha/second"]); + // 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* () { + 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"]); + + 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); + }), + ), + ); +});