inspector: add protocol method Network.dataReceived

PR-URL: https://github.com/nodejs/node/pull/58001
Refs: https://github.com/nodejs/node/issues/53946
Refs: https://github.com/nodejs/undici/issues/4166
Reviewed-By: James M Snell <jasnell@gmail.com>

doc/api/inspector.md

@@ -511,6 +511,19 @@ inspector.Network.requestWillBeSent({
 });
 ```
 
+### `inspector.Network.dataReceived([params])`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `params` {Object}
+
+This feature is only available with the `--experimental-network-inspection` flag enabled.
+
+Broadcasts the `Network.dataReceived` event to connected frontends, or buffers the data if
+`Network.streamResourceContent` command was not invoked for the given request yet.
+
 ### `inspector.Network.requestWillBeSent([params])`
 
 <!-- YAML

lib/inspector.js

@@ -217,6 +217,7 @@ const Network = {
   responseReceived: (params) => broadcastToFrontend('Network.responseReceived', params),
   loadingFinished: (params) => broadcastToFrontend('Network.loadingFinished', params),
   loadingFailed: (params) => broadcastToFrontend('Network.loadingFailed', params),
+  dataReceived: (params) => broadcastToFrontend('Network.dataReceived', params),
 };
 
 module.exports = {

src/inspector/network_agent.cc

@@ -15,6 +15,7 @@ using v8::Maybe;
 using v8::MaybeLocal;
 using v8::Nothing;
 using v8::Object;
+using v8::Uint8Array;
 using v8::Value;
 
 // Get a protocol string property from the object.
@@ -183,6 +184,7 @@ NetworkAgent::NetworkAgent(NetworkInspector* inspector,
   event_notifier_map_["responseReceived"] = &NetworkAgent::responseReceived;
   event_notifier_map_["loadingFailed"] = &NetworkAgent::loadingFailed;
   event_notifier_map_["loadingFinished"] = &NetworkAgent::loadingFinished;
+  event_notifier_map_["dataReceived"] = &NetworkAgent::dataReceived;
 }
 
 void NetworkAgent::emitNotification(v8::Local<v8::Context> context,
@@ -211,6 +213,30 @@ protocol::DispatchResponse NetworkAgent::disable() {
   return protocol::DispatchResponse::Success();
 }
 
+protocol::DispatchResponse NetworkAgent::streamResourceContent(
+    const protocol::String& in_requestId, protocol::Binary* out_bufferedData) {
+  if (!requests_.contains(in_requestId)) {
+    // Request not found, ignore it.
+    return protocol::DispatchResponse::InvalidParams("Request not found");
+  }
+
+  auto& it = requests_[in_requestId];
+
+  it.is_streaming = true;
+
+  // Concat response bodies.
+  *out_bufferedData = protocol::Binary::concat(it.response_data_blobs);
+  // Clear buffered data.
+  it.response_data_blobs.clear();
+
+  if (it.is_finished) {
+    // If the request is finished, remove the entry.
+    requests_.erase(in_requestId);
+  }
+
+  return protocol::DispatchResponse::Success();
+}
+
 void NetworkAgent::requestWillBeSent(v8::Local<v8::Context> context,
                                      v8::Local<v8::Object> params) {
   protocol::String request_id;
@@ -247,6 +273,12 @@ void NetworkAgent::requestWillBeSent(v8::Local<v8::Context> context,
                                std::move(initiator),
                                timestamp,
                                wall_time);
+
+  if (requests_.contains(request_id)) {
+    // Duplicate entry, ignore it.
+    return;
+  }
+  requests_.emplace(request_id, RequestEntry{timestamp, false, false, {}});
 }
 
 void NetworkAgent::responseReceived(v8::Local<v8::Context> context,
@@ -295,6 +327,8 @@ void NetworkAgent::loadingFailed(v8::Local<v8::Context> context,
   }
 
   frontend_->loadingFailed(request_id, timestamp, type, error_text);
+
+  requests_.erase(request_id);
 }
 
 void NetworkAgent::loadingFinished(v8::Local<v8::Context> context,
@@ -309,6 +343,63 @@ void NetworkAgent::loadingFinished(v8::Local<v8::Context> context,
   }
 
   frontend_->loadingFinished(request_id, timestamp);
+
+  auto request_entry = requests_.find(request_id);
+  if (request_entry == requests_.end()) {
+    // No entry found. Ignore it.
+    return;
+  }
+
+  if (request_entry->second.is_streaming) {
+    // Streaming finished, remove the entry.
+    requests_.erase(request_id);
+  } else {
+    request_entry->second.is_finished = true;
+  }
+}
+
+void NetworkAgent::dataReceived(v8::Local<v8::Context> context,
+                                v8::Local<v8::Object> params) {
+  protocol::String request_id;
+  if (!ObjectGetProtocolString(context, params, "requestId").To(&request_id)) {
+    return;
+  }
+
+  auto request_entry = requests_.find(request_id);
+  if (request_entry == requests_.end()) {
+    // No entry found. Ignore it.
+    return;
+  }
+
+  double timestamp;
+  if (!ObjectGetDouble(context, params, "timestamp").To(&timestamp)) {
+    return;
+  }
+  int data_length;
+  if (!ObjectGetInt(context, params, "dataLength").To(&data_length)) {
+    return;
+  }
+  int encoded_data_length;
+  if (!ObjectGetInt(context, params, "encodedDataLength")
+           .To(&encoded_data_length)) {
+    return;
+  }
+  Local<Object> data_obj;
+  if (!ObjectGetObject(context, params, "data").ToLocal(&data_obj)) {
+    return;
+  }
+  if (!data_obj->IsUint8Array()) {
+    return;
+  }
+  Local<Uint8Array> data = data_obj.As<Uint8Array>();
+  auto data_bin = protocol::Binary::fromUint8Array(data);
+
+  if (request_entry->second.is_streaming) {
+    frontend_->dataReceived(
+        request_id, timestamp, data_length, encoded_data_length, data_bin);
+  } else {
+    requests_[request_id].response_data_blobs.push_back(data_bin);
+  }
 }
 
 }  // namespace inspector

src/inspector/network_agent.h

@@ -3,6 +3,7 @@
 
 #include "node/inspector/protocol/Network.h"
 
+#include <map>
 #include <unordered_map>
 
 namespace node {
@@ -10,6 +11,13 @@ namespace inspector {
 
 class NetworkInspector;
 
+struct RequestEntry {
+  double timestamp;
+  bool is_finished;
+  bool is_streaming;
+  std::vector<protocol::Binary> response_data_blobs;
+};
+
 class NetworkAgent : public protocol::Network::Backend {
  public:
   explicit NetworkAgent(NetworkInspector* inspector,
@@ -21,6 +29,10 @@ class NetworkAgent : public protocol::Network::Backend {
 
   protocol::DispatchResponse disable() override;
 
+  protocol::DispatchResponse streamResourceContent(
+      const protocol::String& in_requestId,
+      protocol::Binary* out_bufferedData) override;
+
   void emitNotification(v8::Local<v8::Context> context,
                         const protocol::String& event,
                         v8::Local<v8::Object> params);
@@ -37,6 +49,9 @@ class NetworkAgent : public protocol::Network::Backend {
   void loadingFinished(v8::Local<v8::Context> context,
                        v8::Local<v8::Object> params);
 
+  void dataReceived(v8::Local<v8::Context> context,
+                    v8::Local<v8::Object> params);
+
  private:
   NetworkInspector* inspector_;
   v8_inspector::V8Inspector* v8_inspector_;
@@ -44,6 +59,7 @@ class NetworkAgent : public protocol::Network::Backend {
   using EventNotifier = void (NetworkAgent::*)(v8::Local<v8::Context> context,
                                                v8::Local<v8::Object>);
   std::unordered_map<protocol::String, EventNotifier> event_notifier_map_;
+  std::map<protocol::String, RequestEntry> requests_;
 };
 
 }  // namespace inspector

src/inspector/node_protocol.pdl

@@ -183,6 +183,16 @@ experimental domain Network
   # Enables network tracking, network events will now be delivered to the client.
   command enable
 
+  # Enables streaming of the response for the given requestId.
+  # If enabled, the dataReceived event contains the data that was received during streaming.
+  experimental command streamResourceContent
+    parameters
+      # Identifier of the request to stream.
+      RequestId requestId
+    returns
+      # Data that has been buffered until streaming is enabled.
+      binary bufferedData
+
   # Fired when page is about to send HTTP request.
   event requestWillBeSent
     parameters
@@ -227,6 +237,20 @@ experimental domain Network
       # Timestamp.
       MonotonicTime timestamp
 
+  # Fired when data chunk was received over the network.
+  event dataReceived
+    parameters
+      # Request identifier.
+      RequestId requestId
+      # Timestamp.
+      MonotonicTime timestamp
+      # Data chunk length.
+      integer dataLength
+      # Actual bytes received (might be less than dataLength for compressed encodings).
+      integer encodedDataLength
+      # Data that was received.
+      experimental optional binary data
+
 # Support for inspecting node process state.
 experimental domain NodeRuntime
   # Enable the NodeRuntime events except by `NodeRuntime.waitingForDisconnect`.

src/inspector/node_string.cc

@@ -28,6 +28,20 @@ void ProtocolTypeTraits<std::string>::Serialize(const std::string& value,
   cbor::EncodeString8(SpanFrom(value), bytes);
 }
 
+bool ProtocolTypeTraits<node::inspector::protocol::Binary>::Deserialize(
+    DeserializerState* state, node::inspector::protocol::Binary* value) {
+  CHECK(state->tokenizer()->TokenTag() == cbor::CBORTokenTag::BINARY);
+  span<uint8_t> cbor_span = state->tokenizer()->GetBinary();
+  *value = node::inspector::protocol::Binary::fromSpan(cbor_span);
+  return true;
+}
+
+void ProtocolTypeTraits<node::inspector::protocol::Binary>::Serialize(
+    const node::inspector::protocol::Binary& value,
+    std::vector<uint8_t>* bytes) {
+  cbor::EncodeString8(SpanFrom(value.toBase64()), bytes);
+}
+
 }  // namespace crdtp
 
 namespace node {
@@ -93,6 +107,58 @@ size_t StringUtil::CharacterCount(const std::string_view s) {
   return s.length();
 }
 
+String Binary::toBase64() const {
+  MaybeStackBuffer<char> buffer;
+  size_t str_len = simdutf::base64_length_from_binary(bytes_->size());
+  buffer.SetLength(str_len);
+
+  size_t len =
+      simdutf::binary_to_base64(reinterpret_cast<const char*>(bytes_->data()),
+                                bytes_->size(),
+                                buffer.out());
+  CHECK_EQ(len, str_len);
+  return buffer.ToString();
+}
+
+// static
+Binary Binary::concat(const std::vector<Binary>& binaries) {
+  size_t total_size = 0;
+  for (const auto& binary : binaries) {
+    total_size += binary.size();
+  }
+  auto bytes = std::make_shared<std::vector<uint8_t>>(total_size);
+  uint8_t* data_ptr = bytes->data();
+  for (const auto& binary : binaries) {
+    memcpy(data_ptr, binary.data(), binary.size());
+    data_ptr += binary.size();
+  }
+  return Binary(bytes);
+}
+
+// static
+Binary Binary::fromBase64(const String& base64, bool* success) {
+  Binary binary{};
+  size_t base64_len = simdutf::maximal_binary_length_from_base64(
+      base64.data(), base64.length());
+  binary.bytes_->resize(base64_len);
+
+  simdutf::result result;
+  result =
+      simdutf::base64_to_binary(base64.data(),
+                                base64.length(),
+                                reinterpret_cast<char*>(binary.bytes_->data()));
+  CHECK_EQ(result.error, simdutf::error_code::SUCCESS);
+  return binary;
+}
+
+// static
+Binary Binary::fromUint8Array(v8::Local<v8::Uint8Array> data) {
+  auto bytes = std::make_shared<std::vector<uint8_t>>(data->ByteLength());
+  size_t size = data->CopyContents(bytes->data(), data->ByteLength());
+  CHECK_EQ(size, data->ByteLength());
+  return Binary(bytes);
+}
+
 }  // namespace protocol
 }  // namespace inspector
 }  // namespace node

src/inspector/node_string.h

@@ -3,13 +3,17 @@
 #ifndef SRC_INSPECTOR_NODE_STRING_H_
 #define SRC_INSPECTOR_NODE_STRING_H_
 
+#include <cstring>
+#include <sstream>
+#include <string>
+#include "crdtp/maybe.h"
 #include "crdtp/protocol_core.h"
 #include "util.h"
 #include "v8-inspector.h"
 
-#include <cstring>
-#include <sstream>
-#include <string>
+namespace node::inspector::protocol {
+class Binary;
+}
 
 namespace crdtp {
 
@@ -19,6 +23,19 @@ struct ProtocolTypeTraits<std::string> {
   static void Serialize(const std::string& value, std::vector<uint8_t>* bytes);
 };
 
+template <>
+struct ProtocolTypeTraits<node::inspector::protocol::Binary> {
+  static bool Deserialize(DeserializerState* state,
+                          node::inspector::protocol::Binary* value);
+  static void Serialize(const node::inspector::protocol::Binary& value,
+                        std::vector<uint8_t>* bytes);
+};
+
+template <>
+struct detail::MaybeTypedef<node::inspector::protocol::Binary> {
+  typedef ValueMaybe<node::inspector::protocol::Binary> type;
+};
+
 }  // namespace crdtp
 
 namespace node {
@@ -55,18 +72,32 @@ struct StringUtil {
 };
 
 // A read-only sequence of uninterpreted bytes with reference-counted storage.
-// Though the templates for generating the protocol bindings reference
-// this type, js_protocol.pdl doesn't have a field of type 'binary', so
-// therefore it's unnecessary to provide an implementation here.
 class Binary {
  public:
-  const uint8_t* data() const { UNREACHABLE(); }
-  size_t size() const { UNREACHABLE(); }
-  String toBase64() const { UNREACHABLE(); }
-  static Binary fromBase64(const std::string_view base64, bool* success) {
-    UNREACHABLE();
+  Binary() : bytes_(std::make_shared<std::vector<uint8_t>>()) {}
+
+  const uint8_t* data() const { return bytes_->data(); }
+  size_t size() const { return bytes_->size(); }
+
+  String toBase64() const;
+
+  static Binary concat(const std::vector<Binary>& binaries);
+
+  static Binary fromBase64(const String& base64, bool* success);
+  static Binary fromUint8Array(v8::Local<v8::Uint8Array> data);
+  static Binary fromSpan(const uint8_t* data, size_t size) {
+    return Binary::fromSpan(crdtp::span<uint8_t>(data, size));
   }
-  static Binary fromSpan(const uint8_t* data, size_t size) { UNREACHABLE(); }
+  static Binary fromSpan(crdtp::span<uint8_t> span) {
+    return Binary(
+        std::make_shared<std::vector<uint8_t>>(span.begin(), span.end()));
+  }
+
+ private:
+  std::shared_ptr<std::vector<uint8_t>> bytes_;
+
+  explicit Binary(std::shared_ptr<std::vector<uint8_t>> bytes)
+      : bytes_(bytes) {}
 };
 
 }  // namespace protocol

test/parallel/test-inspector-emit-protocol-event.js

@@ -57,6 +57,11 @@ const EXPECTED_EVENTS = {
         }
       }
     },
+    {
+      name: 'dataReceived',
+      // Network.dataReceived is buffered until Network.streamResourceContent is invoked.
+      skip: true,
+    },
     {
       name: 'loadingFinished',
       params: {
@@ -81,8 +86,8 @@ for (const [domain, events] of Object.entries(EXPECTED_EVENTS)) {
   if (!(domain in inspector)) {
     assert.fail(`Expected domain ${domain} to be present in inspector`);
   }
-  const actualEventNames = Object.keys(inspector[domain]);
-  const expectedEventNames = events.map((event) => event.name);
+  const actualEventNames = Object.keys(inspector[domain]).sort();
+  const expectedEventNames = events.map((event) => event.name).sort();
   assert.deepStrictEqual(actualEventNames, expectedEventNames, `Expected ${domain} to have events ${expectedEventNames}, but got ${actualEventNames}`);
 }
 
@@ -105,6 +110,9 @@ const runAsyncTest = async () => {
   await session.post('Network.enable');
   for (const [domain, events] of Object.entries(EXPECTED_EVENTS)) {
     for (const event of events) {
+      if (event.skip) {
+        continue;
+      }
       session.on(`${domain}.${event.name}`, common.mustCall(({ params }) => {
         if (event.name === 'requestWillBeSent') {
           // Initiator is automatically captured and contains caller info.

test/parallel/test-inspector-network-data-received.js

@@ -0,0 +1,146 @@
+// Flags: --inspect=0 --experimental-network-inspection
+'use strict';
+const common = require('../common');
+
+common.skipIfInspectorDisabled();
+
+const inspector = require('node:inspector/promises');
+const { Network } = require('node:inspector');
+const test = require('node:test');
+const assert = require('node:assert');
+const { waitUntil } = require('../common/inspector-helper');
+const { setTimeout } = require('node:timers/promises');
+
+const session = new inspector.Session();
+session.connect();
+session.post('Network.enable');
+
+async function triggerNetworkEvents(requestId) {
+  const url = 'https://example.com';
+  Network.requestWillBeSent({
+    requestId,
+    timestamp: 1,
+    wallTime: 1,
+    request: {
+      url,
+      method: 'GET',
+      headers: {
+        mKey: 'mValue',
+      },
+    },
+  });
+  await setTimeout(1);
+
+  Network.responseReceived({
+    requestId,
+    timestamp: 2,
+    type: 'Fetch',
+    response: {
+      url,
+      status: 200,
+      statusText: 'OK',
+      headers: {
+        mKey: 'mValue',
+      },
+    },
+  });
+  await setTimeout(1);
+
+  const chunk1 = Buffer.from('Hello, ');
+  Network.dataReceived({
+    requestId,
+    timestamp: 3,
+    dataLength: chunk1.byteLength,
+    encodedDataLength: chunk1.byteLength,
+    data: chunk1,
+  });
+  await setTimeout(1);
+
+  const chunk2 = Buffer.from('world');
+  Network.dataReceived({
+    requestId,
+    timestamp: 4,
+    dataLength: chunk2.byteLength,
+    encodedDataLength: chunk2.byteLength,
+    data: chunk2,
+  });
+  await setTimeout(1);
+
+  Network.loadingFinished({
+    requestId,
+    timestamp: 5,
+  });
+}
+
+test('should stream Network.dataReceived with data chunks', async () => {
+  session.removeAllListeners();
+
+  const requestId = 'my-req-id-1';
+  const chunks = [];
+  let totalDataLength = 0;
+  session.on('Network.requestWillBeSent', common.mustCall(({ params }) => {
+    assert.strictEqual(params.requestId, requestId);
+  }));
+  const responseReceivedFuture = waitUntil(session, 'Network.responseReceived')
+    .then(async ([{ params }]) => {
+      assert.strictEqual(params.requestId, requestId);
+      const { bufferedData } = await session.post('Network.streamResourceContent', {
+        requestId,
+      });
+      const data = Buffer.from(bufferedData, 'base64');
+      totalDataLength += data.byteLength;
+      chunks.push(data);
+    });
+  const loadingFinishedFuture = waitUntil(session, 'Network.loadingFinished')
+    .then(([{ params }]) => {
+      assert.strictEqual(params.requestId, requestId);
+    });
+  session.on('Network.dataReceived', ({ params }) => {
+    assert.strictEqual(params.requestId, requestId);
+    totalDataLength += params.dataLength;
+    chunks.push(Buffer.from(params.data, 'base64'));
+  });
+
+  await triggerNetworkEvents(requestId);
+  await responseReceivedFuture;
+  await loadingFinishedFuture;
+
+  const data = Buffer.concat(chunks);
+  assert.strictEqual(data.byteLength, totalDataLength, data);
+  assert.strictEqual(data.toString('utf8'), 'Hello, world');
+});
+
+test('Network.streamResourceContent should send all buffered chunks', async () => {
+  session.removeAllListeners();
+
+  const requestId = 'my-req-id-2';
+  session.on('Network.requestWillBeSent', common.mustCall(({ params }) => {
+    assert.strictEqual(params.requestId, requestId);
+  }));
+  session.on('Network.responseReceived', common.mustCall(({ params }) => {
+    assert.strictEqual(params.requestId, requestId);
+  }));
+  const loadingFinishedFuture = waitUntil(session, 'Network.loadingFinished')
+    .then(async ([{ params }]) => {
+      assert.strictEqual(params.requestId, requestId);
+    });
+  session.on('Network.dataReceived', common.mustNotCall());
+
+  await triggerNetworkEvents(requestId);
+  await loadingFinishedFuture;
+  const { bufferedData } = await session.post('Network.streamResourceContent', {
+    requestId,
+  });
+  assert.strictEqual(Buffer.from(bufferedData, 'base64').toString('utf8'), 'Hello, world');
+});
+
+test('Network.streamResourceContent should reject if request id not found', async () => {
+  session.removeAllListeners();
+
+  const requestId = 'unknown-request-id';
+  await assert.rejects(session.post('Network.streamResourceContent', {
+    requestId,
+  }), {
+    code: 'ERR_INSPECTOR_COMMAND',
+  });
+});