quic: coalesce received data into fewer buffers

Signed-off-by: James M Snell <jasnell@gmail.com>
Assisted-by: Opencode:Opus 4.6
PR-URL: #63267
Backport-PR-URL: #64675
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: jasnell

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.

doc/api/quic.md

@@ -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

lib/internal/quic/stats.js

@@ -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 {
 returnthis.#handle[this.#offset +IDX_STATS_STREAM_FINAL_SIZE];
 }
 
+/** @type {bigint} Current bytes in the receive accumulation buffer. */
+getbytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_BYTES_ACCUMULATED];
+}
+
+/** @type {bigint} Peak bytes accumulated over the stream's lifetime. */
+getmaxBytesAccumulated(){
+assertIsQuicStreamStats(this);
+returnthis.#handle[this.#offset +IDX_STATS_STREAM_MAX_BYTES_ACCUMULATED];
+}
+
 toString(){
 returnJSONStringify(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](){
-constcount=IDX_STATS_STREAM_FINAL_SIZE+1;
+constcount=IDX_STATS_STREAM_COUNT;
 conststats=newQuicStreamStats(kPrivateConstructor,newBigUint64Array(count),0);
 stats.#disconnected =true;
 returnstats;

src/dataqueue/queue.cc

@@ -179,6 +179,11 @@ class DataQueueImpl final : public DataQueue,
 for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);
  }
 
+voidNotifyBeforePull() {
+if (idempotent_) return;
+for (auto& listener : backpressure_listeners_) listener->BeforePull();
+ }
+
 boolHasBackpressureListeners() constnoexcept {
 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_) {

src/dataqueue/queue.h

@@ -147,6 +147,11 @@ class DataQueue : public MemoryRetainer {
 classBackpressureListener {
 public:
 virtualvoidEntryRead(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.
+virtualvoidBeforePull() {}
  };
 
 // A DataQueue::Entry represents a logical chunk of data in the queue.