From 04360893eb1a089bdf61d01b1b3738ee7a1d9119 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 10:48:57 -0700 Subject: [PATCH 1/6] Cut cold-read latency and probe storms; instrument isolate residency Co-Authored-By: Claude Fable 5.1 --- .changeset/perf-cold-reads.md | 8 + .../src/account/list-members.node.test.ts | 256 ++++++++++++++++++ .../src/account/workos-account-service.ts | 114 ++++++-- apps/cloud/src/auth/jwks-cache.node.test.ts | 40 +++ apps/cloud/src/auth/jwks-cache.ts | 96 +++++-- apps/cloud/src/mcp/agent-handler.ts | 13 +- apps/cloud/src/mcp/responses.ts | 15 + apps/cloud/src/mcp/telemetry.ts | 1 - apps/cloud/src/server.ts | 26 ++ packages/core/sdk/src/executor.ts | 256 ++++++++++-------- .../mcp/agent-session-durable-object.test.ts | 85 ++++++ .../src/mcp/agent-session-durable-object.ts | 22 ++ .../src/mcp/session-runtime-residency.ts | 89 +++++- packages/hosts/mcp/src/tool-server.ts | 54 +--- .../plugins/mcp/src/sdk/catalog-sync.test.ts | 35 ++- .../src/lib/use-connection-health.test.ts | 83 +++++- .../react/src/lib/use-connection-health.ts | 163 +++++++++-- 17 files changed, 1100 insertions(+), 256 deletions(-) create mode 100644 .changeset/perf-cold-reads.md create mode 100644 apps/cloud/src/account/list-members.node.test.ts diff --git a/.changeset/perf-cold-reads.md b/.changeset/perf-cold-reads.md new file mode 100644 index 0000000000..8dd6bcee94 --- /dev/null +++ b/.changeset/perf-cold-reads.md @@ -0,0 +1,8 @@ +--- +"@executor-js/sdk": patch +"@executor-js/cloudflare": patch +"@executor-js/react": patch +"@executor-js/cloud": patch +--- + +Faster reads: a tools read no longer waits on TTL-expired remote catalogs (they re-list behind the read), the members list runs its lookups concurrently, cold isolates race the JWKS store against the upstream fetch, OAuth discovery documents are answered at the Worker entry, automatic connection health probes are deduplicated across remounts, and per-tool MCP registration spans are dropped. diff --git a/apps/cloud/src/account/list-members.node.test.ts b/apps/cloud/src/account/list-members.node.test.ts new file mode 100644 index 0000000000..dbabb870a5 --- /dev/null +++ b/apps/cloud/src/account/list-members.node.test.ts @@ -0,0 +1,256 @@ +import { describe, expect, it } from "@effect/vitest"; +import type { Autumn } from "autumn-js"; +import { Effect, Layer } from "effect"; + +import { AccountProvider } from "@executor-js/api/server"; +import { AccountError } from "@executor-js/api"; + +import { ApiKeyService } from "../auth/api-keys"; +import { UserStoreService } from "../auth/context"; +import { WorkOSError } from "../auth/errors"; +import { ORG_SELECTOR_HEADER } from "../auth/organization"; +import { WorkOSClient, type WorkOSClientService } from "../auth/workos"; +import { AutumnError, AutumnService } from "../extensions/billing/service"; +import { AccountCaller, workosAccountProvider } from "./workos-account-service"; + +// --------------------------------------------------------------------------- +// `GET /api/account/members` at the PROVIDER boundary. +// +// This endpoint used to run strictly sequentially: `getMemberSeats` (Autumn +// getOrCreate, then WorkOS `listOrgMembers`, then WorkOS +// `listPendingInvitations`), THEN a SECOND `listOrgMembers` call for the same +// org, then a per-member `getUser` fan-out. That is the p95 4.4s the +// concurrency rework here fixes. Two properties have to hold for the rework to +// be safe, and only this seam sees both: +// +// 1. `listOrgMembers` is now fetched ONCE and shared between the seat count +// and the member rows — asserted below by counting calls. +// 2. Failure semantics stay split: an Autumn (or listPendingInvitations) +// failure degrades seats to safe defaults but still returns the member +// list; a `listOrgMembers` failure fails the whole request as an +// `AccountError`, exactly as it did when it had its own dedicated call. +// --------------------------------------------------------------------------- + +const ORG = "org_123"; +const USER = "user_admin"; +const createdAt = new Date("2026-01-01T00:00:00.000Z"); +const orgHeaders = { [ORG_SELECTOR_HEADER]: ORG }; + +const session = (accountId: string) => ({ + accountId, + email: `${accountId}@example.test`, + name: null, + avatarUrl: null, + organizationId: ORG, + sealedSession: "sealed", + refreshedSession: null, +}); + +const membership = (userId: string, status: "active" | "pending" = "active") => ({ + id: `om_${userId}`, + userId, + organizationId: ORG, + status, + role: { slug: "member" }, +}); + +const user = (userId: string) => ({ + id: userId, + email: `${userId}@example.test`, + firstName: "First", + lastName: "Last", + profilePictureUrl: null, + lastSignInAt: null, +}); + +/** + * Stub WorkOS client whose `listOrgMembers` and `getUser` calls are counted, + * so the shared-fetch behavior is asserted directly rather than inferred. + */ +const stubWorkOS = (members: ReadonlyArray>) => { + const calls = { listOrgMembers: 0, getUser: 0, listPendingInvitations: 0 }; + const layer = Layer.succeed( + WorkOSClient, + new Proxy({} as WorkOSClientService, { + get: (_target, prop) => { + if (prop === "listUserMemberships") { + return (userId: string) => + Effect.succeed({ data: [{ userId, organizationId: ORG, status: "active" }] }); + } + if (prop === "listOrgMembers") { + return () => { + calls.listOrgMembers += 1; + return Effect.succeed({ data: members }); + }; + } + if (prop === "listPendingInvitations") { + return () => { + calls.listPendingInvitations += 1; + return Effect.succeed({ data: [] }); + }; + } + if (prop === "getUser") { + return (userId: string) => { + calls.getUser += 1; + return Effect.succeed(user(userId)); + }; + } + return () => Effect.die(`unexpected WorkOSClient.${String(prop)} call`); + }, + }), + ); + return { layer, calls }; +}; + +const stubUsers = Layer.succeed(UserStoreService)({ + use: (_op, fn) => + Effect.promise(() => + fn({ + ensureAccount: async (id: string) => ({ id, createdAt }), + getAccount: async (id: string) => ({ id, createdAt }), + upsertOrganization: async (org: { id: string; name: string }) => ({ + ...org, + slug: org.id, + createdAt, + }), + getOrganization: async (id: string) => ({ + id, + name: `Org ${id}`, + slug: id, + createdAt, + }), + getOrganizationBySlug: async (slug: string) => ({ + id: slug, + name: `Org ${slug}`, + slug, + createdAt, + }), + deleteOrganizationCascade: async () => {}, + }), + ), +}); + +const stubApiKeys = Layer.succeed(ApiKeyService)({ + validate: () => Effect.die("listMembers does not validate keys"), + listUserKeys: () => Effect.die("listMembers does not touch api keys"), + createUserKey: () => Effect.die("listMembers does not touch api keys"), + revokeUserKey: () => Effect.die("listMembers does not touch api keys"), + listOrgKeys: () => Effect.die("listMembers does not touch api keys"), + createOrgKey: () => Effect.die("listMembers does not touch api keys"), + revokeOrgKey: () => Effect.die("listMembers does not touch api keys"), +}); + +/** Autumn stub whose `getOrCreate` either succeeds with a plan or fails. */ +const stubAutumn = (mode: "ok" | "fail") => + Layer.succeed(AutumnService)({ + use: (fn: (client: Autumn) => Promise) => + mode === "fail" + ? Effect.fail(new AutumnError({ message: "autumn unreachable" })) + : Effect.promise(() => + fn( + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- test stub narrows the SDK client to what getMemberSeats actually calls + { customers: { getOrCreate: async () => ({ subscriptions: [] }) } } as any, + ), + ), + ensureCustomer: () => Effect.void, + checkExecutionBalance: () => Effect.die("listMembers does not check execution balance"), + trackExecution: () => Effect.void, + setMemberSeats: () => Effect.void, + }); + +const providerWith = ( + members: ReadonlyArray>, + autumnMode: "ok" | "fail", +) => { + const { layer: workosLayer, calls } = stubWorkOS(members); + const provider = AccountProvider.asEffect().pipe( + Effect.provide( + workosAccountProvider.pipe( + Layer.provide( + Layer.mergeAll( + workosLayer, + stubUsers, + stubApiKeys, + stubAutumn(autumnMode), + Layer.succeed(AccountCaller)({ session: session(USER) }), + ), + ), + ), + ), + ); + return { provider, calls }; +}; + +describe("listMembers · provider boundary", () => { + it.effect("fetches listOrgMembers ONCE and shares it between seats and rows", () => + Effect.gen(function* () { + const { provider, calls } = providerWith([membership(USER), membership("user_2")], "ok"); + const account = yield* provider; + + const result = yield* account.listMembers(orgHeaders); + + expect(result.members).toHaveLength(2); + expect(calls.listOrgMembers, "listOrgMembers must be shared, not fetched twice").toBe(1); + expect(calls.getUser).toBe(2); + }), + ); + + it.effect("an Autumn failure degrades seats to defaults but still returns members", () => + Effect.gen(function* () { + const { provider, calls } = providerWith([membership(USER)], "fail"); + const account = yield* provider; + + const result = yield* account.listMembers(orgHeaders); + + expect(result.members).toHaveLength(1); + expect(result.seats).toEqual({ used: 0, granted: 0, unlimited: false }); + expect(calls.listOrgMembers, "still only one shared fetch despite the Autumn failure").toBe( + 1, + ); + }), + ); + + it.effect("a listOrgMembers failure fails the whole request, not just seats", () => + Effect.gen(function* () { + const failingWorkOS = Layer.succeed( + WorkOSClient, + new Proxy({} as WorkOSClientService, { + get: (_target, prop) => { + if (prop === "listUserMemberships") { + return (userId: string) => + Effect.succeed({ data: [{ userId, organizationId: ORG, status: "active" }] }); + } + if (prop === "listOrgMembers") { + return () => Effect.fail(new WorkOSError({ status: 500 })); + } + if (prop === "listPendingInvitations") { + return () => Effect.succeed({ data: [] }); + } + return () => Effect.die(`unexpected WorkOSClient.${String(prop)} call`); + }, + }), + ); + + const provider = AccountProvider.asEffect().pipe( + Effect.provide( + workosAccountProvider.pipe( + Layer.provide( + Layer.mergeAll( + failingWorkOS, + stubUsers, + stubApiKeys, + stubAutumn("ok"), + Layer.succeed(AccountCaller)({ session: session(USER) }), + ), + ), + ), + ), + ); + const account = yield* provider; + + const error = yield* Effect.flip(account.listMembers(orgHeaders)); + + expect(error).toBeInstanceOf(AccountError); + }), + ); +}); diff --git a/apps/cloud/src/account/workos-account-service.ts b/apps/cloud/src/account/workos-account-service.ts index dc8b234b1f..74b1b9a045 100644 --- a/apps/cloud/src/account/workos-account-service.ts +++ b/apps/cloud/src/account/workos-account-service.ts @@ -1,4 +1,4 @@ -import { Context, Effect, Layer } from "effect"; +import { Context, Effect, Layer, Result } from "effect"; import { AccountProvider, type AccountHeaders } from "@executor-js/api/server"; import { @@ -19,6 +19,8 @@ import { countSeatsUsed, getMemberLimitForPlan, selectActiveMemberLimitPlan, + type AutumnSubscriptionSummary, + type SeatMembership, } from "../extensions/billing/plans"; // The per-request resolved caller, injected by the cookie-only session @@ -139,27 +141,47 @@ export const workosAccountProvider: Layer.Layer< } }); + // Pure seat math, shared by `getMemberSeats` (reserveMemberSlot's + // fail-closed lookup) and `listMembers` (which fetches the same three + // inputs concurrently and combines them inline further below). + // `listOrgMembers` returns active members AND pending memberships (an + // invited user shows up as status "pending"); `listPendingInvitations` + // returns the same invited users again. `countSeatsUsed` dedupes them so + // an outstanding invite is not counted twice. + const seatsFrom = ( + memberships: { readonly data: ReadonlyArray }, + pendingInvitationCount: number, + subscriptions: ReadonlyArray, + ) => { + const planId = selectActiveMemberLimitPlan(subscriptions); + const limit = getMemberLimitForPlan(planId); + return { + used: countSeatsUsed(memberships.data, pendingInvitationCount), + granted: limit ?? 0, + unlimited: limit === null, + }; + }; + // Mirror of org/handlers `getMemberSeats` — live seat usage from WorkOS. - const getMemberSeats = (organizationId: string) => + // Accepts an already-fetched `memberships` list so a caller that has one + // on hand (`listMembers`) does not pay for a second `listOrgMembers` + // round trip. `reserveMemberSlot` has no list on hand, so it fetches its + // own — either way, the remaining lookups run concurrently. + const getMemberSeats = ( + organizationId: string, + memberships?: Effect.Success>, + ) => Effect.gen(function* () { - const customer = yield* autumn.use((client) => - client.customers.getOrCreate({ customerId: organizationId }), + const [customer, resolvedMemberships, invitations] = yield* Effect.all( + [ + autumn.use((client) => client.customers.getOrCreate({ customerId: organizationId })), + memberships ? Effect.succeed(memberships) : workos.listOrgMembers(organizationId), + workos.listPendingInvitations(organizationId), + ], + { concurrency: "unbounded" }, ); - const planId = selectActiveMemberLimitPlan(customer.subscriptions); - const limit = getMemberLimitForPlan(planId); - - // `listOrgMembers` returns active members AND pending memberships (an - // invited user shows up as status "pending"); `listPendingInvitations` - // returns the same invited users again. `countSeatsUsed` dedupes them - // so an outstanding invite is not counted twice. - const memberships = yield* workos.listOrgMembers(organizationId); - const invitations = yield* workos.listPendingInvitations(organizationId); - - return { - used: countSeatsUsed(memberships.data, invitations.data.length), - granted: limit ?? 0, - unlimited: limit === null, - }; + + return seatsFrom(resolvedMemberships, invitations.data.length, customer.subscriptions); }); // Mirror of org/handlers `reserveMemberSlot` — fail closed on lookup error. @@ -297,19 +319,53 @@ export const workosAccountProvider: Layer.Layer< Effect.gen(function* () { const { session, org } = yield* requireOrganization(headers); - // Seats fall back to safe display defaults on lookup error — never - // blank the page over a transient Autumn/WorkOS hiccup. The real cap - // gate lives in `reserveMemberSlot`, which fails closed. - const seats = yield* getMemberSeats(org.id).pipe( - Effect.catchCause(() => Effect.succeed({ used: 0, granted: 0, unlimited: false })), + // The three remote lookups below are independent, so they run + // concurrently instead of the old serial chain (getMemberSeats' + // Autumn/listOrgMembers/listPendingInvitations, THEN a second, + // separate listOrgMembers for the rows). `listOrgMembers` is needed + // by both the seat count and the member rows, so it is fetched + // ONCE here and shared instead of twice. + // + // Failure semantics differ per branch, so each branch is captured + // with `Effect.either` rather than left to fail the whole + // `Effect.all`: a `listOrgMembers` failure must still surface as an + // AccountError for the members list (unchanged from before); an + // Autumn or `listPendingInvitations` failure must only degrade + // seats to their safe defaults — never blank the page over a + // transient hiccup, exactly as `getMemberSeats`'s own `catchCause` + // fallback did before. The real cap gate lives in + // `reserveMemberSlot`, which fails closed. + const [membershipsResult, seatInputsResult] = yield* Effect.all( + [ + Effect.result(workos.listOrgMembers(org.id)), + Effect.result( + Effect.all( + [ + autumn.use((client) => client.customers.getOrCreate({ customerId: org.id })), + workos.listPendingInvitations(org.id), + ], + { concurrency: "unbounded" }, + ), + ), + ], + { concurrency: "unbounded" }, ); - const memberships = yield* workos - .listOrgMembers(org.id) - .pipe(Effect.catchTag("WorkOSError", toAccountError)); + if (Result.isFailure(membershipsResult)) { + return yield* toAccountError(); + } + const memberships = membershipsResult.success; + + const seats = Result.isSuccess(seatInputsResult) + ? seatsFrom( + memberships, + seatInputsResult.success[1].data.length, + seatInputsResult.success[0].subscriptions, + ) + : { used: 0, granted: 0, unlimited: false }; const members = yield* Effect.all( - memberships.data.map((m) => + memberships.data.map((m: (typeof memberships.data)[number]) => Effect.gen(function* () { const user = yield* workos.getUser(m.userId); return { @@ -325,7 +381,7 @@ export const workosAccountProvider: Layer.Layer< }; }), ), - { concurrency: 5 }, + { concurrency: 20 }, ).pipe(Effect.catchTag("WorkOSError", toAccountError)); return { members, seats }; diff --git a/apps/cloud/src/auth/jwks-cache.node.test.ts b/apps/cloud/src/auth/jwks-cache.node.test.ts index f4f3e08a64..e2f5090709 100644 --- a/apps/cloud/src/auth/jwks-cache.node.test.ts +++ b/apps/cloud/src/auth/jwks-cache.node.test.ts @@ -228,6 +228,46 @@ describe("createCachedRemoteJWKSet", () => { expect(store.reads()).toBeGreaterThan(0); }); + it("a cold isolate does not wait on a slow store when the upstream answers first", async () => { + const kp = await generateRotatableKeypair("k1"); + const store = makeStoreHarness(); + const warm = makeFetchHarness([kp.publicJwk]); + const first = createCachedRemoteJWKSet(jwksUrl, { fetch: warm.fetch, store }); + const token = await sign(kp); + await jwtVerify(token, first, { issuer, audience }); + + // The production shape: the Workers Cache API read takes seconds on a + // cold isolate while the key server answers in tens of ms. A cold + // resolver must take the upstream answer instead of waiting on the store. + const slowStore: JwksStore = { + get: (url) => new Promise((resolve) => setTimeout(() => resolve(store.get(url)), 2_000)), + put: store.put, + }; + const second = createCachedRemoteJWKSet(jwksUrl, { fetch: warm.fetch, store: slowStore }); + const startedAt = Date.now(); + const { payload } = await jwtVerify(token, second, { issuer, audience }); + expect(payload.sub).toBe("user_test"); + expect(Date.now() - startedAt).toBeLessThan(1_000); + expect(second.inspect().blockingFetchCount).toBe(1); + }); + + it("a cold isolate still answers from the store when the upstream is down", async () => { + const kp = await generateRotatableKeypair("k1"); + const store = makeStoreHarness(); + const warm = makeFetchHarness([kp.publicJwk]); + const first = createCachedRemoteJWKSet(jwksUrl, { fetch: warm.fetch, store }); + const token = await sign(kp); + await jwtVerify(token, first, { issuer, audience }); + + // Upstream rejects immediately (loses the race with nothing); the store + // answer must still win rather than the fetch failure propagating. + const second = createCachedRemoteJWKSet(jwksUrl, { fetch: failingFetch, store }); + const { payload } = await jwtVerify(token, second, { issuer, audience }); + expect(payload.sub).toBe("user_test"); + expect(second.inspect().storeHitCount).toBe(1); + expect(second.inspect().blockingFetchCount).toBe(0); + }); + it("keeps serving the last good keys when the key server is down", async () => { const kp = await generateRotatableKeypair("k1"); const harness = makeFetchHarness([kp.publicJwk]); diff --git a/apps/cloud/src/auth/jwks-cache.ts b/apps/cloud/src/auth/jwks-cache.ts index 3cd1d963d1..d4a20a7fa1 100644 --- a/apps/cloud/src/auth/jwks-cache.ts +++ b/apps/cloud/src/auth/jwks-cache.ts @@ -19,14 +19,24 @@ // // 1. Module-scope memory — free, but dies with the isolate. // 2. A cross-isolate store (the Workers Cache API by default) — colo-local, -// survives isolate recycling, so a cold isolate reads keys in ~1ms -// instead of paying an upstream round trip. +// survives isolate recycling, so a cold isolate still has keys when the +// upstream key server is slow or down. // 3. The upstream JWKS endpoint. // // On top of that it is stale-while-revalidate: once past `ttlMs` a usable key // set is served immediately and refreshed in the background, and if a refresh -// fails we keep serving the last good keys until `staleMaxMs`. Only a fully -// cold path (no memory, no store) ever blocks on the network. +// fails we keep serving the last good keys until `staleMaxMs`. +// +// A cold isolate (no memory) RACES the store read against the upstream fetch +// and takes whichever answers first. The store was added as the fast path, +// but production measured the opposite: `caches.default.match()` on a cold +// isolate takes p50 1.4s / p90 2.8s while the upstream fetch takes p50 26ms +// (`jwks.store_read_ms` vs `jwks.last_fetch_ms` on `workos.session.local_verify`, +// 2026-09). Waiting on the store first put 1.5s on every cold verify — the +// single largest cost of a cold API request. Racing keeps the store's purpose +// (keys survive a slow or dead key server) without its latency: the store +// only decides the outcome when the upstream is slower than it, which is +// exactly the incident it exists for. // // Serving stale keys is safe in a way that serving a stale *token* would not // be: key sets rotate on the order of days, tokens are still signature- and @@ -316,6 +326,8 @@ export const createCachedRemoteJWKSet = ( void ignoreFailure(refresh()); }; + /** Read the cross-isolate store. Pure: the caller decides whether the + * candidate becomes the resolver's entry (see `loadCold`). */ const loadFromStore = async (): Promise => { if (!store) return null; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: the L2 store is an optimization; any failure degrades to an upstream fetch @@ -325,10 +337,7 @@ export const createCachedRemoteJWKSet = ( lastStoreReadMs = Date.now() - storeReadStartedAt; if (!stored) return null; const candidate = entryFrom(stored); - if (!isUsable(candidate)) return null; - entry = candidate; - storeHitCount += 1; - return candidate; + return isUsable(candidate) ? candidate : null; } catch { return null; } @@ -340,6 +349,59 @@ export const createCachedRemoteJWKSet = ( return refresh(); }; + /** Adopt a store candidate unless a fresher entry landed in the meantime + * (the racing upstream fetch may have finished first). */ + const adoptFromStore = (candidate: CacheEntry): CacheEntry => { + storeHitCount += 1; + if (entry === null || entry.fetchedAt < candidate.fetchedAt) entry = candidate; + return entry; + }; + + type ColdWinner = + | { readonly source: "store"; readonly entry: CacheEntry } + | { readonly source: "upstream"; readonly entry: CacheEntry }; + + /** + * The fully cold path: nothing in memory. Race the store read against the + * upstream fetch (see the module header for why the store is not simply + * read first) and answer with whichever produces usable keys first. The + * loser keeps running: an upstream fetch that lands after a store hit still + * refreshes memory and the store, so the isolate converges on fresh keys. + * + * Latency attribution: `blockingFetchCount` moves only when the caller's + * answer actually came from upstream; a fetch that lost the race (or + * failed) is background work the verify did not pay for. + */ + const loadCold = async (): Promise => { + const fromStore = loadFromStore(); + const fromUpstream = refresh().then( + (next) => ({ ok: true as const, entry: next }), + (error: unknown) => ({ ok: false as const, error }), + ); + const first = await Promise.race([ + fromStore.then((candidate) => (candidate ? { source: "store", entry: candidate } : null)), + fromUpstream.then((result) => + result.ok ? { source: "upstream", entry: result.entry } : null, + ), + ]); + if (first?.source === "upstream") { + blockingFetchCount += 1; + return first.entry; + } + if (first?.source === "store") { + return adoptFromStore(first.entry); + } + // The first to settle had nothing: wait for the other side. + const [stored, upstream] = await Promise.all([fromStore, fromUpstream]); + if (upstream.ok) { + blockingFetchCount += 1; + return upstream.entry; + } + if (stored) return adoptFromStore(stored); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: nothing usable anywhere, so the upstream failure is the real answer + throw upstream.error; + }; + const ensureFresh = async (forceRefresh: boolean): Promise => { if (forceRefresh) return refreshBlocking(); if (entry && isFresh(entry)) return entry; @@ -350,22 +412,8 @@ export const createCachedRemoteJWKSet = ( return entry; } - // Cold isolate — the L2 store saves us the upstream round trip. - const stored = await loadFromStore(); - if (stored) { - if (!isFresh(stored)) refreshInBackground(); - return stored; - } - - // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: a failed refresh must fall back to stale keys rather than fail the verify - try { - return await refreshBlocking(); - } catch (error) { - // Upstream is slow or down. Last good keys beat failing every request. - if (entry && isUsable(entry)) return entry; - // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: nothing usable is cached, so the upstream failure is the real answer - throw error; - } + // Cold isolate — store and upstream race; see `loadCold`. + return loadCold(); }; const get: JWTVerifyGetKey = async (protectedHeader, token) => { diff --git a/apps/cloud/src/mcp/agent-handler.ts b/apps/cloud/src/mcp/agent-handler.ts index 3d94d3e9cb..50ea24f397 100644 --- a/apps/cloud/src/mcp/agent-handler.ts +++ b/apps/cloud/src/mcp/agent-handler.ts @@ -31,24 +31,13 @@ import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { wrapMcpSseResponse } from "../observability/memory-metrics"; import { WorkerTelemetryLive } from "../observability/telemetry"; import { cloudMcpAuth } from "./auth-provider"; +import { corsPreflightResponse } from "./responses"; import { isMcpSessionMetaUnavailable } from "./session-meta"; import { McpSessionDOSqlite } from "./session-durable-object"; import { parseTraceparent } from "./traceparent"; const MCP_SESSION_UNAVAILABLE_MESSAGE = "Session storage temporarily unavailable - please retry"; -const corsPreflightResponse = (): Response => - new Response(null, { - status: 204, - headers: { - "access-control-allow-origin": "*", - "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", - "access-control-allow-headers": - "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", - "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", - }, - }); - const jsonRpcResponse = ( status: number, code: number, diff --git a/apps/cloud/src/mcp/responses.ts b/apps/cloud/src/mcp/responses.ts index 8be019e303..17a34d2db7 100644 --- a/apps/cloud/src/mcp/responses.ts +++ b/apps/cloud/src/mcp/responses.ts @@ -47,3 +47,18 @@ export const unauthorized = (auth: UnauthorizedAuth, protectedResourceMetadataUr }, }, ); + +/** CORS preflight for `/mcp` and the OAuth discovery documents: browsers + * preflight the metadata docs during RFC 9728 discovery. Shared by the + * Worker entry (discovery docs) and the Agents bridge (`/mcp`). */ +export const corsPreflightResponse = (): Response => + new Response(null, { + status: 204, + headers: { + ...CORS_ALLOW_ORIGIN, + "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", + "access-control-allow-headers": + "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", + "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", + }, + }); diff --git a/apps/cloud/src/mcp/telemetry.ts b/apps/cloud/src/mcp/telemetry.ts index 3f0c300256..87468f2736 100644 --- a/apps/cloud/src/mcp/telemetry.ts +++ b/apps/cloud/src/mcp/telemetry.ts @@ -134,7 +134,6 @@ const readJsonRpcEnvelope = (request: Request): Effect.Effect (text ? decodeJsonRpcEnvelopeString(text) : Option.none())), Effect.catchCause(() => Effect.succeed(Option.none())), - Effect.withSpan("mcp.request.read_json_rpc"), ); // Managed-cloud capture of the executed script, on the `mcp.request` span diff --git a/apps/cloud/src/server.ts b/apps/cloud/src/server.ts index 4f5bf76289..6e7b1b5a17 100644 --- a/apps/cloud/src/server.ts +++ b/apps/cloud/src/server.ts @@ -7,6 +7,7 @@ import { ATTR_URL_PATH, ATTR_URL_SCHEME, } from "@opentelemetry/semantic-conventions"; +import { Effect } from "effect"; import * as Sentry from "@sentry/cloudflare"; import handler from "@tanstack/react-start/server-entry"; @@ -15,6 +16,11 @@ import { marketingProxyRequest } from "./edge/marketing"; import { passthroughResponse } from "./edge/passthrough"; import { makeCloudMcpAgentHandler } from "./mcp/agent-handler"; import { classifyMcpPath, prepareMcpOrgScope } from "./mcp/mount"; +import { + authorizationServerMetadataResponse, + protectedResourceMetadataResponse, +} from "./mcp/oauth-metadata"; +import { corsPreflightResponse } from "./mcp/responses"; import { parseTraceparent } from "./mcp/traceparent"; import { McpSessionDOSqlite as McpSessionDOBase } from "./mcp/session-durable-object"; import { @@ -291,6 +297,26 @@ const cloudflareHandler: ExportedHandler = { // this entry invokes it for non-MCP paths. const url = new URL(request.url); const mcpRoute = classifyMcpPath(url.pathname); + // The two OAuth discovery documents are static JSON (the protected- + // resource doc is pure; the authorization-server doc is one upstream + // fetch). They are the first thing every MCP client requests, and they + // were dispatched through the app plane, whose first evaluation in a cold + // isolate costs seconds: measured p95 5.0s on `/.well-known/*` from cold + // isolates against 0ms warm. Answer them here, before anything that + // would load the Effect app graph. The envelope still mounts the same + // routes for hosts that serve `/mcp` through it (and for tests). + if (mcpRoute !== null && mcpRoute.kind !== "mcp") { + if (request.method === "OPTIONS") return corsPreflightResponse(); + if (request.method === "GET" || request.method === "HEAD") { + if (mcpRoute.kind === "oauth-protected-resource") { + return protectedResourceMetadataResponse( + mcpRoute.organizationId, + mcpRoute.toolkitSlug ?? null, + ); + } + return Effect.runPromise(authorizationServerMetadataResponse); + } + } if (mcpRoute?.kind === "mcp") { // The Cloudflare Agents MCP bridge needs the platform ExecutionContext // to pass authenticated session props into the hibernatable DO. diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index b01c62dbc7..ee017e62f4 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -5513,116 +5513,156 @@ 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 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)), - }); - // 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[] = []; - 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 + // + // Two of the triggers are SIGNALS that the persisted catalog is wrong + // (stale-marked, config-revised); the third is a GUESS that it might be + // (expired). A read waits for the signalled rebuilds (within its grace + // budget), but never for the guessed ones: an expired catalog is served + // as-is while its re-list runs behind the read, stale-while-revalidate. + // Every remote-catalog connection expires on the same 15-minute clock, + // so before this split a read that happened to trip the TTL paid one + // upstream `tools/list` handshake per connection at once — measured at + // p50 2.2s on the cloud `/api/tools` read, with nothing to show for it + // but a catalog identical to the one it already had. `awaitExpired` is + // the strict mode (`toolsSyncGraceMs: null`): block on everything. + const syncStaleConnectionTools = (options: { readonly awaitExpired: boolean }) => + 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( + (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; - 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[])), - ), - Effect.withSpan("executor.tools.sync_stale", { - attributes: { - "executor.integration": connection.integration, - "executor.connection": connection.name, + 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)), + }); + // 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. One semaphore + // bounds the awaited and the deferred rebuilds together, so a read + // still opens at most `STALE_TOOLS_SYNC_CONCURRENCY` listings. + const permits = Semaphore.makeUnsafe(STALE_TOOLS_SYNC_CONCURRENCY); + const awaited: Effect.Effect[] = []; + const deferred: Effect.Effect[] = []; + 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; + + const target = staleMarked || configRevised || options.awaitExpired ? awaited : deferred; + target.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, + "executor.tools.sync_trigger": staleMarked + ? "stale_marked" + : configRevised + ? "config_revised" + : "expired", + }, + }), + permits.withPermits(1), + ), + ); + } + if (deferred.length > 0) { + const fiber = yield* Effect.forkDetach( + Effect.all(deferred, { concurrency: "unbounded", discard: true }), + ); + // Same keep-alive as the read-side fork: on hosts that cancel + // request-scoped I/O once the response settles, the deferred + // re-lists must outlive the read that tripped the TTL. + config.waitUntil?.( + new Promise((resolve) => fiber.addObserver(() => resolve(undefined))), + ); + } + yield* Effect.annotateCurrentSpan({ + "executor.tools.sync_awaited": awaited.length, + "executor.tools.sync_deferred": deferred.length, + }); + yield* Effect.all(awaited, { concurrency: "unbounded", discard: true }); }); - }); // How long a tools read waits for the stale sync before answering from // the persisted rows (`ExecutorConfig.toolsSyncGraceMs`; `null` blocks @@ -5643,7 +5683,7 @@ export const createExecutor = Effect.gen(function* () { const fiber = yield* Effect.forkDetach( - syncStaleConnectionTools.pipe( + syncStaleConnectionTools({ awaitExpired: false }).pipe( Effect.catch((error) => Effect.logWarning("executor stale tool sync scan failed", { error: describeSyncFailure(error), @@ -5663,7 +5703,7 @@ export const createExecutor = => Effect.gen(function* () { if (toolsSyncGraceMs === null) { - yield* syncStaleConnectionTools; + yield* syncStaleConnectionTools({ awaitExpired: true }); } else { yield* awaitStaleSyncWithinGrace(toolsSyncGraceMs); } diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts index 813f4d5c40..a485163ff5 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts @@ -29,6 +29,8 @@ import { resetInFlightColdBuildCountForTest, resetResidentRuntimeCountForTest, resetResidentSessionRegistryForTest, + residencyAttributes, + residencyCounts, residentSessionIdsForTest, touchResidentSession, } from "./session-runtime-residency"; @@ -1833,6 +1835,89 @@ describe("McpAgentSessionDOBase residency cap eviction", () => { expect(() => markEvictionRequested("never-registered")).not.toThrow(); }); }); + + // These exist because "nothing was evictable" (`mcp.isolate.cap_overflow`) + // does not, on its own, say why: an isolate full of legitimately-streaming + // sessions and one full of sessions that are merely mid-request both trip + // it identically. `residencyCounts`/`residencyAttributes` are the synchronous + // breakdown that tells the two apart from the registry alone. + describe("residencyCounts / residencyAttributes", () => { + it("counts evictable, pinned, eviction-pending, and streaming entries from a mixed registry", () => { + const now = Date.now(); + // Evictable and currently streaming: `canEvict()` only reflects paused/ + // running execution state, not streams, so a registrant can legitimately + // report both. + registerResidentSession({ + sessionId: "evictable-streaming", + lastActivityMs: now, + canEvict: () => true, + dispose: async () => undefined, + isStreaming: () => true, + }); + // Evictable, no stream. + registerResidentSession({ + sessionId: "evictable-idle", + lastActivityMs: now, + canEvict: () => true, + dispose: async () => undefined, + isStreaming: () => false, + }); + // Pinned (not evictable), no stream. + registerResidentSession({ + sessionId: "pinned", + lastActivityMs: now, + canEvict: () => false, + dispose: async () => undefined, + isStreaming: () => false, + }); + // Evictable per `canEvict()`, but inside the post-request grace window — + // this is the case `pickEvictionCandidate` also skips, and exactly the + // gap `evictionPending` exists to surface. + registerResidentSession({ + sessionId: "eviction-pending", + lastActivityMs: now, + canEvict: () => true, + dispose: async () => undefined, + isStreaming: () => false, + }); + markEvictionRequested("eviction-pending", now); + + const counts = residencyCounts(now); + expect(counts.evictable, "3 of 4 entries report canEvict() === true").toBe(3); + expect(counts.pinned, "1 of 4 entries report canEvict() === false").toBe(1); + expect( + counts.evictionPending, + "1 entry has an eviction request still inside the grace window", + ).toBe(1); + expect(counts.streaming, "1 of 4 entries is currently streaming").toBe(1); + + const attributes = residencyAttributes(); + expect(attributes["mcp.isolate.resident_evictable"]).toBe(3); + expect(attributes["mcp.isolate.resident_pinned"]).toBe(1); + expect(attributes["mcp.isolate.resident_eviction_pending"]).toBe(1); + expect(attributes["mcp.isolate.resident_streaming"]).toBe(1); + expect(attributes["mcp.isolate.in_flight_cold_builds"]).toBe(currentInFlightColdBuildCount()); + }); + + it("omits the streaming attribute entirely when no registered entry can answer it", () => { + registerResidentSession({ + sessionId: "no-streaming-signal", + lastActivityMs: Date.now(), + canEvict: () => true, + dispose: async () => undefined, + }); + + const counts = residencyCounts(); + expect(counts.streaming, "no entry supplies isStreaming, so it is left undefined").toBe( + undefined, + ); + expect(residencyAttributes()).not.toHaveProperty("mcp.isolate.resident_streaming"); + }); + + it("reports all-zero counts for an empty registry", () => { + expect(residencyCounts()).toEqual({ evictable: 0, pinned: 0, evictionPending: 0 }); + }); + }); }); // The request-id ledger (`__mcp_stream_reqs__:`, written by the diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index 5a90dc5978..88748ff94d 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -50,6 +50,7 @@ import { reserveColdBuildSlot, type ResidentSessionEntry, residencyAttributes, + residencyCounts, RESIDENT_RUNTIME_SOFT_CAP, touchResidentSession, } from "./session-runtime-residency"; @@ -1025,6 +1026,26 @@ export abstract class McpAgentSessionDOBase< const candidate = pickEvictionCandidate(); if (!candidate) { yield* Effect.annotateCurrentSpan({ "mcp.isolate.cap_overflow": true }); + // Span attributes are sampled, so this condition can be invisible in + // traces even though it happens on every affected init. A structured + // log line is unsampled (Cloudflare logpush -> Axiom), so this is + // the count-independent-of-sampling signal for the same event, and it + // carries the WHY (evictable/pinned/streaming/eviction_pending) that + // `mcp.isolate.cap_overflow: true` alone cannot. + const counts = residencyCounts(); + console.warn( + JSON.stringify({ + event: "mcp_isolate_cap_overflow", + sessionId: self.sessionIdForTelemetry(), + residentRuntimes: currentResidentRuntimeCount(), + inFlightColdBuilds: currentInFlightColdBuildCount(), + evictable: counts.evictable, + pinned: counts.pinned, + streaming: counts.streaming ?? 0, + evictionPending: counts.evictionPending, + softCap: self.residentRuntimeSoftCap(), + }), + ); return; } yield* Effect.sync(() => self.queueCapEvictionRequest(candidate)); @@ -1481,6 +1502,7 @@ export abstract class McpAgentSessionDOBase< lastActivityMs: Date.now(), canEvict: () => self.canEvictResidentRuntime(), dispose: () => self.requestSelfEviction(), + isStreaming: () => self.activeStreamCount() > 0, }); } } diff --git a/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts b/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts index 06ff5c223c..04214468a6 100644 --- a/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts +++ b/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts @@ -132,6 +132,16 @@ export type ResidentSessionEntry = { evictionRequestedAt?: number; readonly canEvict: () => boolean; readonly dispose: (reason: "cap") => Promise; + /** + * Cheap, synchronous "does this session currently have an active stream" + * check, so the registry can explain WHY residency will not budge — a + * cap-overflow isolate full of streaming sessions looks identical to one + * full of paused-but-not-evictable sessions unless this is broken out + * separately. Optional because not every registrant can answer it for + * free; when absent, `residencyCounts` simply omits the streaming count + * rather than reporting a misleading zero. + */ + readonly isStreaming?: () => boolean; }; /** @@ -231,6 +241,58 @@ export const resetResidentSessionRegistryForTest = (): void => { export const residentSessionIdsForTest = (): ReadonlyArray => Array.from(residentSessions.keys()); +/** + * Breaks the registry down by WHY a resident session is or is not eligible + * for eviction right now, computed in one synchronous pass over the same + * entries `pickEvictionCandidate` walks. This exists because "nothing was + * evictable" (`mcp.isolate.cap_overflow`) is not, on its own, actionable: an + * isolate full of legitimately-streaming sessions and one full of sessions + * that are merely mid-request look identical from the outside without this + * breakdown. + * + * `evictable` and `pinned` are computed from `canEvict()` alone and always + * sum to the registry's size. `evictionPending` is a separate, possibly + * overlapping count — an entry can report `canEvict() === true` and still be + * sitting out the post-request grace window that `pickEvictionCandidate` + * also honors, which is exactly the case that makes a healthy-looking + * registry still fail to yield a candidate. `streaming` is omitted entirely + * (rather than reported as zero) unless at least one registered entry + * supplies {@link ResidentSessionEntry.isStreaming}, since a registrant that + * cannot answer it cheaply should not be silently counted as non-streaming. + */ +export const residencyCounts = ( + nowMs = Date.now(), +): { + readonly evictable: number; + readonly pinned: number; + readonly evictionPending: number; + readonly streaming?: number; +} => { + let evictable = 0; + let pinned = 0; + let evictionPending = 0; + let streaming = 0; + let sawStreamingSignal = false; + for (const entry of residentSessions.values()) { + if (entry.canEvict()) { + evictable += 1; + } else { + pinned += 1; + } + if ( + entry.evictionRequestedAt !== undefined && + nowMs - entry.evictionRequestedAt < EVICTION_REQUEST_GRACE_MS + ) { + evictionPending += 1; + } + if (entry.isStreaming) { + sawStreamingSignal = true; + if (entry.isStreaming()) streaming += 1; + } + } + return { evictable, pinned, evictionPending, ...(sawStreamingSignal ? { streaming } : {}) }; +}; + type MemoryCapablePerformance = { readonly memory?: { readonly usedJSHeapSize?: unknown; @@ -271,9 +333,26 @@ export const isolateMemoryAttributes = (): Record => { * every idle disposal, so production can confirm the mechanism directly: * residency should now fall back toward zero as sessions go idle instead of * climbing with the number of connected-but-quiet clients. + * + * The `resident_evictable`/`resident_pinned`/`resident_eviction_pending` + * (and, when available, `resident_streaming`) attributes exist so a sampled + * span can answer WHY residency is or is not shrinking, without needing a + * heap snapshot Workers cannot provide: a cap-overflow isolate full of + * `resident_pinned` looks nothing like one full of `resident_streaming`, and + * only one of those is the eviction policy's own doing. */ -export const residencyAttributes = (): Record => ({ - "mcp.isolate.resident_runtimes": currentResidentRuntimeCount(), - "mcp.isolate.peak_resident_runtimes": peakResidentRuntimeCountInIsolate(), - ...isolateMemoryAttributes(), -}); +export const residencyAttributes = (): Record => { + const counts = residencyCounts(); + return { + "mcp.isolate.resident_runtimes": currentResidentRuntimeCount(), + "mcp.isolate.peak_resident_runtimes": peakResidentRuntimeCountInIsolate(), + "mcp.isolate.resident_evictable": counts.evictable, + "mcp.isolate.resident_pinned": counts.pinned, + ...(counts.streaming === undefined + ? {} + : { "mcp.isolate.resident_streaming": counts.streaming }), + "mcp.isolate.resident_eviction_pending": counts.evictionPending, + "mcp.isolate.in_flight_cold_builds": currentInFlightColdBuildCount(), + ...isolateMemoryAttributes(), + }; +}; diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index f7c1a349f7..c0909c1c13 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -1513,7 +1513,7 @@ const registerPassthroughTools = ( ), ); }); - }).pipe(Effect.withSpan("mcp.host.register_search_invoke")); + }); // --------------------------------------------------------------------------- // Server factory @@ -2062,10 +2062,6 @@ export const createExecutorMcpServer = ( }, ({ code }, extra) => runToolEffect(executeCode(code, extra), extra), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "execute" }, - }), ); yield* Effect.sync(() => @@ -2093,10 +2089,6 @@ export const createExecutorMcpServer = ( ({ name }, extra) => runToolEffect(Effect.succeed(skillsResult(name, executeInventory, skillCatalog)), extra), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "skills" }, - }), ); if (!passthrough) @@ -2161,11 +2153,7 @@ export const createExecutorMcpServer = ( ({ executionId }, extra) => runToolEffect(resumeAfterBrowserApproval(executionId, extra), extra), ); - }).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "resume" }, - }), - ); + }); // --- per-integration search tools (opt-in, `?search_tools=true`) --- // @@ -2191,6 +2179,7 @@ export const createExecutorMcpServer = ( const namespaces = parseIntegrationInventory(description).filter((slug) => TOOL_NAME_SAFE_SLUG.test(slug), ); + yield* Effect.annotateCurrentSpan({ "mcp.namespace_search.count": namespaces.length }); yield* Effect.sync(() => { for (const integration of namespaces) { server.registerTool( @@ -2214,14 +2203,7 @@ export const createExecutorMcpServer = ( ), ); } - }).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { - "mcp.tool.name": "search_", - "mcp.namespace_search.count": namespaces.length, - }, - }), - ); + }); } // --- artifacts / MCP Apps --- @@ -2607,11 +2589,7 @@ export const createExecutorMcpServer = ( ], }), ); - }).pipe( - Effect.withSpan("mcp.host.register_resource", { - attributes: { "mcp.resource.uri": MCP_APPS_SHELL_RESOURCE_URI }, - }), - ); + }); yield* Effect.sync(() => registerAppTool( @@ -2671,10 +2649,6 @@ export const createExecutorMcpServer = ( extra, ), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "create-artifact" }, - }), ); yield* Effect.sync(() => @@ -2740,10 +2714,6 @@ export const createExecutorMcpServer = ( extra, ), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "edit-artifact" }, - }), ); yield* Effect.sync(() => @@ -2758,10 +2728,6 @@ export const createExecutorMcpServer = ( }, (_args, extra) => runToolEffect(listArtifacts(), extra), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "list-artifacts" }, - }), ); yield* Effect.sync(() => @@ -2783,10 +2749,6 @@ export const createExecutorMcpServer = ( }, ({ id }, extra) => runToolEffect(showArtifact(id), extra), ), - ).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "show-artifact" }, - }), ); yield* Effect.sync(() => { @@ -2844,11 +2806,7 @@ export const createExecutorMcpServer = ( extra, ), ); - }).pipe( - Effect.withSpan("mcp.host.register_tool", { - attributes: { "mcp.tool.name": "execute-action" }, - }), - ); + }); } // Client capabilities only exist after `initialize`, and `tools/list` is diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index f3328876b3..b08a4feff8 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -133,11 +133,13 @@ describe("MCP tool-catalog sync (end-to-end)", () => { }), ); - it.effect("expired catalogs re-list on read once older than the freshness TTL", () => + // `it.live` (real clock): the re-list lands on a detached fiber behind the + // read, and the poll below must actually wait for a real HTTP round trip. + it.live("expired catalogs re-list behind the read once older than the freshness TTL", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); const server = yield* serveMcpServer(mutable.factory); - // Everything is instantly stale — every tools read re-lists. + // Everything is instantly stale — every tools read triggers a re-list. const executor = yield* makeCatalogTestExecutor(server.url, { toolsSyncTtlMs: 0 }); expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); @@ -145,6 +147,35 @@ describe("MCP tool-catalog sync (end-to-end)", () => { // Server-side change with no notification and no executor signal at all. mutable.renameTool(); + // Expiry is a guess, not a signal: the read that trips the TTL serves + // the persisted catalog and the re-list lands behind it (a signalled + // stale-mark, by contrast, is awaited — see the two tests above). A + // later read reflects the change once the detached rebuild converges. + expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); + const converged = yield* Effect.gen(function* () { + while (true) { + const names = toolNames(yield* executor.tools.list()); + if (names.includes(mutable.renamedToolName)) return names; + yield* Effect.sleep("50 millis"); + } + }).pipe(Effect.timeoutOption("10 seconds")); + expect(Option.isSome(converged)).toBe(true); + expect(Option.getOrThrow(converged)).not.toContain(mutable.initialToolName); + }), + ); + + it.effect("strict mode (toolsSyncGraceMs: null) awaits expired re-lists too", () => + Effect.gen(function* () { + const mutable = makeMutableCatalogMcpServer(); + const server = yield* serveMcpServer(mutable.factory); + const executor = yield* makeCatalogTestExecutor(server.url, { + toolsSyncTtlMs: 0, + toolsSyncGraceMs: null, + }); + + expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); + mutable.renameTool(); + const refreshed = toolNames(yield* executor.tools.list()); expect(refreshed).toContain(mutable.renamedToolName); expect(refreshed).not.toContain(mutable.initialToolName); diff --git a/packages/react/src/lib/use-connection-health.test.ts b/packages/react/src/lib/use-connection-health.test.ts index 89e5d08405..849229edd6 100644 --- a/packages/react/src/lib/use-connection-health.test.ts +++ b/packages/react/src/lib/use-connection-health.test.ts @@ -1,7 +1,15 @@ -import { describe, expect, it } from "@effect/vitest"; +import { beforeEach, describe, expect, it } from "@effect/vitest"; import type { HealthCheckResult } from "@executor-js/sdk/shared"; -import { HEALTH_REVALIDATE_MS, revalidateQuery } from "./use-connection-health"; +import { + AUTO_PROBE_FLOOR_MS, + HEALTH_REVALIDATE_MS, + clearAutomaticProbeMemory, + recordAutomaticProbe, + resetAutomaticProbeMemoryForTest, + revalidateQuery, + shouldAutoProbe, +} from "./use-connection-health"; const verdict = (status: HealthCheckResult["status"]): HealthCheckResult => ({ status, @@ -46,3 +54,74 @@ describe("revalidateQuery", () => { expect(windows, "only the healthy path is gated").toEqual([undefined, undefined, undefined]); }); }); + +// shouldAutoProbe consults module-scope memory (see automaticProbeMemory in +// use-connection-health.ts), so every test starts from a clean slate and uses +// a fresh key to avoid cross-test interference even under parallel execution. +describe("shouldAutoProbe", () => { + beforeEach(() => { + resetAutomaticProbeMemoryForTest(); + }); + + it("probes on first sight, with no persisted verdict and no memory", () => { + expect(shouldAutoProbe("acme:github:default", null, Date.now())).toBe(true); + }); + + it("does not re-probe a second time inside the floor, even for an expired verdict", () => { + const key = "acme:github:expired-in-floor"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "expired", checkedAt: now }); + + expect( + shouldAutoProbe(key, verdict("expired"), now + AUTO_PROBE_FLOOR_MS - 1), + "a remount inside the floor must not re-arm the probe", + ).toBe(false); + }); + + it("probes again once the floor has elapsed, for a non-healthy verdict", () => { + const key = "acme:github:expired-after-floor"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "expired", checkedAt: now }); + + expect( + shouldAutoProbe(key, verdict("expired"), now + AUTO_PROBE_FLOOR_MS + 1), + "the floor elapsing re-arms the probe so recovery can still show", + ).toBe(true); + }); + + it("suppresses a remembered healthy result younger than HEALTH_REVALIDATE_MS, even past the floor", () => { + const key = "acme:github:healthy-remembered"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "healthy", checkedAt: now }); + + expect( + shouldAutoProbe(key, null, now + AUTO_PROBE_FLOOR_MS + 1), + "a fresh healthy verdict must not probe just because the floor elapsed", + ).toBe(false); + }); + + it("probes again once a remembered healthy result ages past HEALTH_REVALIDATE_MS", () => { + const key = "acme:github:healthy-stale"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "healthy", checkedAt: now }); + + expect( + shouldAutoProbe(key, null, now + HEALTH_REVALIDATE_MS + 1), + "a healthy verdict must revalidate once it goes stale", + ).toBe(true); + }); + + it("re-arms immediately once the entry is cleared, ignoring the floor", () => { + const key = "acme:github:cleared"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "expired", checkedAt: now }); + expect(shouldAutoProbe(key, verdict("expired"), now + 1), "still inside the floor").toBe(false); + + clearAutomaticProbeMemory(key); + + expect( + shouldAutoProbe(key, null, now + 1), + "clearing the memory re-arms the probe even inside the floor", + ).toBe(true); + }); +}); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 9bfc1fe36a..bfec920de1 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -26,13 +26,18 @@ const connectionParams = (connection: Connection) => ({ name: connection.name, }); +const probeKey = (connection: Connection): string => + `${connection.owner}:${connection.integration}:${connection.name}`; + /** Whether a persisted verdict may render as-is without a background probe. * Healthy-and-fresh renders untouched. Everything else revalidates: stale or * never-checked for obvious reasons, and NON-healthy always; an expired dot * is exactly the verdict the user is waiting to see change, so recovery must * show on the next load, not after the freshness window. */ -const healthyAndFresh = (last: HealthCheckResult | null | undefined): boolean => - last?.status === "healthy" && Date.now() - last.checkedAt < HEALTH_REVALIDATE_MS; +const healthyAndFresh = ( + last: HealthCheckResult | null | undefined, + now: number = Date.now(), +): boolean => last?.status === "healthy" && now - last.checkedAt < HEALTH_REVALIDATE_MS; /** The revalidation query: a healthy (but stale) verdict defers to the * server-enforced window so N open tabs can't stampede the upstream; a @@ -81,6 +86,76 @@ const freshestVerdict = ( return persisted.checkedAt > live.checkedAt ? persisted : live; }; +/** Module-scope memory of automatic probes, keyed by `probeKey`. Unlike the + * per-hook `useRef` guards below (which reset whenever a row remounts), this + * map survives remounts: it is what stops an org-wide reactivity bump from + * remounting a row and re-arming its probe every time. `at` is recorded from + * the LOCAL wall clock, never `result.checkedAt` — the server may answer + * from its 5-minute cache with an old `checkedAt`, which would under-count + * elapsed time and defeat the floor below. */ +const automaticProbeMemory = new Map< + string, + { readonly at: number; readonly result: HealthCheckResult } +>(); + +/** How long a remembered automatic probe blocks another automatic probe for + * the same connection, regardless of verdict. A non-healthy verdict must + * still eventually re-probe so recovery can show (see `revalidateQuery`), + * but "eventually" must not mean "every remount": this floor is what turns a + * per-second remount storm into at most one probe per floor, for healthy and + * non-healthy verdicts alike. */ +export const AUTO_PROBE_FLOOR_MS = 30 * 1000; + +/** + * Whether an automatic probe should fire for `key` right now, given the + * persisted verdict. Consults the module memory together with `persisted`: + * - if the freshest of the two (see `freshestVerdict`) is healthy-and-fresh, + * never probe — the existing freshness contract. + * - otherwise, a remembered probe younger than `AUTO_PROBE_FLOOR_MS` blocks + * another probe no matter its verdict — the anti-storm floor. + * - otherwise (no memory yet, or memory older than the floor) probe. + * Pure with respect to its arguments and `now`; the only state it reads is + * the shared module memory, which only `recordAutomaticProbe` and + * `clearAutomaticProbeMemory` mutate. + */ +export function shouldAutoProbe( + key: string, + persisted: HealthCheckResult | null | undefined, + now: number = Date.now(), +): boolean { + const remembered = automaticProbeMemory.get(key); + const freshest = freshestVerdict(remembered?.result ?? null, persisted); + if (healthyAndFresh(freshest, now)) return false; + if (remembered !== undefined && now - remembered.at < AUTO_PROBE_FLOOR_MS) return false; + return true; +} + +/** Records a successful probe — automatic or manual — into the module + * memory, so a later remount or automatic pass can see it. See + * `automaticProbeMemory` for why `at` is the local clock, not the server's + * `checkedAt`. Exported (not test-only) so `shouldAutoProbe`'s decision logic + * can be exercised directly, without rendering the hooks that normally call + * it. */ +export function recordAutomaticProbe(key: string, result: HealthCheckResult): void { + automaticProbeMemory.set(key, { at: Date.now(), result }); +} + +/** Deletes the remembered probe for `key`, forcing the next `shouldAutoProbe` + * call to return `true` regardless of the floor. Called on the reconnect + * ("cleared verdict") transition, which must always re-probe: an OAuth + * re-mint is the one case where the anti-storm floor must not apply. + * Exported for the same testability reason as `recordAutomaticProbe`. */ +export function clearAutomaticProbeMemory(key: string): void { + automaticProbeMemory.delete(key); +} + +/** Test-only escape hatch: clears every remembered automatic probe. The + * memory is module-scope, so without this, probes recorded by one test + * would leak into the next. */ +export function resetAutomaticProbeMemoryForTest(): void { + automaticProbeMemory.clear(); +} + /** * Imperative invalidation of the connections cache for one owner. The server * persists every verdict on `last_health`, so after a check we must re-read the @@ -99,10 +174,12 @@ function useInvalidateConnections(): (owner: Owner) => void { /** * Health for ONE connection, stale-while-revalidate. The persisted verdict - * renders instantly; a background probe on mount corrects it in place (once - * per mount, quiet on failure: the persisted verdict is still the best known - * state). `runCheck` is the manual path ("Check now"): it always forces a - * fresh probe and folds the result into the same live state. + * renders instantly; a background probe corrects it in place, guarded by + * `shouldAutoProbe` so a row that remounts (e.g. from an org-wide reactivity + * bump) does not re-probe every time, quiet on failure: the persisted verdict + * is still the best known state. `runCheck` is the manual path ("Check + * now"): it always forces a fresh probe and folds the result into the same + * live state. */ export function useConnectionHealth(connection: Connection): { readonly probe: HealthCheckResult | null; @@ -111,7 +188,14 @@ export function useConnectionHealth(connection: Connection): { } { // A live probe result, once a check has run; merged with the persisted // verdict by freshness (see freshestVerdict for why not live-always-wins). - const [liveProbe, setLiveProbe] = useState(null); + // Seeded from the module memory on mount, not `null`: without this, a + // remount (any connections-write in the org bumps the org-wide reactivity + // key and can remount this row) would render the OLDER persisted verdict + // until the background probe resolves, even though we already know the + // last automatic probe's result. + const [liveProbe, setLiveProbe] = useState( + () => automaticProbeMemory.get(probeKey(connection))?.result ?? null, + ); const doCheck = useAtomSet(checkConnectionHealth, { mode: "promiseExit" }); const invalidateConnections = useInvalidateConnections(); @@ -121,13 +205,16 @@ export function useConnectionHealth(connection: Connection): { // Health checks are AUTOMATIC: loading the list revalidates any verdict // older than the freshness window (or never checked), stale-while-revalidate // style: the persisted verdict renders instantly, the probe corrects it in - // place. The guard is once per mount PLUS once per clearing: the ref holds - // the last epoch seen, and a verdict giving way to `null` (an OAuth re-mint - // cleared it) re-arms the probe — that is how a completed reconnect gets its - // recovery probe without a page reload. Only the clearing transition - // re-arms; every other epoch change (a probe's own verdict echoed back by - // the refetch, a concurrent surface's fresher verdict) stays quiet, keeping - // the no-probe-storm invariant of the original once-per-mount guard. + // place. The per-mount part of the guard is once per mount PLUS once per + // clearing (the ref holds the last epoch seen this mount, and a verdict + // giving way to `null` -- an OAuth re-mint -- re-arms it, which is how a + // completed reconnect gets its recovery probe without a page reload). But a + // fresh `useRef` starts at `undefined` on every remount, so that guard alone + // re-arms on every remount too. `shouldAutoProbe` is the guard that survives + // remounts: it consults the module-scope `automaticProbeMemory`, so a row + // remounted a second later -- before its own last probe even resolved, or + // resolved with a non-healthy verdict -- does not re-probe. Only the + // clearing transition bypasses that floor (see `clearAutomaticProbeMemory`). const seenEpoch = useRef(undefined); useEffect(() => { const last = connection.lastHealth; @@ -136,7 +223,9 @@ export function useConnectionHealth(connection: Connection): { const cleared = epoch === null && seenEpoch.current !== null && !firstSight; seenEpoch.current = epoch; if (!firstSight && !cleared) return; - if (healthyAndFresh(last)) return; + const key = probeKey(connection); + if (cleared) clearAutomaticProbeMemory(key); + if (!shouldAutoProbe(key, last)) return; void doCheck({ params: connectionParams(connection), query: revalidateQuery(last), @@ -148,6 +237,7 @@ export function useConnectionHealth(connection: Connection): { // churns the cache (which would refetch connections, re-run this // effect, and, but for the epoch guard, risk a probe loop). if (!Exit.isSuccess(exit)) return; + recordAutomaticProbe(key, exit.value); seenEpoch.current = exit.value.checkedAt; setLiveProbe(exit.value); if (exit.value.status !== (last?.status ?? "unknown")) { @@ -159,13 +249,17 @@ export function useConnectionHealth(connection: Connection): { const runCheck = useCallback(async () => { // Manual "Check now": invalidate the connections cache unconditionally so // every surface picks up the freshly persisted verdict. Adopting the - // result's epoch keeps the resulting refetch from re-probing. + // result's epoch keeps the resulting refetch from re-probing. This path + // always bypasses `shouldAutoProbe` -- the user explicitly asked for a + // fresh check -- but still records into the module memory, so a remount + // right after a manual check doesn't immediately fire an automatic one. const exit = await doCheck({ params: connectionParams(connection), query: {}, reactivityKeys: connectionCheckKeys, }); if (Exit.isSuccess(exit)) { + recordAutomaticProbe(probeKey(connection), exit.value); seenEpoch.current = exit.value.checkedAt; setLiveProbe(exit.value); } @@ -175,9 +269,6 @@ export function useConnectionHealth(connection: Connection): { return { probe, status, runCheck }; } -const probeKey = (connection: Connection): string => - `${connection.owner}:${connection.integration}:${connection.name}`; - /** * Health for MANY connections at once (the integrations-list summary), where * hooks-in-a-loop is illegal. One effect walks the list and fires the same @@ -189,32 +280,54 @@ const probeKey = (connection: Connection): string => export function useConnectionsHealth( connections: readonly Connection[], ): (connection: Connection) => HealthCheckResult | null { - const [liveProbes, setLiveProbes] = useState>(new Map()); + // Seeded from the module memory for whichever connections are known at + // mount time, for the same reason as `useConnectionHealth`'s `liveProbe`: + // a remount must show the last automatic probe's verdict, not fall back to + // the older persisted one while a new probe is (or isn't, thanks to + // `shouldAutoProbe`) in flight. + const [liveProbes, setLiveProbes] = useState>(() => { + const seeded = new Map(); + for (const connection of connections) { + const remembered = automaticProbeMemory.get(probeKey(connection)); + if (remembered) seeded.set(probeKey(connection), remembered.result); + } + return seeded; + }); const doCheck = useAtomSet(checkConnectionHealth, { mode: "promiseExit" }); const invalidateConnections = useInvalidateConnections(); // Once per VERDICT per connection (same epoch guard as the single-connection // hook): the list streams in asynchronously, so the effect re-runs as rows - // arrive; each row probes once per persisted-verdict epoch, and a re-minted - // connection (epoch cleared to null) probes again without a remount. + // arrive; each row is considered once per persisted-verdict epoch, and a + // re-minted connection (epoch cleared to null) is considered again without + // a remount. As with the single-connection hook, this `useRef` guard alone + // would re-arm on every remount of the owning component, so whether a + // considered row actually probes is decided by `shouldAutoProbe` against + // the module-scope `automaticProbeMemory`, which survives that remount. const revalidated = useRef(new Map()); useEffect(() => { for (const connection of connections) { const key = probeKey(connection); const last = connection.lastHealth; const epoch = verdictEpoch(last); - if (revalidated.current.has(key) && revalidated.current.get(key) === epoch) continue; + const firstSight = !revalidated.current.has(key); + const previousEpoch = revalidated.current.get(key) ?? null; + if (!firstSight && previousEpoch === epoch) continue; + const cleared = !firstSight && previousEpoch !== null && epoch === null; revalidated.current.set(key, epoch); - if (healthyAndFresh(last)) continue; + if (cleared) clearAutomaticProbeMemory(key); + if (!shouldAutoProbe(key, last)) continue; void doCheck({ params: connectionParams(connection), query: revalidateQuery(last), }).then((exit) => { // Same automatic-path rule as the single-connection hook: reflect the - // verdict, adopt its epoch so the refetch doesn't re-probe, and + // verdict, adopt its epoch so the refetch doesn't re-probe, record it + // into the module memory so a remount respects the floor, and // invalidate the connections cache only when the verdict changed so an // unchanged reconfirm never churns the cache. if (!Exit.isSuccess(exit)) return; + recordAutomaticProbe(key, exit.value); revalidated.current.set(key, exit.value.checkedAt); setLiveProbes((current) => new Map(current).set(key, exit.value)); if (exit.value.status !== (last?.status ?? "unknown")) { From b12dce07ea0ad4947b269bd1d037289f179708a5 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:20:58 -0700 Subject: [PATCH 2/6] Keep the request socket open for deferred catalog rebuilds; scope health-probe memory by identity Co-Authored-By: Claude Fable 5.1 --- .../src/account/list-members.node.test.ts | 35 +++++++---- .../src/account/workos-account-service.ts | 45 +++++++------- apps/cloud/src/auth/jwks-cache.node.test.ts | 10 ++++ apps/cloud/src/auth/jwks-cache.ts | 4 ++ apps/cloud/src/db/db.close.test.ts | 60 ++++++++++++++++++- apps/cloud/src/db/db.ts | 47 ++++++++++++++- apps/cloud/src/db/fuma.ts | 6 +- .../core/api/src/server/scoped-executor.ts | 28 ++++++++- packages/core/sdk/src/executor-fuma-db.ts | 9 +++ .../src/lib/use-connection-health.test.ts | 43 ++++++++++++- .../react/src/lib/use-connection-health.ts | 53 ++++++++++++---- 11 files changed, 285 insertions(+), 55 deletions(-) diff --git a/apps/cloud/src/account/list-members.node.test.ts b/apps/cloud/src/account/list-members.node.test.ts index dbabb870a5..b7b725234f 100644 --- a/apps/cloud/src/account/list-members.node.test.ts +++ b/apps/cloud/src/account/list-members.node.test.ts @@ -140,18 +140,21 @@ const stubApiKeys = Layer.succeed(ApiKeyService)({ revokeOrgKey: () => Effect.die("listMembers does not touch api keys"), }); -/** Autumn stub whose `getOrCreate` either succeeds with a plan or fails. */ -const stubAutumn = (mode: "ok" | "fail") => +/** Autumn stub whose `getOrCreate` succeeds with a plan, fails as a typed + * error, or dies (a defect — the SDK throwing something untyped). */ +const stubAutumn = (mode: "ok" | "fail" | "die") => Layer.succeed(AutumnService)({ use: (fn: (client: Autumn) => Promise) => - mode === "fail" - ? Effect.fail(new AutumnError({ message: "autumn unreachable" })) - : Effect.promise(() => - fn( - // eslint-disable-next-line @typescript-eslint/no-explicit-any -- test stub narrows the SDK client to what getMemberSeats actually calls - { customers: { getOrCreate: async () => ({ subscriptions: [] }) } } as any, + mode === "die" + ? Effect.die("autumn sdk threw") + : mode === "fail" + ? Effect.fail(new AutumnError({ message: "autumn unreachable" })) + : Effect.promise(() => + fn( + // eslint-disable-next-line @typescript-eslint/no-explicit-any -- test stub narrows the SDK client to what getMemberSeats actually calls + { customers: { getOrCreate: async () => ({ subscriptions: [] }) } } as any, + ), ), - ), ensureCustomer: () => Effect.void, checkExecutionBalance: () => Effect.die("listMembers does not check execution balance"), trackExecution: () => Effect.void, @@ -160,7 +163,7 @@ const stubAutumn = (mode: "ok" | "fail") => const providerWith = ( members: ReadonlyArray>, - autumnMode: "ok" | "fail", + autumnMode: "ok" | "fail" | "die", ) => { const { layer: workosLayer, calls } = stubWorkOS(members); const provider = AccountProvider.asEffect().pipe( @@ -210,6 +213,18 @@ describe("listMembers · provider boundary", () => { }), ); + it.effect("an Autumn DEFECT also degrades seats to defaults (the old catchCause contract)", () => + Effect.gen(function* () { + const { provider } = providerWith([membership(USER)], "die"); + const account = yield* provider; + + const result = yield* account.listMembers(orgHeaders); + + expect(result.members).toHaveLength(1); + expect(result.seats).toEqual({ used: 0, granted: 0, unlimited: false }); + }), + ); + it.effect("a listOrgMembers failure fails the whole request, not just seats", () => Effect.gen(function* () { const failingWorkOS = Layer.succeed( diff --git a/apps/cloud/src/account/workos-account-service.ts b/apps/cloud/src/account/workos-account-service.ts index 74b1b9a045..471d785e89 100644 --- a/apps/cloud/src/account/workos-account-service.ts +++ b/apps/cloud/src/account/workos-account-service.ts @@ -327,26 +327,24 @@ export const workosAccountProvider: Layer.Layer< // ONCE here and shared instead of twice. // // Failure semantics differ per branch, so each branch is captured - // with `Effect.either` rather than left to fail the whole - // `Effect.all`: a `listOrgMembers` failure must still surface as an - // AccountError for the members list (unchanged from before); an - // Autumn or `listPendingInvitations` failure must only degrade - // seats to their safe defaults — never blank the page over a - // transient hiccup, exactly as `getMemberSeats`'s own `catchCause` - // fallback did before. The real cap gate lives in - // `reserveMemberSlot`, which fails closed. - const [membershipsResult, seatInputsResult] = yield* Effect.all( + // separately rather than left to fail the whole `Effect.all`: a + // `listOrgMembers` failure must still surface as an AccountError for + // the members list (unchanged from before); an Autumn or + // `listPendingInvitations` failure — error OR defect, exactly the + // `catchCause` fallback `getMemberSeats` had before — must only + // degrade seats to their safe defaults, never blank the page over a + // transient hiccup. The real cap gate lives in `reserveMemberSlot`, + // which fails closed. + const [membershipsResult, seatInputs] = yield* Effect.all( [ Effect.result(workos.listOrgMembers(org.id)), - Effect.result( - Effect.all( - [ - autumn.use((client) => client.customers.getOrCreate({ customerId: org.id })), - workos.listPendingInvitations(org.id), - ], - { concurrency: "unbounded" }, - ), - ), + Effect.all( + [ + autumn.use((client) => client.customers.getOrCreate({ customerId: org.id })), + workos.listPendingInvitations(org.id), + ], + { concurrency: "unbounded" }, + ).pipe(Effect.catchCause(() => Effect.succeed(null))), ], { concurrency: "unbounded" }, ); @@ -356,13 +354,10 @@ export const workosAccountProvider: Layer.Layer< } const memberships = membershipsResult.success; - const seats = Result.isSuccess(seatInputsResult) - ? seatsFrom( - memberships, - seatInputsResult.success[1].data.length, - seatInputsResult.success[0].subscriptions, - ) - : { used: 0, granted: 0, unlimited: false }; + const seats = + seatInputs === null + ? { used: 0, granted: 0, unlimited: false } + : seatsFrom(memberships, seatInputs[1].data.length, seatInputs[0].subscriptions); const members = yield* Effect.all( memberships.data.map((m: (typeof memberships.data)[number]) => diff --git a/apps/cloud/src/auth/jwks-cache.node.test.ts b/apps/cloud/src/auth/jwks-cache.node.test.ts index e2f5090709..e66c49b52f 100644 --- a/apps/cloud/src/auth/jwks-cache.node.test.ts +++ b/apps/cloud/src/auth/jwks-cache.node.test.ts @@ -268,6 +268,16 @@ describe("createCachedRemoteJWKSet", () => { expect(second.inspect().blockingFetchCount).toBe(0); }); + it("a cold isolate with no store entry and a dead upstream fails, and counts the wait", async () => { + const kp = await generateRotatableKeypair("k1"); + const store = makeStoreHarness(); + const cold = createCachedRemoteJWKSet(jwksUrl, { fetch: failingFetch, store }); + const token = await sign(kp); + await expect(jwtVerify(token, cold, { issuer, audience })).rejects.toThrow(); + expect(cold.inspect().blockingFetchCount).toBe(1); + expect(cold.inspect().storeHitCount).toBe(0); + }); + it("keeps serving the last good keys when the key server is down", async () => { const kp = await generateRotatableKeypair("k1"); const harness = makeFetchHarness([kp.publicJwk]); diff --git a/apps/cloud/src/auth/jwks-cache.ts b/apps/cloud/src/auth/jwks-cache.ts index d4a20a7fa1..499b918132 100644 --- a/apps/cloud/src/auth/jwks-cache.ts +++ b/apps/cloud/src/auth/jwks-cache.ts @@ -398,6 +398,10 @@ export const createCachedRemoteJWKSet = ( return upstream.entry; } if (stored) return adoptFromStore(stored); + // The caller waited on the upstream and it failed: that is a blocking + // fetch too, so `jwks.fetched_during_verify` stays honest on the failure + // path the attribute exists to explain. + blockingFetchCount += 1; // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: nothing usable anywhere, so the upstream failure is the real answer throw upstream.error; }; diff --git a/apps/cloud/src/db/db.close.test.ts b/apps/cloud/src/db/db.close.test.ts index 761448b6a1..2ff867c53f 100644 --- a/apps/cloud/src/db/db.close.test.ts +++ b/apps/cloud/src/db/db.close.test.ts @@ -21,7 +21,7 @@ import { describe, expect, it } from "@effect/vitest"; import { Effect, Exit } from "effect"; -import { POSTGRES_END_TIMEOUT_SECONDS, closePostgres } from "./db"; +import { POSTGRES_END_TIMEOUT_SECONDS, closePostgres, closePostgresAfter } from "./db"; describe("closePostgres", () => { it.effect("passes a non-zero drain window to sql.end (clean Terminate, not abandon)", () => @@ -107,3 +107,61 @@ describe("closePostgres", () => { }), ); }); + +describe("closePostgresAfter", () => { + it.effect("with nothing retained, closes in the scope like closePostgres", () => + Effect.gen(function* () { + let ended = 0; + const fakeSql = { end: () => Promise.resolve(void (ended += 1)) }; + const extended: Promise[] = []; + yield* closePostgresAfter(fakeSql, [], (work) => void extended.push(work)); + expect(ended).toBe(1); + expect(extended).toHaveLength(0); + }), + ); + + it.effect("with retained work, returns immediately and closes only after it settles", () => + Effect.gen(function* () { + // The bug this pins: a read detaches a stale catalog re-list, answers, + // and the request scope closes the socket. The re-list then persists + // into an ended pool. Retained work must keep the socket open past the + // scope finalizer, and the finalizer itself must not wait on it. + const order: string[] = []; + let release!: () => void; + const retained = new Promise((resolve) => { + release = () => { + order.push("work-settled"); + resolve(); + }; + }); + const fakeSql = { + end: () => { + order.push("end-called"); + return Promise.resolve(); + }, + }; + const extended: Promise[] = []; + yield* closePostgresAfter(fakeSql, [retained], (work) => void extended.push(work)); + order.push("finalizer-returned"); + expect(order).toEqual(["finalizer-returned"]); + expect(extended).toHaveLength(1); + + release(); + yield* Effect.promise(() => extended[0]!); + expect(order).toEqual(["finalizer-returned", "work-settled", "end-called"]); + }), + ); + + it.effect("closes even when the retained work fails", () => + Effect.gen(function* () { + let ended = 0; + const fakeSql = { end: () => Promise.resolve(void (ended += 1)) }; + const extended: Promise[] = []; + // oxlint-disable-next-line executor/no-promise-reject -- test fake: model a rejected background rebuild + const failed = Promise.reject("rebuild failed"); + yield* closePostgresAfter(fakeSql, [failed], (work) => void extended.push(work)); + yield* Effect.promise(() => extended[0]!); + expect(ended).toBe(1); + }), + ); +}); diff --git a/apps/cloud/src/db/db.ts b/apps/cloud/src/db/db.ts index ca764dd020..f0c635427e 100644 --- a/apps/cloud/src/db/db.ts +++ b/apps/cloud/src/db/db.ts @@ -14,7 +14,7 @@ // Migrations are run out-of-band (e.g. via a separate script or CI step), // not at request time — Cloudflare Workers cannot read the filesystem. -import { env } from "cloudflare:workers"; +import { env, waitUntil } from "cloudflare:workers"; import { Context, Effect, Layer } from "effect"; import { drizzle } from "drizzle-orm/postgres-js"; import type { PgDatabase } from "drizzle-orm/pg-core"; @@ -34,6 +34,16 @@ export type DrizzleDb = PgDatabase; export type DbServiceShape = { readonly sql?: Sql; readonly db: DrizzleDb; + /** + * Keep this request's driver open until `work` settles. The socket is + * request-scoped (see `closePostgres`), so background work an executor + * detaches from the request — a stale tool-catalog re-list that lands after + * the read answered — would otherwise persist into an already-ended pool + * and fail. Retained work defers the close (handed to the platform + * `waitUntil`, which keeps the invocation alive) instead of delaying the + * response; see `closePostgresAfter`. + */ + readonly keepAlive?: (work: Promise) => void; }; type DbResource = DbServiceShape & { @@ -108,12 +118,43 @@ export const closePostgres = (sql: Pick): Effect.Effect => }), ); -const makePostgresResource = (): DbResource => { +/** + * Close a postgres pool once every retained piece of background work has + * settled. With nothing retained this IS `closePostgres` (awaited, in the + * request scope). With retained work the close cannot be awaited there — that + * would hold the response until the background work finished, which is the + * exact cost the work was detached to avoid — so the teardown is handed to + * `extend` (the platform `waitUntil`) to run after the response, still inside + * this request's I/O context, and the scope finalizer returns immediately. + */ +export const closePostgresAfter = ( + sql: Pick, + retained: ReadonlyArray>, + extend: (work: Promise) => void, +): Effect.Effect => + retained.length === 0 + ? closePostgres(sql) + : Effect.sync(() => + extend( + Promise.allSettled(retained).then(() => + sql.end({ timeout: POSTGRES_END_TIMEOUT_SECONDS }).then( + () => undefined, + () => undefined, + ), + ), + ), + ); + +const makePostgresResource = (extend: (work: Promise) => void = waitUntil): DbResource => { const sql = makeSql(); + const retained: Promise[] = []; return { sql, db: drizzle(sql, { schema: combinedSchema }) as DrizzleDb, - close: () => closePostgres(sql), + keepAlive: (work) => { + retained.push(work); + }, + close: () => closePostgresAfter(sql, retained, extend), }; }; diff --git a/apps/cloud/src/db/fuma.ts b/apps/cloud/src/db/fuma.ts index 4fe70562f6..f823e92f4b 100644 --- a/apps/cloud/src/db/fuma.ts +++ b/apps/cloud/src/db/fuma.ts @@ -57,7 +57,7 @@ export const cloudDbProviderLayer = ( tables: FumaTables, ): Layer.Layer => Layer.effect(DbProvider)( - Effect.map(DbService.asEffect(), ({ db }): ExecutorDbHandle => { + Effect.map(DbService.asEffect(), ({ db, keepAlive }): ExecutorDbHandle => { const fuma = createDrizzleFumaDb({ db, tables, @@ -68,6 +68,10 @@ export const cloudDbProviderLayer = ( db: fuma.db, fuma: fuma.fuma, close: async () => {}, + // Background work the executor detaches from a read (stale catalog + // re-lists) must not outlive the request's socket: retaining it here + // defers `DbService`'s close past it. See `DbServiceShape.keepAlive`. + ...(keepAlive === undefined ? {} : { keepAlive }), // Plugin blobs (multi-MB resolved specs) live in R2, not Postgres. // Guarded because test workers / local dev may run without the // binding — the executor then falls back to the FumaDB `blob` table. diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index 5e10bbbfd1..8cfe689330 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -50,7 +50,7 @@ import { touchSubject, } from "@executor-js/sdk/host-internal"; -import { DbProvider } from "./executor-fuma-db"; +import { DbProvider, type ExecutorDbHandle } from "./executor-fuma-db"; // --------------------------------------------------------------------------- // HostConfig seam — the two host scalars that vary the `createExecutor` options. @@ -246,6 +246,26 @@ export class PluginsProvider extends Context.Service, + config: Pick, +): ((work: Promise) => void) | undefined => { + const keepAlive = handle.keepAlive; + const platform = config.waitUntil; + if (keepAlive === undefined) return platform; + return (work) => { + keepAlive(work); + platform?.(work); + }; +}; + export const makeScopedExecutor = < const TPlugins extends readonly AnyPlugin[] = readonly AnyPlugin[], >( @@ -264,9 +284,11 @@ export const makeScopedExecutor = < }, ): Effect.Effect, StorageFailure, DbProvider | PluginsProvider | HostConfig> => Effect.gen(function* () { - const { db, blobs } = yield* DbProvider.asEffect(); + const dbHandle = yield* DbProvider.asEffect(); + const { db, blobs } = dbHandle; const { plugins: pluginsFactory } = yield* PluginsProvider.asEffect(); const config = yield* HostConfig.asEffect(); + const waitUntil = composeWaitUntil(dbHandle, config); // Explicit config wins; otherwise fall back to the request origin if a host // provided one (HTTP middleware / MCP session DO). Stays `undefined` for // non-request callers — `coreTools.webBaseUrl` is optional and only the @@ -323,7 +345,7 @@ export const makeScopedExecutor = < fetch: hostedFetch, onIntegrationChange: config.onIntegrationChange, ...(config.toolsSyncTtlMs !== undefined ? { toolsSyncTtlMs: config.toolsSyncTtlMs } : {}), - ...(config.waitUntil !== undefined ? { waitUntil: config.waitUntil } : {}), + ...(waitUntil !== undefined ? { waitUntil } : {}), onElicitation: "accept-all", ...(options?.orgWrites === undefined ? {} : { orgWrites: options.orgWrites }), redirectUri, diff --git a/packages/core/sdk/src/executor-fuma-db.ts b/packages/core/sdk/src/executor-fuma-db.ts index 33a4e7e395..a005fbda92 100644 --- a/packages/core/sdk/src/executor-fuma-db.ts +++ b/packages/core/sdk/src/executor-fuma-db.ts @@ -114,4 +114,13 @@ export interface ExecutorDbHandle< * `blob` table over `db`. */ readonly blobs?: BlobStore; + /** + * Keep this handle's driver open until `work` settles. Hosts whose driver + * is REQUEST-scoped (cloud: one postgres socket per request) supply it so + * background work the executor detaches from a request — a stale tool + * catalog re-list that lands after the read answered — can still persist. + * `makeScopedExecutor` folds it into `ExecutorConfig.waitUntil`. Hosts with + * a long-lived driver omit it. + */ + readonly keepAlive?: (work: Promise) => void; } diff --git a/packages/react/src/lib/use-connection-health.test.ts b/packages/react/src/lib/use-connection-health.test.ts index 849229edd6..bc14878aa4 100644 --- a/packages/react/src/lib/use-connection-health.test.ts +++ b/packages/react/src/lib/use-connection-health.test.ts @@ -1,10 +1,19 @@ import { beforeEach, describe, expect, it } from "@effect/vitest"; -import type { HealthCheckResult } from "@executor-js/sdk/shared"; +import { + AuthTemplateSlug, + ConnectionAddress, + ConnectionName, + IntegrationSlug, + ProviderKey, + type Connection, + type HealthCheckResult, +} from "@executor-js/sdk/shared"; import { AUTO_PROBE_FLOOR_MS, HEALTH_REVALIDATE_MS, clearAutomaticProbeMemory, + probeMemoryKey, recordAutomaticProbe, resetAutomaticProbeMemoryForTest, revalidateQuery, @@ -16,6 +25,17 @@ const verdict = (status: HealthCheckResult["status"]): HealthCheckResult => ({ checkedAt: Date.now(), }); +const githubDefault: Connection = { + owner: "org", + name: ConnectionName.make("default"), + integration: IntegrationSlug.make("github"), + template: AuthTemplateSlug.make("default"), + provider: ProviderKey.make("default"), + address: ConnectionAddress.make("tools.github.org.default"), + identityLabel: null, + expiresAt: null, +}; + describe("revalidateQuery", () => { it("defers a healthy verdict to the server-enforced freshness window", () => { expect(revalidateQuery(verdict("healthy")).ifStaleMs, "the healthy window is sent").toBe( @@ -95,7 +115,7 @@ describe("shouldAutoProbe", () => { recordAutomaticProbe(key, { status: "healthy", checkedAt: now }); expect( - shouldAutoProbe(key, null, now + AUTO_PROBE_FLOOR_MS + 1), + shouldAutoProbe(key, { status: "healthy", checkedAt: now }, now + AUTO_PROBE_FLOOR_MS + 1), "a fresh healthy verdict must not probe just because the floor elapsed", ).toBe(false); }); @@ -111,6 +131,25 @@ describe("shouldAutoProbe", () => { ).toBe(true); }); + it("re-arms when the persisted verdict was cleared while unmounted, inside the floor", () => { + // An OAuth reconnect clears `last_health` server-side. If that landed + // while the row was unmounted, the remount's first sight sees `null` + // against a remembered pre-reconnect verdict: that IS the clearing + // transition, and it must fire the recovery probe despite the floor. + const key = "u|org|org:github:default"; + recordAutomaticProbe(key, verdict("expired")); + expect(shouldAutoProbe(key, null, Date.now() + 1_000)).toBe(true); + }); + + it("keys the memory by identity, so two orgs' same-named connections do not collide", () => { + const a = probeMemoryKey("user_1|org_a", githubDefault); + const b = probeMemoryKey("user_1|org_b", githubDefault); + expect(a).not.toBe(b); + recordAutomaticProbe(a, verdict("healthy")); + expect(shouldAutoProbe(a, verdict("healthy"))).toBe(false); + expect(shouldAutoProbe(b, verdict("expired"))).toBe(true); + }); + it("re-arms immediately once the entry is cleared, ignoring the floor", () => { const key = "acme:github:cleared"; const now = Date.now(); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index bfec920de1..0bf920229d 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -12,7 +12,9 @@ import * as Exit from "effect/Exit"; import type { Connection, HealthCheckResult, HealthStatus, Owner } from "@executor-js/sdk/shared"; import { checkConnectionHealth, connectionsOptimisticAtom } from "../api/atoms"; +import { useOrganizationId } from "../api/organization-context"; import { connectionCheckKeys } from "../api/reactivity-keys"; +import { useAuth, type AuthState } from "../multiplayer/auth-context"; /** Freshness window for automatic revalidation: a HEALTHY verdict younger * than this renders as-is; anything else (stale, missing, or non-healthy) @@ -26,6 +28,10 @@ const connectionParams = (connection: Connection) => ({ name: connection.name, }); +/** Identity of a connection within the current scope. `owner` is only + * `"org"`/`"user"`, so on its own two organizations' `org:github:default` + * rows are the same key — the module memory below must therefore be + * partitioned by the active identity as well; see `probeScope`. */ const probeKey = (connection: Connection): string => `${connection.owner}:${connection.integration}:${connection.name}`; @@ -98,6 +104,21 @@ const automaticProbeMemory = new Map< { readonly at: number; readonly result: HealthCheckResult } >(); +/** The identity partition of the module memory: the signed-in user and the + * active organization. A connection row's own key (`probeKey`) does not + * carry either — `owner` is only `"org"`/`"user"` — so without this an org + * switch in the same tab would read the previous org's remembered verdict + * for a same-named connection. Hosts without an org (local, desktop) and + * the loading/unauthenticated states fall into a single default partition, + * which on those hosts IS one identity. */ +const probeScope = (auth: AuthState, organizationId: string | null): string => + auth.status === "authenticated" ? `${auth.user.id}|${organizationId ?? ""}` : ""; + +/** Memory key for a connection under an identity partition. Exported so the + * decision logic is testable without rendering the hooks. */ +export const probeMemoryKey = (scope: string, connection: Connection): string => + `${scope}|${probeKey(connection)}`; + /** How long a remembered automatic probe blocks another automatic probe for * the same connection, regardless of verdict. A non-healthy verdict must * still eventually re-probe so recovery can show (see `revalidateQuery`), @@ -124,6 +145,15 @@ export function shouldAutoProbe( now: number = Date.now(), ): boolean { const remembered = automaticProbeMemory.get(key); + // A persisted verdict of `null` next to a remembered one means the grant + // was re-minted (an OAuth reconnect clears `last_health`) since that probe. + // The hooks catch this transition while mounted; this catches it when the + // reconnect landed while the row was UNMOUNTED — a remount inside the floor + // must still fire the recovery probe, not keep the pre-reconnect verdict. + if (persisted === null && remembered !== undefined) { + automaticProbeMemory.delete(key); + return true; + } const freshest = freshestVerdict(remembered?.result ?? null, persisted); if (healthyAndFresh(freshest, now)) return false; if (remembered !== undefined && now - remembered.at < AUTO_PROBE_FLOOR_MS) return false; @@ -193,8 +223,9 @@ export function useConnectionHealth(connection: Connection): { // key and can remount this row) would render the OLDER persisted verdict // until the background probe resolves, even though we already know the // last automatic probe's result. + const scope = probeScope(useAuth(), useOrganizationId()); const [liveProbe, setLiveProbe] = useState( - () => automaticProbeMemory.get(probeKey(connection))?.result ?? null, + () => automaticProbeMemory.get(probeMemoryKey(scope, connection))?.result ?? null, ); const doCheck = useAtomSet(checkConnectionHealth, { mode: "promiseExit" }); const invalidateConnections = useInvalidateConnections(); @@ -223,7 +254,7 @@ export function useConnectionHealth(connection: Connection): { const cleared = epoch === null && seenEpoch.current !== null && !firstSight; seenEpoch.current = epoch; if (!firstSight && !cleared) return; - const key = probeKey(connection); + const key = probeMemoryKey(scope, connection); if (cleared) clearAutomaticProbeMemory(key); if (!shouldAutoProbe(key, last)) return; void doCheck({ @@ -244,7 +275,7 @@ export function useConnectionHealth(connection: Connection): { invalidateConnections(connection.owner); } }); - }, [connection, doCheck, invalidateConnections]); + }, [connection, doCheck, invalidateConnections, scope]); const runCheck = useCallback(async () => { // Manual "Check now": invalidate the connections cache unconditionally so @@ -259,12 +290,12 @@ export function useConnectionHealth(connection: Connection): { reactivityKeys: connectionCheckKeys, }); if (Exit.isSuccess(exit)) { - recordAutomaticProbe(probeKey(connection), exit.value); + recordAutomaticProbe(probeMemoryKey(scope, connection), exit.value); seenEpoch.current = exit.value.checkedAt; setLiveProbe(exit.value); } return exit; - }, [connection, doCheck]); + }, [connection, doCheck, scope]); return { probe, status, runCheck }; } @@ -285,10 +316,11 @@ export function useConnectionsHealth( // a remount must show the last automatic probe's verdict, not fall back to // the older persisted one while a new probe is (or isn't, thanks to // `shouldAutoProbe`) in flight. + const scope = probeScope(useAuth(), useOrganizationId()); const [liveProbes, setLiveProbes] = useState>(() => { const seeded = new Map(); for (const connection of connections) { - const remembered = automaticProbeMemory.get(probeKey(connection)); + const remembered = automaticProbeMemory.get(probeMemoryKey(scope, connection)); if (remembered) seeded.set(probeKey(connection), remembered.result); } return seeded; @@ -315,8 +347,9 @@ export function useConnectionsHealth( if (!firstSight && previousEpoch === epoch) continue; const cleared = !firstSight && previousEpoch !== null && epoch === null; revalidated.current.set(key, epoch); - if (cleared) clearAutomaticProbeMemory(key); - if (!shouldAutoProbe(key, last)) continue; + const memoryKey = probeMemoryKey(scope, connection); + if (cleared) clearAutomaticProbeMemory(memoryKey); + if (!shouldAutoProbe(memoryKey, last)) continue; void doCheck({ params: connectionParams(connection), query: revalidateQuery(last), @@ -327,7 +360,7 @@ export function useConnectionsHealth( // invalidate the connections cache only when the verdict changed so an // unchanged reconfirm never churns the cache. if (!Exit.isSuccess(exit)) return; - recordAutomaticProbe(key, exit.value); + recordAutomaticProbe(memoryKey, exit.value); revalidated.current.set(key, exit.value.checkedAt); setLiveProbes((current) => new Map(current).set(key, exit.value)); if (exit.value.status !== (last?.status ?? "unknown")) { @@ -335,7 +368,7 @@ export function useConnectionsHealth( } }); } - }, [connections, doCheck, invalidateConnections]); + }, [connections, doCheck, invalidateConnections, scope]); return useCallback( (connection: Connection) => From 62aa82112b8fe819a674bdb0f2a123c600939a34 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:25:23 -0700 Subject: [PATCH 3/6] Drain late-retained work before closing the socket; keep unknown verdicts under the probe floor Co-Authored-By: Claude Fable 5.1 --- apps/cloud/src/db/db.close.test.ts | 38 +++++++++++++++++++ apps/cloud/src/db/db.ts | 19 +++++++++- .../src/lib/use-connection-health.test.ts | 12 ++++++ .../react/src/lib/use-connection-health.ts | 16 +++++--- 4 files changed, 78 insertions(+), 7 deletions(-) diff --git a/apps/cloud/src/db/db.close.test.ts b/apps/cloud/src/db/db.close.test.ts index 2ff867c53f..d4caf7468f 100644 --- a/apps/cloud/src/db/db.close.test.ts +++ b/apps/cloud/src/db/db.close.test.ts @@ -152,6 +152,44 @@ describe("closePostgresAfter", () => { }), ); + it.live("keeps waiting for work retained while earlier retained work was settling", () => + Effect.gen(function* () { + // A stale-catalog scan past its grace deadline registers each expired + // rebuild AFTER the scope finalizer already handed the close off. The + // close must not fire until those late registrations settle too. + const order: string[] = []; + const retained: Promise[] = []; + let releaseLate!: () => void; + const late = new Promise((resolve) => { + releaseLate = () => { + order.push("late-settled"); + resolve(); + }; + }); + const early = Promise.resolve().then(() => { + order.push("early-settled"); + retained.push(late); + }); + retained.push(early); + const fakeSql = { + end: () => { + order.push("end-called"); + return Promise.resolve(); + }, + }; + const extended: Promise[] = []; + yield* closePostgresAfter(fakeSql, retained, (work) => void extended.push(work)); + // Let the early work settle and register the late work. + yield* Effect.promise(() => early); + yield* Effect.sleep("10 millis"); + expect(order).toEqual(["early-settled"]); + + releaseLate(); + yield* Effect.promise(() => extended[0]!); + expect(order).toEqual(["early-settled", "late-settled", "end-called"]); + }), + ); + it.effect("closes even when the retained work fails", () => Effect.gen(function* () { let ended = 0; diff --git a/apps/cloud/src/db/db.ts b/apps/cloud/src/db/db.ts index f0c635427e..b72c08511e 100644 --- a/apps/cloud/src/db/db.ts +++ b/apps/cloud/src/db/db.ts @@ -136,7 +136,7 @@ export const closePostgresAfter = ( ? closePostgres(sql) : Effect.sync(() => extend( - Promise.allSettled(retained).then(() => + drainRetained(retained).then(() => sql.end({ timeout: POSTGRES_END_TIMEOUT_SECONDS }).then( () => undefined, () => undefined, @@ -145,6 +145,23 @@ export const closePostgresAfter = ( ), ); +/** + * Settle every retained promise, INCLUDING ones registered while earlier + * ones were still settling. Retained work can register more retained work — + * a stale-catalog scan that runs past its grace deadline registers each + * expired rebuild it discovers — so a single `Promise.allSettled` snapshot + * would let the socket close under a rebuild registered after the snapshot. + * Re-check the list after each round until it stops growing. + */ +const drainRetained = async (retained: ReadonlyArray>): Promise => { + let settled = 0; + while (settled < retained.length) { + const round = retained.slice(settled); + settled = retained.length; + await Promise.allSettled(round); + } +}; + const makePostgresResource = (extend: (work: Promise) => void = waitUntil): DbResource => { const sql = makeSql(); const retained: Promise[] = []; diff --git a/packages/react/src/lib/use-connection-health.test.ts b/packages/react/src/lib/use-connection-health.test.ts index bc14878aa4..acb1207ea1 100644 --- a/packages/react/src/lib/use-connection-health.test.ts +++ b/packages/react/src/lib/use-connection-health.test.ts @@ -141,6 +141,18 @@ describe("shouldAutoProbe", () => { expect(shouldAutoProbe(key, null, Date.now() + 1_000)).toBe(true); }); + it("does not treat a never-persisted unknown verdict as a reconnect clear", () => { + // A plugin with no health probe answers `unknown` and the server persists + // nothing, so `persisted` stays `null` for that connection forever. That + // is not a clearing transition: the floor must still apply, or every + // remount would re-probe. + const key = "u|org|org:noprobe:default"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "unknown", checkedAt: now }); + expect(shouldAutoProbe(key, null, now + 1_000)).toBe(false); + expect(shouldAutoProbe(key, null, now + AUTO_PROBE_FLOOR_MS + 1)).toBe(true); + }); + it("keys the memory by identity, so two orgs' same-named connections do not collide", () => { const a = probeMemoryKey("user_1|org_a", githubDefault); const b = probeMemoryKey("user_1|org_b", githubDefault); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 0bf920229d..887a820dd6 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -145,12 +145,16 @@ export function shouldAutoProbe( now: number = Date.now(), ): boolean { const remembered = automaticProbeMemory.get(key); - // A persisted verdict of `null` next to a remembered one means the grant - // was re-minted (an OAuth reconnect clears `last_health`) since that probe. - // The hooks catch this transition while mounted; this catches it when the - // reconnect landed while the row was UNMOUNTED — a remount inside the floor - // must still fire the recovery probe, not keep the pre-reconnect verdict. - if (persisted === null && remembered !== undefined) { + // A persisted verdict of `null` next to a remembered REAL verdict means + // the grant was re-minted (an OAuth reconnect clears `last_health`) since + // that probe. The hooks catch this transition while mounted; this catches + // it when the reconnect landed while the row was UNMOUNTED — a remount + // inside the floor must still fire the recovery probe, not keep the + // pre-reconnect verdict. `unknown` is excluded on purpose: the server never + // persists a no-capability probe, so for such a connection `null` next to a + // remembered `unknown` is the steady state, not a clearing — treating it as + // one would re-probe on every remount, the storm this memory exists to end. + if (persisted === null && remembered !== undefined && remembered.result.status !== "unknown") { automaticProbeMemory.delete(key); return true; } From 0fead03e717736b612901b4c1d6391bd2270f1e0 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:28:32 -0700 Subject: [PATCH 4/6] Decide reconnect clears by whether the server held a verdict, not by status Co-Authored-By: Claude Fable 5.1 --- .../src/lib/use-connection-health.test.ts | 15 ++++- .../react/src/lib/use-connection-health.ts | 63 ++++++++++++++----- 2 files changed, 60 insertions(+), 18 deletions(-) diff --git a/packages/react/src/lib/use-connection-health.test.ts b/packages/react/src/lib/use-connection-health.test.ts index acb1207ea1..0adc15137b 100644 --- a/packages/react/src/lib/use-connection-health.test.ts +++ b/packages/react/src/lib/use-connection-health.test.ts @@ -141,18 +141,29 @@ describe("shouldAutoProbe", () => { expect(shouldAutoProbe(key, null, Date.now() + 1_000)).toBe(true); }); - it("does not treat a never-persisted unknown verdict as a reconnect clear", () => { + it("does not treat a never-persisted verdict as a reconnect clear", () => { // A plugin with no health probe answers `unknown` and the server persists // nothing, so `persisted` stays `null` for that connection forever. That // is not a clearing transition: the floor must still apply, or every // remount would re-probe. const key = "u|org|org:noprobe:default"; const now = Date.now(); - recordAutomaticProbe(key, { status: "unknown", checkedAt: now }); + recordAutomaticProbe(key, { status: "unknown", checkedAt: now }, { persisted: false }); expect(shouldAutoProbe(key, null, now + 1_000)).toBe(false); expect(shouldAutoProbe(key, null, now + AUTO_PROBE_FLOOR_MS + 1)).toBe(true); }); + it("treats a cleared PERSISTED unknown verdict as a reconnect clear", () => { + // A plugin health check can legitimately answer `unknown`, and the server + // persists that. If a reconnect then clears it while the row is + // unmounted, the remount must probe: the status alone cannot tell the + // two `unknown`s apart, only whether the server held a verdict. + const key = "u|org|org:flaky:default"; + const now = Date.now(); + recordAutomaticProbe(key, { status: "unknown", checkedAt: now }, { persisted: true }); + expect(shouldAutoProbe(key, null, now + 1_000)).toBe(true); + }); + it("keys the memory by identity, so two orgs' same-named connections do not collide", () => { const a = probeMemoryKey("user_1|org_a", githubDefault); const b = probeMemoryKey("user_1|org_b", githubDefault); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 887a820dd6..61232c64b7 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -101,7 +101,18 @@ const freshestVerdict = ( * elapsed time and defeat the floor below. */ const automaticProbeMemory = new Map< string, - { readonly at: number; readonly result: HealthCheckResult } + { + readonly at: number; + readonly result: HealthCheckResult; + /** Whether the SERVER held a persisted verdict for this connection when + * the probe was recorded — the row's `lastHealth` after the probe, or + * the probe's own result when it was one the server persists. A later + * `null` from the server is then a clearing (an OAuth re-mint wiped a + * verdict that existed), not the steady state of a connection whose + * probes are never persisted (a plugin with no health check answers + * `unknown` and the server writes nothing). */ + readonly persisted: boolean; + } >(); /** The identity partition of the module memory: the signed-in user and the @@ -145,16 +156,16 @@ export function shouldAutoProbe( now: number = Date.now(), ): boolean { const remembered = automaticProbeMemory.get(key); - // A persisted verdict of `null` next to a remembered REAL verdict means - // the grant was re-minted (an OAuth reconnect clears `last_health`) since - // that probe. The hooks catch this transition while mounted; this catches - // it when the reconnect landed while the row was UNMOUNTED — a remount - // inside the floor must still fire the recovery probe, not keep the - // pre-reconnect verdict. `unknown` is excluded on purpose: the server never - // persists a no-capability probe, so for such a connection `null` next to a - // remembered `unknown` is the steady state, not a clearing — treating it as - // one would re-probe on every remount, the storm this memory exists to end. - if (persisted === null && remembered !== undefined && remembered.result.status !== "unknown") { + // A persisted verdict of `null` where the server HELD one when the probe + // was recorded means the grant was re-minted (an OAuth reconnect clears + // `last_health`) since that probe. The hooks catch this transition while + // mounted; this catches it when the reconnect landed while the row was + // UNMOUNTED — a remount inside the floor must still fire the recovery + // probe, not keep the pre-reconnect verdict. A connection the server never + // persisted a verdict for (`remembered.persisted === false`) is the steady + // state, not a clearing: it stays under the floor, or every remount would + // re-probe it — the storm this memory exists to end. + if (persisted === null && remembered !== undefined && remembered.persisted) { automaticProbeMemory.delete(key); return true; } @@ -170,10 +181,26 @@ export function shouldAutoProbe( * `checkedAt`. Exported (not test-only) so `shouldAutoProbe`'s decision logic * can be exercised directly, without rendering the hooks that normally call * it. */ -export function recordAutomaticProbe(key: string, result: HealthCheckResult): void { - automaticProbeMemory.set(key, { at: Date.now(), result }); +export function recordAutomaticProbe( + key: string, + result: HealthCheckResult, + options: { readonly persisted: boolean } = { persisted: true }, +): void { + automaticProbeMemory.set(key, { at: Date.now(), result, persisted: options.persisted }); } +/** Whether the server persists a probe with this status. Mirrors the SDK's + * `connectionCheckHealth`: every real probe verdict is written to the row, + * and the one that is not is the no-capability `unknown` (a plugin without a + * health check), which the server answers without writing. A plugin CAN + * legitimately probe to `unknown` and have it persisted, so the row's own + * `lastHealth` after the probe is the better signal when it is available; + * this is the fallback for a probe recorded before the row refetches. */ +const probePersisted = ( + result: HealthCheckResult, + persistedAfter: HealthCheckResult | null | undefined, +): boolean => (persistedAfter != null ? true : result.status !== "unknown"); + /** Deletes the remembered probe for `key`, forcing the next `shouldAutoProbe` * call to return `true` regardless of the floor. Called on the reconnect * ("cleared verdict") transition, which must always re-probe: an OAuth @@ -272,7 +299,7 @@ export function useConnectionHealth(connection: Connection): { // churns the cache (which would refetch connections, re-run this // effect, and, but for the epoch guard, risk a probe loop). if (!Exit.isSuccess(exit)) return; - recordAutomaticProbe(key, exit.value); + recordAutomaticProbe(key, exit.value, { persisted: probePersisted(exit.value, last) }); seenEpoch.current = exit.value.checkedAt; setLiveProbe(exit.value); if (exit.value.status !== (last?.status ?? "unknown")) { @@ -294,7 +321,9 @@ export function useConnectionHealth(connection: Connection): { reactivityKeys: connectionCheckKeys, }); if (Exit.isSuccess(exit)) { - recordAutomaticProbe(probeMemoryKey(scope, connection), exit.value); + recordAutomaticProbe(probeMemoryKey(scope, connection), exit.value, { + persisted: probePersisted(exit.value, connection.lastHealth), + }); seenEpoch.current = exit.value.checkedAt; setLiveProbe(exit.value); } @@ -364,7 +393,9 @@ export function useConnectionsHealth( // invalidate the connections cache only when the verdict changed so an // unchanged reconfirm never churns the cache. if (!Exit.isSuccess(exit)) return; - recordAutomaticProbe(memoryKey, exit.value); + recordAutomaticProbe(memoryKey, exit.value, { + persisted: probePersisted(exit.value, last), + }); revalidated.current.set(key, exit.value.checkedAt); setLiveProbes((current) => new Map(current).set(key, exit.value)); if (exit.value.status !== (last?.status ?? "unknown")) { From 0526dc863c2453cddae85f272317569a5219beac Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:31:36 -0700 Subject: [PATCH 5/6] Bound probe-floor suppression with an in-mount retry; test the persistence inference Co-Authored-By: Claude Fable 5.1 --- .../src/lib/use-connection-health.test.ts | 42 ++++++ .../react/src/lib/use-connection-health.ts | 124 ++++++++++++------ 2 files changed, 125 insertions(+), 41 deletions(-) diff --git a/packages/react/src/lib/use-connection-health.test.ts b/packages/react/src/lib/use-connection-health.test.ts index 0adc15137b..27db63aea2 100644 --- a/packages/react/src/lib/use-connection-health.test.ts +++ b/packages/react/src/lib/use-connection-health.test.ts @@ -13,7 +13,9 @@ import { AUTO_PROBE_FLOOR_MS, HEALTH_REVALIDATE_MS, clearAutomaticProbeMemory, + autoProbeRetryDelayMs, probeMemoryKey, + probePersisted, recordAutomaticProbe, resetAutomaticProbeMemoryForTest, revalidateQuery, @@ -187,3 +189,43 @@ describe("shouldAutoProbe", () => { ).toBe(true); }); }); + +describe("probePersisted", () => { + it("is true whenever the row already held a verdict before the probe", () => { + expect(probePersisted(verdict("unknown"), verdict("healthy"))).toBe(true); + expect(probePersisted(verdict("unknown"), verdict("expired"))).toBe(true); + }); + + it("on a row with no verdict, infers from the status: only unknown may be unpersisted", () => { + // The server's no-capability path is the only one that answers without + // writing, and it always answers `unknown`. A plugin probe that answers + // `unknown` on a first-ever check IS persisted, so this is a conservative + // guess for that case: the floor applies, and the retry timer bounds it. + expect(probePersisted(verdict("healthy"), null)).toBe(true); + expect(probePersisted(verdict("expired"), undefined)).toBe(true); + expect(probePersisted(verdict("unknown"), null)).toBe(false); + }); +}); + +describe("autoProbeRetryDelayMs", () => { + beforeEach(() => { + resetAutomaticProbeMemoryForTest(); + }); + + it("is null with no memory and null once the floor has elapsed", () => { + const key = "u|org|org:github:retry"; + const now = Date.now(); + expect(autoProbeRetryDelayMs(key, now)).toBeNull(); + recordAutomaticProbe(key, verdict("expired")); + expect(autoProbeRetryDelayMs(key, now + AUTO_PROBE_FLOOR_MS + 1)).toBeNull(); + }); + + it("is the time left on the floor while it is suppressing", () => { + const key = "u|org|org:github:retry"; + recordAutomaticProbe(key, verdict("expired")); + const delay = autoProbeRetryDelayMs(key, Date.now() + 10_000); + expect(delay).not.toBeNull(); + expect(delay!).toBeGreaterThan(AUTO_PROBE_FLOOR_MS - 10_000 - 50); + expect(delay!).toBeLessThanOrEqual(AUTO_PROBE_FLOOR_MS - 10_000); + }); +}); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 61232c64b7..18c212e9a3 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -196,10 +196,22 @@ export function recordAutomaticProbe( * legitimately probe to `unknown` and have it persisted, so the row's own * `lastHealth` after the probe is the better signal when it is available; * this is the fallback for a probe recorded before the row refetches. */ -const probePersisted = ( +export const probePersisted = ( result: HealthCheckResult, - persistedAfter: HealthCheckResult | null | undefined, -): boolean => (persistedAfter != null ? true : result.status !== "unknown"); + persistedBefore: HealthCheckResult | null | undefined, +): boolean => (persistedBefore != null ? true : result.status !== "unknown"); + +/** How long until the floor stops suppressing an automatic probe for `key`, + * or `null` when nothing is suppressing it. The hooks arm a one-shot timer + * for this so a suppressed probe still fires once the floor elapses WHILE + * THE ROW STAYS MOUNTED — without it, a verdict cleared by a reconnect that + * landed inside the floor would wait for the next remount to recover. */ +export function autoProbeRetryDelayMs(key: string, now: number = Date.now()): number | null { + const remembered = automaticProbeMemory.get(key); + if (remembered === undefined) return null; + const remaining = AUTO_PROBE_FLOOR_MS - (now - remembered.at); + return remaining > 0 ? remaining : null; +} /** Deletes the remembered probe for `key`, forcing the next `shouldAutoProbe` * call to return `true` regardless of the floor. Called on the reconnect @@ -287,25 +299,39 @@ export function useConnectionHealth(connection: Connection): { if (!firstSight && !cleared) return; const key = probeMemoryKey(scope, connection); if (cleared) clearAutomaticProbeMemory(key); - if (!shouldAutoProbe(key, last)) return; - void doCheck({ - params: connectionParams(connection), - query: revalidateQuery(last), - }).then((exit) => { - // Background refresh: update the dot on success, stay quiet on failure - // (the persisted verdict is still the best known state). Invalidate the - // connections cache ONLY when the verdict actually changed: on the - // common no-change reconfirm we skip it, so an automatic probe never - // churns the cache (which would refetch connections, re-run this - // effect, and, but for the epoch guard, risk a probe loop). - if (!Exit.isSuccess(exit)) return; - recordAutomaticProbe(key, exit.value, { persisted: probePersisted(exit.value, last) }); - seenEpoch.current = exit.value.checkedAt; - setLiveProbe(exit.value); - if (exit.value.status !== (last?.status ?? "unknown")) { - invalidateConnections(connection.owner); - } - }); + const probe = () => + void doCheck({ + params: connectionParams(connection), + query: revalidateQuery(last), + }).then((exit) => { + // Background refresh: update the dot on success, stay quiet on failure + // (the persisted verdict is still the best known state). Invalidate the + // connections cache ONLY when the verdict actually changed: on the + // common no-change reconfirm we skip it, so an automatic probe never + // churns the cache (which would refetch connections, re-run this + // effect, and, but for the epoch guard, risk a probe loop). + if (!Exit.isSuccess(exit)) return; + recordAutomaticProbe(key, exit.value, { persisted: probePersisted(exit.value, last) }); + seenEpoch.current = exit.value.checkedAt; + setLiveProbe(exit.value); + if (exit.value.status !== (last?.status ?? "unknown")) { + invalidateConnections(connection.owner); + } + }); + if (shouldAutoProbe(key, last)) { + probe(); + return; + } + // Suppressed by the floor: re-decide once it elapses, while still + // mounted, so the suppression is bounded by the floor and never by the + // next remount. A healthy-and-fresh verdict has no timer to arm (the + // freshness window, not the floor, is what suppressed it). + const delay = autoProbeRetryDelayMs(key); + if (delay === null) return; + const timer = setTimeout(() => { + if (shouldAutoProbe(key, last)) probe(); + }, delay); + return () => clearTimeout(timer); }, [connection, doCheck, invalidateConnections, scope]); const runCheck = useCallback(async () => { @@ -371,6 +397,7 @@ export function useConnectionsHealth( // the module-scope `automaticProbeMemory`, which survives that remount. const revalidated = useRef(new Map()); useEffect(() => { + const timers: ReturnType[] = []; for (const connection of connections) { const key = probeKey(connection); const last = connection.lastHealth; @@ -382,27 +409,42 @@ export function useConnectionsHealth( revalidated.current.set(key, epoch); const memoryKey = probeMemoryKey(scope, connection); if (cleared) clearAutomaticProbeMemory(memoryKey); - if (!shouldAutoProbe(memoryKey, last)) continue; - void doCheck({ - params: connectionParams(connection), - query: revalidateQuery(last), - }).then((exit) => { - // Same automatic-path rule as the single-connection hook: reflect the - // verdict, adopt its epoch so the refetch doesn't re-probe, record it - // into the module memory so a remount respects the floor, and - // invalidate the connections cache only when the verdict changed so an - // unchanged reconfirm never churns the cache. - if (!Exit.isSuccess(exit)) return; - recordAutomaticProbe(memoryKey, exit.value, { - persisted: probePersisted(exit.value, last), + const probe = () => + void doCheck({ + params: connectionParams(connection), + query: revalidateQuery(last), + }).then((exit) => { + // Same automatic-path rule as the single-connection hook: reflect the + // verdict, adopt its epoch so the refetch doesn't re-probe, record it + // into the module memory so a remount respects the floor, and + // invalidate the connections cache only when the verdict changed so an + // unchanged reconfirm never churns the cache. + if (!Exit.isSuccess(exit)) return; + recordAutomaticProbe(memoryKey, exit.value, { + persisted: probePersisted(exit.value, last), + }); + revalidated.current.set(key, exit.value.checkedAt); + setLiveProbes((current) => new Map(current).set(key, exit.value)); + if (exit.value.status !== (last?.status ?? "unknown")) { + invalidateConnections(connection.owner); + } }); - revalidated.current.set(key, exit.value.checkedAt); - setLiveProbes((current) => new Map(current).set(key, exit.value)); - if (exit.value.status !== (last?.status ?? "unknown")) { - invalidateConnections(connection.owner); - } - }); + if (shouldAutoProbe(memoryKey, last)) { + probe(); + continue; + } + // Same bounded suppression as the single-connection hook. + const delay = autoProbeRetryDelayMs(memoryKey); + if (delay === null) continue; + timers.push( + setTimeout(() => { + if (shouldAutoProbe(memoryKey, last)) probe(); + }, delay), + ); } + return () => { + for (const timer of timers) clearTimeout(timer); + }; }, [connections, doCheck, invalidateConnections, scope]); return useCallback( From 0265b4fa64c76470d43edecd2bc5b8d8a03df9f8 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:34:37 -0700 Subject: [PATCH 6/6] Re-arm the probe retry when an effect rerun cancels it Co-Authored-By: Claude Fable 5.1 --- packages/react/src/lib/use-connection-health.ts | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 18c212e9a3..077b00319b 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -328,7 +328,14 @@ export function useConnectionHealth(connection: Connection): { // freshness window, not the floor, is what suppressed it). const delay = autoProbeRetryDelayMs(key); if (delay === null) return; + // The decision is NOT final while the timer is pending: roll the epoch + // back to "unseen" so an effect rerun (a new `connection` identity from a + // list refetch, a scope change) that cancels this timer re-enters the + // decision above and re-arms it, instead of treating the epoch as already + // handled and leaving the row suppressed until the next remount. + seenEpoch.current = undefined; const timer = setTimeout(() => { + seenEpoch.current = epoch; if (shouldAutoProbe(key, last)) probe(); }, delay); return () => clearTimeout(timer); @@ -433,11 +440,16 @@ export function useConnectionsHealth( probe(); continue; } - // Same bounded suppression as the single-connection hook. + // Same bounded suppression as the single-connection hook, and the same + // rule for a pending timer: the row is NOT yet considered for this + // epoch, so a rerun (the list re-merging as owners' rows arrive) that + // clears the timers re-enters the decision and re-arms it. const delay = autoProbeRetryDelayMs(memoryKey); if (delay === null) continue; + revalidated.current.delete(key); timers.push( setTimeout(() => { + revalidated.current.set(key, epoch); if (shouldAutoProbe(memoryKey, last)) probe(); }, delay), );