http: add diagnostics channel for outgoing body chunks

doc/api/diagnostics_channel.md

@@ -1627,6 +1627,28 @@ The event is emitted before the response is sent.
 
 Emitted when server sends a response.
 
+##### Event: `'http.client.request.bodyChunkSent'`
+
+* `message` {http.ClientRequest}
+* `data` {Buffer|string}
+* `encoding` {string}
+
+Emitted when a chunk of a client request body is being sent. The byte length
+of the chunk can be computed from `data` and `encoding` with
+`Buffer.byteLength(data, encoding)` (or `data.byteLength` for buffers), which
+denotes the amount of data sent without Node having to measure it itself.
+
+##### Event: `'http.server.response.bodyChunkSent'`
+
+* `message` {http.ServerResponse}
+* `data` {Buffer|string}
+* `encoding` {string}
+
+Emitted when a chunk of a server response body is being sent. The byte length
+of the chunk can be computed from `data` and `encoding` with
+`Buffer.byteLength(data, encoding)` (or `data.byteLength` for buffers), which
+denotes the amount of data sent without Node having to measure it itself.
+
 #### HTTP/2
 
 > Stability: 1 - Experimental

lib/_http_client.js

@@ -56,6 +56,7 @@ const {
   kHighWaterMark,
   kUniqueHeaders,
   parseUniqueHeadersOption,
+  setupBodyChunkSentDiagnostics,
   OutgoingMessage,
 } = require('_http_outgoing');
 const Agent = require('_http_agent');
@@ -336,6 +337,8 @@ function ClientRequest(input, options, cb) {
 
   OutgoingMessage.call(this);
 
+  setupBodyChunkSentDiagnostics(this, false);
+
   if (typeof input === 'string') {
     const urlStr = input;
     input = urlToHttpOptions(new URL(urlStr));

lib/_http_outgoing.js

@@ -77,6 +77,25 @@ const {
 } = require('internal/util');
 const { isUint8Array } = require('internal/util/types');
 
+const dc = require('diagnostics_channel');
+const onClientRequestBodyChunkSentChannel =
+  dc.channel('http.client.request.bodyChunkSent');
+const onServerResponseBodyChunkSentChannel =
+  dc.channel('http.server.response.bodyChunkSent');
+const kBodyChunkSentChannel = Symbol('kBodyChunkSentChannel');
+
+// Resolve the correct body-chunk-sent diagnostics channel for a message and
+// cache it on the instance when it has subscribers, so the hot `write_` path
+// only reads a cached reference instead of querying `channel.hasSubscribers`
+// on every chunk.
+function setupBodyChunkSentDiagnostics(msg, isServer) {
+  const channel = isServer ? onServerResponseBodyChunkSentChannel :
+    onClientRequestBodyChunkSentChannel;
+  if (channel.hasSubscribers) {
+    msg[kBodyChunkSentChannel] = channel;
+  }
+}
+
 let debug = require('internal/util/debuglog').debuglog('http', (fn) => {
   debug = fn;
 });
@@ -1011,6 +1030,18 @@ function write_(msg, chunk, encoding, callback, fromEnd) {
     process.nextTick(connectionCorkNT, msg.socket);
   }
 
+  // Expose the outgoing body chunk on a diagnostics channel so that subscribed
+  // tools can account for the actual bytes being sent without Node paying to
+  // measure them when nobody is reading. `data` and `encoding` carry enough
+  // information to compute the byte length (`Buffer.byteLength(data, encoding)`
+  // for strings, `data.byteLength` for buffers). Whether the channel has
+  // subscribers is resolved once in the constructor; `kBodyChunkSentChannel` is
+  // only set when there is a subscriber, so `write_` can short-circuit cheaply.
+  const channel = msg[kBodyChunkSentChannel];
+  if (channel) {
+    channel.publish({ message: msg, data: chunk, encoding });
+  }
+
   let ret;
   if (msg.chunkedEncoding && chunk.length !== 0) {
     len ??= typeof chunk === 'string' ? Buffer.byteLength(chunk, encoding) : chunk.byteLength;
@@ -1341,5 +1372,6 @@ module.exports = {
   parseUniqueHeadersOption,
   validateHeaderName,
   validateHeaderValue,
+  setupBodyChunkSentDiagnostics,
   OutgoingMessage,
 };

lib/_http_server.js

@@ -54,6 +54,7 @@ const { ConnectionsList } = internalBinding('http_parser');
 const {
   kUniqueHeaders,
   parseUniqueHeadersOption,
+  setupBodyChunkSentDiagnostics,
   OutgoingMessage,
   validateHeaderName,
   validateHeaderValue,
@@ -212,6 +213,8 @@ function ServerResponse(req, options) {
 
   OutgoingMessage.call(this, options);
 
+  setupBodyChunkSentDiagnostics(this, true);
+
   if (req.method === 'HEAD') this._hasBody = false;
 
   this.req = req;

test/parallel/test-diagnostic-channel-http-body-chunk-sent.js

@@ -0,0 +1,73 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+const dc = require('diagnostics_channel');
+
+// The outgoing body chunk diagnostics channels let tools account for the actual
+// bytes being sent without Node paying to measure them when nobody is reading.
+// Consumers compute the byte length themselves from `data` and `encoding`.
+// Refs: https://github.com/nodejs/node/pull/66039
+
+const requestBodyChunks = [];
+dc.subscribe('http.client.request.bodyChunkSent', ({ message, data, encoding }) => {
+  assert.ok(message instanceof http.ClientRequest);
+  requestBodyChunks.push({ data, encoding });
+});
+
+const responseBodyChunks = [];
+dc.subscribe('http.server.response.bodyChunkSent', ({ message, data, encoding }) => {
+  assert.ok(message instanceof http.ServerResponse);
+  responseBodyChunks.push({ data, encoding });
+});
+
+const requestBody = 'é'.repeat(50) + '漢'.repeat(20);
+const responseBody = 'é'.repeat(100) + '😀'.repeat(10);
+
+function byteLength(data, encoding) {
+  return typeof data === 'string' ? Buffer.byteLength(data, encoding) : data.byteLength;
+}
+
+const server = http.createServer(common.mustCall((req, res) => {
+  req.on('data', () => {});
+  req.on('end', common.mustCall(() => {
+    res.write('é'.repeat(100));
+    res.end('😀'.repeat(10));
+  }));
+}));
+
+server.listen(0, common.mustCall(() => {
+  const { port } = server.address();
+
+  const req = http.request({
+    port,
+    method: 'POST',
+  }, common.mustCall((res) => {
+    res.on('data', () => {});
+    res.on('end', common.mustCall(() => {
+      // Each user-visible body write is published once, carrying the chunk and
+      // its encoding, so the byte length can be recovered without Node needing
+      // to measure it.
+      const requestSent = requestBodyChunks.reduce(
+        (total, { data, encoding }) => total + byteLength(data, encoding), 0);
+      assert.strictEqual(requestSent, Buffer.byteLength(requestBody));
+
+      const responseSent = responseBodyChunks.reduce(
+        (total, { data, encoding }) => total + byteLength(data, encoding), 0);
+      assert.strictEqual(responseSent, Buffer.byteLength(responseBody));
+
+      // The published chunks preserve the exact bytes that were written.
+      assert.deepStrictEqual(
+        requestBodyChunks.map(({ data }) => data).join(''),
+        requestBody);
+      assert.deepStrictEqual(
+        responseBodyChunks.map(({ data }) => data).join(''),
+        responseBody);
+
+      server.close();
+    }));
+  }));
+
+  req.write('é'.repeat(50));
+  req.end('漢'.repeat(20));
+}));