Connect: parallel gate lookups, zombie-socket freshness, gate fonts, per-message compression (B29-B32)

apps/connect/src/session.test.ts

@@ -4,13 +4,14 @@ import { fileURLToPath } from "node:url";
 import Database from "better-sqlite3";
 import { eq } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/better-sqlite3";
-import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import {
   labelClaim,
   machine,
   profile,
   schema,
   server,
+  session,
   user,
 } from "@bb/connect-db";
 
@@ -19,6 +20,7 @@ import {
   markMachineSeen,
   resolveLabel,
   verifyMachineCredentialDetails,
+  verifySessionCookie,
 } from "./session.js";
 import { assignMachineLabel } from "./machine-label.js";
 
@@ -408,3 +410,97 @@ describe("machine credential presence", () => {
     ).toBe(true);
   });
 });
+
+// A page load fans out dozens of gate requests within milliseconds, all
+// before the first D1 answer lands. Value-only caches let every one of them
+// miss; the pending lookup itself must be shared so a burst costs one query.
+describe("in-flight lookup sharing", () => {
+  it("collapses concurrent resolves of one label into a single query", async () => {
+    seedUser("acct-burst");
+    seedServer({
+      id: "srv-burst",
+      userId: "acct-burst",
+      name: "default",
+      subdomain: "burst-label",
+    });
+    const select = vi.spyOn(db, "select");
+    const results = await Promise.all(
+      Array.from({ length: 5 }, () => resolveLabel("burst-label", db)),
+    );
+    expect(select).toHaveBeenCalledTimes(1);
+    for (const result of results) {
+      expect(result).toMatchObject({ kind: "server", userId: "acct-burst" });
+    }
+    // Settled: later callers hit the value cache, not a retained promise.
+    await expect(resolveLabel("burst-label", db)).resolves.toMatchObject({
+      kind: "server",
+    });
+    expect(select).toHaveBeenCalledTimes(1);
+  });
+
+  it("does not share a fresh (cache-bypassing) resolve with the pending one", async () => {
+    seedUser("acct-fresh");
+    seedServer({
+      id: "srv-fresh",
+      userId: "acct-fresh",
+      name: "default",
+      subdomain: "fresh-label",
+    });
+    const select = vi.spyOn(db, "select");
+    await Promise.all([
+      resolveLabel("fresh-label", db),
+      resolveLabel("fresh-label", db, { fresh: true }),
+    ]);
+    expect(select).toHaveBeenCalledTimes(2);
+  });
+
+  it("drops a rejected lookup so the next request retries", async () => {
+    const failing = vi.spyOn(db, "select").mockImplementationOnce(() => {
+      throw new Error("D1 unavailable");
+    });
+    await expect(resolveLabel("flaky-label", db)).rejects.toThrow(
+      "D1 unavailable",
+    );
+    failing.mockRestore();
+    await expect(resolveLabel("flaky-label", db)).resolves.toBeNull();
+  });
+
+  it("collapses concurrent verifications of one session cookie into a single query", async () => {
+    seedUser("acct-session");
+    const token = "sess_token_burst";
+    const secret = "test-better-auth-secret";
+    const key = await crypto.subtle.importKey(
+      "raw",
+      new TextEncoder().encode(secret),
+      { name: "HMAC", hash: "SHA-256" },
+      false,
+      ["sign"],
+    );
+    const sigBuf = await crypto.subtle.sign(
+      "HMAC",
+      key,
+      new TextEncoder().encode(token),
+    );
+    const sig = btoa(String.fromCharCode(...new Uint8Array(sigBuf)));
+    const cookieValue = `${token}.${sig}`;
+    db.insert(session)
+      .values({
+        id: "sess-burst",
+        token,
+        expiresAt: new Date(Date.now() + 60_000),
+        userId: "acct-session",
+        createdAt: now,
+        updatedAt: now,
+      })
+      .run();
+
+    const select = vi.spyOn(db, "select");
+    const results = await Promise.all(
+      Array.from({ length: 5 }, () =>
+        verifySessionCookie(cookieValue, secret, db),
+      ),
+    );
+    expect(results).toEqual(Array(5).fill("acct-session"));
+    expect(select).toHaveBeenCalledTimes(1);
+  });
+});

apps/connect/src/session.ts

@@ -16,6 +16,13 @@ import {
 // D1. TTLs are short so sign-out / disconnect take effect quickly (and the DO
 // already severs a live tunnel on revoke, so a stale-cached label still can't
 // reach a disconnected server).
+//
+// The caches also hold the in-flight lookup, not only the settled value: a
+// page load fans out ~40 asset requests within a few milliseconds, all before
+// the first D1 answer lands. Value-only caching let every one of them miss and
+// issue its own round trip; sharing the pending promise collapses the burst
+// into one query per key. A rejected lookup is dropped so the next request
+// retries instead of replaying a transient D1 error for the whole TTL.
 const LABEL_TTL_MS = 15_000;
 const SESSION_TTL_MS = 20_000;
 
@@ -25,6 +32,8 @@ interface CacheEntry<T> {
 }
 const labelCache = new Map<string, CacheEntry<ResolvedLabel | null>>();
 const sessionCache = new Map<string, CacheEntry<string | null>>();
+const labelInflight = new Map<string, Promise<ResolvedLabel | null>>();
+const sessionInflight = new Map<string, Promise<string | null>>();
 
 function cacheGet<T>(
   map: Map<string, CacheEntry<T>>,
@@ -37,6 +46,26 @@ function cacheGet<T>(
   return undefined;
 }
 
+/**
+ * Return the pending lookup for `key` when one exists, else start `lookup`
+ * and share it until it settles. The value cache is written by `lookup`
+ * itself, so a settled promise is dropped here and later callers hit that
+ * cache instead.
+ */
+function shareInflight<T>(
+  map: Map<string, Promise<T>>,
+  key: string,
+  lookup: () => Promise<T>,
+): Promise<T> {
+  const pending = map.get(key);
+  if (pending !== undefined) return pending;
+  const started = lookup().finally(() => {
+    if (map.get(key) === started) map.delete(key);
+  });
+  map.set(key, started);
+  return started;
+}
+
 export interface ResolvedServer {
   kind: "server";
   /**
@@ -87,12 +116,17 @@ export async function resolveLabel(
   db: ConnectDb,
   options?: { fresh?: boolean },
 ): Promise<ResolvedLabel | null> {
-  const now = Date.now();
-  if (!options?.fresh) {
-    const cached = cacheGet(labelCache, label, now);
-    if (cached !== undefined) return cached;
-  }
+  if (options?.fresh) return queryLabel(label, db);
+  const cached = cacheGet(labelCache, label, Date.now());
+  if (cached !== undefined) return cached;
+  return shareInflight(labelInflight, label, () => queryLabel(label, db));
+}
 
+async function queryLabel(
+  label: string,
+  db: ConnectDb,
+): Promise<ResolvedLabel | null> {
+  const now = Date.now();
   const serverRow = await db
     .select({
       userId: server.userId,
@@ -185,15 +219,26 @@ export async function verifySessionCookie(
   const token = decoded.slice(0, dot);
   const providedSig = decoded.slice(dot + 1);
 
-  const now = Date.now();
   // Cache on the full `token.sig` value, not the token alone: keying on the
   // token would return a cached userId before the signature is checked, so a
   // valid `token` with a forged signature would authenticate (and a forged
   // one would negative-poison the real token). The full-cookie key makes the
   // cache reflect exactly what passed verification.
-  const cached = cacheGet(sessionCache, decoded, now);
+  const cached = cacheGet(sessionCache, decoded, Date.now());
   if (cached !== undefined) return cached;
+  return shareInflight(sessionInflight, decoded, () =>
+    querySession(token, providedSig, decoded, secret, db),
+  );
+}
 
+async function querySession(
+  token: string,
+  providedSig: string,
+  cacheKey: string,
+  secret: string,
+  db: ConnectDb,
+): Promise<string | null> {
+  const now = Date.now();
   const key = await crypto.subtle.importKey(
     "raw",
     new TextEncoder().encode(secret),
@@ -208,7 +253,7 @@ export async function verifySessionCookie(
   );
   const expectedSig = btoa(String.fromCharCode(...new Uint8Array(sigBuf)));
   if (!constantTimeEqual(providedSig, expectedSig)) {
-    sessionCache.set(decoded, { value: null, expires: now + SESSION_TTL_MS });
+    sessionCache.set(cacheKey, { value: null, expires: now + SESSION_TTL_MS });
     return null;
   }
 
@@ -218,7 +263,7 @@ export async function verifySessionCookie(
     .where(and(eq(session.token, token), gt(session.expiresAt, new Date())))
     .get();
   const userId = row?.userId ?? null;
-  sessionCache.set(decoded, { value: userId, expires: now + SESSION_TTL_MS });
+  sessionCache.set(cacheKey, { value: userId, expires: now + SESSION_TTL_MS });
   return userId;
 }
 

apps/connect/src/tunnel-do.ts

@@ -28,6 +28,19 @@ const RESP_HEAD_TIMEOUT_MS = 30_000;
 // dashboard shows accurate presence. Alarm-driven (auto-response pings don't
 // run JS), kept under the 90s offline window.
 const PRESENCE_INTERVAL_MS = 50_000;
+// A tunnel socket counts as live only while its client keeps proving it: the
+// client pings every 20s (@bb/tunnel-client HEARTBEAT_INTERVAL_MS) and the
+// runtime auto-responds, stamping the socket's auto-response timestamp
+// without waking this object. A socket that has neither been accepted nor
+// pinged within this window is a zombie — the client's network died without a
+// close frame ever reaching Cloudflare (laptop lid, cellular handoff, NAT
+// timeout). Its readyState still reads OPEN and send() still succeeds, so
+// without this check every visitor request was proxied into the void and
+// hung for RESP_HEAD_TIMEOUT_MS before a 504, for as long as the TCP zombie
+// lingered (~80s). Two missed pings plus slack; the client's own deadline
+// (45s, checked per 20s tick) makes it redial at its next tick after this
+// window closes, so the offline page shows only briefly on a real drop.
+export const TUNNEL_STALE_MS = 50_000;
 
 // Standard WebSocket readyState numbering (workerd's READY_STATE_OPEN; the
 // constant itself is Cloudflare-only, so tests in Node use the number).
@@ -187,13 +200,60 @@ export class TunnelDO {
     // (abrupt network drop), leaving it tagged but unusable — send() on it
     // throws. And after a reconnect the runtime can briefly list both the
     // stale socket and the replacement. Pick the most recently accepted OPEN
-    // socket; a dead-but-lingering socket must read as "offline", never be
-    // proxied to (that turns every visitor request into an uncaught 1101).
+    // socket that is still heartbeat-fresh; a dead-but-lingering socket must
+    // read as "offline", never be proxied to (that turns every visitor request
+    // into an uncaught 1101, or a 30s hang when the zombie still accepts
+    // writes). Stale sockets are closed here so they stop being listed and
+    // the presence alarm stops advertising a tunnel nobody is behind.
+    const now = Date.now();
     const sockets = this.state.getWebSockets(TUNNEL_TAG);
+    let live: WebSocket | null = null;
+    let closedStale = false;
     for (let i = sockets.length - 1; i >= 0; i--) {
-      if (sockets[i].readyState === WS_READY_STATE_OPEN) return sockets[i];
+      const socket = sockets[i];
+      if (socket.readyState !== WS_READY_STATE_OPEN) continue;
+      if (this.isTunnelFresh(socket, now)) {
+        live ??= socket;
+        continue;
+      }
+      closedStale = true;
+      try {
+        socket.close(1001, "tunnel heartbeat stale");
+      } catch {
+        // Already gone — nothing to close.
+      }
+    }
+    // Every candidate was a zombie: its in-flight streams can never complete
+    // (the client behind it is gone), so fail them now rather than letting
+    // each visitor wait out the response-head timeout. With a live
+    // replacement the streams belong to it (acceptTunnel already abandoned
+    // the old socket's), so they stay.
+    if (live === null && closedStale) {
+      this.abandonStreams(
+        "tunnel disconnected mid-request",
+        "tunnel disconnected",
+      );
     }
-    return null;
+    return live;
+  }
+
+  /**
+   * Freshness = the later of when the socket was accepted and when the runtime
+   * last auto-answered a client heartbeat on it, within TUNNEL_STALE_MS. A
+   * socket with neither timestamp was accepted by a build before this check
+   * existed; stamp it now so it earns one grace window and must then ping.
+   */
+  private isTunnelFresh(socket: WebSocket, now: number): boolean {
+    const acceptedAt = readAcceptedAt(socket.deserializeAttachment());
+    const lastHeartbeat =
+      this.state.getWebSocketAutoResponseTimestamp(socket)?.getTime() ?? null;
+    if (acceptedAt === null && lastHeartbeat === null) {
+      socket.serializeAttachment({ acceptedAt: now });
+      return true;
+    }
+    return (
+      now - Math.max(acceptedAt ?? 0, lastHeartbeat ?? 0) <= TUNNEL_STALE_MS
+    );
   }
 
   /**
@@ -300,6 +360,9 @@ export class TunnelDO {
       void this.state.storage.setAlarm(Date.now() + PRESENCE_INTERVAL_MS);
     }
     const pair = new WebSocketPair();
+    // acceptedAt seeds the freshness check until the first heartbeat lands;
+    // it lives in the attachment so it survives hibernation.
+    pair[1].serializeAttachment({ acceptedAt: Date.now() });
     this.state.acceptWebSocket(pair[1], [TUNNEL_TAG]);
     return new Response(null, { status: 101, webSocket: pair[0] });
   }
@@ -689,6 +752,16 @@ export class TunnelDO {
   }
 }
 
+/** Narrow a tunnel socket's attachment to its accept timestamp, if present. */
+function readAcceptedAt(attachment: unknown): number | null {
+  if (typeof attachment !== "object" || attachment === null) return null;
+  if (!("acceptedAt" in attachment)) return null;
+  const { acceptedAt } = attachment;
+  return typeof acceptedAt === "number" && Number.isFinite(acceptedAt)
+    ? acceptedAt
+    : null;
+}
+
 /** Clamp arbitrary close codes to ones close() is allowed to send. */
 function safeCloseCode(code: number): number {
   return code === 1000 || (code >= 3000 && code <= 4999) ? code : 1000;