perf(api): accelerate cold starts

apps/api/src/deps.ts

@@ -22,18 +22,23 @@ const state: {
 
 export const getPool = () => state.pool
 
-export const initPool = async (): Promise<void> => {
+const initSessionCache = async (): Promise<void> => {
   try {
-    state.sessionCache = new SessionCache({
+    const sessionCache = new SessionCache({
       redisUrl: REDIS_URL,
       ttlSeconds: SESSION_TTL,
     })
+    await sessionCache.connect()
+    state.sessionCache = sessionCache
     console.log("[api] session cache connected  (Tier 2 fast-path enabled)")
   } catch (err) {
+    state.sessionCache = undefined
     console.warn("[api] session cache unavailable — Tier 2 disabled:", err instanceof Error ? err.message : err)
   }
+}
 
-  state.pool = new BrowserPool({
+export const initPool = async (): Promise<void> => {
+  const pool = new BrowserPool({
     poolSize: POOL_SIZE,
     acquireTimeoutMs: ACQUIRE_TIMEOUT_MS,
     recycleAfterTemporaryContexts: RECYCLE_AFTER_TEMPORARY_CONTEXTS,
@@ -42,8 +47,12 @@ export const initPool = async (): Promise<void> => {
     closeTimeoutMs: CLOSE_TIMEOUT_MS,
     launchTimeoutMs: LAUNCH_TIMEOUT_MS,
   })
-  await state.pool.init()
-  state.pool.startHealthCheck()
+  // Publish the pool before its first await. Tier 1 can serve immediately and
+  // browser-backed requests can wait in acquire() while capacity warms.
+  state.pool = pool
+
+  await Promise.all([initSessionCache(), pool.init()])
+  pool.startHealthCheck()
 
   console.log(`[api] ready — all ${POOL_SIZE} browser${POOL_SIZE === 1 ? "" : "s"} warm`)
 }
@@ -51,13 +60,15 @@ export const initPool = async (): Promise<void> => {
 export const getDeps = (): OrchestratorDeps => {
   if (!state.pool) throw new Error("pool not ready")
   const p = state.pool
-  const sc = state.sessionCache
   return {
     acquireBrowser: (d: string, budgetMs?: number) => p.acquire(d, budgetMs),
     releaseBrowser: (id: number, lease?: number) => p.release(id, lease),
-    loadSession: (d: string) => (sc ? sc.load(d).catch(() => undefined) : Promise.resolve(undefined)),
-    saveSession: (d: string, data: SessionData) => (sc ? sc.save(d, data).catch(() => {}) : Promise.resolve()),
-    invalidateSession: (d: string) => (sc ? sc.invalidate(d).catch(() => {}) : Promise.resolve()),
+    loadSession: (d: string) =>
+      state.sessionCache ? state.sessionCache.load(d).catch(() => undefined) : Promise.resolve(undefined),
+    saveSession: (d: string, data: SessionData) =>
+      state.sessionCache ? state.sessionCache.save(d, data).catch(() => {}) : Promise.resolve(),
+    invalidateSession: (d: string) =>
+      state.sessionCache ? state.sessionCache.invalidate(d).catch(() => {}) : Promise.resolve(),
     proxyPool,
     residentialProxyPool,
   }

apps/api/src/index.ts

@@ -32,23 +32,25 @@ console.log(`[api] TRAWL starting on :${PORT}  (pool: ${POOL_SIZE} browser${POOL
 
 const state: { proxyHandle?: MitmProxyHandle } = {}
 
-initPool()
-  .then(() => {
-    if (MITM_PROXY_ENABLED) {
-      state.proxyHandle = startMitmProxy({
-        port: MITM_PROXY_PORT,
-        host: MITM_PROXY_HOST,
-        caDir: MITM_PROXY_CA_DIR,
-        deps: getDeps(),
-        maxTier: MITM_PROXY_MAX_TIER,
-        debug: MITM_PROXY_DEBUG,
-      })
-    }
-  })
-  .catch((err) => {
-    console.error("[api] startup failed:", err)
-    process.exit(1)
+const poolReady = initPool()
+
+// Tier 0 does not need a browser, and browser-backed requests already have a
+// bounded acquire queue. Start accepting proxy traffic while the pool warms.
+if (MITM_PROXY_ENABLED) {
+  state.proxyHandle = startMitmProxy({
+    port: MITM_PROXY_PORT,
+    host: MITM_PROXY_HOST,
+    caDir: MITM_PROXY_CA_DIR,
+    deps: getDeps(),
+    maxTier: MITM_PROXY_MAX_TIER,
+    debug: MITM_PROXY_DEBUG,
   })
+}
+
+poolReady.catch((err) => {
+  console.error("[api] startup failed:", err)
+  process.exit(1)
+})
 
 registerLifecycleHandlers({
   onShutdown: async () => {

apps/api/src/proxy/__tests__/directForward.test.ts

@@ -1,4 +1,6 @@
 import { afterAll, describe, expect, test } from "bun:test"
+import { once } from "node:events"
+import net from "node:net"
 import { gzipSync } from "node:zlib"
 import { directForwardHttp } from "../directForward"
 
@@ -137,6 +139,43 @@ describe("directForwardHttp — Range / 206 Partial Content", () => {
 })
 
 describe("directForwardHttp — buffered by default", () => {
+  test("skips 103 Early Hints and escalates cf-mitigated without waiting for an open body", async () => {
+    const sockets = new Set<net.Socket>()
+    const hangingServer = net.createServer((socket) => {
+      sockets.add(socket)
+      socket.once("close", () => sockets.delete(socket))
+      socket.write(
+        "HTTP/1.1 103 Early Hints\r\nLink: </style.css>; rel=preload\r\n\r\n" +
+          "HTTP/1.1 403 Forbidden\r\nContent-Type: text/html\r\nCF-Mitigated: Challenge\r\nConnection: keep-alive\r\n\r\n",
+      )
+    })
+    hangingServer.listen(0, "127.0.0.1")
+    await once(hangingServer, "listening")
+    const address = hangingServer.address()
+    if (!address || typeof address === "string") throw new Error("test server did not bind a TCP port")
+
+    try {
+      const startedAt = performance.now()
+      const result = await directForwardHttp({
+        url: `http://127.0.0.1:${address.port}/challenge`,
+        method: "GET",
+        headers: {},
+        timeoutMs: 2_000,
+      })
+
+      expect(performance.now() - startedAt).toBeLessThan(500)
+      expect(result.mode).toBe("buffer")
+      if (result.mode !== "buffer") return
+      expect(result.status).toBe(403)
+      expect(result.challengeDetected).toBe(true)
+      expect(result.headers["cf-mitigated"]).toBe("Challenge")
+      expect(result.body.length).toBe(0)
+    } finally {
+      for (const socket of sockets) socket.destroy()
+      await new Promise<void>((resolve, reject) => hangingServer.close((error) => (error ? reject(error) : resolve())))
+    }
+  })
+
   test("buffers and de-chunks small HTML instead of treating it as a stream", async () => {
     const result = await directForwardHttp({
       url: `${baseUrl}/chunked-html`,

apps/api/src/proxy/directForward.ts

@@ -192,8 +192,9 @@ async function readHttpResponse(
   socket: net.Socket,
   url: string,
   skipChallengeDetection: boolean,
+  initialResponseBytes: Buffer = Buffer.alloc(0),
 ): Promise<ForwardResult> {
-  const headerBuf = await readUpTo(socket, MAX_HEADER_BYTES, "\r\n\r\n")
+  const headerBuf = await readUpTo(socket, MAX_HEADER_BYTES, "\r\n\r\n", initialResponseBytes)
   if (!headerBuf.found) {
     socket.destroy()
     return { mode: "error", error: new Error("upstream response headers exceeded 64 KiB") }
@@ -209,6 +210,13 @@ async function readHttpResponse(
   }
   const status = Number(statusMatch[1])
 
+  // Informational responses (most commonly Cloudflare's 103 Early Hints) are
+  // followed by another HTTP response on the same connection. Do not treat the
+  // final response headers/body as the body of the 1xx response.
+  if (status >= 100 && status < 200 && status !== 101) {
+    return readHttpResponse(socket, url, skipChallengeDetection, headerBuf.leftover)
+  }
+
   const headers: Record<string, string> = {}
   for (const line of lines.slice(1)) {
     const idx = line.indexOf(":")
@@ -225,6 +233,25 @@ async function readHttpResponse(
   const contentType = headers["content-type"] ?? "application/octet-stream"
   const rawLen = headers["content-length"]
   const contentLength = rawLen ? Number(rawLen) : undefined
+
+  // Cloudflare defines `cf-mitigated: challenge` as an authoritative Challenge
+  // Page signal. Escalate as soon as the headers arrive instead of waiting for
+  // an unbounded/keep-alive response body to finish (or hit the 30s socket timeout).
+  // Body-based detection below remains the fallback for challenge variants that
+  // do not send this header.
+  if (!skipChallengeDetection && detectChallengeType("", headers) === "cloudflare-interstitial") {
+    socket.destroy()
+    return {
+      mode: "buffer",
+      status,
+      headers,
+      contentType,
+      contentLength,
+      body: Buffer.alloc(0),
+      challengeDetected: true,
+    }
+  }
+
   const streamDecision = shouldStream(url, contentLength, contentType)
 
   const isChunked = (headers["transfer-encoding"] ?? "").toLowerCase().includes("chunked")
@@ -331,14 +358,13 @@ async function readUpTo(
   socket: net.Socket,
   maxBytes: number,
   delimiter: string,
+  initial: Buffer = Buffer.alloc(0),
 ): Promise<{ found: boolean; text: string; leftover?: Buffer }> {
   return new Promise((resolve, reject) => {
-    const chunks: Buffer[] = []
-    let total = 0
+    const chunks: Buffer[] = initial.length > 0 ? [initial] : []
+    let total = initial.length
 
-    const onData = (chunk: Buffer) => {
-      chunks.push(chunk)
-      total += chunk.length
+    const inspect = (): boolean => {
       const buf = Buffer.concat(chunks, total)
       const idx = buf.indexOf(delimiter)
       if (idx >= 0) {
@@ -351,13 +377,20 @@ async function readUpTo(
         const endIdx = idx + delimiter.length
         const leftover = endIdx < buf.length ? buf.subarray(endIdx) : undefined
         resolve({ found: true, text: buf.subarray(0, endIdx).toString("latin1"), leftover })
-        return
+        return true
       }
       if (total > maxBytes) {
         socket.off("data", onData)
         socket.off("error", onError)
         resolve({ found: false, text: buf.toString("latin1") })
+        return true
       }
+      return false
+    }
+    const onData = (chunk: Buffer) => {
+      chunks.push(chunk)
+      total += chunk.length
+      inspect()
     }
     const onError = (err: Error) => {
       socket.off("data", onData)
@@ -367,6 +400,7 @@ async function readUpTo(
 
     socket.on("data", onData)
     socket.once("error", onError)
+    if (!inspect()) socket.resume()
   })
 }