Skip to content
Open
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion packages/bun-usockets/src/loop.c
Original file line number Diff line number Diff line change
Expand Up @@ -700,10 +700,14 @@
}
if (us_socket_is_shut_down(s)) {
/* We got FIN back after sending it */
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) {
/* on_data paused us mid-drain; kqueue EV_EOF fires with bytes
* still buffered. Defer on_end so resume can drain first. The
* shut_down close above must stay ungated or EPOLLHUP spins. */
} else if(s->flags.allow_half_open) {

Check notice on line 710 in packages/bun-usockets/src/loop.c

View check run for this annotation

Claude / Claude Code Review

shut_down fast-close truncates bytes the is_paused defer is meant to preserve

Pre-existing, same bug class at the same line, so noting as another sibling for #32257 rather than a merge blocker. The ungated `us_socket_is_shut_down` fast-close at :701 runs *before* the new `is_paused` check, so the exact scenario this PR fixes is still not fixed when the client has already `end()`ed — `c.end(req)` → server writes >512KB + FIN → first `recv()` fills the buffer, `push()` returns false → pause → break → `is_shut_down` is true → `close_raw`, and the remaining bytes are silently
Comment thread
robobun marked this conversation as resolved.
Outdated
/* 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
Expand Down
7 changes: 4 additions & 3 deletions test/js/node/net/net-mongodb-pattern-leak.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
64 changes: 62 additions & 2 deletions test/js/node/net/node-net.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -842,7 +842,7 @@
}

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()}`));
Expand All @@ -857,7 +857,7 @@
}
}
await Promise.all(batch);
expectMaxObjectTypeCount(expect, "TCPSocket", before);
await expectMaxObjectTypeCount(expect, "TCPSocket", before);
},
20_000,
);
Expand Down Expand Up @@ -999,3 +999,63 @@
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();
queueMicrotask(() => c.resume());

Check warning on line 1025 in test/js/node/net/node-net.test.ts

View check run for this annotation

Claude / Claude Code Review

queueMicrotask resume drains before loop.c sees is_paused, so the test never exercises the new guard

This test doesn't actually exercise the new `is_paused` branch in loop.c: `on_data`'s `exit_scope` drains microtasks before returning to the recv loop, so the `queueMicrotask`'d `c.resume()` clears `is_paused` back to 0 before loop.c ever checks it — the test passes with or without the loop.c change on every platform. Swap `queueMicrotask` for `setImmediate` so the pause survives past `us_dispatch_data`'s return and the test becomes a real fail-before guard on macOS.
Comment thread
robobun marked this conversation as resolved.
Outdated
});
c.on("end", () => {
ended = true;
resolve();
});
await promise;
} finally {
c.destroy();
server.close();
}
expect({ received, dataAfterEnd }).toEqual({ received: TOTAL, dataAfterEnd: 0 });
});

// The is_paused guard in loop.c must not skip the shut_down fast-close: on
// epoll a paused+shut-down socket registers only EPOLLHUP|EPOLLERR, and if the
// eof handler does nothing the level-triggered EPOLLHUP spins the loop.
it("paused socket that ends then receives the peer's FIN closes instead of busy-looping", 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();
}
});
Loading