nodejs
diff --git a/‎doc/api/quic.md‎
Lines changed: 28 additions & 0 deletions b/‎doc/api/quic.md‎
Lines changed: 28 additions & 0 deletions
diff --git a/‎lib/internal/quic/stats.js‎
Lines changed: 25 additions & 1 deletion b/‎lib/internal/quic/stats.js‎
Lines changed: 25 additions & 1 deletion
diff --git a/‎src/dataqueue/queue.cc‎
Lines changed: 10 additions & 0 deletions b/‎src/dataqueue/queue.cc‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎src/dataqueue/queue.h‎
Lines changed: 5 additions & 0 deletions b/‎src/dataqueue/queue.h‎
Lines changed: 5 additions & 0 deletions
@@ -2076,6 +2076,20 @@ added: v23.8.0
 
 * Type: {bigint}
 
+### `streamStats.bytesAccumulated`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {bigint}
+
+The current number of bytes sitting in the stream's receive accumulation
+buffer, awaiting delivery to the application. A value near zero indicates
+the reader is keeping up with incoming data. A value near the stream's
+flow control window indicates the application is not consuming data fast
+enough.
+
 ### `streamStats.bytesReceived`
 
 <!-- YAML
@@ -2124,6 +2138,20 @@ added: v23.8.0
 
 * Type: {bigint}
 
+### `streamStats.maxBytesAccumulated`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* Type: {bigint}
+
+The peak number of bytes that were accumulated in the stream's receive
+buffer at any point during the stream's lifetime. This value only
+increases monotonically. It is useful for diagnosing whether a stream
+experienced backpressure episodes and whether the accumulation buffer
+sizing is appropriate for the workload.
+
 ### `streamStats.maxOffset`
 
 <!-- YAML
 
@@ -108,6 +108,8 @@ const {
   IDX_STATS_STREAM_MAX_OFFSET_ACK,
   IDX_STATS_STREAM_MAX_OFFSET_RECV,
   IDX_STATS_STREAM_FINAL_SIZE,
+  IDX_STATS_STREAM_BYTES_ACCUMULATED,
+  IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED,
   IDX_STATS_STREAM_COUNT,
 } = internalBinding('quic');
 
@@ -166,6 +168,8 @@ assert(IDX_STATS_STREAM_MAX_OFFSET !== undefined);
 assert(IDX_STATS_STREAM_MAX_OFFSET_ACK !== undefined);
 assert(IDX_STATS_STREAM_MAX_OFFSET_RECV !== undefined);
 assert(IDX_STATS_STREAM_FINAL_SIZE !== undefined);
+assert(IDX_STATS_STREAM_BYTES_ACCUMULATED !== undefined);
+assert(IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED !== undefined);
 assert(IDX_STATS_STREAM_COUNT !== undefined);
 assert(IDX_STATS_SESSION_COUNT !== undefined);
 
@@ -889,6 +893,18 @@ class QuicStreamStats {
     return this.#handle[this.#offset + IDX_STATS_STREAM_FINAL_SIZE];
   }
 
+  /** @type {bigint} Current bytes in the receive accumulation buffer. */
+  get bytesAccumulated() {
+    assertIsQuicStreamStats(this);
+    return this.#handle[this.#offset + IDX_STATS_STREAM_BYTES_ACCUMULATED];
+  }
+
+  /** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+  get maxBytesAccumulated() {
+    assertIsQuicStreamStats(this);
+    return this.#handle[this.#offset + IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+  }
+
   toString() {
     return JSONStringify(this.toJSON());
   }
@@ -907,6 +923,8 @@ class QuicStreamStats {
       maxOffsetAcknowledged,
       maxOffsetReceived,
       finalSize,
+      bytesAccumulated,
+      maxBytesAccumulated,
     } = this;
     return {
       __proto__: null,
@@ -924,6 +942,8 @@ class QuicStreamStats {
       maxOffsetAcknowledged: `${maxOffsetAcknowledged}`,
       maxOffsetReceived: `${maxOffsetReceived}`,
       finalSize: `${finalSize}`,
+      bytesAccumulated: `${bytesAccumulated}`,
+      maxBytesAccumulated: `${maxBytesAccumulated}`,
     };
   }
 
@@ -950,6 +970,8 @@ class QuicStreamStats {
       maxOffsetAcknowledged,
       maxOffsetReceived,
       finalSize,
+      bytesAccumulated,
+      maxBytesAccumulated,
     } = this;
 
     return `QuicStreamStats ${inspect({
@@ -965,6 +987,8 @@ class QuicStreamStats {
       maxOffsetAcknowledged,
       maxOffsetReceived,
       finalSize,
+      bytesAccumulated,
+      maxBytesAccumulated,
     }, opts)}`;
   }
 
@@ -990,7 +1014,7 @@ class QuicStreamStats {
   // Creates an immediately disconnected QuicStreamStats object. Used when
   // lazily creating stats for a stream that has already been destroyed.
   static [kCreateDisconnected]() {
-    const count = IDX_STATS_STREAM_FINAL_SIZE + 1;
+    const count = IDX_STATS_STREAM_COUNT;
     const stats = new QuicStreamStats(kPrivateConstructor, new BigUint64Array(count), 0);
     stats.#disconnected = true;
     return stats;
 
@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
     for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
   }
 
+  void NotifyBeforePull() {
+    if (idempotent_) return;
+    for (auto& listener : backpressure_listeners_) listener->BeforePull();
+  }
+
   bool HasBackpressureListeners() const noexcept {
     return !backpressure_listeners_.empty();
   }
@@ -381,6 +386,11 @@ class NonIdempotentDataQueueReader final
            size_t max_count_hint = bob::kMaxCountHint) override {
     std::shared_ptr<DataQueue::Reader> self = shared_from_this();
 
+    // Let listeners flush pending data before we check the entries list.
+    // This allows, for example, the QUIC Stream to flush its receive
+    // accumulation buffer into the queue before the pull proceeds.
+    data_queue_->NotifyBeforePull();
+
     // If ended is true, this reader has already reached the end and cannot
     // provide any more data.
     if (ended_) {
 
@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
   class BackpressureListener {
    public:
     virtual void EntryRead(size_t amount) = 0;
+
+    // Called before the reader pulls from the DataQueue. Gives
+    // the listener a chance to flush pending data into the queue
+    // before the pull checks the entries list.
+    virtual void BeforePull() {}
   };
 
   // A DataQueue::Entry represents a logical chunk of data in the queue.