http2: fix write deadlock exposed by larger window sizes

This removes a guard (no reads while write pending) that creates
this deadlock, which was added as a security mechanism. This guard is
redundant given then other existing mechanisms, and a test is added to
demonstrate that.

Signed-off-by: Tim Perry <pimterry@gmail.com>
PR-URL: #65440
Backport-PR-URL: #65953
Signed-off-by: Filip Skokan <panva.ip@gmail.com>
Assisted-by: Codex
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Richard Lau <richard.lau@ibm.com>
Reviewed-By: Tim Perry <pimterry@gmail.com>

Author: pimterry

benchmark/http2/full-duplex.js

@@ -0,0 +1,71 @@
+'use strict';
+
+const common = require('../common.js');
+const fixtures = require('../../test/common/fixtures');
+
+const bench = common.createBenchmark(main, {
+  n: [100],
+  streams: [2],
+  size: [4 * 1024 * 1024],
+  // Use the HTTP/2 protocol default.
+  window: [65535],
+}, {
+  test: { size: 128 * 1024, window: 65535 },
+});
+
+function main({ n, streams, size, window }) {
+  const http2 = require('http2');
+  const payload = Buffer.alloc(size);
+  const server = http2.createSecureServer({
+    key: fixtures.readKey('agent1-key.pem'),
+    cert: fixtures.readKey('agent1-cert.pem'),
+    settings: { initialWindowSize: window },
+  });
+
+  let completed = 0;
+  let batches = 0;
+
+  function onTransferComplete() {
+    if (++completed !== streams * 2)
+      return;
+
+    if (++batches === n) {
+      // Report combined upload and download throughput in MiB/s.
+      bench.end(n * streams * size * 2 / (1024 * 1024));
+      client.close();
+      server.close();
+      return;
+    }
+
+    startBatch();
+  }
+
+  server.on('stream', (stream) => {
+    stream.resume();
+    stream.on('end', onTransferComplete);
+    stream.respond();
+    stream.end(payload);
+  });
+
+  let client;
+  function startBatch() {
+    completed = 0;
+    for (let i = 0; i < streams; i++) {
+      const request = client.request({ ':method': 'POST' });
+      request.resume();
+      request.on('end', onTransferComplete);
+      request.end(payload);
+    }
+  }
+
+  server.listen(0, () => {
+    client = http2.connect(`https://localhost:${server.address().port}`, {
+      rejectUnauthorized: false,
+      settings: { initialWindowSize: window },
+    });
+    client.on('connect', () => {
+      bench.start();
+      startBatch();
+    });
+  });
+}

src/node_http2.cc

@@ -970,45 +970,24 @@ ssize_t Http2Session::OnMaxFrameSizePadding(size_t frameLen,
 // quite expensive. This is a potential performance optimization target later.
 void Http2Session::ConsumeHTTP2Data() {
   CHECK_NOT_NULL(stream_buf_.base);
-  CHECK_LE(stream_buf_offset_, stream_buf_.len);
-  size_t read_len = stream_buf_.len - stream_buf_offset_;
 
   // multiple side effects.
-  Debug(this, "receiving %d bytes [wants data? %d]",
-        read_len,
+  Debug(this,
+        "receiving %d bytes [wants data? %d]",
+        stream_buf_.len,
         nghttp2_session_want_read(session_.get()));
-  set_receive_paused(false);
   custom_recv_error_code_ = nullptr;
   set_receiving();
   ssize_t ret =
-    nghttp2_session_mem_recv(session_.get(),
-                             reinterpret_cast<uint8_t*>(stream_buf_.base) +
-                                 stream_buf_offset_,
-                             read_len);
+      nghttp2_session_mem_recv(session_.get(),
+                               reinterpret_cast<uint8_t*>(stream_buf_.base),
+                               stream_buf_.len);
   set_receiving(false);
   CHECK_NE(ret, NGHTTP2_ERR_NOMEM);
   CHECK_IMPLIES(custom_recv_error_code_ != nullptr, ret < 0);
 
-  if (is_receive_paused()) {
-    CHECK(is_reading_stopped());
-
-    CHECK_GT(ret, 0);
-    CHECK_LE(static_cast<size_t>(ret), read_len);
-
-    // Mark the remainder of the data as available for later consumption.
-    // Even if all bytes were received, a paused stream may delay the
-    // nghttp2_on_frame_recv_callback which may have an END_STREAM flag.
-    stream_buf_offset_ += ret;
-    // Still complete a Close() deferred during mem_recv; do not fall through
-    // to SendPendingData() here (paused receives historically skip that flush
-    // because a write may already be in progress).
-    MaybeFinishPendingClose();
-    goto done;
-  }
-
   // We are done processing the current input chunk.
   DecrementCurrentSessionMemory(stream_buf_.len);
-  stream_buf_offset_ = 0;
   stream_buf_ab_.Reset();
   stream_buf_allocation_.reset();
   stream_buf_ = uv_buf_init(nullptr, 0);
@@ -1017,14 +996,6 @@ void Http2Session::ConsumeHTTP2Data() {
   // not written after pending RST_STREAM frames.
   MaybeFinishPendingClose();
 
-done:
-  // Finish a Close() deferred above before flushing, so GOAWAY is not written
-  // after pending RST_STREAM frames.
-  if (is_close_pending() && !is_destroyed()) {
-    set_close_pending(false);
-    FinishClose(pending_close_code_, pending_close_socket_closed_);
-  }
-
   // Send any data that was queued up while processing the received data.
   if (ret >= 0 && !is_destroyed()) {
     SendPendingData();
@@ -1478,15 +1449,6 @@ int Http2Session::OnDataChunkReceived(nghttp2_session* handle,
     }
   } while (len != 0);
 
-  // If we are currently waiting for a write operation to finish, we should
-  // tell nghttp2 that we want to wait before we process more input data.
-  if (session->is_write_in_progress()) {
-    CHECK(session->is_reading_stopped());
-    session->set_receive_paused();
-    Debug(session, "receive paused");
-    return NGHTTP2_ERR_PAUSE;
-  }
-
   return 0;
 }
 
@@ -1569,7 +1531,6 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   size_t offset = buf.base - session->stream_buf_.base;
 
   // Verify that the data offset is inside the current read buffer.
-  CHECK_GE(offset, session->stream_buf_offset_);
   CHECK_LE(offset, session->stream_buf_.len);
   CHECK_LE(offset + buf.len, session->stream_buf_.len);
 
@@ -1883,11 +1844,6 @@ void Http2Session::OnStreamAfterWrite(WriteWrap* w, int status) {
     return;
   }
 
-  // If there is more incoming data queued up, consume it.
-  if (stream_buf_offset_ > 0) {
-    ConsumeHTTP2Data();
-  }
-
   if (!is_write_scheduled() && !is_destroyed()) {
     // Schedule a new write if nghttp2 wants to send data.
     MaybeScheduleWrite();
@@ -1934,7 +1890,7 @@ void Http2Session::MaybeStopReading() {
   if (is_reading_stopped() || is_closing()) return;
   int want_read = nghttp2_session_want_read(session_.get());
   Debug(this, "wants read? %d", want_read);
-  if (want_read == 0 || is_write_in_progress()) {
+  if (want_read == 0) {
     set_reading_stopped();
     stream_->ReadStop();
   }
@@ -2199,7 +2155,7 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
   Context::Scope context_scope(env()->context());
   Http2Scope h2scope(this);
   CHECK_NOT_NULL(stream_);
-  Debug(this, "receiving %d bytes, offset %d", nread, stream_buf_offset_);
+  Debug(this, "receiving %d bytes", nread);
   std::unique_ptr<BackingStore> bs = env()->release_managed_buffer(buf_);
 
   // Only pass data on if nread > 0
@@ -2214,40 +2170,18 @@ void Http2Session::OnStreamRead(ssize_t nread, const uv_buf_t& buf_) {
 
   statistics_.data_received += nread;
 
-  if (stream_buf_offset_ == 0 && static_cast<size_t>(nread) != bs->ByteLength())
-      [[likely]] {
+  // ConsumeHTTP2Data() always consumes the whole chunk, so there is never a
+  // partially processed buffer left over from a previous read.
+  DCHECK_NULL(stream_buf_.base);
+
+  if (static_cast<size_t>(nread) != bs->ByteLength()) [[likely]] {
     // Shrink to the actual amount of used data.
     std::unique_ptr<BackingStore> old_bs = std::move(bs);
     bs = ArrayBuffer::NewBackingStore(
         env()->isolate(),
         nread,
         BackingStoreInitializationMode::kUninitialized);
     memcpy(bs->Data(), old_bs->Data(), nread);
-  } else {
-    // This is a very unlikely case, and should only happen if the ReadStart()
-    // call in OnStreamAfterWrite() immediately provides data. If that does
-    // happen, we concatenate the data we received with the already-stored
-    // pending input data, slicing off the already processed part.
-    size_t pending_len = stream_buf_.len - stream_buf_offset_;
-    std::unique_ptr<BackingStore> new_bs = ArrayBuffer::NewBackingStore(
-        env()->isolate(),
-        pending_len + nread,
-        BackingStoreInitializationMode::kUninitialized);
-    memcpy(static_cast<char*>(new_bs->Data()),
-           stream_buf_.base + stream_buf_offset_,
-           pending_len);
-    memcpy(static_cast<char*>(new_bs->Data()) + pending_len,
-           bs->Data(),
-           nread);
-
-    bs = std::move(new_bs);
-    nread = bs->ByteLength();
-    stream_buf_offset_ = 0;
-    stream_buf_ab_.Reset();
-
-    // We have now fully processed the stream_buf_ input chunk (by moving the
-    // remaining part into buf, which will be accounted for below).
-    DecrementCurrentSessionMemory(stream_buf_.len);
   }
 
   IncrementCurrentSessionMemory(nread);

src/node_http2.h

@@ -75,9 +75,8 @@ constexpr int kSessionStateClosing = 0x8;
 constexpr int kSessionStateSending = 0x10;
 constexpr int kSessionStateWriteInProgress = 0x20;
 constexpr int kSessionStateReadingStopped = 0x40;
-constexpr int kSessionStateReceivePaused = 0x80;
-constexpr int kSessionStateReceiving = 0x100;
-constexpr int kSessionStateClosePending = 0x200;
+constexpr int kSessionStateReceiving = 0x80;
+constexpr int kSessionStateClosePending = 0x100;
 
 // The Padding Strategy determines the method by which extra padding is
 // selected for HEADERS and DATA frames. These are configurable via the
@@ -664,7 +663,6 @@ class Http2Session : public AsyncWrap,
   IS_FLAG(sending, kSessionStateSending)
   IS_FLAG(write_in_progress, kSessionStateWriteInProgress)
   IS_FLAG(reading_stopped, kSessionStateReadingStopped)
-  IS_FLAG(receive_paused, kSessionStateReceivePaused)
   IS_FLAG(receiving, kSessionStateReceiving)
   IS_FLAG(close_pending, kSessionStateClosePending)
 
@@ -945,7 +943,6 @@ class Http2Session : public AsyncWrap,
   // will be set. stream_buf_ab_ is lazily created from stream_buf_allocation_.
   v8::Global<v8::ArrayBuffer> stream_buf_ab_;
   std::unique_ptr<v8::BackingStore> stream_buf_allocation_;
-  size_t stream_buf_offset_ = 0;
   // Custom error code for errors that originated inside one of the callbacks
   // called by nghttp2_session_mem_recv.
   const char* custom_recv_error_code_ = nullptr;

test/parallel/test-http2-bidirectional-write-deadlock.js

@@ -0,0 +1,107 @@
+'use strict';
+
+// Regression test against deadlocks between two HTTP/2 peers that are both
+// writing at the same time.
+//
+// To bound how much it buffered while output was backed up, an Http2Session
+// used to stop reading from its socket whenever a write was in flight, and
+// resume only once that write completed. When the peer was itself blocked
+// writing, that write never completed, so the session never read again and
+// the connection hung forever with no error and no timeout.
+//
+// Rather than relying on kernel socket buffers filling up - which depends on
+// the platform and configured window sizes - this models one half of that
+// cycle directly. The client's socket forwards a write but does not report it
+// as complete, substituting for a write blocked because the peer is not
+// reading. Only after that write is stalled does the server send its response
+// body. A session that stops reading while writing never sees it.
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const net = require('net');
+const { Duplex } = require('stream');
+
+const BODY = 'the response body';
+
+const heldCallbacks = [];
+let serverStream;
+
+let stallWrites = false;
+
+// Client-side socket that forwards writes to a real connection, but can leave
+// their completion callbacks pending to model a transport-blocked write.
+class StalledClientSocket extends Duplex {
+  constructor(port) {
+    super();
+    this.inner = net.connect(port, common.localhostIPv4);
+    this.inner.on('data', (chunk) => this.push(chunk));
+  }
+  _read() {
+    // Incoming data is pushed as it arrives.
+  }
+  _write(chunk, encoding, callback) {
+    this.inner.write(chunk, encoding);
+    if (stallWrites) {
+      heldCallbacks.push(callback);
+      // Avoid writing from the server re-entrantly inside _write(). The
+      // ordering is still explicit: this callback is already held.
+      setImmediate(() => serverStream.end(BODY));
+      return;
+    }
+    callback();
+  }
+  _final(callback) {
+    callback();
+  }
+  _destroy(err, callback) {
+    this.inner.destroy();
+    callback(err);
+  }
+}
+
+const server = http2.createServer();
+
+server.on('stream', common.mustCall((stream) => {
+  // Send headers first. Their response event starts the stalled client write.
+  stream.respond();
+  serverStream = stream;
+}));
+
+server.listen(0, common.mustCall(() => {
+  const port = server.address().port;
+
+  const client = http2.connect(`http://${common.localhostIPv4}:${port}`, {
+    createConnection: () => new StalledClientSocket(port),
+  });
+
+  const req = client.request({ ':method': 'POST' });
+
+  let received = '';
+
+  req.on('response', common.mustCall(() => {
+    // _write() will schedule the response body only after it has retained the
+    // callback, guaranteeing that the native write is still in progress.
+    stallWrites = true;
+    req.write(Buffer.alloc(256));
+  }));
+
+  req.on('data', (chunk) => {
+    received += chunk;
+  });
+
+  req.on('end', common.mustCall(() => {
+    assert.strictEqual(received, BODY);
+    assert.ok(heldCallbacks.length > 0,
+              'test did not actually stall a socket write');
+
+    // Let the stalled writes complete so that everything can shut down.
+    stallWrites = false;
+    for (const callback of heldCallbacks) callback();
+
+    client.destroy();
+    server.close();
+  }));
+}));