Fix two CI e2e flakes: stylesheet-count race and dev-db queue wedge

apps/cloud/scripts/dev-db.ts

@@ -104,8 +104,14 @@ 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;
-// src/db/dev-db-socket-concurrency.node.test.ts is the regression test.
+// 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.
 const server = new PGLiteSocketServer({
   db,
   port: PORT,
@@ -115,7 +121,11 @@ const server = new PGLiteSocketServer({
   // sent, no Sync) with its socket still OPEN would hold the queue's handler
   // affinity forever and starve every other connection, since affinity only
   // releases on detach and detach needs close/error/idle-timeout. In ms; the
-  // timer resets on every data event, so only a genuinely dead client trips it.
+  // timer resets on every data event. The patch scopes the reap to connections
+  // actually HOLDING affinity (open pipeline or transaction): an idle-at-rest
+  // connection is the normal state of a healthy postgres.js pool held by a
+  // long-lived scope (SSE), and reaping those raced live queries into
+  // sporadic `write CONNECTION_ENDED` 500s.
   idleTimeout: Number(process.env.DEV_DB_IDLE_TIMEOUT_MS ?? 30_000),
 });
 

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

@@ -21,6 +21,8 @@
 // with DIFFERENT parameter counts (the exact drizzle/postgres-js shape) through
 // one PGLiteSocketServer and asserts zero protocol corruption.
 
+import { setTimeout as sleep } from "node:timers/promises";
+import { connect, type Socket } from "node:net";
 import { describe, expect, it } from "@effect/vitest";
 import { PGlite } from "@electric-sql/pglite";
 import { PGLiteSocketServer } from "@electric-sql/pglite-socket";
@@ -30,6 +32,16 @@ const PORT = 45998;
 const CLIENTS = 6;
 const QUERIES_PER_CLIENT = 40;
 
+const makeClient = (port: number, connectTimeout = 5) =>
+  postgres(`postgres://postgres:[email protected]:${port}/postgres`, {
+    max: 1,
+    idle_timeout: 0,
+    connect_timeout: connectTimeout,
+    fetch_types: false,
+    prepare: true,
+    onnotice: () => undefined,
+  });
+
 describe("dev-db PGlite socket under concurrent connections", () => {
   it(
     "serves interleaved multi-connection pipelines without protocol corruption",
@@ -48,14 +60,7 @@ describe("dev-db PGlite socket under concurrent connections", () => {
       const errors: string[] = [];
 
       const worker = async (id: number) => {
-        const sql = postgres(`postgres://postgres:[email protected]:${PORT}/postgres`, {
-          max: 1,
-          idle_timeout: 0,
-          connect_timeout: 10,
-          fetch_types: false,
-          prepare: true,
-          onnotice: () => undefined,
-        });
+        const sql = makeClient(PORT, 10);
         // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: postgres.js is promise-native and the socket must be closed on every path
         try {
           for (let q = 0; q < QUERIES_PER_CLIENT; q++) {
@@ -91,4 +96,153 @@ describe("dev-db PGlite socket under concurrent connections", () => {
       expect(ok).toBe(CLIENTS * QUERIES_PER_CLIENT);
     },
   );
+
+  // Regression for the CI e2e "cloud signIn: callback set no session (500)"
+  // cascade: QueryQueueManager.processQueue used to `return` out of its drain
+  // loop when a query REJECTED (as opposed to returning a wire-level
+  // ErrorResponse), leaving `processing` latched true. From then on every
+  // enqueue — including brand-new connections' startup packets — sat in the
+  // queue forever: in-flight requests hung, postgres.js reconnects died with
+  // CONNECT_TIMEOUT, and the whole dev stack was bricked until restart. The
+  // patch rejects the one entry, drops pipeline affinity, and keeps draining.
+  it(
+    "a rejected query fails one client, not the whole socket server",
+    { timeout: 30_000 },
+    async () => {
+      const port = 45997;
+      const db = await PGlite.create();
+      const server = new PGLiteSocketServer({ db, port, host: "127.0.0.1", maxConnections: 100 });
+      await server.start();
+
+      const first = makeClient(port);
+      // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
+      try {
+        expect((await first.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 });
+
+        // Force the NEXT protocol exchange to reject at the JS level, the shape
+        // PGlite produces when the shared session is broken mid-run.
+        const real = db.execProtocolRawStream.bind(db);
+        let arm = true;
+        (db as { execProtocolRawStream: typeof real }).execProtocolRawStream = (...args) => {
+          if (arm) {
+            arm = false;
+            // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: simulating a PGlite internal failure requires a raw throw
+            throw new Error("synthetic PGlite failure");
+          }
+          return real(...args);
+        };
+
+        await expect(first.unsafe(`select 2 as two`)).rejects.toThrow();
+
+        // The poisoned entry must take down only its own connection: a fresh
+        // client (new socket, full startup handshake) still gets served.
+        const second = makeClient(port);
+        // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
+        try {
+          expect((await second.unsafe(`select 3 as three`))[0]).toEqual({ three: 3 });
+        } finally {
+          // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
+          await second.end({ timeout: 5 }).catch(() => {});
+        }
+      } finally {
+        // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
+        await first.end({ timeout: 5 }).catch(() => {});
+        await server.stop();
+        await db.close();
+      }
+    },
+  );
+
+  // Regression for the sporadic `write CONNECTION_ENDED` 500s: the server's
+  // idleTimeout backstop used to kill ANY connection with no traffic for the
+  // window, which is the resting state of every healthy postgres.js pool
+  // connection (idle_timeout: 0) held by a long-lived scope. The backstop now
+  // only fires on a connection that is actually blocking the shared session —
+  // an open pipeline or an open transaction.
+  it("an idle-at-rest connection outlives the idle backstop", { timeout: 30_000 }, async () => {
+    const port = 45996;
+    const db = await PGlite.create();
+    const server = new PGLiteSocketServer({
+      db,
+      port,
+      host: "127.0.0.1",
+      maxConnections: 100,
+      idleTimeout: 250,
+    });
+    await server.start();
+
+    const sql = makeClient(port);
+    // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
+    try {
+      expect((await sql.unsafe(`select 1 as one`))[0]).toEqual({ one: 1 });
+      await sleep(900);
+      expect((await sql.unsafe(`select 2 as two`))[0]).toEqual({ two: 2 });
+    } finally {
+      // oxlint-disable-next-line executor/no-promise-catch -- test boundary: a failed teardown must not mask the assertion
+      await sql.end({ timeout: 5 }).catch(() => {});
+      await server.stop();
+      await db.close();
+    }
+  });
+
+  // The backstop's actual job still works: a client that opens a pipeline
+  // (Parse sent, never Sync) and goes silent holds queue affinity, which
+  // starves every other connection. The idle timer must reap exactly that
+  // client and hand the queue back.
+  it(
+    "a client stalled mid-pipeline is reaped and the queue recovers",
+    { timeout: 30_000 },
+    async () => {
+      const port = 45995;
+      const db = await PGlite.create();
+      const server = new PGLiteSocketServer({
+        db,
+        port,
+        host: "127.0.0.1",
+        maxConnections: 100,
+        idleTimeout: 250,
+      });
+      await server.start();
+
+      // 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 bystander = makeClient(port, 10);
+      // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: sockets must be closed on every path
+      try {
+        // Connects and queries only once the staller is reaped (~250ms).
+        expect((await bystander.unsafe(`select 4 as four`))[0]).toEqual({ four: 4 });
+      } 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(() => {});
+        staller.destroy();
+        await server.stop();
+        await db.close();
+      }
+    },
+  );
 });

e2e/scenarios/artifacts.test.ts

@@ -143,18 +143,31 @@ const recordHandshakeOrdering = async (page: Page): Promise<void> => {
 const readHandshakeOrdering = (page: Page): Promise<ReadonlyArray<string>> =>
   page.evaluate(() => globalThis.__handshakeOrder ?? []);
 
-const readConsoleStyle = (
-  page: Page,
-): Promise<{ primary: string; buttonBg: string; styleSheets: number }> =>
+const readConsoleStyle = (page: Page): Promise<{ primary: string; buttonBg: string }> =>
   page.evaluate(() => {
     const button = document.querySelector("button");
     return {
       primary: getComputedStyle(document.documentElement).getPropertyValue("--primary").trim(),
       buttonBg: button ? getComputedStyle(button).backgroundColor : "",
-      styleSheets: document.styleSheets.length,
     };
   });
 
+// The shell's compiled stylesheet declares `--mcp-apps-shell-stylesheet: 1`
+// on `:root` as a provenance marker (see the shell's globals.css): the shell's
+// tokens deliberately mirror the console's, so this marker is the only
+// declaration that identifies the sheet. Reading it as a computed value on a
+// document's root element answers "did the shell's stylesheet land in THIS
+// document?" — unlike counting document.styleSheets, which moves on its own in
+// dev (TanStack Start swaps its route-styles <link> as matches settle, and a
+// swapped-in link only counts once loaded), which made an equality-of-counts
+// assertion flaky.
+const readShellStylesheetMarker = (page: Page): Promise<string> =>
+  page.evaluate(() =>
+    getComputedStyle(document.documentElement)
+      .getPropertyValue("--mcp-apps-shell-stylesheet")
+      .trim(),
+  );
+
 scenario(
   "Artifacts · create-artifact hands a non-Apps client a deep link that renders the live component",
   { timeout: 180_000 },
@@ -242,10 +255,11 @@ scenario(
 
       yield* browser.session(identity, async ({ page, step }) => {
         // The console's own styling, sampled BEFORE any artifact is opened.
-        // The shell ships its own Tailwind build and its own palette (a teal
-        // `--primary` against the console's near-black), so if its stylesheet
-        // ever reaches the top-level document again these values move.
-        let consoleStyleBefore: { primary: string; buttonBg: string; styleSheets: number };
+        // The shell ships its own Tailwind build; if its stylesheet ever
+        // reaches the top-level document again, its base/utility layers move
+        // these computed values (and its provenance marker appears, asserted
+        // below).
+        let consoleStyleBefore: { primary: string; buttonBg: string };
 
         await step("Open the artifact link the agent handed over", async () => {
           await recordHandshakeOrdering(page);
@@ -338,29 +352,33 @@ scenario(
           expect(after.buttonBg, "a console button keeps its own background").toBe(
             consoleStyleBefore.buttonBg,
           );
-          expect(
-            after.styleSheets,
-            "the shell injected no stylesheet into the console document",
-          ).toBe(consoleStyleBefore.styleSheets);
 
           // And positively: the shell's stylesheet IS present, one document
           // down. Without this the assertions above would also pass if the
           // shell had simply failed to load.
           const shellHasOwnStyles = await page
             .frameLocator('[data-testid="artifact-shell-frame"]')
             .locator("html")
-            .evaluate((html) => {
-              const primary = getComputedStyle(html).getPropertyValue("--primary").trim();
-              return { primary, sheets: html.ownerDocument.styleSheets.length };
-            });
+            .evaluate((html) => ({
+              marker: getComputedStyle(html).getPropertyValue("--mcp-apps-shell-stylesheet").trim(),
+              sheets: html.ownerDocument.styleSheets.length,
+            }));
           expect(
             shellHasOwnStyles.sheets,
             "the shell document carries its own stylesheets",
           ).toBeGreaterThan(0);
           expect(
-            shellHasOwnStyles.primary,
-            "the shell keeps its own palette inside its own document",
-          ).not.toBe("");
+            shellHasOwnStyles.marker,
+            "the shell document carries the shell's own compiled stylesheet",
+          ).toBe("1");
+
+          // The marker is the injection fingerprint: even a shell sheet that
+          // lost the cascade race (so the computed values above stayed put)
+          // would still surface it on the console's root element.
+          expect(
+            await readShellStylesheetMarker(page),
+            "the shell injected no stylesheet into the console document",
+          ).toBe("");
         });
 
         await step("The artifact fills the page and scrolls inside itself", async () => {