fix(http) mark completed true (#11926)

src/js/node/http.ts

@@ -765,6 +765,7 @@ async function consumeStream(self, reader: ReadableStreamDefaultReader) {
     var { done, value } = await reader.readMany();
     if (self[abortedSymbol]) return;
     if (done) {
+      self.complete = true;
       self.push(null);
       break;
     }
@@ -776,11 +777,12 @@ async function consumeStream(self, reader: ReadableStreamDefaultReader) {
 
 IncomingMessage.prototype._read = function (size) {
   if (this[noBodySymbol]) {
-    this.push(null);
     this.complete = true;
+    this.push(null);
   } else if (this[bodyStreamSymbol] == null) {
     const reader = this[reqSymbol].body?.getReader() as ReadableStreamDefaultReader;
     if (!reader) {
+      this.complete = true;
       this.push(null);
       return;
     }
@@ -795,17 +797,6 @@ Object.defineProperty(IncomingMessage.prototype, "aborted", {
   },
 });
 
-function abort(self) {
-  if (self[abortedSymbol]) return;
-  self[abortedSymbol] = true;
-  var bodyStream = self[bodyStreamSymbol];
-  if (!bodyStream) return;
-  bodyStream.cancel();
-  self.complete = true;
-  self[bodyStreamSymbol] = undefined;
-  self.push(null);
-}
-
 Object.defineProperty(IncomingMessage.prototype, "connection", {
   get() {
     return (this[fakeSocketSymbol] ??= new FakeSocket());
@@ -875,74 +866,6 @@ IncomingMessage.prototype.setTimeout = function (msecs, callback) {
   return this;
 };
 
-function emitErrorNt(msg, err, callback) {
-  callback(err);
-  if (typeof msg.emit === "function" && !msg._closed) {
-    msg.emit("error", err);
-  }
-}
-
-function onError(self, err, cb) {
-  process.nextTick(() => emitErrorNt(self, err, cb));
-}
-
-function write_(msg, chunk, encoding, callback, fromEnd) {
-  if (typeof callback !== "function") callback = nop;
-
-  let len;
-  if (chunk === null) {
-    // throw new ERR_STREAM_NULL_VALUES();
-    throw new Error("ERR_STREAM_NULL_VALUES");
-  } else if (typeof chunk === "string") {
-    len = Buffer.byteLength(chunk, encoding);
-  } else {
-    throw new Error("Invalid arg type for chunk");
-    // throw new ERR_INVALID_ARG_TYPE(
-    //   "chunk",
-    //   ["string", "Buffer", "Uint8Array"],
-    //   chunk,
-    // );
-  }
-
-  let err;
-  if (msg.finished) {
-    // err = new ERR_STREAM_WRITE_AFTER_END();
-    err = new Error("ERR_STREAM_WRITE_AFTER_END");
-  } else if (msg.destroyed) {
-    // err = new ERR_STREAM_DESTROYED("write");
-    err = new Error("ERR_STREAM_DESTROYED");
-  }
-
-  if (err) {
-    if (!msg.destroyed) {
-      onError(msg, err, callback);
-    } else {
-      process.nextTick(callback, err);
-    }
-    return false;
-  }
-
-  if (!msg._header) {
-    if (fromEnd) {
-      msg._contentLength = len;
-    }
-    // msg._implicitHeader();
-  }
-
-  if (!msg._hasBody) {
-    $debug("This type of response MUST NOT have a body. " + "Ignoring write() calls.");
-    process.nextTick(callback);
-    return true;
-  }
-
-  // if (!fromEnd && msg.socket && !msg.socket.writableCorked) {
-  //   msg.socket.cork();
-  //   process.nextTick(connectionCorkNT, msg.socket);
-  // }
-
-  return true;
-}
-
 const headersSymbol = Symbol("headers");
 const finishedSymbol = Symbol("finished");
 const timeoutTimerSymbol = Symbol("timeoutTimer");

test/js/node/http/node-http.test.ts

@@ -2152,3 +2152,48 @@ it("should error with faulty args", async () => {
   }
   server.close();
 });
+
+it("should mark complete true", async () => {
+  const { promise: serve, resolve: resolveServe } = Promise.withResolvers();
+  const server = createServer(async (req, res) => {
+    let count = 0;
+    let data = "";
+    req.on("data", chunk => {
+      data += chunk.toString();
+    });
+    while (!req.complete) {
+      await Bun.sleep(100);
+      count++;
+      if (count > 10) {
+        res.writeHead(500, { "Content-Type": "text/plain" });
+        res.end("Request timeout");
+        return;
+      }
+    }
+    res.writeHead(200, { "Content-Type": "text/plain" });
+    res.end(data);
+  });
+
+  server.listen(0, () => {
+    resolveServe(`http://localhost:${server.address().port}`);
+  });
+
+  const url = await serve;
+  try {
+    const response = await fetch(url, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        name: "Hotel 1",
+        price: 100,
+      }),
+    });
+
+    expect(response.status).toBe(200);
+    expect(await response.text()).toBe('{"name":"Hotel 1","price":100}');
+  } finally {
+    server.close();
+  }
+});

test/js/third_party/express/express.test.ts

@@ -0,0 +1,44 @@
+// @ts-nocheck
+// can't use @types/express or @types/body-parser because they
+// depend on @types/node which conflicts with bun-types
+import { test, expect } from "bun:test";
+import { isIPv6 } from "node:net";
+import express from "express";
+// https://github.com/oven-sh/bun/issues/8926
+test("should respond with 404 when wrong method is used", async () => {
+  const { promise: serve, resolve } = Promise.withResolvers();
+  const app = express();
+  app.use(express.json());
+
+  app.get("/api/hotels", (req, res) => {
+    res.json({
+      success: true,
+    });
+  });
+
+  const server = app.listen(0, (_, host, port) => {
+    if (isIPv6(host)) {
+      resolve(`http://[${host}]:${port}`);
+    } else {
+      resolve(`http://${host}:${port}`);
+    }
+  });
+
+  try {
+    const url = await serve;
+    const response = await fetch(`${url}/api/hotels`, {
+      method: "POST",
+      signal: AbortSignal.timeout(500),
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        name: "Hotel 1",
+        price: 100,
+      }),
+    });
+    expect(response.status).toBe(404);
+  } finally {
+    server.close();
+  }
+});

test/js/third_party/express/package.json

@@ -0,0 +1,7 @@
+{
+  "name": "express-test",
+  "version": "1.0.0",
+  "dependencies": {
+    "express": "4.18.2"
+  }
+}