fix(server): answer requests that arrive before the HTTP handler attaches

apps/server/src/httpResponseErrorGuard.test.ts

@@ -1,36 +1,80 @@
+// Exercise the raw Node HTTP boundary before Effect attaches serving handlers.
 // @effect-diagnostics nodeBuiltinImport:off
 import * as NodeHttp from "node:http";
 import * as NodeNet from "node:net";
 import { afterEach, describe, expect, it } from "vite-plus/test";
 
-import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts";
+import * as HttpResponseErrorGuard from "./httpResponseErrorGuard.ts";
+
+const { guardHttpResponseWriteErrors } = HttpResponseErrorGuard;
 
 const servers: NodeHttp.Server[] = [];
+const requests: NodeHttp.ClientRequest[] = [];
+const sockets: NodeNet.Socket[] = [];
 
 function listen(server: NodeHttp.Server): Promise<number> {
   servers.push(server);
-  return new Promise((resolve) => {
+  server.on("connection", (socket) => sockets.push(socket));
+  return new Promise((resolve, reject) => {
+    server.once("error", reject);
     server.listen(0, "127.0.0.1", () => {
       resolve((server.address() as NodeNet.AddressInfo).port);
     });
   });
 }
 
-function fetchStatus(port: number, path: string): Promise<number> {
+function fetchResponse(
+  port: number,
+  path: string,
+  upgrade = false,
+): Promise<NodeHttp.IncomingMessage> {
   return new Promise((resolve, reject) => {
-    const request = NodeHttp.get({ host: "127.0.0.1", port, path }, (response) => {
-      response.resume();
-      resolve(response.statusCode ?? 0);
-    });
+    const request = NodeHttp.get(
+      {
+        host: "127.0.0.1",
+        port,
+        path,
+        agent: false,
+        headers: upgrade
+          ? {
+              connection: "Upgrade",
+              upgrade: "websocket",
+              "sec-websocket-key": "dGhlIHNhbXBsZSBub25jZQ==",
+              "sec-websocket-version": "13",
+            }
+          : {},
+      },
+      (response) => {
+        response.on("error", reject);
+        response.on("end", () => resolve(response));
+        response.resume();
+      },
+    );
+    requests.push(request);
     request.on("error", reject);
-    request.setTimeout(5_000, () => reject(new Error("request timed out")));
+    request.on("upgrade", (response, socket) => {
+      socket.destroy();
+      resolve(response);
+    });
   });
 }
 
-afterEach(() => {
-  for (const server of servers.splice(0)) {
-    server.close();
+function fetchStatus(port: number, path: string): Promise<number> {
+  return fetchResponse(port, path).then((response) => response.statusCode ?? 0);
+}
+
+afterEach(async () => {
+  for (const request of requests.splice(0)) {
+    request.destroy();
+  }
+  for (const socket of sockets.splice(0)) {
+    socket.destroy();
   }
+  await Promise.all(
+    servers
+      .splice(0)
+      .map((server) => new Promise<void>((resolve) => server.close(() => resolve()))),
+  );
 });
 
 describe("guardHttpResponseWriteErrors", () => {
@@ -65,6 +109,7 @@ describe("guardHttpResponseWriteErrors", () => {
         ].join("\r\n"),
       );
     });
+    sockets.push(client);
     client.on("error", () => {});
 
     await failureObserved.promise;
@@ -101,3 +146,42 @@ describe("guardHttpResponseWriteErrors", () => {
     expect(writeErrors).toEqual([]);
   });
 });
+
+describe("rejectRequestsUntilServing", () => {
+  it("rejects early HTTP and upgrades, then hands each off to its serving handler", async () => {
+    const writeErrors: unknown[] = [];
+    const server = HttpResponseErrorGuard.rejectRequestsUntilServing(
+      guardHttpResponseWriteErrors(NodeHttp.createServer(), (error) => writeErrors.push(error)),
+    );
+    const port = await listen(server);
+
+    const earlyHttp = await fetchResponse(port, "/");
+    expect(earlyHttp.statusCode).toBe(503);
+    expect(earlyHttp.headers["retry-after"]).toBe("1");
+
+    const earlyUpgrade = await fetchResponse(port, "/rpc", true);
+    expect(earlyUpgrade.statusCode).toBe(503);
+    expect(earlyUpgrade.headers["retry-after"]).toBe("1");
+
+    server.on("request", (_request, response) => response.writeHead(204).end());
+
+    const servingHttp = await fetchResponse(port, "/");
+    expect(servingHttp.statusCode).toBe(204);
+    expect(servingHttp.headers["retry-after"]).toBeUndefined();
+
+    const pendingUpgrade = await fetchResponse(port, "/rpc", true);
+    expect(pendingUpgrade.statusCode).toBe(503);
+    expect(pendingUpgrade.headers["retry-after"]).toBe("1");
+
+    server.on("upgrade", (_request, socket) => {
+      socket.end(
+        "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
+      );
+    });
+
+    const servingUpgrade = await fetchResponse(port, "/rpc", true);
+    expect(servingUpgrade.statusCode).toBe(101);
+    expect(servingUpgrade.headers["retry-after"]).toBeUndefined();
+    expect(writeErrors).toEqual([]);
+  }, 5_000);
+});

apps/server/src/httpResponseErrorGuard.ts

@@ -1,5 +1,7 @@
+// These guards run on the raw Node HTTP server before Effect handles requests.
 // @effect-diagnostics nodeBuiltinImport:off
 import type * as NodeHttp from "node:http";
+import type * as NodeStream from "node:stream";
 
 /**
  * An "error" event without a listener escalates into an uncaught exception and
@@ -22,3 +24,33 @@ export function guardHttpResponseWriteErrors<T extends NodeHttp.Server>(
   });
   return server;
 }
+
+/**
+ * NodeHttpServer listens before serve() attaches its handlers, so requests in
+ * between would hang. Install this after the write-error guards so only the
+ * serving handlers trigger the handoff.
+ */
+export function rejectRequestsUntilServing<T extends NodeHttp.Server>(server: T): T {
+  const rejectRequest = (_request: NodeHttp.IncomingMessage, response: NodeHttp.ServerResponse) => {
+    response.writeHead(503, { "retry-after": "1", connection: "close" }).end();
+  };
+  const rejectUpgrade = (_request: NodeHttp.IncomingMessage, socket: NodeStream.Duplex) => {
+    socket.end(
+      "HTTP/1.1 503 Service Unavailable\r\nRetry-After: 1\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
+    );
+  };
+  server.on("request", rejectRequest);
+  server.on("upgrade", rejectUpgrade);
+  // Node emits newListener before adding the handler, making the handoff synchronous.
+  server.on("newListener", function onNewListener(event) {
+    if (event === "request") server.off("request", rejectRequest);
+    if (event === "upgrade") server.off("upgrade", rejectUpgrade);
+    if (
+      !server.listeners("request").includes(rejectRequest) &&
+      !server.listeners("upgrade").includes(rejectUpgrade)
+    ) {
+      server.off("newListener", onNewListener);
+    }
+  });
+  return server;
+}

apps/server/src/server.ts

@@ -8,6 +8,7 @@ import * as StorageCleanup from "./storageCleanup.ts";
 import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts";
 import * as PullRequestWatchReactor from "./orchestration-v2/PullRequestWatchReactor.ts";
 import * as DelegatedTaskQuestionReactor from "./orchestration-v2/DelegatedTaskQuestionReactor.ts";
+// Effect wraps this raw Node HTTP server after its startup and write-error guards attach.
 // @effect-diagnostics nodeBuiltinImport:off
 import * as NodeHttp from "node:http";
 
@@ -30,7 +31,10 @@ import * as HostPowerMonitor from "./background/HostPowerMonitor.ts";
 import * as ServerConfig from "./config.ts";
 import { withUntracedRequests } from "./http.ts";
 import * as ServerHttp from "./http.ts";
-import { guardHttpResponseWriteErrors } from "./httpResponseErrorGuard.ts";
+import {
+  guardHttpResponseWriteErrors,
+  rejectRequestsUntilServing,
+} from "./httpResponseErrorGuard.ts";
 import { fixPath } from "./os-jank.ts";
 import * as Ws from "./ws.ts";
 import * as AgentScopeLive from "./process/agentScope.ts";
@@ -287,16 +291,19 @@ const layerRelayClient = Layer.unwrap(
 const layerHttpServer = Layer.unwrap(
   Effect.gen(function* () {
     const config = yield* ServerConfig.ServerConfig;
-    return NodeHttpServer.layer(() => guardHttpResponseWriteErrors(NodeHttp.createServer()), {
-      host: config.host ?? "127.0.0.1",
-      port: config.port,
-      gracefulShutdownTimeout: HTTP_PREEMPTIVE_SHUTDOWN_GRACE_MS,
-      // Context takeover stays enabled (ws default) so the compression
-      // window is shared across frames — that also makes small frames cheap
-      // to compress, so no size threshold is set (ws only honors
-      // `threshold` when context takeover is disabled).
-      websocket: { perMessageDeflate: true },
-    });
+    return NodeHttpServer.layer(
+      () => rejectRequestsUntilServing(guardHttpResponseWriteErrors(NodeHttp.createServer())),
+      {
+        host: config.host ?? "127.0.0.1",
+        port: config.port,
+        gracefulShutdownTimeout: HTTP_PREEMPTIVE_SHUTDOWN_GRACE_MS,
+        // Context takeover stays enabled (ws default) so the compression
+        // window is shared across frames — that also makes small frames cheap
+        // to compress, so no size threshold is set (ws only honors
+        // `threshold` when context takeover is disabled).
+        websocket: { perMessageDeflate: true },
+      },
+    );
   }),
 );