Optimize Node HTTP response handling

.changeset/optimize-node-http-server-response.md

@@ -0,0 +1,5 @@
+---
+"@effect/platform-node": patch
+---
+
+Optimize Node HTTP streaming responses and ensure HEAD completion and stream backpressure are handled once.

packages/platform-node/src/NodeHttpServer.ts

@@ -347,8 +347,9 @@ class ServerRequestImpl extends NodeHttpIncomingMessage<HttpServerError> impleme
     return this.source.url!
   }
 
+  private cachedMethod: HttpMethod | undefined
   get method(): HttpMethod {
-    return this.source.method!.toUpperCase() as HttpMethod
+    return this.cachedMethod ??= this.source.method!.toUpperCase() as HttpMethod
   }
 
   override get headers(): Headers.Headers {
@@ -524,7 +525,10 @@ const handleResponse = (
   if (request.method === "HEAD") {
     nodeResponse.writeHead(response.status, headers)
     return Effect.callback<void>((resume) => {
+      let completed = false
       const done = () => {
+        if (completed) return
+        completed = true
         nodeResponse.off("close", done)
         resume(Effect.void)
       }
@@ -614,17 +618,9 @@ const handleResponse = (
       return body.stream.pipe(
         Stream.orDie,
         Stream.runForEachArray((array) => {
-          let needDrain = false
-          for (let i = 0; i < array.length; i++) {
-            const written = nodeResponse.write(array[i])
-            if (!written && !needDrain) {
-              needDrain = true
-              drainLatch.closeUnsafe()
-            } else if (written && needDrain) {
-              needDrain = false
-            }
-          }
-          if (!needDrain) return Effect.void
+          const chunk = array.length > 1 ? Buffer.concat(array) : array[0]
+          if (nodeResponse.write(chunk)) return Effect.void
+          drainLatch.closeUnsafe()
           return drainLatch.await
         }),
         Effect.interruptible,

packages/platform-node/test/NodeHttpServer.test.ts

@@ -29,6 +29,7 @@ import {
 } from "effect/unstable/http"
 import * as HttpApiError from "effect/unstable/httpapi/HttpApiError"
 import * as Buffer from "node:buffer"
+import { EventEmitter } from "node:events"
 import * as Http from "node:http"
 
 const Todo = Schema.Struct({
@@ -509,6 +510,120 @@ describe("HttpServer", () => {
       assert.strictEqual(res.status, 204)
     }).pipe(Effect.provide(NodeHttpServer.layerTest)))
 
+  it.effect("completes a HEAD response once when close precedes the end callback", () =>
+    Effect.gen(function*() {
+      const scope = yield* Effect.scope
+      const handler = yield* NodeHttpServer.makeHandler(
+        Effect.succeed(HttpServerResponse.empty()),
+        { scope }
+      )
+      const completed = Latch.makeUnsafe()
+      let writableEnded = false
+      const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
+        get: () => writableEnded
+      }) as Http.ServerResponse
+      let closeListenerRemovals = 0
+      nodeResponse.writeHead = () => nodeResponse
+      nodeResponse.off = ((event: string | symbol, listener: (...args: Array<unknown>) => void) => {
+        if (event === "close") {
+          closeListenerRemovals++
+        }
+        return EventEmitter.prototype.off.call(nodeResponse, event, listener) as Http.ServerResponse
+      }) as Http.ServerResponse["off"]
+      nodeResponse.end = ((callback: () => void) => {
+        writableEnded = true
+        nodeResponse.emit("close")
+        callback()
+        completed.openUnsafe()
+        return nodeResponse
+      }) as Http.ServerResponse["end"]
+
+      handler(
+        { method: "HEAD", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
+        nodeResponse
+      )
+      yield* completed.await
+
+      assert.strictEqual(closeListenerRemovals, 1)
+    }).pipe(Effect.scoped))
+
+  it.effect("coalesces streaming chunks from the same pull", () =>
+    Effect.gen(function*() {
+      const scope = yield* Effect.scope
+      const handler = yield* NodeHttpServer.makeHandler(
+        Effect.succeed(HttpServerResponse.stream(Stream.make(
+          Buffer.Buffer.from("a"),
+          Buffer.Buffer.from("b")
+        ))),
+        { scope }
+      )
+      const completed = Latch.makeUnsafe()
+      const writes: Array<Uint8Array> = []
+      let writableEnded = false
+      const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
+        get: () => writableEnded
+      }) as Http.ServerResponse
+      nodeResponse.writeHead = () => nodeResponse
+      nodeResponse.write = ((chunk: Uint8Array) => {
+        writes.push(chunk)
+        return true
+      }) as Http.ServerResponse["write"]
+      nodeResponse.end = (() => {
+        writableEnded = true
+        completed.openUnsafe()
+        return nodeResponse
+      }) as Http.ServerResponse["end"]
+
+      handler(
+        { method: "GET", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
+        nodeResponse
+      )
+      yield* completed.await
+
+      assert.deepStrictEqual(writes.map((chunk) => Buffer.Buffer.from(chunk).toString()), ["ab"])
+    }).pipe(Effect.scoped))
+
+  it.effect("waits for drain after a streaming write applies backpressure", () =>
+    Effect.gen(function*() {
+      const scope = yield* Effect.scope
+      const handler = yield* NodeHttpServer.makeHandler(
+        Effect.succeed(HttpServerResponse.stream(Stream.make(
+          Buffer.Buffer.from("a"),
+          Buffer.Buffer.from("b")
+        ))),
+        { scope }
+      )
+      const writeObserved = Latch.makeUnsafe()
+      const completed = Latch.makeUnsafe()
+      let writableEnded = false
+      const nodeResponse = Object.defineProperty(new EventEmitter(), "writableEnded", {
+        get: () => writableEnded
+      }) as Http.ServerResponse
+      let writeCount = 0
+      nodeResponse.writeHead = () => nodeResponse
+      nodeResponse.write = (() => {
+        writeCount++
+        queueMicrotask(() => writeObserved.openUnsafe())
+        return writeCount > 1
+      }) as Http.ServerResponse["write"]
+      nodeResponse.end = (() => {
+        writableEnded = true
+        completed.openUnsafe()
+        return nodeResponse
+      }) as Http.ServerResponse["end"]
+
+      handler(
+        { method: "GET", url: "/", headers: {}, socket: {} } as Http.IncomingMessage,
+        nodeResponse
+      )
+      yield* writeObserved.await
+      assert.strictEqual(nodeResponse.writableEnded, false)
+
+      nodeResponse.emit("drain")
+      yield* completed.await
+      assert.strictEqual(writeCount, 1)
+    }).pipe(Effect.scoped))
+
   it.live("disposes after a client aborts a handler awaiting an upstream request", () => {
     const upstreamStarted = Latch.makeUnsafe()
     const upstream = Http.createServer(() => {