test: widen net-mongodb-pattern-leak RSS bound; await the named-pipe GC check

packages/bun-usockets/src/loop.c

@@ -703,7 +703,11 @@ void us_internal_dispatch_ready_poll(struct us_poll_t *p, int error, int eof, in
                     s = us_internal_socket_close_raw(s, LIBUS_SOCKET_CLOSE_CODE_CLEAN_SHUTDOWN, NULL);
                     return;
                 }
-                if(s->flags.allow_half_open) {
+                if (s->flags.is_paused && (events & LIBUS_SOCKET_READABLE)) {
+                    /* on_data paused us mid-drain this dispatch; kqueue EV_EOF
+                     * fires with bytes still buffered so defer on_end. Gated on
+                     * READABLE so a prior pause (AF_UNIX EPOLLHUP) still ends. */
+                } else if(s->flags.allow_half_open) {
                     /* EOF with half-open allowed: stop polling readable but KEEP
                      * polling writable. Masking with the current events dropped
                      * writable when the EOF landed before the poll had been

test/js/node/net/net-mongodb-pattern-leak.test.ts

@@ -202,9 +202,10 @@ describe.each([
       expect(messages.listenerCount("data")).toBe(0);
       expect(messages.listenerCount("error")).toBeLessThanOrEqual(1);
 
-      // RSS round-2 vs round-1: weak signal (mimalloc segment noise) but
-      // catches anything egregious that heapStats can't see.
-      const rssBound = isASAN || isDebug ? 32 * 1024 * 1024 : 8 * 1024 * 1024;
+      // RSS round-2 vs round-1: weak backstop for native leaks heapStats can't
+      // see. Release CI observed 8-13 MB here with flat heapSize and flat object
+      // counts (mimalloc + JIT noise); a real native leak at ITER=5000 is 25+ MB.
+      const rssBound = isASAN || isDebug ? 32 * 1024 * 1024 : 24 * 1024 * 1024;
       expect(after2.rss - after1.rss).toBeLessThan(rssBound);
     } finally {
       sock.destroy();

test/js/node/net/node-net.test.ts

@@ -842,7 +842,7 @@ it.if(isWindows)(
     }
 
     const batch = [];
-    const before = heapStats().objectTypeCounts.TLSSocket || 0;
+    const before = heapStats().objectTypeCounts.TCPSocket || 0;
     for (let i = 0; i < 100; i++) {
       batch.push(test(`\\\\.\\pipe\\test\\${randomUUID()}`));
       batch.push(test(`\\\\?\\pipe\\test\\${randomUUID()}`));
@@ -857,7 +857,7 @@ it.if(isWindows)(
       }
     }
     await Promise.all(batch);
-    expectMaxObjectTypeCount(expect, "TCPSocket", before);
+    await expectMaxObjectTypeCount(expect, "TCPSocket", before);
   },
   20_000,
 );
@@ -999,3 +999,94 @@ describe("paused socket whose peer sends RST", () => {
     expect(errors.map(e => e.code)).not.toContain("ENOEXEC");
   });
 });
+
+// Regression for test-net-write-slow.js (macOS): kqueue's EV_EOF fires with
+// bytes still buffered, and loop.c dispatched end while on_data had paused the
+// recv loop mid-drain, so resume pushed the remainder after push(null).
+it("paused socket whose peer wrote >LIBUS_RECV_BUFFER_LENGTH then closed delivers every byte before end", async () => {
+  const TOTAL = 2 * 1024 * 1024;
+  const server = createServer(c => {
+    c.on("error", () => {});
+    c.write(Buffer.alloc(TOTAL, 0x61));
+    c.end();
+  });
+  await new Promise<void>(r => server.listen(0, "127.0.0.1", r));
+  const { promise, resolve, reject } = Promise.withResolvers<void>();
+  let received = 0;
+  let dataAfterEnd = 0;
+  let ended = false;
+  const c = connect((server.address() as import("node:net").AddressInfo).port, "127.0.0.1");
+  try {
+    c.on("error", reject);
+    c.on("data", chunk => {
+      if (ended) dataAfterEnd += chunk.length;
+      received += chunk.length;
+      c.pause();
+      // setImmediate, not a microtask: on_data's exit_scope drains microtasks
+      // before returning to loop.c, so a microtask resume would clear is_paused
+      // before loop.c reads it and the branch under test would never run.
+      setImmediate(() => c.resume());
+    });
+    c.on("end", () => {
+      ended = true;
+      resolve();
+    });
+    await promise;
+  } finally {
+    c.destroy();
+    server.close();
+  }
+  expect({ received, dataAfterEnd }).toEqual({ received: TOTAL, dataAfterEnd: 0 });
+});
+
+// Deferring on_end while is_paused must not busy-loop on a poll-level eof
+// (EPOLLHUP). Windows uv_poll never passes eof from its callback and stops
+// at events==0, so pause();end() has nothing to observe there; see #32257.
+describe.skipIf(isWindows)("paused socket whose peer closed the connection", () => {
+  // pause() then end(): our SHUT_WR plus the peer's FIN asserts EPOLLHUP on TCP.
+  // The shut_down fast-close in the eof block must stay ungated by is_paused.
+  it("closes after end() once the peer's FIN arrives (TCP, shut_down)", async () => {
+    const server = createServer(c => {
+      c.on("error", () => {});
+      c.on("end", () => c.end());
+    });
+    await new Promise<void>(r => server.listen(0, "127.0.0.1", r));
+    const { promise, resolve, reject } = Promise.withResolvers<string>();
+    const c = connect((server.address() as import("node:net").AddressInfo).port, "127.0.0.1");
+    try {
+      c.on("error", reject);
+      c.on("close", () => resolve("close"));
+      await new Promise<void>(r => c.once("connect", () => r()));
+      c.pause();
+      c.end("x");
+      expect(await promise).toBe("close");
+    } finally {
+      c.destroy();
+      server.close();
+    }
+  });
+
+  // pause() only (not shut_down), AF_UNIX peer close sets sk_shutdown=MASK so
+  // EPOLLHUP fires without our SHUT_WR. The defer must only apply when on_data
+  // paused this dispatch (events had READABLE); otherwise on_end must still run.
+  it("closes when the AF_UNIX peer closes (not shut_down)", async () => {
+    const sockPath = join(socket_domain, `paused-unix-${randomUUID()}.sock`);
+    const server = createServer(c => {
+      c.on("error", () => {});
+      setImmediate(() => c.destroy());
+    });
+    await new Promise<void>(r => server.listen(sockPath, r));
+    const { promise, resolve, reject } = Promise.withResolvers<string>();
+    const c = connect({ path: sockPath });
+    try {
+      c.on("error", e => reject(e));
+      c.on("close", () => resolve("close"));
+      await new Promise<void>(r => c.once("connect", () => r()));
+      c.pause();
+      expect(await promise).toBe("close");
+    } finally {
+      c.destroy();
+      server.close();
+    }
+  });
+});