fix(stack): log target failures and handle disconnections quietly

docs/adr/0017-simplified-managed-stack-architecture.md

@@ -103,7 +103,8 @@ public and private claims before one state commit; successful public sockets
 remain held and are adopted directly. Temporary private TCP listeners remain
 held until commit and then close, so a private workload gap remains possible.
 Fresh automatic claims draw from
-`20000..32767` with a random start and stride `257`, making up to 64 bounded
+`20000..32767` with a start derived from the stack's project root, identifier, and
+listener key, and stride `257`, making up to 64 bounded
 `EADDRINUSE`/`EACCES` attempts per newly selected binding while skipping
 durable sibling claims. Sticky values do not migrate. Failed acquisition
 preserves the

packages/stack/src/HostProcess.ts

@@ -56,7 +56,7 @@ export const acquireHost = Effect.fn("HostProcess.acquireHost")(function* (
     readonly closeConnections: Effect.Effect<void>;
   },
   HostProcessError | PortError | State.StateError,
-  Scope.Scope | import("effect").Crypto.Crypto
+  Scope.Scope
 > {
   if ((yield* state.read(stackId)) === undefined)
     return yield* error("acquire", "Stack is not registered");
@@ -307,10 +307,7 @@ export const launchHost = Effect.fn("HostProcess.launchHost")(function* (
 ): Effect.fn.Return<
   HostEndpoint,
   HostProcessError | State.StateError,
-  | Scope.Scope
-  | HttpClient.HttpClient
-  | import("effect").Crypto.Crypto
-  | ChildProcessSpawner.ChildProcessSpawner
+  Scope.Scope | HttpClient.HttpClient | ChildProcessSpawner.ChildProcessSpawner
 > {
   const existing = yield* connectHost(state, options.stackId).pipe(
     Effect.map(Option.some),

packages/stack/src/HttpProxy.integration.test.ts

@@ -1,6 +1,6 @@
 import { NodeHttpClient, NodeServices } from "@effect/platform-node";
 import { expect, it } from "@effect/vitest";
-import { Data, Deferred, Effect, Fiber, Layer } from "effect";
+import { Data, Deferred, Effect, Fiber, Layer, Logger } from "effect";
 import { HttpClient, HttpClientRequest } from "effect/unstable/http";
 import { createServer, type Server, type ServerResponse } from "node:http"; // oxlint-disable-line effecttsgo/node-builtin-import -- raw server fixture.
 import { Socket } from "node:net"; // oxlint-disable-line effecttsgo/node-builtin-import -- raw disconnect fixture.
@@ -35,6 +35,14 @@ class HttpProxyTestError extends Data.TaggedError("HttpProxyTestError")<{
   readonly cause?: unknown;
 }> {}
 
+const captureErrors = (lines: Array<string>) =>
+  Logger.layer([
+    Logger.make(({ logLevel, message }) => {
+      if (logLevel === "Error")
+        lines.push((Array.isArray(message) ? message : [message]).map(String).join(" "));
+    }),
+  ]);
+
 const request = (port: number, path: string, body: Uint8Array) =>
   Effect.gen(function* () {
     const client = yield* HttpClient.HttpClient;
@@ -128,8 +136,9 @@ it.live("keeps the retained listener and remaining route after one route is remo
   ).pipe(Effect.provide(NodeServices.layer)),
 );
 
-it.live("interrupts target acquisition when a waiting client disconnects", () =>
-  Effect.scoped(
+it.live("interrupts target acquisition quietly when a waiting client disconnects", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
     Effect.gen(function* () {
       const backend = createServer((_request, response) => response.end("unused"));
       const backendAddress = yield* listen(backend);
@@ -173,9 +182,10 @@ it.live("interrupts target acquisition when a waiting client disconnects", () =>
       yield* Deferred.await(acquired).pipe(Effect.timeout("5 seconds"));
       yield* Effect.sync(() => client.destroy());
       yield* Deferred.await(released);
+      expect(logs).toEqual([]);
     }),
-  ).pipe(Effect.provide(NodeServices.layer)),
-);
+  ).pipe(Effect.provide(Layer.merge(NodeServices.layer, captureErrors(logs))));
+});
 
 it.live("forwards raw WebSocket upgrades, subprotocols, and echo frames", () =>
   Effect.scoped(
@@ -259,13 +269,14 @@ it.live("overrides the upstream host for HTTP routes when configured", () =>
   ).pipe(Effect.provide(NodeServices.layer)),
 );
 
-it.live("returns a gateway error when a managed target cannot become ready", () =>
-  Effect.scoped(
+it.live("returns a gateway error naming the route and cause when a target cannot wake", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
     Effect.gen(function* () {
       const proxy = yield* makeHttpProxy({ host: "127.0.0.1", port: 0 });
       yield* proxy.setRoutes([
         {
-          id: "failed",
+          id: "rest",
           prefix: "/",
           target: Effect.fail(new ProxyError({ message: "readiness failed" })),
         },
@@ -275,17 +286,71 @@ it.live("returns a gateway error when a managed target cannot become ready", ()
       expect(response.status).toBe(502);
       expect(response.headers["access-control-allow-origin"]).toBe("*");
       expect(yield* response.text).toBe("Bad Gateway");
+      expect(logs).toHaveLength(1);
+      expect(logs[0]).toContain("Route rest request failed");
+      expect(logs[0]).toContain("readiness failed");
     }),
-  ).pipe(Effect.provide(Layer.merge(NodeHttpClient.layerNodeHttp, NodeServices.layer))),
-);
+  ).pipe(
+    Effect.provide(
+      Layer.mergeAll(NodeHttpClient.layerNodeHttp, NodeServices.layer, captureErrors(logs)),
+    ),
+  );
+});
 
-it.live("disconnects a pending upstream response when its client closes", () =>
-  Effect.scoped(
+it.live("closes an upgrade naming the route and cause when a target cannot wake", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
+    Effect.gen(function* () {
+      const proxy = yield* makeHttpProxy({ host: "127.0.0.1", port: 0 });
+      yield* proxy.setRoutes([
+        {
+          id: "realtime",
+          prefix: "/socket",
+          target: Effect.fail(new ProxyError({ message: "wake failed" })),
+        },
+      ]);
+      const socket = yield* Effect.acquireRelease(
+        Effect.sync(() => new Socket()),
+        (value) => Effect.sync(() => value.destroy()),
+      );
+      const received: Array<Buffer> = [];
+      yield* Effect.callback<void, HttpProxyTestError>((resume) => {
+        socket.on("data", (chunk: Buffer) => received.push(chunk));
+        // A destroyed upgrade reaches the client as a reset, which closes the socket either way.
+        socket.on("error", () => undefined);
+        socket.once("close", () => resume(Effect.void));
+        socket.connect(proxy.port, "127.0.0.1", () =>
+          socket.write(
+            "GET /socket HTTP/1.1\r\nHost: localhost\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n",
+          ),
+        );
+        return Effect.void;
+      }).pipe(Effect.timeout("5 seconds"));
+      expect(Buffer.concat(received)).toHaveLength(0);
+      expect(logs).toHaveLength(1);
+      expect(logs[0]).toContain("Route realtime upgrade failed");
+      expect(logs[0]).toContain("wake failed");
+    }),
+  ).pipe(Effect.provide(Layer.merge(NodeServices.layer, captureErrors(logs))));
+});
+
+it.live("disconnects a pending upstream response quietly when its client closes", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
     Effect.gen(function* () {
       const backend = createServer();
       const address = yield* listen(backend);
       const proxy = yield* makeHttpProxy({ host: "127.0.0.1", port: 0 });
-      yield* proxy.setRoutes([{ id: "pending", prefix: "/", target: Effect.succeed(address) }]);
+      const released = yield* Deferred.make<void>();
+      yield* proxy.setRoutes([
+        {
+          id: "pending",
+          prefix: "/",
+          target: Effect.acquireRelease(Effect.succeed(address), () =>
+            Deferred.succeed(released, undefined),
+          ),
+        },
+      ]);
       yield* Effect.callback<void, HttpProxyTestError>((resume) => {
         const client = new Socket();
         client.on("error", (cause) =>
@@ -300,12 +365,60 @@ it.live("disconnects a pending upstream response when its client closes", () =>
         );
         return Effect.sync(() => client.destroy());
       }).pipe(Effect.timeout("5 seconds"));
+      yield* Deferred.await(released).pipe(Effect.timeout("5 seconds"));
+      expect(logs).toEqual([]);
     }),
-  ).pipe(Effect.provide(NodeServices.layer)),
-);
+  ).pipe(Effect.provide(Layer.merge(NodeServices.layer, captureErrors(logs))));
+});
 
-it.live("releases a waiting WebSocket target when its client resets the connection", () =>
-  Effect.scoped(
+it.live("stays quiet when a client closes after receiving part of the response", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
+    Effect.gen(function* () {
+      const backend = createServer((incoming, outgoing) => {
+        incoming.resume();
+        outgoing.writeHead(200, { "content-type": "application/octet-stream" });
+        outgoing.write("first-chunk");
+      });
+      const address = yield* listen(backend);
+      const proxy = yield* makeHttpProxy({ host: "127.0.0.1", port: 0 });
+      const released = yield* Deferred.make<void>();
+      yield* proxy.setRoutes([
+        {
+          id: "streaming",
+          prefix: "/",
+          target: Effect.acquireRelease(Effect.succeed(address), () =>
+            Deferred.succeed(released, undefined),
+          ),
+        },
+      ]);
+      const client = yield* Effect.acquireRelease(
+        Effect.sync(() => new Socket()),
+        (socket) => Effect.sync(() => socket.destroy()),
+      );
+      yield* Effect.callback<void, HttpProxyTestError>((resume) => {
+        // Closing mid-response reaches the client as a reset, which is the disconnect under test.
+        client.on("error", () => undefined);
+        const onData = (chunk: Buffer) => {
+          if (!chunk.includes("first-chunk")) return;
+          client.destroy();
+          resume(Effect.void);
+        };
+        client.on("data", onData);
+        client.connect(proxy.port, "127.0.0.1", () =>
+          client.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"),
+        );
+        return Effect.sync(() => client.off("data", onData));
+      }).pipe(Effect.timeout("5 seconds"));
+      yield* Deferred.await(released).pipe(Effect.timeout("5 seconds"));
+      expect(logs).toEqual([]);
+    }),
+  ).pipe(Effect.provide(Layer.merge(NodeServices.layer, captureErrors(logs))));
+});
+
+it.live("releases a waiting WebSocket target quietly when its client resets", () => {
+  const logs: Array<string> = [];
+  return Effect.scoped(
     Effect.gen(function* () {
       const proxy = yield* makeHttpProxy({ host: "127.0.0.1", port: 0 });
       const acquiring = yield* Deferred.make<void>();
@@ -338,6 +451,7 @@ it.live("releases a waiting WebSocket target when its client resets the connecti
       yield* Deferred.await(acquiring);
       yield* Effect.sync(() => socket.resetAndDestroy());
       yield* Deferred.await(released).pipe(Effect.timeout("5 seconds"));
+      expect(logs).toEqual([]);
     }),
-  ).pipe(Effect.provide(NodeServices.layer)),
-);
+  ).pipe(Effect.provide(Layer.merge(NodeServices.layer, captureErrors(logs))));
+});

packages/stack/src/HttpProxy.ts

@@ -15,6 +15,9 @@ class HttpProxyError extends Data.TaggedError("HttpProxyError")<{
   readonly cause?: unknown;
 }> {}
 
+/** Distinguishes a client that went away first from a genuine proxy failure. */
+class HttpProxyDisconnected extends Data.TaggedError("HttpProxyDisconnected") {}
+
 export interface HttpRoute {
   readonly id: string;
   readonly prefix: string;
@@ -85,8 +88,8 @@ const setCors = (response: ServerResponse, request: IncomingMessage) => {
 };
 
 const disconnected = (request: IncomingMessage, response: ServerResponse) =>
-  Effect.callback<never, HttpProxyError>((resume) => {
-    const onAbort = () => resume(Effect.fail(errorFor("client disconnected")));
+  Effect.callback<never, HttpProxyDisconnected>((resume) => {
+    const onAbort = () => resume(Effect.fail(new HttpProxyDisconnected()));
     const onRequestClose = () => {
       if (!request.complete) onAbort();
     };
@@ -131,33 +134,37 @@ const proxyRequest = Effect.fn("HttpProxy.proxyRequest")(
   (request: IncomingMessage, response: ServerResponse, route: HttpRoute) =>
     Effect.gen(function* () {
       const backend = yield* Effect.raceFirst(route.target, disconnected(request, response));
-      yield* Effect.callback<void, HttpProxyError>((resume) => {
+      yield* Effect.callback<void, HttpProxyError | HttpProxyDisconnected>((resume) => {
         let outgoing: ReturnType<typeof upstreamRequest> | undefined;
         let incoming: IncomingMessage | undefined;
         let settled = false;
         // Error listeners remain until collection because destroy may emit errors asynchronously.
         const cleanup = () => {
-          request.off("aborted", onError);
+          request.off("aborted", onClientGone);
           response.off("close", onResponseClose);
           response.off("finish", onFinish);
 
           incoming?.off("aborted", onError);
         };
-        const finish = (result: Effect.Effect<void, HttpProxyError>) => {
+        const finish = (result: Effect.Effect<void, HttpProxyError | HttpProxyDisconnected>) => {
           if (settled) return;
           settled = true;
           cleanup();
           resume(result);
         };
-        const onError = (cause: Error) => {
+        // Settling first keeps the outcome: destroying a partial upstream response emits
+        // `aborted` synchronously, which would otherwise resettle as a proxy failure.
+        const abandon = (result: Effect.Effect<void, HttpProxyError | HttpProxyDisconnected>) => {
           if (settled) return;
+          finish(result);
           outgoing?.destroy();
           incoming?.destroy();
-          finish(Effect.fail(errorFor(cause)));
         };
+        const onError = (cause: Error) => abandon(Effect.fail(errorFor(cause)));
+        const onClientGone = () => abandon(Effect.fail(new HttpProxyDisconnected()));
         const onFinish = () => finish(Effect.void);
         const onResponseClose = () => {
-          if (!response.writableEnded) onError(new Error("client response closed"));
+          if (!response.writableEnded) onClientGone();
         };
         outgoing = upstreamRequest(
           {
@@ -183,7 +190,7 @@ const proxyRequest = Effect.fn("HttpProxy.proxyRequest")(
           },
         );
         outgoing.on("error", onError);
-        request.once("aborted", onError);
+        request.once("aborted", onClientGone);
         response.once("close", onResponseClose);
         request.pipe(outgoing);
         return Effect.sync(() => {
@@ -201,15 +208,15 @@ const upgrade = Effect.fn("HttpProxy.upgrade")(
     Effect.gen(function* () {
       const backend = yield* Effect.raceFirst(
         route.target,
-        Effect.callback<never, HttpProxyError>((resume) => {
-          const onClose = () => resume(Effect.fail(errorFor("client disconnected")));
+        Effect.callback<never, HttpProxyDisconnected>((resume) => {
+          const onClose = () => resume(Effect.fail(new HttpProxyDisconnected()));
           client.once("close", onClose);
           if (client.destroyed) onClose();
           return Effect.sync(() => client.off("close", onClose));
         }),
       );
       const upstream = yield* connectInterruptibly(backend);
-      yield* Effect.callback<void, HttpProxyError>((resume) => {
+      yield* Effect.callback<void, HttpProxyError | HttpProxyDisconnected>((resume) => {
         let settled = false;
         const cleanup = () => {
           client.off("close", onClose);
@@ -218,25 +225,23 @@ const upgrade = Effect.fn("HttpProxy.upgrade")(
           client.off("end", onClientEnd);
           upstream.off("end", onUpstreamEnd);
         };
-        const finish = (result: Effect.Effect<void, HttpProxyError>) => {
+        const finish = (result: Effect.Effect<void, HttpProxyError | HttpProxyDisconnected>) => {
           if (settled) return;
           settled = true;
           cleanup();
           resume(result);
         };
-        const onError = (cause: Error) => {
-          client.destroy();
-          upstream.destroy();
-          finish(Effect.fail(errorFor(cause)));
-        };
-        const onClose = () => {
+        const abandon = (result: Effect.Effect<void, HttpProxyError | HttpProxyDisconnected>) => {
+          finish(result);
           client.destroy();
           upstream.destroy();
-          finish(Effect.void);
         };
+        const onError = (cause: Error) => abandon(Effect.fail(errorFor(cause)));
+        const onClientGone = () => abandon(Effect.fail(new HttpProxyDisconnected()));
+        const onClose = () => abandon(Effect.void);
         const onClientEnd = () => upstream.end();
         const onUpstreamEnd = () => client.end();
-        client.on("error", onError);
+        client.on("error", onClientGone);
         client.once("close", onClose);
         upstream.on("error", onError);
         upstream.once("close", onClose);
@@ -286,6 +291,11 @@ export const makeHttpProxy = (options: {
               response.end("Not Found");
             } else {
               yield* proxyRequest(request, response, route).pipe(
+                Effect.tapError((cause) =>
+                  cause._tag === "HttpProxyDisconnected"
+                    ? Effect.void
+                    : Effect.logError(`Route ${route.id} request failed`, cause),
+                ),
                 Effect.catch(() =>
                   Effect.sync(() => {
                     if (response.destroyed) return;
@@ -318,6 +328,11 @@ export const makeHttpProxy = (options: {
             if (route === undefined) socket.destroy();
             else
               yield* upgrade(request, socket, head, route).pipe(
+                Effect.tapError((cause) =>
+                  cause._tag === "HttpProxyDisconnected"
+                    ? Effect.void
+                    : Effect.logError(`Route ${route.id} upgrade failed`, cause),
+                ),
                 Effect.catch(() => Effect.sync(() => socket.destroy())),
               );
           }),