quic: additional statistics updates

doc/api/quic.md

@@ -441,6 +441,19 @@ added: REPLACEME
 A `BigInt` representing the total number of bidirectional streams
 created for this `QuicSession`.
 
+#### quicsession.blockCount
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {BigInt}
+
+A `BigInt` representing the total number of times the `QuicSession` has
+been blocked from sending stream data due to flow control.
+
+Such blocks indicate that transmitted stream data is not being consumed
+quickly enough by the connected peer.
+
 #### quicsession.bytesInFlight
 <!-- YAML
 added: REPLACEME
@@ -1615,6 +1628,26 @@ added: REPLACEME
 
 Set to `true` if the `QuicStream` is bidirectional.
 
+#### quicstream.bytesReceived
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {BigInt}
+
+A `BigInt` representing the total number of bytes received for this
+`QuicStream`.
+
+#### quicstream.bytesSent
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {BigInt}
+
+A `BigInt` representing the total number of bytes sent by this
+`QuicStream`.
+
 #### quicstream.clientInitiated
 <!-- YAML
 added: REPLACEME
@@ -1654,6 +1687,15 @@ added: REPLACEME
 -->
 TBD
 
+#### quicstream.duration
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {BigInt}
+
+A `BigInt` representing the length of time the `QuicStream` has been active.
+
 #### quicstream.id
 <!-- YAML
 added: REPLACEME
@@ -1663,6 +1705,16 @@ added: REPLACEME
 
 The numeric identifier of the `QuicStream`.
 
+#### quicstream.maxExtendedOffset
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {BigInt}
+
+A `BigInt` representing the maximum extended data offset that has been
+reported to the connected peer.
+
 #### quicstream.pending
 <!-- YAML
 added: REPLACEME

lib/internal/quic/core.js

@@ -153,6 +153,15 @@ const {
     IDX_QUIC_SESSION_STATS_LOSS_RETRANSMIT_COUNT,
     IDX_QUIC_SESSION_STATS_ACK_DELAY_RETRANSMIT_COUNT,
     IDX_QUIC_SESSION_STATS_MAX_BYTES_IN_FLIGHT,
+    IDX_QUIC_SESSION_STATS_BLOCK_COUNT,
+    IDX_QUIC_STREAM_STATS_CREATED_AT,
+    IDX_QUIC_STREAM_STATS_SENT_AT,
+    IDX_QUIC_STREAM_STATS_RECEIVED_AT,
+    IDX_QUIC_STREAM_STATS_ACKED_AT,
+    IDX_QUIC_STREAM_STATS_CLOSING_AT,
+    IDX_QUIC_STREAM_STATS_BYTES_RECEIVED,
+    IDX_QUIC_STREAM_STATS_BYTES_SENT,
+    IDX_QUIC_STREAM_STATS_MAX_OFFSET,
     ERR_INVALID_REMOTE_TRANSPORT_PARAMS,
     ERR_INVALID_TLS_SESSION_TICKET,
     NGTCP2_PATH_VALIDATION_RESULT_FAILURE,
@@ -1702,6 +1711,11 @@ class QuicSession extends EventEmitter {
       this[kHandle].state[IDX_QUIC_SESSION_STATE_BYTES_IN_FLIGHT] : 0;
   }
 
+  get blockCount() {
+    return this[kHandle] ?
+      this[kHandle].state[IDX_QUIC_SESSION_STATS_BLOCK_COUNT] : 0;
+  }
+
   get authenticated() {
     // Specifically check for null. Undefined means the check has not
     // been performed yet, another other value other than null means
@@ -2209,6 +2223,7 @@ class QuicStream extends Duplex {
   #dataRateHistogram = undefined;
   #dataSizeHistogram = undefined;
   #dataAckHistogram = undefined;
+  #stats = undefined;
 
   constructor(options, session) {
     super({
@@ -2572,8 +2587,10 @@ class QuicStream extends Duplex {
     // Do not use handle after this point as the underlying C++
     // object has been destroyed. Any attempt to use the object
     // will segfault and crash the process.
-    if (handle !== undefined)
+    if (handle !== undefined) {
+      this.#stats = new BigInt64Array(handle.stats);
       handle.destroy();
+    }
     callback(error);
   }
 
@@ -2683,6 +2700,27 @@ class QuicStream extends Duplex {
     return this[kHandle].submitTrailingHeaders(
       mapToHeaders(headers, assertValidPseudoHeaderTrailer));
   }
+
+  get duration() {
+    const now = process.hrtime.bigint();
+    const stats = this.#stats || this[kHandle].stats;
+    return now - stats[IDX_QUIC_STREAM_STATS_CREATED_AT];
+  }
+
+  get bytesReceived() {
+    const stats = this.#stats || this[kHandle].stats;
+    return stats[IDX_QUIC_STREAM_STATS_BYTES_RECEIVED];
+  }
+
+  get bytesSent() {
+    const stats = this.#stats || this[kHandle].stats;
+    return stats[IDX_QUIC_STREAM_STATS_BYTES_SENT];
+  }
+
+  get maxExtendedOffset() {
+    const stats = this.#stats || this[kHandle].stats;
+    return stats[IDX_QUIC_STREAM_STATS_MAX_OFFSET];
+  }
 }
 
 function createSocket(options = {}) {

src/node_quic.cc

@@ -213,6 +213,15 @@ void Initialize(Local<Object> target,
   NODE_DEFINE_CONSTANT(constants,
                        IDX_QUIC_SESSION_STATS_PATH_VALIDATION_FAILURE_COUNT);
   NODE_DEFINE_CONSTANT(constants, IDX_QUIC_SESSION_STATS_MAX_BYTES_IN_FLIGHT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_SESSION_STATS_BLOCK_COUNT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_CREATED_AT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_SENT_AT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_RECEIVED_AT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_ACKED_AT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_CLOSING_AT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_BYTES_RECEIVED);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_BYTES_SENT);
+  NODE_DEFINE_CONSTANT(constants, IDX_QUIC_STREAM_STATS_MAX_OFFSET);
 
   NODE_DEFINE_CONSTANT(constants, IDX_HTTP3_QPACK_MAX_TABLE_CAPACITY);
   NODE_DEFINE_CONSTANT(constants, IDX_HTTP3_QPACK_BLOCKED_STREAMS);

src/node_quic_default_application.cc

@@ -177,6 +177,7 @@ bool DefaultApplication::SendStreamData(QuicStream* stream) {
           return false;
         case NGTCP2_ERR_STREAM_DATA_BLOCKED:
           Debug(stream, "Stream data blocked");
+          Session()->StreamDataBlocked(stream->GetID());
           return true;
         case NGTCP2_ERR_STREAM_SHUT_WR:
           Debug(stream, "Stream writable side is closed");

src/node_quic_http3_application.cc

@@ -460,6 +460,7 @@ bool Http3Application::SendPendingData() {
     if (nwrite < 0) {
       switch (nwrite) {
         case NGTCP2_ERR_STREAM_DATA_BLOCKED:
+          Session()->StreamDataBlocked(stream_id);
           if (Session()->GetMaxDataLeft() == 0)
             return true;
           // Fall through

src/node_quic_session.cc

@@ -1378,6 +1378,11 @@ void QuicSession::RemoveListener(QuicSessionListener* listener) {
   listener->previous_listener_ = nullptr;
 }
 
+void QuicSession::StreamDataBlocked(int64_t stream_id) {
+  // Increments the block count
+  IncrementStat(1, &session_stats_, &session_stats::block_count);
+}
+
 std::string QuicSession::diagnostic_name() const {
   return std::string("QuicSession ") +
       (IsServer() ? "Server" : "Client") +

src/node_quic_session.h

@@ -197,7 +197,8 @@ enum QuicSessionStatsIdx : int {
   IDX_QUIC_SESSION_STATS_ACK_DELAY_RETRANSMIT_COUNT,
   IDX_QUIC_SESSION_STATS_PATH_VALIDATION_SUCCESS_COUNT,
   IDX_QUIC_SESSION_STATS_PATH_VALIDATION_FAILURE_COUNT,
-  IDX_QUIC_SESSION_STATS_MAX_BYTES_IN_FLIGHT
+  IDX_QUIC_SESSION_STATS_MAX_BYTES_IN_FLIGHT,
+  IDX_QUIC_SESSION_STATS_BLOCK_COUNT,
 };
 
 class QuicSessionListener {
@@ -933,6 +934,9 @@ class QuicSession : public AsyncWrap,
 
   void SetConnectionIDStrategory(ConnectionIDStrategy* strategy);
 
+  // Report that the stream data is flow control blocked
+  void StreamDataBlocked(int64_t stream_id);
+
   // Tracks whether or not we are currently within an ngtcp2 callback
   // function. Certain ngtcp2 APIs are not supposed to be called when
   // within a callback. We use this as a gate to check.
@@ -1370,6 +1374,9 @@ class QuicSession : public AsyncWrap,
     uint64_t path_validation_failure_count;
     // The max number of in flight bytes recorded
     uint64_t max_bytes_in_flight;
+    // The total number of times the session has been
+    // flow control blocked.
+    uint64_t block_count;
   };
   session_stats session_stats_{};
 

src/node_quic_stream.cc

@@ -85,6 +85,13 @@ QuicStream::QuicStream(
           FIXED_ONE_BYTE_STRING(env()->isolate(), "data_rx_ack"),
           data_rx_ack_->object(),
           PropertyAttribute::ReadOnly).IsNothing()) return;
+
+  ngtcp2_transport_params params;
+  ngtcp2_conn_get_local_transport_params(Session()->Connection(), &params);
+  IncrementStat(
+      params.initial_max_data,
+      &stream_stats_,
+      &stream_stats::max_offset);
 }
 
 std::string QuicStream::diagnostic_name() const {
@@ -246,6 +253,10 @@ int QuicStream::ReadStart() {
   CHECK(IsReadable());
   SetReadStart();
   SetReadResume();
+  IncrementStat(
+      inbound_consumed_data_while_paused_,
+      &stream_stats_,
+      &stream_stats::max_offset);
   session_->ExtendStreamOffset(this, inbound_consumed_data_while_paused_);
   return 0;
 }
@@ -327,8 +338,13 @@ void QuicStream::ReceiveData(
       // so that pausing does not throw off our flow control.
       if (read_paused)
         inbound_consumed_data_while_paused_ += avail;
-      else
+      else {
+        IncrementStat(
+            avail,
+            &stream_stats_,
+            &stream_stats::max_offset);
         session_->ExtendStreamOffset(this, avail);
+      }
     }
   }
 

src/node_quic_stream.h

@@ -37,6 +37,17 @@ enum QuicStreamHeadersKind : int {
   QUICSTREAM_HEADERS_KIND_TRAILING
 };
 
+enum QuicStreamStatsIdx : int {
+  IDX_QUIC_STREAM_STATS_CREATED_AT,
+  IDX_QUIC_STREAM_STATS_SENT_AT,
+  IDX_QUIC_STREAM_STATS_RECEIVED_AT,
+  IDX_QUIC_STREAM_STATS_ACKED_AT,
+  IDX_QUIC_STREAM_STATS_CLOSING_AT,
+  IDX_QUIC_STREAM_STATS_BYTES_RECEIVED,
+  IDX_QUIC_STREAM_STATS_BYTES_SENT,
+  IDX_QUIC_STREAM_STATS_MAX_OFFSET
+};
+
 // QuicHeader is a base class for implementing QUIC application
 // specific headers. Each type of QUIC application may have
 // different internal representations for a header name+value
@@ -447,8 +458,10 @@ class QuicStream : public AsyncWrap, public StreamBase {
     uint64_t bytes_received;
     // The total number of bytes sent
     uint64_t bytes_sent;
+    // The maximum extended stream offset
+    uint64_t max_offset;
   };
-  stream_stats stream_stats_{0, 0, 0, 0, 0, 0, 0};
+  stream_stats stream_stats_{};
 
   // data_rx_rate_ measures the elapsed time between data packets
   // for this stream. When used in combination with the data_rx_size,