Skip to content
Open
Show file tree
Hide file tree
Changes from all 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
20 changes: 0 additions & 20 deletions src/js/node/http2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1275,15 +1275,6 @@ function onServerStream(Http2ServerRequest, Http2ServerResponse, stream, headers
server.emit("request", request, response);
}

// The h2 frame writer flushes through the native socket directly, bypassing the JS
// Writable accounting net.Socket's bytesWritten getter is built on — the native
// counter is the ground truth once frames hit the wire.
function socketBytesWritten(socket) {
const native = socket._handle?.bytesWritten || 0;
const js = socket.bytesWritten || 0;
return native > js ? native : js;
}

const proxyCompatSocketHandler = {
has(stream, prop) {
const ref = stream.session !== undefined ? stream.session[bunHTTP2Socket] : stream;
Expand Down Expand Up @@ -1316,10 +1307,6 @@ const proxyCompatSocketHandler = {
case "pause":
case "resume":
throw $ERR_HTTP2_NO_SOCKET_MANIPULATION();
case "bytesWritten": {
const ref = stream.session !== undefined ? stream.session[bunHTTP2Socket] : stream;
return socketBytesWritten(ref);
}
default: {
const ref = stream.session !== undefined ? stream.session[bunHTTP2Socket] : stream;
const value = ref[prop];
Expand Down Expand Up @@ -1380,13 +1367,6 @@ const proxySocketHandler = {
case "setKeepAlive":
case "setNoDelay":
throw $ERR_HTTP2_NO_SOCKET_MANIPULATION();
case "bytesWritten": {
const socket = session[bunHTTP2Socket];
if (!socket) {
throw $ERR_HTTP2_SOCKET_UNBOUND();
}
return socketBytesWritten(socket);
}
default: {
const socket = session[bunHTTP2Socket];
if (!socket) {
Expand Down
16 changes: 6 additions & 10 deletions src/js/node/net.ts
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,8 @@ function closeAdoptedTLSRawNowNT(handle, self, isException) {
}
function detachSocket(self) {
if (!self) self = this;
const handle = self._handle;
if (handle) self[kBytesWritten] = handle.bytesWritten;
self._handle = null;
}
function destroyNT(self, err) {
Expand Down Expand Up @@ -502,8 +504,6 @@ const SocketHandlers: SocketHandler = {
} else {
self._pendingData = null;
}

self[kBytesWritten] = socket.bytesWritten;
}
},
end(socket) {
Expand Down Expand Up @@ -582,7 +582,6 @@ const SocketHandlers: SocketHandler = {
}

if (!self[kupgraded]) {
self[kBytesWritten] = socket.bytesWritten;
// this is not actually emitted on nodejs when socket used on the connection
// this is already emmited on non-TLS socket and on TLS socket is emmited secureConnect after handshake
self.emit("connect", self);
Expand Down Expand Up @@ -1328,15 +1327,12 @@ const SocketHandlers2: SocketHandler<NonNullable<import("node:net").Socket["_han
const res = socket.$write(writeChunk || "", self._pendingEncoding || "utf8");
if (res < 0) {
// The retried send failed for good (peer gone): $write returned -errno.
self[kBytesWritten] = socket.bytesWritten;
failWrite(self, res, callback);
} else if (res) {
self[kBytesWritten] = socket.bytesWritten;
self._pendingData = self[kwriteCallback] = null;
unrefAfterDrain(self, socket);
callback(null);
} else {
self[kBytesWritten] = socket.bytesWritten;
self._pendingData = null;
}
}
Expand Down Expand Up @@ -1843,15 +1839,17 @@ Object.defineProperty(Socket.prototype, "bufferSize", {
},
});

// https://github.com/nodejs/node/blob/v26.3.0/lib/net.js#L1043-L1045
Object.defineProperty(Socket.prototype, "_bytesDispatched", {
get: function () {
return this[kBytesWritten] || 0;
const handle = this._handle;
return (handle ? handle.bytesWritten : this[kBytesWritten]) || 0;
Comment thread
robobun marked this conversation as resolved.
Resolved
},
});

Object.defineProperty(Socket.prototype, "bytesWritten", {
get: function () {
let bytes = this[kBytesWritten] || 0;
let bytes = this._bytesDispatched;
const data = this._pendingData;
const writableBuffer = this.writableBuffer;
if (!writableBuffer) return undefined;
Expand Down Expand Up @@ -1895,7 +1893,6 @@ Socket.prototype[kAttach] = function (port, socket) {
}

if (!this[kupgraded]) {
this[kBytesWritten] = socket.bytesWritten;
// this is not actually emitted on nodejs when socket used on the connection
// this is already emmited on non-TLS socket and on TLS socket is emmited secureConnect after handshake
this.emit("connect", this);
Expand Down Expand Up @@ -2833,7 +2830,6 @@ Socket.prototype._write = function _write(chunk, encoding, callback) {
return false;
}
const res = socket.$write(chunk, encoding);
this[kBytesWritten] = socket.bytesWritten;
if (res < 0) {
// The kernel rejected the send outright (peer reset): $write returned the
// negative errno; deliver it like the EBADF/EPIPE branch above.
Expand Down
48 changes: 48 additions & 0 deletions test/js/node/http2/node-http2.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -6329,3 +6329,51 @@ describe("frames issued from inside a user-supplied Duplex transport's _write",
},
);
});

describe("net.Socket bytesWritten through an http2 session", () => {
// The h2 frame writer writes through the native socket handle, not through
// the JS write path. The raw socket that the user passes to createConnection
// must count those bytes too, like the session.socket proxy does.
it("counts the frames the session writes on the raw createConnection socket", async () => {
const server = http2.createServer();
server.on("stream", stream => {
stream.respond({ ":status": 200 });
stream.end("hello world from h2");
});
await new Promise(resolve => server.listen(0, "127.0.0.1", resolve));
const port = server.address().port;

const raw = net.connect(port, "127.0.0.1");
const client = http2.connect(`http://127.0.0.1:${port}`, { createConnection: () => raw });
const failed = new Promise((_, reject) => {
raw.once("error", reject);
client.once("error", reject);
});
try {
const req = client.request({ ":path": "/" });
req.resume();
await Promise.race([
failed,
new Promise((resolve, reject) => {
req.once("end", resolve);
req.once("error", reject);
}),
]);

const proxy = client.socket.bytesWritten;
expect(proxy).toBeGreaterThan(0);
expect(raw.bytesWritten).toBe(proxy);
expect(raw._bytesDispatched).toBe(proxy);

const closed = new Promise(resolve => raw.once("close", resolve));
client.close();
await Promise.race([failed, closed]);
// The count survives the destroy of the handle.
expect(raw.bytesWritten).toBeGreaterThanOrEqual(proxy);
expect(raw._bytesDispatched).toBe(raw.bytesWritten);
} finally {
client.destroy();
await new Promise(resolve => server.close(resolve));
}
});
});
30 changes: 30 additions & 0 deletions test/js/node/net/node-net.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3130,3 +3130,33 @@ describe.concurrent("uncaughtException from socket listeners", () => {
expect(exitCode).toBe(1);
});
});

describe("bytesWritten after the peer resets the connection", () => {
// A reset reaches the server socket through the native close handler, which
// drops the handle before destroy() runs. The count must survive that.
it("keeps the count on a server socket that the client reset", async () => {
const { promise: serverClosed, resolve, reject } = Promise.withResolvers<Socket>();
const server = createServer(c => {
c.on("error", () => {});
c.on("close", () => resolve(c));
c.write("hello", () => c.write("world"));
});
server.on("error", reject);
await once(server.listen(0, "127.0.0.1"), "listening");
try {
const client = connect(server.address().port, "127.0.0.1");
client.on("error", reject);
let received = 0;
client.on("data", chunk => {
received += chunk.length;
if (received === 10) client.resetAndDestroy();
});
const c = await serverClosed;
expect(c._handle).toBeNull();
expect(c.bytesWritten).toBe(10);
expect(c._bytesDispatched).toBe(10);
} finally {
await new Promise(resolve => server.close(resolve));
}
});
});