diff --git a/apps/cloud/src/server.ts b/apps/cloud/src/server.ts index fc9c146f3..71905e352 100644 --- a/apps/cloud/src/server.ts +++ b/apps/cloud/src/server.ts @@ -243,19 +243,64 @@ const markStartGraphEntered = (): void => { // `servedByAppPlane` (./app-paths) decides which paths qualify โ€” two under // `/api` are claimed by Start's middleware first and must keep their old route. -// Instantiated on the first request that needs it and memoized per isolate, -// mirroring `start.ts`'s `getApp`. The import stays dynamic so an isolate that -// only serves pages or proxies never evaluates the app graph at all. -let appPlane: ReturnType | undefined; +// Instantiated once per isolate and memoized as a promise, mirroring +// `start.ts`'s `getApp`. The import stays dynamic so the Worker's static +// startup closure does not include the app graph; the promise memo means a +// pre-warm and a real request racing on a fresh isolate share one import. +type AppPlane = ReturnType; +let appPlanePromise: Promise | undefined; let appGraphEntered = false; -const getAppPlane = async (): Promise> => { - if (appPlane === undefined) { - const { cloudApiHandler } = await import("./app"); - appPlane = cloudApiHandler(); - appGraphEntered = true; +const getAppPlane = (): Promise => { + if (appPlanePromise === undefined) { + appPlanePromise = import("./app").then( + ({ cloudApiHandler }) => { + const plane = cloudApiHandler(); + appGraphEntered = true; + return plane; + }, + (cause: unknown) => { + // Do not memoize a failure: the next request re-imports, as the + // un-memoized version did, instead of failing every request after. + appPlanePromise = undefined; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: re-raise the import failure to the awaiting request + throw cause; + }, + ); } - return appPlane; + return appPlanePromise; +}; + +// --------------------------------------------------------------------------- +// Pre-warming the app plane. +// --------------------------------------------------------------------------- +// +// Measured on production 2026-09-18: 30% of `/api/*` dispatches landed on an +// isolate that had not yet evaluated the app graph, and paid ~2s (p50) for it +// against ~100ms warm. Only 43% of those isolates were under 5s old. The rest +// had been alive for seconds to minutes serving `/mcp`, discovery documents, +// or proxies, none of which enter the app graph, so the dashboard's first +// call was the one that paid. Isolates live about a minute at the median, so +// there is rarely a second dashboard request to benefit. +// +// So any request that does NOT need the app plane starts its import in the +// background. The request itself returns as before; the import runs under +// `waitUntil`, so the isolate stays up until it finishes. A dashboard call +// arriving afterwards finds the graph evaluated. The truly fresh isolate +// (first request IS a dashboard call) still pays; that cost is the graph's +// evaluation itself, addressed separately. +// +// Failure is swallowed on purpose: a pre-warm that fails must not fail the +// request that triggered it, and the next real app-plane request re-imports +// through the same memo and surfaces the error where it belongs. +const prewarmAppPlane = (ctx: ExecutionContext): void => { + if (appGraphEntered) return; + ctx.waitUntil( + getAppPlane().then( + () => undefined, + () => undefined, + ), + ); }; const cloudflareHandler: ExportedHandler = { @@ -266,6 +311,12 @@ const cloudflareHandler: ExportedHandler = { // import loads the entire React + Effect server graph and can take seconds // on a cold isolate. Classify and service-bind marketing at the Worker // entry, before telemetry or fetchHandler touches that graph. + // Everything that returns before the app-plane dispatch below leaves the + // graph unevaluated for the next request; warm it in the background. + if (!servedByAppPlane(new URL(request.url).pathname, request.method)) { + prewarmAppPlane(ctx); + } + const marketingRequest = marketingProxyRequest(request); const marketing: Fetcher | undefined = env.MARKETING; if (marketingRequest && marketing) return marketing.fetch(marketingRequest); @@ -445,6 +496,9 @@ const cloudflareHandler: ExportedHandler = { // isolate goes idle. scheduled: async (_controller, _env, ctx) => { installTracerProvider(); + // The cron fires every minute, often on an isolate that has served no + // dashboard request yet: the cheapest pre-warm there is. + prewarmAppPlane(ctx); await runWorkOsEventsSync(); ctx.waitUntil(flushTracerProvider()); }, diff --git a/e2e/scenarios/health-probe-churn.test.ts b/e2e/scenarios/health-probe-churn.test.ts new file mode 100644 index 000000000..53cd64ee8 --- /dev/null +++ b/e2e/scenarios/health-probe-churn.test.ts @@ -0,0 +1,297 @@ +// Cross-target (browser): the health-probe churn loop seen in production. +// +// Production symptom (2026-09-18): one signed-in browser with many +// non-healthy or never-checked connections spread across many integrations +// sent hundreds of `/api/connections/.../health` POSTs and `/api/connections` +// GETs per minute for as long as the dashboard stayed open. Most were +// interrupted client-side before completing (the shared health mutation atom +// cancels the previous in-flight call), but hundreds per minute still reached +// the server and the upstreams. The server annotated every verdict as +// unchanged, so the loop lives in the client. +// +// The existing verdict scenario pins "one broken connection probes exactly +// once per surface". This one reproduces the production SHAPE: many +// connections, several integrations, a mix of verdicts (`unknown` from a +// connection with no probe configured, `degraded` from an upstream that +// answers 401), and the page journey a user actually makes (integrations +// list โ†’ integration page โ†’ integrations list). It then watches a quiet +// window. A settled page must stop asking. A count that keeps climbing is the +// loop. +// +// Skips on targets with no browser surface. +import { randomBytes } from "node:crypto"; +import { createServer } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import type { HttpApiClient } from "effect/unstable/httpapi"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { AuthTemplateSlug, ConnectionName, IntegrationSlug } from "@executor-js/sdk/shared"; + +import { scenario } from "../src/scenario"; +import { Api, Browser, Target } from "../src/services"; +import { visit } from "../src/surfaces/browser"; + +const api = composePluginApi([openApiHttpPlugin()] as const); +type Client = HttpApiClient.ForApi; + +const TEMPLATE = AuthTemplateSlug.make("apiKey"); + +/** Integrations whose connection has a probe configured and an upstream that + * rejects the key: the probe persists `degraded`. */ +const DEGRADED_COUNT = 4; +/** Integrations whose connection has NO probe configured: every automatic + * check answers `unknown` with a fresh `checkedAt`. */ +const UNKNOWN_COUNT = 4; + +/** How long a settled page is watched for further probes. Production cycled + * every 0.5-1s, so a loop shows up well inside this. */ +const QUIET_WINDOW_MS = 10_000; + +const unique = (prefix: string) => `${prefix}${randomBytes(3).toString("hex")}`; + +/** Upstream on 127.0.0.1 whose `GET /me` rejects every key. */ +const serveRejectingUpstream = () => + Effect.acquireRelease( + Effect.callback<{ readonly url: string; readonly close: () => void }>((resume) => { + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith("/me")) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "invalid_token" })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const identitySpec = (baseUrl: string, title: string): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title, version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/me": { + get: { + operationId: "getMe", + summary: "The current account", + responses: { + "200": { + description: "The authenticated account", + content: { + "application/json": { + schema: { type: "object", properties: { email: { type: "string" } } }, + }, + }, + }, + }, + }, + }, + }, + }); + +/** One OpenAPI integration with one saved org connection. With `probe`, the + * identity GET is configured as the health check. */ +const seedIntegration = ( + client: Client, + upstreamUrl: string, + options: { readonly prefix: string; readonly probe: boolean }, +) => + Effect.gen(function* () { + const slug = IntegrationSlug.make(unique(options.prefix)); + const name = ConnectionName.make(`${options.prefix}conn`); + + yield* Effect.addFinalizer(() => + Effect.all( + [ + client.connections + .remove({ params: { owner: "org", integration: slug, name } }) + .pipe(Effect.ignore), + client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore), + ], + { discard: true }, + ), + ); + + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: identitySpec(upstreamUrl, `Churn ${slug}`) }, + slug, + baseUrl: upstreamUrl, + authenticationTemplate: [ + { + slug: "apiKey", + type: "apiKey", + headers: { authorization: ["Bearer ", { type: "variable", name: "token" }] }, + }, + ], + }, + }); + + if (options.probe) { + const candidates = yield* client.integrations.healthCheckCandidates({ params: { slug } }); + const getMe = candidates.find((candidate) => candidate.method === "get"); + if (!getMe) return yield* Effect.die("identity spec exposed no GET candidate"); + yield* client.integrations.healthCheckSet({ + params: { slug }, + payload: { spec: { operation: getMe.operation, identityField: "email" } }, + }); + } + + yield* client.connections.create({ + payload: { owner: "org", name, integration: slug, template: TEMPLATE, value: "bad-key" }, + }); + + return { slug, name }; + }); + +scenario( + "Health checks (UI) ยท many broken connections across many integrations settle instead of looping", + {}, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const browser = yield* Browser; + const { client: makeClient } = yield* Api; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveRejectingUpstream(); + + const degraded = yield* Effect.all( + Array.from({ length: DEGRADED_COUNT }, () => + seedIntegration(client, upstream.url, { prefix: "churndeg", probe: true }), + ), + { concurrency: 2 }, + ); + const unknown = yield* Effect.all( + Array.from({ length: UNKNOWN_COUNT }, () => + seedIntegration(client, upstream.url, { prefix: "churnunk", probe: false }), + ), + { concurrency: 2 }, + ); + const seeded = [...degraded, ...unknown]; + const connectionCount = seeded.length; + + // Persist a verdict on every connection BEFORE the browser opens, the + // way a returning user finds them: `degraded` on the probed ones, + // `unknown` on the rest. The list then renders each row with a + // persisted verdict, and every automatic revalidation compares its + // result against one. + for (const { slug, name } of seeded) { + yield* client.connections.checkHealth({ + params: { owner: "org", integration: slug, name }, + query: {}, + }); + } + + yield* browser.session(identity, async ({ page, step }) => { + // The client's wire contract, observed from outside: every health POST + // and every connections-list GET the app sends, plus how many the + // browser reports as failed (an aborted request shows up here as a + // failure, which is how the interrupted mutation atom looks on the wire). + const health: string[] = []; + const lists: string[] = []; + let aborted = 0; + // Only THIS scenario's connections count. Targets that share one org + // across scenarios (selfhost) can carry rows another scenario left + // behind, and those revalidate too; they are not ours to bound. + const seededSlugs = new Set(seeded.map(({ slug }) => String(slug))); + const isHealth = (method: string, url: string) => + method === "POST" && + url.includes("/health") && + [...seededSlugs].some((slug) => url.includes(`/api/connections/org/${slug}/`)); + const isList = (method: string, url: string) => + method === "GET" && /\/api\/connections(\?|$)/.test(url); + page.on("request", (request) => { + const url = request.url(); + if (isHealth(request.method(), url)) health.push(url); + else if (isList(request.method(), url)) lists.push(url); + }); + page.on("requestfailed", (request) => { + const url = request.url(); + if (isHealth(request.method(), url) || isList(request.method(), url)) aborted += 1; + }); + + const snapshot = (label: string) => { + const line = `[churn] ${label}: health=${String(health.length)} lists=${String(lists.length)} aborted=${String(aborted)}`; + console.log(line); + return { health: health.length, lists: lists.length, aborted }; + }; + + await step("Open the integrations list with every broken connection on it", async () => { + await visit(page, "/"); + for (const { slug } of seeded) { + await page + .getByRole("link", { name: new RegExp(slug, "i") }) + .first() + .waitFor({ + timeout: 30_000, + }); + } + // Let the automatic revalidation land for every row. + await page.waitForTimeout(3_000); + }); + const afterList = snapshot("after integrations list"); + + await step("Open one integration page, then return to the list", async () => { + await visit(page, `/integrations/${degraded[0]!.slug}`); + await page.waitForTimeout(2_000); + await visit(page, "/"); + await page.waitForTimeout(3_000); + }); + const afterJourney = snapshot("after journey"); + + // The quiet window: nothing changed on the page, nothing changed on + // the server, so nothing more should be asked. + await step("Leave the settled list open and watch the wire", async () => { + await page.waitForTimeout(QUIET_WINDOW_MS); + }); + const afterQuiet = snapshot("after quiet window"); + + const quietHealth = afterQuiet.health - afterJourney.health; + const quietLists = afterQuiet.lists - afterJourney.lists; + console.log( + `[churn] quiet window (${String(QUIET_WINDOW_MS)}ms): +${String(quietHealth)} health, +${String(quietLists)} list fetches over ${String(connectionCount)} connections`, + ); + + // One list mount revalidates each non-healthy connection once, and + // reads the connections list once per owner (plus the unscoped read). + // A probe count above the connection count means rows are being + // re-probed; a list count above three means verdicts are being + // treated as changes and refetching the list. + expect( + afterList.health, + `the first list mount probes each broken connection at most once (sent ${String(afterList.health)} for ${String(connectionCount)} connections)`, + ).toBeLessThanOrEqual(connectionCount); + expect( + afterList.lists, + `the first list mount reads the connections list at most once per owner (sent ${String(afterList.lists)})`, + ).toBeLessThanOrEqual(3); + + // THE production symptom: a settled page keeps asking. Zero is the + // contract; anything else is the loop. + expect(quietHealth, "a settled integrations list sends no further health probes").toBe(0); + expect( + quietLists, + "a settled integrations list does not refetch the connections list", + ).toBe(0); + }); + }), + ), +); diff --git a/packages/react/src/api/atoms.tsx b/packages/react/src/api/atoms.tsx index 0d6fb9af7..366cd5299 100644 --- a/packages/react/src/api/atoms.tsx +++ b/packages/react/src/api/atoms.tsx @@ -17,6 +17,7 @@ import { } from "@executor-js/sdk/shared"; import * as Atom from "effect/unstable/reactivity/Atom"; import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"; +import * as Reactivity from "effect/unstable/reactivity/Reactivity"; import * as Effect from "effect/Effect"; import { ExecutorApiClient } from "./client"; @@ -198,6 +199,37 @@ export const refreshConnection = ExecutorApiClient.mutation("connections", "refr * cache on every load). */ export const checkConnectionHealth = ExecutorApiClient.mutation("connections", "checkHealth"); +export interface CheckConnectionHealthArgs { + readonly params: { + readonly owner: Owner; + readonly integration: IntegrationSlug; + readonly name: ConnectionName; + }; + readonly query: { readonly ifStaleMs?: number }; + readonly reactivityKeys?: ReadonlyArray; +} + +/** The AUTOMATIC health probe, one atom PER CONNECTION. + * + * `checkConnectionHealth` above is one shared mutation atom. Awaiting it + * (`useAtomSet(..., { mode: "promiseExit" })`) resolves with the atom's next + * settled result, whichever call produced it, and a new call interrupts the + * one in flight. A surface that probes every row of a list in one pass + * therefore cancels all but the last probe and hands every row the LAST + * row's verdict. Each row then reads a foreign verdict as a change to its own + * connection, refreshes the connections cache, and re-probes: the probe storm + * the automatic path was built to avoid. Keying the atom by connection address + * gives every probe its own fiber and its own result. */ +export const checkConnectionHealthFor = Atom.family((address: ConnectionAddress) => + ExecutorApiClient.runtime.fn()((args) => { + const probe = Effect.gen(function* () { + const client = yield* ExecutorApiClient; + return yield* client.connections.checkHealth({ params: args.params, query: args.query }); + }).pipe(Effect.withSpan("connection.health.probe", { attributes: { address } })); + return args.reactivityKeys ? Reactivity.mutation(probe, args.reactivityKeys) : probe; + }), +); + /** Validate an IN-FLIGHT credential without saving it (the key-first connect * flow). Returns the probe result the UI derives a connection name from. */ export const validateConnection = ExecutorApiClient.mutation("connections", "validate"); diff --git a/packages/react/src/lib/use-connection-health.ts b/packages/react/src/lib/use-connection-health.ts index 9bfc1fe36..52100b5fe 100644 --- a/packages/react/src/lib/use-connection-health.ts +++ b/packages/react/src/lib/use-connection-health.ts @@ -7,11 +7,17 @@ // drift apart. import { useCallback, useContext, useEffect, useRef, useState } from "react"; -import { RegistryContext, useAtomSet } from "@effect/atom-react"; +import { RegistryContext } from "@effect/atom-react"; +import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as AtomRegistry from "effect/unstable/reactivity/AtomRegistry"; import type { Connection, HealthCheckResult, HealthStatus, Owner } from "@executor-js/sdk/shared"; -import { checkConnectionHealth, connectionsOptimisticAtom } from "../api/atoms"; +import { + checkConnectionHealthFor, + connectionsOptimisticAtom, + type CheckConnectionHealthArgs, +} from "../api/atoms"; import { connectionCheckKeys } from "../api/reactivity-keys"; /** Freshness window for automatic revalidation: a HEALTHY verdict younger @@ -97,6 +103,32 @@ function useInvalidateConnections(): (owner: Owner) => void { ); } +/** + * Run one probe against its OWN per-connection atom and await that atom's + * result. The shared `checkConnectionHealth` mutation cannot be awaited from a + * loop: every `set` interrupts the previous call and every waiter resolves + * with whichever call settled last, so a list of N rows would hand N-1 rows a + * verdict for a connection that is not theirs (see `checkConnectionHealthFor`). + * This is the same set-then-await that `useAtomSet` performs in promise mode, + * addressed at the connection's atom, and usable from a loop. + */ +function useProbeConnection(): ( + connection: Connection, + args: CheckConnectionHealthArgs, +) => Promise> { + const registry = useContext(RegistryContext); + return useCallback( + (connection: Connection, args: CheckConnectionHealthArgs) => { + const atom = checkConnectionHealthFor(connection.address); + registry.set(atom, args); + return Effect.runPromiseExit( + AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }), + ); + }, + [registry], + ); +} + /** * Health for ONE connection, stale-while-revalidate. The persisted verdict * renders instantly; a background probe on mount corrects it in place (once @@ -112,7 +144,7 @@ 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); - const doCheck = useAtomSet(checkConnectionHealth, { mode: "promiseExit" }); + const doCheck = useProbeConnection(); const invalidateConnections = useInvalidateConnections(); const probe = freshestVerdict(liveProbe, connection.lastHealth); @@ -137,7 +169,7 @@ export function useConnectionHealth(connection: Connection): { seenEpoch.current = epoch; if (!firstSight && !cleared) return; if (healthyAndFresh(last)) return; - void doCheck({ + void doCheck(connection, { params: connectionParams(connection), query: revalidateQuery(last), }).then((exit) => { @@ -160,7 +192,7 @@ export function useConnectionHealth(connection: Connection): { // 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. - const exit = await doCheck({ + const exit = await doCheck(connection, { params: connectionParams(connection), query: {}, reactivityKeys: connectionCheckKeys, @@ -190,7 +222,7 @@ export function useConnectionsHealth( connections: readonly Connection[], ): (connection: Connection) => HealthCheckResult | null { const [liveProbes, setLiveProbes] = useState>(new Map()); - const doCheck = useAtomSet(checkConnectionHealth, { mode: "promiseExit" }); + const doCheck = useProbeConnection(); const invalidateConnections = useInvalidateConnections(); // Once per VERDICT per connection (same epoch guard as the single-connection @@ -206,7 +238,7 @@ export function useConnectionsHealth( if (revalidated.current.has(key) && revalidated.current.get(key) === epoch) continue; revalidated.current.set(key, epoch); if (healthyAndFresh(last)) continue; - void doCheck({ + void doCheck(connection, { params: connectionParams(connection), query: revalidateQuery(last), }).then((exit) => {