diff --git a/apps/connect/src/session.test.ts b/apps/connect/src/session.test.ts index 4596464019..fbc94ecb51 100644 --- a/apps/connect/src/session.test.ts +++ b/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); + }); +}); diff --git a/apps/connect/src/session.ts b/apps/connect/src/session.ts index be7731a948..25d470650a 100644 --- a/apps/connect/src/session.ts +++ b/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 { } const labelCache = new Map>(); const sessionCache = new Map>(); +const labelInflight = new Map>(); +const sessionInflight = new Map>(); function cacheGet( map: Map>, @@ -37,6 +46,26 @@ function cacheGet( 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( + map: Map>, + key: string, + lookup: () => Promise, +): Promise { + 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 { - 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 { + 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 { + 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; } diff --git a/apps/connect/src/tunnel-do.ts b/apps/connect/src/tunnel-do.ts index 7eca8777af..aa9606e79a 100644 --- a/apps/connect/src/tunnel-do.ts +++ b/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; diff --git a/apps/connect/src/worker.test.ts b/apps/connect/src/worker.test.ts index e6a9bdd078..03b0b4a361 100644 --- a/apps/connect/src/worker.test.ts +++ b/apps/connect/src/worker.test.ts @@ -163,7 +163,11 @@ import { SECURE_DESKTOP_SESSION_COOKIE as DESKTOP_SESSION_COOKIE } from "./cloud import { handleAssignMachineLabel } from "./machine-label.js"; import { serveWithCache } from "./cache.js"; import worker, { offlinePage, relativeTime, wantsHtml } from "./worker.js"; -import { TUNNEL_OFFLINE_HEADER, TunnelDO } from "./tunnel-do.js"; +import { + TUNNEL_OFFLINE_HEADER, + TUNNEL_STALE_MS, + TunnelDO, +} from "./tunnel-do.js"; const mockParseCookie = vi.mocked(parseCookie); const mockResolveLabel = vi.mocked(resolveLabel); @@ -1095,6 +1099,120 @@ function offlineDoResponse(): Response { }); } +// ── gate lookup overlap ───────────────────────────────────────────────────── +// +// Label resolution and session verification are independent D1 round trips. +// Awaiting them back to back put two single-region D1 RTTs on every uncached +// visitor request; the gate must issue the session check before the label +// answer arrives. + +describe("gate lookup overlap", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockParseCookie.mockReturnValue("session-token"); + }); + + it("verifies the visitor session while the label lookup is still pending", async () => { + let resolveLabel!: (value: ReturnType) => void; + mockResolveLabel.mockReturnValue( + new Promise((resolve) => { + resolveLabel = resolve; + }), + ); + let resolveSession!: (value: string) => void; + mockVerifySession.mockReturnValue( + new Promise((resolve) => { + resolveSession = resolve; + }), + ); + const { env, ctx, captured } = makeEnv(() => new Response("origin")); + const pending = worker.fetch( + visitorRequest("sawyer.getbb.app", "/app.js"), + env as never, + ctx, + ); + // Let the handler run up to its first await. + await Promise.resolve(); + expect(mockVerifySession).toHaveBeenCalledTimes(1); + expect(mockResolveLabel).toHaveBeenCalledTimes(1); + + resolveSession(OWNER); + resolveLabel(resolvedServer()); + const res = await pending; + expect(res.status).toBe(200); + expect(captured).toHaveLength(1); + }); + + it("does not verify a session on a tunnel dial", async () => { + mockResolveLabel.mockResolvedValue(resolvedServer()); + const { env, ctx } = makeEnv(() => new Response("upgraded")); + await worker.fetch( + visitorRequest("sawyer.getbb.app", "/__tunnel?v=1", { + headers: { authorization: "Bearer nope" }, + }), + env as never, + ctx, + ); + expect(mockVerifySession).not.toHaveBeenCalled(); + }); + + it("keeps the overlapped verification alive past an early return", async () => { + // A machine's bare label answers from the label cache before the session + // read lands. Without waitUntil the request context would close on that + // pending D1 read, and session.ts shares the pending lookup with the next + // request for the same cookie — which would then wait on it forever. + let resolveSession!: (value: string) => void; + mockVerifySession.mockReturnValue( + new Promise((resolve) => { + resolveSession = resolve; + }), + ); + mockResolveLabel.mockResolvedValue(resolvedMachine()); + const { env, ctx, captured } = makeEnv(() => new Response("origin")); + const res = await worker.fetch( + visitorRequest("mac.getbb.app", "/"), + env as never, + ctx, + ); + expect(res.status).toBe(200); + expect(captured).toHaveLength(0); + expect(mockVerifySession).toHaveBeenCalledTimes(1); + const waited = vi.mocked(ctx.waitUntil).mock.calls; + expect(waited).toHaveLength(1); + let settled = false; + void (waited[0][0] as Promise).then(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + resolveSession(OWNER); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(settled).toBe(true); + }); + + it("returns the 404 for an unknown label even when the session check fails", async () => { + // The early return must not leave the overlapped verification as an + // unhandled rejection. + mockResolveLabel.mockResolvedValue(null); + mockVerifySession.mockRejectedValue(new Error("D1 unavailable")); + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + const { env, ctx } = makeEnv(() => new Response("origin")); + const res = await worker.fetch( + visitorRequest("ghost.getbb.app", "/"), + env as never, + ctx, + ); + expect(res.status).toBe(404); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(unhandled).not.toHaveBeenCalled(); + } finally { + process.off("unhandledRejection", unhandled); + } + }); +}); + describe("gate offline page", () => { beforeEach(() => { vi.clearAllMocks(); @@ -1229,6 +1347,16 @@ describe("gate page helpers", () => { expect(res.headers.get("content-type")).toContain("text/html"); expect(await res.text()).toContain("Retry now"); }); + + it("gate pages load no third-party stylesheet or font before first paint", async () => { + // A cross-origin font stylesheet is render-blocking; on a phone on + // cellular it delayed the offline/sign-in card by a full RTT or more. + const html = await offlinePage(null, "server").text(); + expect(html).not.toContain("fonts.googleapis.com"); + expect(html).not.toContain("fonts.gstatic.com"); + expect(html).not.toMatch(/ void; + /** Stamp the runtime's last auto-response (client heartbeat) time for `ws`. */ + setHeartbeat: (ws: WebSocket, at: Date) => void; storage: Map; restore: Promise; api: DurableObjectState; @@ -1252,6 +1382,7 @@ type MockState = { function mockDoState(initialStorage: Record = {}): MockState { const storage = new Map(Object.entries(initialStorage)); const entries: Array<{ ws: WebSocket; tags: string[] }> = []; + const heartbeats = new Map(); let restore = Promise.resolve(); const api = { getWebSockets: (tag?: string) => @@ -1264,6 +1395,8 @@ function mockDoState(initialStorage: Record = {}): MockState { entries.push({ ws, tags }); }, setWebSocketAutoResponse: vi.fn(), + getWebSocketAutoResponseTimestamp: (ws: WebSocket) => + heartbeats.get(ws) ?? null, blockConcurrencyWhile: (fn: () => Promise) => { restore = fn(); return restore; @@ -1283,6 +1416,9 @@ function mockDoState(initialStorage: Record = {}): MockState { addSocket: (ws: WebSocket, tags: string[]) => { entries.push({ ws, tags }); }, + setHeartbeat: (ws: WebSocket, at: Date) => { + heartbeats.set(ws, at); + }, storage, get restore() { return restore; @@ -1303,11 +1439,16 @@ function makeDoEnv() { function fakeTunnelSocket( send?: (data: ArrayBuffer | ArrayBufferView | string) => void, readyState = 1, // READY_STATE_OPEN + attachment: unknown = null, ) { + let stored = attachment; return { send: send ?? vi.fn(), close: vi.fn(), - deserializeAttachment: () => null, + deserializeAttachment: () => stored, + serializeAttachment: (value: unknown) => { + stored = value; + }, readyState, } as unknown as WebSocket; } @@ -1744,3 +1885,188 @@ describe("TunnelDO dead tunnel sockets", () => { expect(res.headers.get("x-bb-tunnel-offline")).toBe("1"); }); }); + +// ── TunnelDO zombie tunnel sockets ────────────────────────────────────────── +// +// A tunnel whose client vanished without a close frame (laptop lid, cellular +// handoff, NAT timeout) keeps readyState OPEN and accepts send() for as long +// as the TCP zombie lingers. Every visitor request proxied into it hung for +// the full response-head timeout before a 504. Liveness now also requires a +// recent accept or heartbeat auto-response. + +describe("TunnelDO zombie tunnel sockets", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-19T12:00:00Z")); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + function acceptedAgo(ms: number): { acceptedAt: number } { + return { acceptedAt: Date.now() - ms }; + } + + it("answers 503 offline and closes a socket that has not pinged within the window", async () => { + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + const zombie = fakeTunnelSocket( + captureSent(sent), + 1, + acceptedAgo(TUNNEL_STALE_MS + 1_000), + ); + state.addSocket(zombie, ["tunnel"]); + state.setHeartbeat(zombie, new Date(Date.now() - TUNNEL_STALE_MS - 500)); + + const res = await dob.fetch(new Request("https://do.internal/")); + expect(res.status).toBe(503); + expect(res.headers.get(TUNNEL_OFFLINE_HEADER)).toBe("1"); + expect(sent).toHaveLength(0); + expect(vi.mocked(zombie.close)).toHaveBeenCalledWith( + 1001, + "tunnel heartbeat stale", + ); + }); + + it("treats a long-lived socket as live while heartbeats keep arriving", async () => { + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + const tunnel = fakeTunnelSocket( + captureSent(sent), + 1, + acceptedAgo(6 * 60 * 60_000), + ); + state.addSocket(tunnel, ["tunnel"]); + state.setHeartbeat(tunnel, new Date(Date.now() - 15_000)); + + void dob.fetch(new Request("https://do.internal/app.js")); + expect(sent).toHaveLength(1); + expect(decodeFrame(sent[0]).type).toBe("open-http"); + expect(vi.mocked(tunnel.close)).not.toHaveBeenCalled(); + }); + + it("treats a just-accepted socket as live before its first heartbeat", async () => { + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + const tunnel = fakeTunnelSocket(captureSent(sent), 1, acceptedAgo(5_000)); + state.addSocket(tunnel, ["tunnel"]); + + void dob.fetch(new Request("https://do.internal/app.js")); + expect(sent).toHaveLength(1); + expect(decodeFrame(sent[0]).type).toBe("open-http"); + }); + + it("gives a socket accepted before freshness tracking one grace window", async () => { + // No acceptedAt and no heartbeat: stamp now, proxy, and require a ping + // within the window from here on. + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + const legacy = fakeTunnelSocket(captureSent(sent)); + state.addSocket(legacy, ["tunnel"]); + + void dob.fetch(new Request("https://do.internal/a")); + expect(sent).toHaveLength(1); + expect(legacy.deserializeAttachment()).toEqual({ acceptedAt: Date.now() }); + + vi.advanceTimersByTime(TUNNEL_STALE_MS + 1); + const res = await dob.fetch(new Request("https://do.internal/b")); + expect(res.status).toBe(503); + expect(vi.mocked(legacy.close)).toHaveBeenCalledWith( + 1001, + "tunnel heartbeat stale", + ); + }); + + it("fails requests already in flight on the zombie instead of waiting out the timeout", async () => { + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + // Accepted 45s ago, never pinged: still inside the window, so a request + // arriving now is proxied — but the client is already gone. + const tunnel = fakeTunnelSocket(captureSent(sent), 1, acceptedAgo(45_000)); + state.addSocket(tunnel, ["tunnel"]); + const visitor = fakeTunnelSocket(); + state.addSocket(visitor, ["visitor:7"]); + + const stranded = dob.fetch(new Request("https://do.internal/api/threads")); + expect(sent).toHaveLength(1); + + // Six seconds later the window closes; the next request must both get the + // offline answer and take the stranded one down with it (well before its + // own 30s response-head timeout). + vi.advanceTimersByTime(TUNNEL_STALE_MS - 45_000 + 1); + const next = await dob.fetch(new Request("https://do.internal/api/next")); + expect(next.status).toBe(503); + + const strandedResponse = await stranded; + expect(strandedResponse.status).toBe(502); + expect(await strandedResponse.text()).toContain( + "tunnel disconnected mid-request", + ); + expect(vi.mocked(visitor.close)).toHaveBeenCalledWith( + 1001, + "tunnel disconnected", + ); + }); + + it("routes around a zombie to a fresh replacement without abandoning its streams", async () => { + const sent: Uint8Array[] = []; + const state = mockDoState({ protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + const zombie = fakeTunnelSocket( + captureSent([]), + 1, + acceptedAgo(TUNNEL_STALE_MS + 60_000), + ); + state.addSocket(zombie, ["tunnel"]); + const fresh = fakeTunnelSocket(captureSent(sent), 1, acceptedAgo(1_000)); + state.addSocket(fresh, ["tunnel"]); + const visitor = fakeTunnelSocket(); + state.addSocket(visitor, ["visitor:9"]); + + const pending = dob.fetch(new Request("https://do.internal/app.js")); + expect(sent).toHaveLength(1); + expect(vi.mocked(zombie.close)).toHaveBeenCalledWith( + 1001, + "tunnel heartbeat stale", + ); + expect(vi.mocked(visitor.close)).not.toHaveBeenCalled(); + + const streamId = openHttpStreamId(sent, 0); + dob.webSocketMessage( + fresh, + frameBuffer({ type: "resp-head", streamId, status: 204, headers: [] }), + ); + expect((await pending).status).toBe(204); + }); + + it("stops advertising presence for a zombie on the alarm", async () => { + const run = vi.fn(async () => {}); + const where = vi.fn(() => ({ run })); + const set = vi.fn(() => ({ where })); + const update = vi.fn(() => ({ set })); + vi.mocked(drizzle).mockReturnValue({ update } as never); + const state = mockDoState({ serverId: "srv1", protocolVersion: 1 }); + const dob = new TunnelDO(state.api, makeDoEnv()); + await state.restore; + state.addSocket( + fakeTunnelSocket(undefined, 1, acceptedAgo(TUNNEL_STALE_MS + 1)), + ["tunnel"], + ); + + await dob.alarm(); + + expect(run).not.toHaveBeenCalled(); + expect(state.storage.has("serverId")).toBe(false); + }); +}); diff --git a/apps/connect/src/worker.ts b/apps/connect/src/worker.ts index 9935622663..a6ec7f7631 100644 --- a/apps/connect/src/worker.ts +++ b/apps/connect/src/worker.ts @@ -1,5 +1,10 @@ import { drizzle } from "drizzle-orm/d1"; -import { RESERVED_HANDLES, parseVisitorHost, schema } from "@bb/connect-db"; +import { + RESERVED_HANDLES, + parseVisitorHost, + schema, + type ConnectDb, +} from "@bb/connect-db"; import { TUNNEL_OFFLINE_HEADER, TunnelDO, type Env } from "./tunnel-do.js"; import { parseCookie, @@ -50,8 +55,8 @@ function text(body: string, status: number): Response { }); } -// Matches the bb dashboard's visual language (Inter, --canvas/--ink tokens, -// dark primary button, bb logo) since this plain worker can't bundle React. +// Matches the bb dashboard's visual language (--canvas/--ink tokens, dark +// primary button, bb logo) since this plain worker can't bundle React. export function dashboardSignInUrl(appUrl: string, returnTo: string): string { const url = new URL("/dashboard", appUrl); url.searchParams.set("returnTo", returnTo); @@ -60,9 +65,12 @@ export function dashboardSignInUrl(appUrl: string, returnTo: string): string { // Shared gate-page shell. The 401 sign-in and 503 offline pages render through // one template so they can never drift apart the way the dashboard and gate -// once did. Matches the bb dashboard's visual language (Inter, --canvas/--ink -// tokens derived from two anchors, dark-mode media query, inlined bb icon, -// centered card) since this plain worker can't bundle React. +// once did. Matches the bb dashboard's visual language (--canvas/--ink tokens +// derived from two anchors, dark-mode media query, inlined bb icon, centered +// card) since this plain worker can't bundle React. The pages use the system +// font stack on purpose: a third-party font stylesheet is a render-blocking +// cross-origin fetch on the phone's first paint, and the gate is what a +// visitor sees on a cold, often cellular, connection. const GATE_STYLE = ` :root{--canvas:oklch(1 0 0);--ink:oklch(0.3211 0 0); --muted:color-mix(in oklch,var(--ink) 55%,var(--canvas)); @@ -76,7 +84,7 @@ const GATE_STYLE = ` *{box-sizing:border-box} body{margin:0;min-height:100dvh;display:flex;align-items:center;justify-content:center; background:var(--canvas);color:var(--ink); - font:15px/1.6 "Inter",-apple-system,system-ui,sans-serif;-webkit-font-smoothing:antialiased} + font:15px/1.6 -apple-system,system-ui,"Segoe UI",sans-serif;-webkit-font-smoothing:antialiased} .wrap{width:100%;max-width:420px;padding:24px} .brand{display:flex;align-items:center;gap:10px;margin-bottom:18px} .brand img{width:28px;height:28px} @@ -89,7 +97,7 @@ const GATE_STYLE = ` code{font-family:"Fira Code",ui-monospace,monospace;font-size:.92em} .btn{display:flex;align-items:center;justify-content:center;width:100%;padding:11px 16px; border-radius:8px;border:1px solid var(--border);background:var(--card);color:var(--ink); - font:500 14px/1 "Inter",-apple-system,system-ui,sans-serif;text-decoration:none;cursor:pointer} + font:500 14px/1 -apple-system,system-ui,"Segoe UI",sans-serif;text-decoration:none;cursor:pointer} .btn.primary{background:var(--ink);border-color:var(--ink);color:var(--canvas)} .glyph{width:34px;height:34px;border-radius:999px;background:var(--warn-bg); border:1px solid var(--warn-border);color:var(--warn); @@ -111,9 +119,6 @@ function gatePage( ` ${refresh}bb connect - - -
bb
bb connect
Your bb, reachable anywhere
@@ -259,6 +264,42 @@ function isHostManagementMutation(request: Request, pathname: string): boolean { ); } +interface VisitorAuth { + sessionUserId: string | null; + desktopUserId: string | null; +} + +/** + * Start visitor cookie verification without awaiting it, so it overlaps the + * label lookup. Returns null when the request carries no visitor cookie + * (nothing to verify). A caller that returns early (404 label, machine page) + * leaves the promise to settle in the background: it is registered with + * `ctx.waitUntil` so the request context outlives it (a D1 read whose request + * context ends never settles, and the session module shares the pending + * lookup with later requests for the same cookie, which would then wait on + * it forever), and a D1 failure is observed here rather than surfacing as an + * unhandled rejection; the visitor path awaits the same promise and still + * sees the failure. + */ +function startVisitorAuth( + cookie: string | null, + desktopCookie: string | null, + secret: string, + db: ConnectDb, + ctx: ExecutionContext, +): Promise | null { + if (!cookie && !desktopCookie) return null; + const auth = Promise.all([ + cookie ? verifySessionCookie(cookie, secret, db) : null, + desktopCookie ? verifyDesktopSessionCookie(desktopCookie, secret) : null, + ]).then(([sessionUserId, desktopUserId]) => ({ + sessionUserId, + desktopUserId, + })); + ctx.waitUntil(auth.catch(() => undefined)); + return auth; +} + /** Cache namespace for a resolved routing key plus optional share target. */ export function cacheNamespace( routingKey: string, @@ -315,6 +356,26 @@ export default { // avoids both stale credentials and a cached negative immediately after a // machine label is assigned. const isTunnelDial = url.pathname === "/__tunnel"; + // Visitor cookies do not depend on the label, so their verification starts + // now and overlaps the label lookup: on a cold isolate both are D1 round + // trips (single-region, ~100+ ms each from a phone far from the DB), and + // running them back to back doubled the gate's cost on every uncached + // request. Tunnel dials never carry a session, so they skip the read. + const cookieHeader = request.headers.get("cookie"); + const cookie = parseCookie(cookieHeader, runtime.sessionCookieName); + const desktopCookie = parseCookie( + cookieHeader, + runtime.desktopSessionCookieName, + ); + const visitorAuth = isTunnelDial + ? null + : startVisitorAuth( + cookie, + desktopCookie, + env.BETTER_AUTH_SECRET, + db, + ctx, + ); const resolved = await resolveLabel( label, db, @@ -430,21 +491,9 @@ export default { // Visitor request — require a session owned by this label's account. // Identical auth for bare-label and share hosts. Because this check passed, // only the owner ever reaches the DO below (and thus its offline 503). - const cookieHeader = request.headers.get("cookie"); - const cookie = parseCookie(cookieHeader, runtime.sessionCookieName); - const desktopCookie = parseCookie( - cookieHeader, - runtime.desktopSessionCookieName, - ); const appUrl = runtime.accountAppUrl; - if (!cookie && !desktopCookie) - return signInPage(label, appUrl, url.toString()); - const sessionUserId = cookie - ? await verifySessionCookie(cookie, env.BETTER_AUTH_SECRET, db) - : null; - const desktopUserId = desktopCookie - ? await verifyDesktopSessionCookie(desktopCookie, env.BETTER_AUTH_SECRET) - : null; + if (visitorAuth === null) return signInPage(label, appUrl, url.toString()); + const { sessionUserId, desktopUserId } = await visitorAuth; if (!sessionUserId && !desktopUserId) { return signInPage(label, appUrl, url.toString()); } diff --git a/apps/host-daemon/src/ws.d.ts b/apps/host-daemon/src/ws.d.ts index 54213d36c5..7ec42425e7 100644 --- a/apps/host-daemon/src/ws.d.ts +++ b/apps/host-daemon/src/ws.d.ts @@ -19,7 +19,10 @@ declare module "ws" { static readonly OPEN: number; readonly readyState: number; readonly protocol: string; - send(data: string | Buffer | Uint8Array): void; + send( + data: string | Buffer | Uint8Array, + options?: { compress?: boolean }, + ): void; close(code?: number, reason?: string): void; terminate(): void; on(event: "open", listener: () => void): this; diff --git a/packages/tunnel-client/src/session.ts b/packages/tunnel-client/src/session.ts index 61417e617d..76f243a59f 100644 --- a/packages/tunnel-client/src/session.ts +++ b/packages/tunnel-client/src/session.ts @@ -22,7 +22,13 @@ import { headersForLoopbackRequest } from "./headers.js"; import type { TunnelClientLogger } from "./logger.js"; const HEARTBEAT_INTERVAL_MS = 20_000; -const HEARTBEAT_DEADLINE_MS = 60_000; +// Two missed acks plus slack, evaluated on each 20s tick: a dead link is +// declared at the third tick after the last ack (60s; it used to be the +// fourth, 80s). The relay stops treating the socket as live 50s after the +// last heartbeat it answered (TUNNEL_STALE_MS in apps/connect), so the +// client redials within one tick of the relay showing its visitors the +// offline page instead of ~30s later. +const HEARTBEAT_DEADLINE_MS = 45_000; const UNREGISTERED_PORT_BODY = "this port is not shared"; const textEncoder = new TextEncoder(); @@ -89,6 +95,20 @@ function isInitialThreadLoad(path: string): boolean { return !new URL(path, "http://bb.local").searchParams.has("afterSequence"); } +/** True when the origin already encoded the body (anything but identity). */ +export function isPrecompressedResponse( + contentEncoding: string | string[] | undefined, +): boolean { + if (contentEncoding === undefined) return false; + const value = Array.isArray(contentEncoding) + ? contentEncoding.join(",") + : contentEncoding; + return value + .split(",") + .map((token) => token.trim().toLowerCase()) + .some((token) => token !== "" && token !== "identity"); +} + function roundDurationMs(durationMs: number): number { return Math.round(durationMs * 10) / 10; } @@ -210,9 +230,18 @@ export class TunnelSession { this.setRemoteClients(Math.max(0, this.remoteClientCount + delta)); } - private send(frame: Frame): void { + /** + * `compress: false` opts a frame out of permessage-deflate when the dial + * negotiated it. Body chunks whose origin response is already encoded + * (brotli/gzip static assets, gzip API JSON) gain nothing from a second + * deflate pass and pay per-chunk CPU plus a few bytes of expansion; identity + * bodies and control frames keep the default. + */ + private send(frame: Frame, options: { compress?: boolean } = {}): void { if (this.options.tunnel.readyState === NodeWebSocket.OPEN) { - this.options.tunnel.send(encodeFrame(frame)); + this.options.tunnel.send(encodeFrame(frame), { + compress: options.compress ?? true, + }); } } @@ -320,6 +349,9 @@ export class TunnelSession { }); const originTtfbMs = performance.now() - startedAt; const respHeaders = responseHeaderPairs(res); + const compress = !isPrecompressedResponse( + res.headers["content-encoding"], + ); const initialThreadLoad = isInitialThreadLoad(meta.path); if (initialThreadLoad) { respHeaders.push([ @@ -338,7 +370,9 @@ export class TunnelSession { const value = chunk instanceof Uint8Array ? chunk : Buffer.from(String(chunk)); responseBytes += value.byteLength; - for (const frame of chunkBody(streamId, value)) this.send(frame); + for (const frame of chunkBody(streamId, value)) { + this.send(frame, { compress }); + } } this.send({ type: "body-end", streamId }); if (initialThreadLoad) { diff --git a/packages/tunnel-client/test/session-compress.test.ts b/packages/tunnel-client/test/session-compress.test.ts new file mode 100644 index 0000000000..6eabba9dac --- /dev/null +++ b/packages/tunnel-client/test/session-compress.test.ts @@ -0,0 +1,125 @@ +import { EventEmitter } from "node:events"; +import { createServer, type Server } from "node:http"; +import { gzipSync } from "node:zlib"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { WebSocket as NodeWebSocket } from "ws"; +import { decodeFrame, encodeFrame, type Frame } from "@bb/tunnel-contract"; +import { TunnelSession, isPrecompressedResponse } from "../src/session.js"; + +// The tunnel dial negotiates permessage-deflate. Static assets and API JSON +// arrive from the origin already brotli/gzip encoded; deflating those chunks +// again costs CPU per chunk and grows them. Only identity bodies (and every +// control frame) should ride the extension. + +interface SentMessage { + frame: Frame; + compress: boolean | undefined; +} + +class FakeTunnel extends EventEmitter { + readyState: number = NodeWebSocket.OPEN; + readonly sent: SentMessage[] = []; + send(data: Uint8Array, options?: { compress?: boolean }): void { + this.sent.push({ frame: decodeFrame(data), compress: options?.compress }); + } + terminate(): void { + this.readyState = NodeWebSocket.CLOSED; + } +} + +let server: Server; +let origin: string; + +beforeAll(async () => { + server = createServer((request, response) => { + if (request.url === "/precompressed.js") { + response.writeHead(200, { + "content-type": "text/javascript", + "content-encoding": "gzip", + }); + response.end(gzipSync(Buffer.from("console.log('hi')".repeat(64)))); + return; + } + response.writeHead(200, { "content-type": "text/plain" }); + response.end("plain body ".repeat(64)); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const address = server.address(); + if (address === null || typeof address === "string") { + throw new Error("test server has no port"); + } + origin = `http://127.0.0.1:${address.port}`; +}); + +afterAll(async () => { + await new Promise((resolve) => server.close(() => resolve())); +}); + +function startSession(): FakeTunnel { + const tunnel = new FakeTunnel(); + const session = new TunnelSession({ + // The session only uses the EventEmitter + send/readyState surface. + tunnel: tunnel as unknown as NodeWebSocket, + log: { info: vi.fn(), warn: vi.fn() }, + resolveOrigin: () => ({ + kind: "ok", + resolved: { origin, publicOrigin: "https://sawyer.getbb.app" }, + }), + }); + session.start(); + return tunnel; +} + +async function relay(tunnel: FakeTunnel, path: string): Promise { + tunnel.emit( + "message", + Buffer.from( + encodeFrame({ + type: "open-http", + streamId: 1, + method: "GET", + path, + headers: [], + hasBody: false, + }), + ), + true, + ); + await vi.waitFor(() => { + expect(tunnel.sent.some((m) => m.frame.type === "body-end")).toBe(true); + }); + return tunnel.sent; +} + +describe("TunnelSession body-chunk compression", () => { + it("opts precompressed origin bodies out of permessage-deflate", async () => { + const sent = await relay(startSession(), "/precompressed.js"); + const chunks = sent.filter((m) => m.frame.type === "body-chunk"); + expect(chunks.length).toBeGreaterThan(0); + expect(chunks.every((m) => m.compress === false)).toBe(true); + // Control frames still compress. + const head = sent.find((m) => m.frame.type === "resp-head"); + expect(head?.compress).toBe(true); + const end = sent.find((m) => m.frame.type === "body-end"); + expect(end?.compress).toBe(true); + }); + + it("keeps deflate for identity bodies", async () => { + const sent = await relay(startSession(), "/plain.txt"); + const chunks = sent.filter((m) => m.frame.type === "body-chunk"); + expect(chunks.length).toBeGreaterThan(0); + expect(chunks.every((m) => m.compress === true)).toBe(true); + }); +}); + +describe("isPrecompressedResponse", () => { + it("treats only identity (or absent) encodings as compressible", () => { + expect(isPrecompressedResponse(undefined)).toBe(false); + expect(isPrecompressedResponse("identity")).toBe(false); + expect(isPrecompressedResponse("")).toBe(false); + expect(isPrecompressedResponse("br")).toBe(true); + expect(isPrecompressedResponse("GZIP")).toBe(true); + expect(isPrecompressedResponse("identity, gzip")).toBe(true); + expect(isPrecompressedResponse(["gzip"])).toBe(true); + }); +});