Fix two more CI e2e flakes: picker hydration race and dev-db ghost affinity

apps/cloud/scripts/dev-db.ts

@@ -104,14 +104,20 @@ await migrate(drizzle(db), { migrationsFolder: MIGRATIONS_FOLDER });
 // but prepared statement requires M" -> random 500s on whichever request lost
 // the race). The patch in patches/@electric-sql%[email protected]
 // batches each socket data event into one queue entry and holds handler
-// affinity while a pipeline is open. The patch also fixes the queue's failure
-// path: stock 0.1.4 `return`ed out of the drain loop when a query REJECTED at
-// the JS level, leaving its `processing` flag latched true — after one such
-// throw nothing was ever dequeued again, so new connections' startup packets
-// sat unanswered (postgres.js CONNECT_TIMEOUT) and the whole stack was bricked
-// until restart: the CI e2e "cloud signIn: callback set no session (500)"
-// cascade. src/db/dev-db-socket-concurrency.node.test.ts is the regression
-// test for all of the above.
+// affinity while a pipeline is open. The patch also fixes the queue's two
+// self-bricking failure paths — both surfaced in CI as the e2e "cloud signIn:
+// callback set no session (500)" cascade, where new connections' startup
+// packets sat unanswered (postgres.js CONNECT_TIMEOUT) until restart:
+//   1. Stock 0.1.4 `return`ed out of the drain loop when a query REJECTED at
+//      the JS level, leaving its `processing` flag latched true; nothing was
+//      ever dequeued again.
+//   2. A client whose socket died WHILE its pipeline-opening entry executed:
+//      detach() cleared affinity before the entry finished, the queue then
+//      took affinity for the already-dead handler, and no timer was left to
+//      release it. The queue now tracks detached handlers and repairs any
+//      transaction or pipeline affinity they can no longer release.
+// src/db/dev-db-socket-concurrency.node.test.ts is the regression test for
+// all of the above.
 const server = new PGLiteSocketServer({
   db,
   port: PORT,

apps/cloud/src/db/dev-db-socket-concurrency.node.test.ts

@@ -42,6 +42,40 @@ const makeClient = (port: number, connectTimeout = 5) =>
     onnotice: () => undefined,
   });
 
+// Hand-rolled wire client: connect and complete the trust-auth startup, so a
+// test can then speak raw protocol frames (e.g. a lone Parse) that postgres.js
+// would never emit on its own. Resolves after ReadyForQuery so the next write
+// is its own data event — and its own queue entry — on the server.
+const openWireClient = async (port: number): Promise<Socket> => {
+  const socket: Socket = connect(port, "127.0.0.1");
+  await new Promise<void>((res, rej) => {
+    socket.once("connect", res);
+    socket.once("error", rej);
+  });
+  const startupBody = Buffer.concat([
+    Buffer.from([0, 3, 0, 0]),
+    Buffer.from("user\0postgres\0database\0postgres\0\0"),
+  ]);
+  const startup = Buffer.concat([Buffer.alloc(4), startupBody]);
+  startup.writeInt32BE(startup.length, 0);
+  socket.write(startup);
+  await new Promise<void>((res) => {
+    socket.on("data", (chunk: Buffer) => {
+      if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery
+    });
+  });
+  return socket;
+};
+
+// A Parse frame for an unnamed statement: opens an extended-protocol pipeline
+// that only a later Sync (or the server's recovery) closes.
+const parseFrame = (query: string): Buffer => {
+  const body = Buffer.concat([Buffer.from(`\0${query}\0`), Buffer.from([0, 0])]);
+  const frame = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), body]);
+  frame.writeInt32BE(4 + body.length, 1);
+  return frame;
+};
+
 describe("dev-db PGlite socket under concurrent connections", () => {
   it(
     "serves interleaved multi-connection pipelines without protocol corruption",
@@ -207,29 +241,8 @@ describe("dev-db PGlite socket under concurrent connections", () => {
       // Hand-rolled wire client: complete the trust-auth startup, then send a
       // lone Parse. Its last frame type ('P') marks the pipeline open, so the
       // handler takes affinity and every other connection queues behind it.
-      const staller: Socket = connect(port, "127.0.0.1");
-      await new Promise<void>((res, rej) => {
-        staller.once("connect", res);
-        staller.once("error", rej);
-      });
-      const startupBody = Buffer.concat([
-        Buffer.from([0, 3, 0, 0]),
-        Buffer.from("user\0postgres\0database\0postgres\0\0"),
-      ]);
-      const startup = Buffer.concat([Buffer.alloc(4), startupBody]);
-      startup.writeInt32BE(startup.length, 0);
-      staller.write(startup);
-      // Wait for AuthenticationOk + ReadyForQuery before opening the pipeline,
-      // so the Parse is its own data event (and its own queue entry).
-      await new Promise<void>((res) => {
-        staller.on("data", (chunk: Buffer) => {
-          if (chunk.includes(0x5a)) res(); // 'Z' = ReadyForQuery
-        });
-      });
-      const parseBody = Buffer.from("\0select 1\0\0\0");
-      const parse = Buffer.concat([Buffer.from("P"), Buffer.alloc(4), parseBody]);
-      parse.writeInt32BE(4 + parseBody.length, 1);
-      staller.write(parse);
+      const staller = await openWireClient(port);
+      staller.write(parseFrame("select 1"));
 
       const bystander = makeClient(port, 10);
       // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
@@ -245,4 +258,50 @@ describe("dev-db PGlite socket under concurrent connections", () => {
       }
     },
   );
+
+  // Regression for the second wedge mode behind the same CI cascade: a client
+  // whose socket dies WHILE its pipeline-opening entry is executing. detach()
+  // clears pipeline affinity before the entry finishes, so the queue then
+  // assigned affinity to the already-dead handler — and nothing ever cleared
+  // it: the dead handler has no timers left, and every other connection
+  // (including fresh startups) queued behind the ghost forever. The queue now
+  // tracks detached handlers and repairs affinity they can no longer release.
+  it(
+    "a client that dies mid-execution does not leave the queue pinned to its ghost",
+    { timeout: 30_000 },
+    async () => {
+      const port = 45994;
+      const db = await PGlite.create();
+
+      // Hold the marker query in flight long enough that the disconnect below
+      // reliably lands while the entry is EXECUTING (after detach's cleanup,
+      // before the queue takes affinity for it).
+      const real = db.execProtocolRawStream.bind(db);
+      (db as { execProtocolRawStream: typeof real }).execProtocolRawStream = async (...args) => {
+        if (Buffer.from(args[0]).includes("ghost_marker")) await sleep(300);
+        return real(...args);
+      };
+
+      const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 });
+      await server.start();
+
+      const ghost = await openWireClient(port);
+      ghost.write(parseFrame("select 'ghost_marker'"));
+      // Give the data event time to reach the queue and start executing, then
+      // die without a trace mid-flight.
+      await sleep(100);
+      ghost.destroy();
+
+      const bystander = makeClient(port);
+      // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
+      try {
+        expect((await bystander.unsafe(`select 5 as five`))[0]).toEqual({ five: 5 });
+      } finally {
+        // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
+        await bystander.end({ timeout: 5 }).catch(() => {});
+        await server.stop();
+        await db.close();
+      }
+    },
+  );
 });

e2e/scenarios/google-health-checks.test.ts

@@ -16,7 +16,7 @@ import {
 import { scenario } from "../src/scenario";
 import { Api, Browser, Target } from "../src/services";
 import type { Identity, Target as TargetShape } from "../src/target";
-import type { BrowserSurface } from "../src/surfaces/browser";
+import { clickToReveal, type BrowserSurface } from "../src/surfaces/browser";
 
 const api = composePluginApi([openApiHttpPlugin()] as const);
 type Client = HttpApiClient.ForApi<typeof api>;
@@ -88,12 +88,8 @@ const addGooglePresetFromCatalog = (
   browser.session(identity, async ({ page, step }) => {
     await step(`Open ${presetName} from the connect catalog`, async () => {
       await page.goto("/integrations", { waitUntil: "networkidle" });
-      await page
-        .getByRole("button", { name: /Connect/ })
-        .first()
-        .click();
       const dialog = page.getByRole("dialog", { name: "Connect an integration" });
-      await dialog.waitFor();
+      await clickToReveal(page.getByRole("button", { name: /Connect/ }).first(), dialog);
       await dialog.getByPlaceholder(/Search or paste a URL/).fill(presetName);
       await dialog.getByRole("link", { name: new RegExp(`^${presetName}\\b`) }).click();
     });

e2e/scenarios/google-photos-preset-ui.test.ts

@@ -3,6 +3,7 @@ import { Effect } from "effect";
 
 import { scenario } from "../src/scenario";
 import { Browser, Target } from "../src/services";
+import { clickToReveal } from "../src/surfaces/browser";
 
 scenario(
   "Google Photos: separated catalog presets open a Photos service add flow",
@@ -17,9 +18,8 @@ scenario(
         "Find the separated Google Photos presets from the integrations picker",
         async () => {
           await page.goto("/integrations", { waitUntil: "networkidle" });
-          await page.getByRole("button", { name: "Connect" }).click();
           const dialog = page.getByRole("dialog", { name: "Connect an integration" });
-          await dialog.waitFor();
+          await clickToReveal(page.getByRole("button", { name: "Connect" }), dialog);
           await dialog.getByPlaceholder(/Search or paste a URL/).fill("google photos");
           await dialog.getByRole("link", { name: /^Google Photos Library\b/ }).waitFor();
           await dialog.getByRole("link", { name: /^Google Photos Picker\b/ }).waitFor();

e2e/scenarios/provider-plugins-ui.test.ts

@@ -3,6 +3,7 @@ import { Effect } from "effect";
 
 import { scenario } from "../src/scenario";
 import { Browser, Target } from "../src/services";
+import { clickToReveal } from "../src/surfaces/browser";
 
 scenario(
   "Provider catalog · Google and Microsoft services are OpenAPI presets",
@@ -15,8 +16,10 @@ scenario(
     yield* browser.session(identity, async ({ page, step }) => {
       await step("Open the integrations picker", async () => {
         await page.goto("/integrations", { waitUntil: "networkidle" });
-        await page.getByRole("button", { name: "Connect" }).click();
-        await page.getByRole("dialog", { name: "Connect an integration" }).waitFor();
+        await clickToReveal(
+          page.getByRole("button", { name: "Connect" }),
+          page.getByRole("dialog", { name: "Connect an integration" }),
+        );
       });
 
       await step("The picker exposes OpenAPI plus provider service presets", async () => {

e2e/src/surfaces/browser.ts

@@ -9,7 +9,7 @@ import { join } from "node:path";
 import { promisify } from "node:util";
 
 import { Effect } from "effect";
-import { chromium, type Page } from "playwright";
+import { chromium, type Locator, type Page } from "playwright";
 
 import { beat, enterFocus, markNavigation, markRecordingStart } from "../timeline";
 import { appendTraces, type TraceEntry } from "../trace-harvest";
@@ -36,6 +36,35 @@ const slug = (text: string): string =>
     .replace(/^-+|-+$/g, "")
     .slice(0, 60);
 
+/**
+ * Click `trigger` until `revealed` is visible.
+ *
+ * `waitUntil: "networkidle"` does not mean the console has hydrated: a click
+ * that lands between the SSR paint and React attaching the handler is
+ * swallowed without a trace, and whatever the click was meant to open never
+ * appears (the "Connect an integration" dialog no-show flake). Re-clicking a
+ * reveal-style trigger is idempotent, so retry until the result is actually
+ * on screen; the final attempt waits with the full timeout so the failure
+ * surfaces as the ordinary locator error.
+ */
+export const clickToReveal = async (
+  trigger: Locator,
+  revealed: Locator,
+  { attempts = 5, revealTimeoutMs = 4_000 }: { attempts?: number; revealTimeoutMs?: number } = {},
+): Promise<void> => {
+  for (let attempt = 1; attempt < attempts; attempt++) {
+    await trigger.click();
+    const shown = await revealed
+      .waitFor({ timeout: revealTimeoutMs })
+      .then(() => true)
+      // oxlint-disable-next-line executor/no-promise-catch -- retry boundary: a missed reveal is the signal to click again, not a failure
+      .catch(() => false);
+    if (shown) return;
+  }
+  await trigger.click();
+  await revealed.waitFor({ timeout: revealTimeoutMs });
+};
+
 // acquireUseRelease so a vitest timeout (fiber interruption) still closes the
 // browser and flushes video + trace — a bare promise would leak Chromium.
 export const makeBrowserSurface = (dir: string, target: Target): BrowserSurface => ({