History

Probe connection health through a per-connection atom (#2071)

* e2e: reproduce health-probe churn on the integrations list Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * Probe connection health through a per-connection atom Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * e2e: bound the churn scenario to its own connections Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> ---------

Rhys Sullivan committed 1 month agomain

Commit 09c19e41 parent 1aa6fd93 files · +368 −7Browse files

Showing the changes to e2e (1 of 3 files). Show all

Unified · Split

e2e/scenarios/health-probe-churn.test.ts+297−0
@@ -0,0 +1,297 @@
1// Cross-target (browser): the health-probe churn loop seen in production.
2//
3// Production symptom (2026-09-18): one signed-in browser with many
4// non-healthy or never-checked connections spread across many integrations
5// sent hundreds of `/api/connections/.../health` POSTs and `/api/connections`
6// GETs per minute for as long as the dashboard stayed open. Most were
7// interrupted client-side before completing (the shared health mutation atom
8// cancels the previous in-flight call), but hundreds per minute still reached
9// the server and the upstreams. The server annotated every verdict as
10// unchanged, so the loop lives in the client.
11//
12// The existing verdict scenario pins "one broken connection probes exactly
13// once per surface". This one reproduces the production SHAPE: many
14// connections, several integrations, a mix of verdicts (`unknown` from a
15// connection with no probe configured, `degraded` from an upstream that
16// answers 401), and the page journey a user actually makes (integrations
17// list → integration page → integrations list). It then watches a quiet
18// window. A settled page must stop asking. A count that keeps climbing is the
19// loop.
20//
21// Skips on targets with no browser surface.
22import { randomBytes } from "node:crypto";
23import { createServer } from "node:http";
24
25import { expect } from "@effect/vitest";
26import { Effect } from "effect";
27import type { HttpApiClient } from "effect/unstable/httpapi";
28import { composePluginApi } from "@executor-js/api/server";
29import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api";
30import { AuthTemplateSlug, ConnectionName, IntegrationSlug } from "@executor-js/sdk/shared";
31
32import { scenario } from "../src/scenario";
33import { Api, Browser, Target } from "../src/services";
34import { visit } from "../src/surfaces/browser";
35
36const api = composePluginApi([openApiHttpPlugin()] as const);
37type Client = HttpApiClient.ForApi<typeof api>;
38
39const TEMPLATE = AuthTemplateSlug.make("apiKey");
40
41/** Integrations whose connection has a probe configured and an upstream that
42 * rejects the key: the probe persists `degraded`. */
43const DEGRADED_COUNT = 4;
44/** Integrations whose connection has NO probe configured: every automatic
45 * check answers `unknown` with a fresh `checkedAt`. */
46const UNKNOWN_COUNT = 4;
47
48/** How long a settled page is watched for further probes. Production cycled
49 * every 0.5-1s, so a loop shows up well inside this. */
50const QUIET_WINDOW_MS = 10_000;
51
52const unique = (prefix: string) => `${prefix}${randomBytes(3).toString("hex")}`;
53
54/** Upstream on 127.0.0.1 whose `GET /me` rejects every key. */
55const serveRejectingUpstream = () =>
56 Effect.acquireRelease(
57 Effect.callback<{ readonly url: string; readonly close: () => void }>((resume) => {
58 const server = createServer((request, response) => {
59 if (request.method === "GET" && (request.url ?? "").startsWith("/me")) {
60 response.writeHead(401, { "content-type": "application/json" });
61 response.end(JSON.stringify({ error: "invalid_token" }));
62 return;
63 }
64 response.writeHead(404, { "content-type": "application/json" });
65 response.end(JSON.stringify({ error: "not_found" }));
66 });
67 server.listen(0, "127.0.0.1", () => {
68 const address = server.address();
69 const port = typeof address === "object" && address ? address.port : 0;
70 resume(
71 Effect.succeed({
72 url: `http://127.0.0.1:${port}`,
73 close: () => {
74 server.close();
75 server.closeAllConnections();
76 },
77 }),
78 );
79 });
80 }),
81 (server) => Effect.sync(server.close),
82 );
83
84const identitySpec = (baseUrl: string, title: string): string =>
85 JSON.stringify({
86 openapi: "3.0.3",
87 info: { title, version: "1.0.0" },
88 servers: [{ url: baseUrl }],
89 paths: {
90 "/me": {
91 get: {
92 operationId: "getMe",
93 summary: "The current account",
94 responses: {
95 "200": {
96 description: "The authenticated account",
97 content: {
98 "application/json": {
99 schema: { type: "object", properties: { email: { type: "string" } } },
100 },
101 },
102 },
103 },
104 },
105 },
106 },
107 });
108
109/** One OpenAPI integration with one saved org connection. With `probe`, the
110 * identity GET is configured as the health check. */
111const seedIntegration = (
112 client: Client,
113 upstreamUrl: string,
114 options: { readonly prefix: string; readonly probe: boolean },
115) =>
116 Effect.gen(function* () {
117 const slug = IntegrationSlug.make(unique(options.prefix));
118 const name = ConnectionName.make(`${options.prefix}conn`);
119
120 yield* Effect.addFinalizer(() =>
121 Effect.all(
122 [
123 client.connections
124 .remove({ params: { owner: "org", integration: slug, name } })
125 .pipe(Effect.ignore),
126 client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore),
127 ],
128 { discard: true },
129 ),
130 );
131
132 yield* client.openapi.addSpec({
133 payload: {
134 spec: { kind: "blob", value: identitySpec(upstreamUrl, `Churn ${slug}`) },
135 slug,
136 baseUrl: upstreamUrl,
137 authenticationTemplate: [
138 {
139 slug: "apiKey",
140 type: "apiKey",
141 headers: { authorization: ["Bearer ", { type: "variable", name: "token" }] },
142 },
143 ],
144 },
145 });
146
147 if (options.probe) {
148 const candidates = yield* client.integrations.healthCheckCandidates({ params: { slug } });
149 const getMe = candidates.find((candidate) => candidate.method === "get");
150 if (!getMe) return yield* Effect.die("identity spec exposed no GET candidate");
151 yield* client.integrations.healthCheckSet({
152 params: { slug },
153 payload: { spec: { operation: getMe.operation, identityField: "email" } },
154 });
155 }
156
157 yield* client.connections.create({
158 payload: { owner: "org", name, integration: slug, template: TEMPLATE, value: "bad-key" },
159 });
160
161 return { slug, name };
162 });
163
164scenario(
165 "Health checks (UI) · many broken connections across many integrations settle instead of looping",
166 {},
167 Effect.scoped(
168 Effect.gen(function* () {
169 const target = yield* Target;
170 const browser = yield* Browser;
171 const { client: makeClient } = yield* Api;
172 const identity = yield* target.newIdentity();
173 const client = yield* makeClient(api, identity);
174 const upstream = yield* serveRejectingUpstream();
175
176 const degraded = yield* Effect.all(
177 Array.from({ length: DEGRADED_COUNT }, () =>
178 seedIntegration(client, upstream.url, { prefix: "churndeg", probe: true }),
179 ),
180 { concurrency: 2 },
181 );
182 const unknown = yield* Effect.all(
183 Array.from({ length: UNKNOWN_COUNT }, () =>
184 seedIntegration(client, upstream.url, { prefix: "churnunk", probe: false }),
185 ),
186 { concurrency: 2 },
187 );
188 const seeded = [...degraded, ...unknown];
189 const connectionCount = seeded.length;
190
191 // Persist a verdict on every connection BEFORE the browser opens, the
192 // way a returning user finds them: `degraded` on the probed ones,
193 // `unknown` on the rest. The list then renders each row with a
194 // persisted verdict, and every automatic revalidation compares its
195 // result against one.
196 for (const { slug, name } of seeded) {
197 yield* client.connections.checkHealth({
198 params: { owner: "org", integration: slug, name },
199 query: {},
200 });
201 }
202
203 yield* browser.session(identity, async ({ page, step }) => {
204 // The client's wire contract, observed from outside: every health POST
205 // and every connections-list GET the app sends, plus how many the
206 // browser reports as failed (an aborted request shows up here as a
207 // failure, which is how the interrupted mutation atom looks on the wire).
208 const health: string[] = [];
209 const lists: string[] = [];
210 let aborted = 0;
211 // Only THIS scenario's connections count. Targets that share one org
212 // across scenarios (selfhost) can carry rows another scenario left
213 // behind, and those revalidate too; they are not ours to bound.
214 const seededSlugs = new Set(seeded.map(({ slug }) => String(slug)));
215 const isHealth = (method: string, url: string) =>
216 method === "POST" &&
217 url.includes("/health") &&
218 [...seededSlugs].some((slug) => url.includes(`/api/connections/org/${slug}/`));
219 const isList = (method: string, url: string) =>
220 method === "GET" && /\/api\/connections(\?|$)/.test(url);
221 page.on("request", (request) => {
222 const url = request.url();
223 if (isHealth(request.method(), url)) health.push(url);
224 else if (isList(request.method(), url)) lists.push(url);
225 });
226 page.on("requestfailed", (request) => {
227 const url = request.url();
228 if (isHealth(request.method(), url) || isList(request.method(), url)) aborted += 1;
229 });
230
231 const snapshot = (label: string) => {
232 const line = `[churn] ${label}: health=${String(health.length)} lists=${String(lists.length)} aborted=${String(aborted)}`;
233 console.log(line);
234 return { health: health.length, lists: lists.length, aborted };
235 };
236
237 await step("Open the integrations list with every broken connection on it", async () => {
238 await visit(page, "/");
239 for (const { slug } of seeded) {
240 await page
241 .getByRole("link", { name: new RegExp(slug, "i") })
242 .first()
243 .waitFor({
244 timeout: 30_000,
245 });
246 }
247 // Let the automatic revalidation land for every row.
248 await page.waitForTimeout(3_000);
249 });
250 const afterList = snapshot("after integrations list");
251
252 await step("Open one integration page, then return to the list", async () => {
253 await visit(page, `/integrations/${degraded[0]!.slug}`);
254 await page.waitForTimeout(2_000);
255 await visit(page, "/");
256 await page.waitForTimeout(3_000);
257 });
258 const afterJourney = snapshot("after journey");
259
260 // The quiet window: nothing changed on the page, nothing changed on
261 // the server, so nothing more should be asked.
262 await step("Leave the settled list open and watch the wire", async () => {
263 await page.waitForTimeout(QUIET_WINDOW_MS);
264 });
265 const afterQuiet = snapshot("after quiet window");
266
267 const quietHealth = afterQuiet.health - afterJourney.health;
268 const quietLists = afterQuiet.lists - afterJourney.lists;
269 console.log(
270 `[churn] quiet window (${String(QUIET_WINDOW_MS)}ms): +${String(quietHealth)} health, +${String(quietLists)} list fetches over ${String(connectionCount)} connections`,
271 );
272
273 // One list mount revalidates each non-healthy connection once, and
274 // reads the connections list once per owner (plus the unscoped read).
275 // A probe count above the connection count means rows are being
276 // re-probed; a list count above three means verdicts are being
277 // treated as changes and refetching the list.
278 expect(
279 afterList.health,
280 `the first list mount probes each broken connection at most once (sent ${String(afterList.health)} for ${String(connectionCount)} connections)`,
281 ).toBeLessThanOrEqual(connectionCount);
282 expect(
283 afterList.lists,
284 `the first list mount reads the connections list at most once per owner (sent ${String(afterList.lists)})`,
285 ).toBeLessThanOrEqual(3);
286
287 // THE production symptom: a settled page keeps asking. Zero is the
288 // contract; anything else is the loop.
289 expect(quietHealth, "a settled integrations list sends no further health probes").toBe(0);
290 expect(
291 quietLists,
292 "a settled integrations list does not refetch the connections list",
293 ).toBe(0);
294 });
295 }),
296 ),
297);