quic: update writedesiredsize on maxdataframe

This commit will react on a callback from
ngtcp2, when a maxdata frame arrives and
update the scheduled streams accordingly.
A PR to get this callback is pending.

Fixes: #64835

Signed-off-by: Marten Richter <marten.richter@freenet.de>

Author: martenrichter

src/quic/application.cc

@@ -419,7 +419,7 @@ class DefaultApplication final : public Session::Application {
   void BlockStream(stream_id id) override {
     if (auto stream = session().FindStream(id)) [[likely]] {
       // Remove the stream from the send queue. It will be re-scheduled
-      // via ExtendMaxStreamData when the peer grants more flow control.
+      // via ExtendMax(Stream)Data when the peer grants more flow control.
       // Without this, SendPendingData would repeatedly pop and retry
       // the same blocked stream in an infinite loop.
       stream->Unschedule();
@@ -434,6 +434,15 @@ class DefaultApplication final : public Session::Application {
     stream->Schedule(&stream_queue_);
   }
 
+  void ExtendMaxData(uint64_t max_data) override {
+    // The peer granted more flow control for session. Re-schedule
+    // all streams so SendPendingData will resume writing.
+    for (auto& [id, stream] : session().streams()) {
+      stream->Schedule(&stream_queue_);
+    }
+  }
+  
+
   bool StreamCommit(Session::StreamData* stream_data, size_t datalen) override {
     DCHECK_NOT_NULL(stream_data);
     CHECK(stream_data->stream);

src/quic/application.h

@@ -149,6 +149,14 @@ class Session::Application : public MemoryRetainer {
     // By default do nothing.
   }
 
+  // Called when the Session determines that the flow control window for the
+  // session has been expanded. Not all Application types will require
+  // this notification so the default is to do nothing.
+  virtual void ExtendMaxData(uint64_t max_data) {
+    Debug(session_, "Application extending max data");
+    // By default do nothing.
+  }
+
   // Different Applications may wish to set some application data in the
   // session ticket (e.g. http/3 would set server settings in the application
   // data). The first byte written MUST be the Application::Type enum value.

src/quic/http3.cc

@@ -378,6 +378,16 @@ class Http3ApplicationImpl final : public Session::Application {
     nghttp3_conn_unblock_stream(*this, stream->id());
   }
 
+  void ExtendMaxData(uint64_t max_data) override {
+    Debug(&session(),
+          "HTTP/3 application extending max data to %" PRIu64,
+          max_data);
+    for (auto& [id, stream] : session().streams()) {
+      stream->UpdateWriteDesiredSize();  // the stream might be blocked on js side
+      // is unblock stream also required?
+    }
+  }
+
   void CollectSessionTicketAppData(
       SessionTicket::AppData* app_data) const override {
     uint8_t buf[kSessionTicketAppDataSize];

src/quic/session.cc

@@ -1358,6 +1358,14 @@ struct Session::Impl final : public MemoryRetainer {
     return NGTCP2_SUCCESS;
   }
 
+  static int on_extend_max_data(ngtcp2_conn* conn,
+                                uint64_t max_data,
+                                void* user_data) {
+    NGTCP2_CALLBACK_SCOPE(session)
+    session->application().ExtendMaxData(max_data);
+    return NGTCP2_SUCCESS;
+  }
+
   static int on_get_new_cid(ngtcp2_conn* conn,
                             ngtcp2_cid* cid,
                             ngtcp2_stateless_reset_token* token,
@@ -1700,6 +1708,9 @@ struct Session::Impl final : public MemoryRetainer {
       on_receive_stream_stop_sending,
 #ifdef NGTCP2_CALLBACKS_V5
       nullptr,
+#ifdef NGTCP2_CALLBACKS_V6
+      on_extend_max_data,
+#endif
 #endif  // NGTCP2_CALLBACKS_V5
 #endif  // NGTCP2_CALLBACKS_V4
   };
@@ -1754,6 +1765,9 @@ struct Session::Impl final : public MemoryRetainer {
       on_receive_stream_stop_sending,
 #ifdef NGTCP2_CALLBACKS_V5
       nullptr,
+#ifdef NGTCP2_CALLBACKS_V6
+      on_extend_max_data,
+#endif
 #endif  // NGTCP2_CALLBACKS_V5
 #endif  // NGTCP2_CALLBACKS_V4
   };

test/parallel/test-quic-h3-maxdata-external-buffer-failure.mjs

@@ -0,0 +1,76 @@
+// Flags: --experimental-quic --experimental-stream-iter --no-warnings
+
+// Test: Quic maxdata updates on http/3
+// Client sends a body that precisely fills the session window size,
+// and verifies that it is data transfer is not stalled.
+
+import { hasQuic, skip } from '../common/index.mjs';
+import { readFile } from 'node:fs/promises';
+import { setTimeout as sleep } from 'node:timers/promises';
+
+if (!hasQuic) {
+  skip('QUIC is not enabled');
+}
+const { listen, connect } = await import('node:quic');
+const { createPrivateKey } = await import('node:crypto');
+const { drainableProtocol } = await import('stream/iter');
+
+const keys = 'test/fixtures/keys';
+const key = createPrivateKey(await readFile(`${keys}/agent1-key.pem`));
+const cert = await readFile(`${keys}/agent1-cert.pem`);
+
+const WINDOW = 4096;
+// Fills the window exactly:
+// considers all framing including some initial session capsules
+const BODY = WINDOW - 38;
+
+let letServerRead;
+const serverMayRead = new Promise((resolve) => { letServerRead = resolve; });
+
+const endpoint = await listen((session) => {
+  session.onstream = async (stream) => {
+    await serverMayRead;
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of stream) { /* reading extends the window */ }
+  };
+}, {
+  sni: { '*': { keys: [key], certs: [cert] } },
+  transportParams: {
+    initialMaxStreamDataBidiRemote: 1024 * 1024,  // Make sure maxstreamdata does not block
+    initialMaxData: WINDOW,
+  },
+  onheaders() { this.sendHeaders({ ':status': '200' }); },
+});
+
+const session = await connect(endpoint.address, {
+  servername: 'localhost',
+  verifyPeer: 'manual',
+});
+await session.opened;
+
+// Budget well above the window, so the window is what stops the writer.
+const stream = await session.createBidirectionalStream({ budget: 1024 * 1024 });
+stream.sendHeaders({
+  ':method': 'POST',
+  ':path': '/',
+  ':scheme': 'https',
+  ':authority': 'localhost',
+}, { terminal: false });
+
+const writer = stream.writer;
+writer.writeSync(new Uint8Array(BODY));
+
+// Long enough for every byte to be acked. The peer acks as data arrives,
+// whether or not its application has read any of it, so by now the window is
+// exhausted, the send buffer is empty, and no further ACK can arrive.
+await sleep(500);
+
+const watchdog = setTimeout(() => {
+  console.error('STALLED: no drain after MAX_STREAM_DATA');
+  process.exit(1);
+}, 5000);
+letServerRead();                 // Extend the window, with no ack attached
+await writer[drainableProtocol]();
+
+clearTimeout(watchdog);
+process.exit(0);