usockets: dispatch on_end before the poll change in the half-open EOF arm

packages/bun-usockets/src/loop.c

@@ -930,9 +930,14 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                      * stream's 'finish' never happened - the FIN-terminated
                      * http response tests hung on every Linux target. The
                      * writable dispatch disables writable polling again once
-                     * the buffer is drained, so this does not busy-poll. */
-                    us_poll_change(&s->p, loop, LIBUS_SOCKET_WRITABLE);
+                     * the buffer is drained, so this does not busy-poll.
+                     * on_end runs first: an owner that closes there needs no poll
+                     * change at all. */
                     s = s->ssl ? us_internal_ssl_on_end(s) : us_dispatch_end(s);
+                    s = us_internal_socket_follow_adopted(s);
+                    if (s && !us_socket_is_closed(s) && !error) {
+                        us_poll_change(&s->p, loop, us_internal_poll_type(&s->p) == POLL_TYPE_SOCKET_SHUT_DOWN ? 0 : LIBUS_SOCKET_WRITABLE);
+                    }
                 } else {
                     /* Half-open not allowed, or a hangup (both directions down, level-triggered):
                      * emit end unless a FIN already did, then close so EPOLLHUP stops re-firing. */

test/js/node/net/node-net-allowHalfOpen.test.js

@@ -1,5 +1,7 @@
-import { expect, test } from "bun:test";
+import { describe, expect, test } from "bun:test";
 import { bunRun, isWindows, nodeExe, tempDir, tempDirWithFiles } from "harness";
+import { randomFillSync } from "node:crypto";
+import { once } from "node:events";
 import net from "node:net";
 import { join } from "node:path";
 
@@ -127,3 +129,46 @@ test.skipIf(isWindows)("allowHalfOpen: paused socket whose unix peer closed deli
   );
   expect(result).toEqual({ stdout: JSON.stringify(expected) + "\nidle", stderr: "", exitCode: 0, signalCode: null });
 });
+
+// The client sends its request and its FIN at once, then reads. The reply is larger than
+// the kernel takes in one send, so part of it waits in user space across the peer's FIN.
+describe.each([
+  { allowHalfOpen: true, replyOn: "end" },
+  { allowHalfOpen: true, replyOn: "data" },
+  { allowHalfOpen: false, replyOn: "data" },
+])("server with allowHalfOpen: $allowHalfOpen that replies on '$replyOn'", ({ allowHalfOpen, replyOn }) => {
+  test("sends the whole reply after the client's FIN", async () => {
+    const N = 8 * 1024 * 1024;
+    const payload = randomFillSync(Buffer.allocUnsafe(N));
+    const serverSide = Promise.withResolvers();
+    const server = net.createServer({ allowHalfOpen }, socket => {
+      socket.on("error", serverSide.reject);
+      socket.on("close", serverSide.resolve);
+      socket.on("data", () => {
+        if (replyOn === "data") socket.end(payload);
+      });
+      socket.on("end", () => {
+        if (replyOn === "end") socket.end(payload);
+      });
+    });
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    try {
+      const clientSide = Promise.withResolvers();
+      let got = 0;
+      let mismatchAt = -1;
+      const client = net.connect({ port: server.address().port, host: "127.0.0.1", allowHalfOpen: true });
+      client.on("error", clientSide.reject);
+      client.on("connect", () => client.end("request\n"));
+      client.on("data", chunk => {
+        if (mismatchAt === -1 && !chunk.equals(payload.subarray(got, got + chunk.length))) mismatchAt = got;
+        got += chunk.length;
+      });
+      client.on("close", clientSide.resolve);
+      await Promise.all([clientSide.promise, serverSide.promise]);
+      expect({ got, mismatchAt }).toEqual({ got: N, mismatchAt: -1 });
+    } finally {
+      server.close();
+    }
+  });
+});