http2: exclude queued data from response memory checks

doc/api/http2.md

@@ -3354,7 +3354,11 @@ changes:
     the current memory use of the header compression tables, header blocks
     retained by open streams, current data queued to be sent, and
     unacknowledged `PING` and `SETTINGS` frames are all counted towards the
-    current limit. **Default:** `10`.
+    current limit. For received headers on locally initiated streams, including
+    informational responses and trailers, queued outbound data across the session
+    is excluded from the memory check. Retained headers still count towards the
+    limit. This exception does not apply to server push streams.
+    **Default:** `10`.
   * `maxHeaderListPairs` {number} Sets the maximum number of header entries.
     This is similar to [`server.maxHeadersCount`][] or
     [`request.maxHeadersCount`][] in the `node:http` module. The minimum value

src/node_http2.cc

@@ -2782,9 +2782,11 @@ bool Http2Stream::AddHeader(nghttp2_rcbuf* name,
 
   Http2Header header(env(), name, value, flags);
   size_t length = header.length() + 32;
-  // A header can only be added if we have not exceeded the maximum number
-  // of headers and the session has memory available for it.
-  if (!session_->has_available_session_memory(length) ||
+  // Queued request bodies must not block response headers on local streams.
+  // Push streams have even IDs and retain the full session memory check.
+  const bool exclude_outbound =
+      session_->type() == NGHTTP2_SESSION_CLIENT && (id_ & 1) != 0;
+  if (!session_->has_available_session_memory(length, exclude_outbound) ||
       current_headers_.size() == max_header_pairs_ ||
       current_headers_length_ + length > max_header_length_) {
     return false;
@@ -2895,12 +2897,13 @@ ssize_t Http2Stream::Provider::Stream::OnRead(nghttp2_session* handle,
 
 void Http2Stream::IncrementAvailableOutboundLength(size_t amount) {
   available_outbound_length_ += amount;
-  session_->IncrementCurrentSessionMemory(amount);
+  session_->IncrementQueuedOutboundMemory(amount);
 }
 
 void Http2Stream::DecrementAvailableOutboundLength(size_t amount) {
+  DCHECK_LE(amount, available_outbound_length_);
   available_outbound_length_ -= amount;
-  session_->DecrementCurrentSessionMemory(amount);
+  session_->DecrementQueuedOutboundMemory(amount);
 }
 
 

src/node_http2.h

@@ -811,6 +811,17 @@ class Http2Session : public AsyncWrap,
   void UpdateLocalCustomSettings(size_t count_,
                                  nghttp2_settings_entry* entries_);
 
+  void IncrementQueuedOutboundMemory(uint64_t amount) {
+    queued_outbound_memory_ += amount;
+    IncrementCurrentSessionMemory(amount);
+  }
+
+  void DecrementQueuedOutboundMemory(uint64_t amount) {
+    DCHECK_LE(amount, queued_outbound_memory_);
+    queued_outbound_memory_ -= amount;
+    DecrementCurrentSessionMemory(amount);
+  }
+
   // Tell our custom memory allocator that this rcbuf is independent of
   // this session now, and may outlive it.
   void StopTrackingRcbuf(nghttp2_rcbuf* buf);
@@ -824,9 +835,17 @@ class Http2Session : public AsyncWrap,
     return total;
   }
 
-  // Return true if current_session_memory + amount is less than the max
-  bool has_available_session_memory(uint64_t amount) const {
-    return current_session_memory() + amount <= max_session_memory_;
+  // Headers on local client streams may exclude queued outbound DATA, but
+  // retained headers and all other session memory still count towards the
+  // limit.
+  bool has_available_session_memory(uint64_t amount,
+                                    bool exclude_outbound = false) const {
+    uint64_t memory = current_session_memory();
+    if (exclude_outbound) {
+      DCHECK_LE(queued_outbound_memory_, memory);
+      memory -= queued_outbound_memory_;
+    }
+    return memory + amount <= max_session_memory_;
   }
 
   struct Statistics {
@@ -966,6 +985,8 @@ class Http2Session : public AsyncWrap,
   // The maximum amount of memory allocated for this session
   uint64_t max_session_memory_ = kDefaultMaxSessionMemory;
   uint64_t current_session_memory_ = 0;
+  // Part of current_session_memory_: DATA not yet consumed by nghttp2.
+  uint64_t queued_outbound_memory_ = 0;
   // The amount of memory allocated by nghttp2 internals
   uint64_t current_nghttp2_memory_ = 0;
 

test/parallel/test-http2-max-session-memory-cancel.js

@@ -11,9 +11,14 @@ const { once } = require('events');
 (async () => {
   // Keep the entire upload queued until it is canceled.
   const server = http2.createServer({ settings: { initialWindowSize: 0 } });
+  let serverPush;
   server.on('stream', common.mustCall((stream, headers) => {
     stream.on('error', () => {});
     if (headers[':path'] === '/cancel') return;
+    stream.pushStream({ ':path': '/push' }, common.mustSucceed((push) => {
+      serverPush = push;
+      push.on('error', () => {});
+    }));
     stream.respond({ ':status': 200 });
     stream.end();
   }, 2));
@@ -35,7 +40,9 @@ const { once } = require('events');
   // Native write queue cleanup runs on an immediate after stream destruction.
   await new Promise(setImmediate);
 
-  // The canceled upload must no longer count towards maxSessionMemory.
+  // Push admission uses the full memory check, so leaked outbound accounting
+  // cannot be hidden by the DATA exception on local streams.
+  const promised = once(client, 'stream');
   const req = client.request({ ':path': '/check' });
   const response = once(req, 'response');
   req.end();
@@ -44,6 +51,14 @@ const { once } = require('events');
   const ended = once(req, 'end');
   req.resume();
   await ended;
+  const [push] = await promised;
+  const pushedResponse = once(push, 'push');
+  const pushedEnd = once(push, 'end');
+  push.resume();
+  serverPush.respond({ ':status': 200 });
+  serverPush.end();
+  const [[pushedHeaders]] = await Promise.all([pushedResponse, pushedEnd]);
+  assert.strictEqual(pushedHeaders[':status'], 200);
   client.close();
   server.close();
 })().then(common.mustCall());

test/parallel/test-http2-max-session-memory-response-limits.js

@@ -0,0 +1,75 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const assert = require('assert');
+const http2 = require('http2');
+const { once } = require('events');
+
+// Excluding queued DATA must preserve the cumulative header budget and
+// the independent header count and size limits.
+async function test(kind, bodySize = 4 * 1024 * 1024) {
+  const sessions = new Set();
+  const server = http2.createServer({ settings: { initialWindowSize: 0 } });
+  server.on('session', (session) => sessions.add(session));
+  server.on('stream', (stream) => {
+    stream.on('error', () => {});
+    switch (kind) {
+      case 'repeated-informational':
+        for (let i = 0; i < 80; i++) {
+          stream.additionalHeaders({ ':status': 103,
+                                     'x-hint': 'x'.repeat(16 * 1024) });
+        }
+        break;
+      case 'pairs':
+        stream.respond({ ':status': 413, 'a': '1', 'b': '2', 'c': '3', 'd': '4' });
+        stream.end();
+        return;
+      case 'size':
+        stream.respond({ ':status': 413, 'x-large': 'x'.repeat(2048) });
+        stream.end();
+        return;
+    }
+    stream.respond({ ':status': 413 });
+    stream.end();
+  });
+  server.listen(0);
+  await once(server, 'listening');
+  const client = http2.connect(`http://localhost:${server.address().port}`, {
+    maxSessionMemory: 1,
+    maxHeaderListPairs: kind === 'pairs' ? 4 : 128,
+    settings: { maxHeaderListSize: kind === 'size' ? 1024 : 65535 },
+  });
+  await Promise.all([
+    once(client, 'remoteSettings'),
+    once(client, 'localSettings'),
+  ]);
+  const req = client.request({ ':method': 'POST' });
+  const failure = once(req, 'error');
+  let hints = 0;
+  if (kind === 'repeated-informational') {
+    req.on('headers', () => hints++);
+  } else {
+    req.on('headers', common.mustNotCall());
+  }
+  req.on('response', common.mustNotCall());
+  req.on('trailers', common.mustNotCall());
+  req.end(Buffer.alloc(bodySize));
+  const [err] = await failure;
+  assert.strictEqual(err.code, 'ERR_HTTP2_STREAM_ERROR');
+  if (kind === 'repeated-informational') {
+    assert(hints > 0 && hints < 80);
+  }
+  client.destroy();
+  for (const session of sessions) session.destroy();
+  server.close();
+}
+
+(async () => {
+  await test('pairs');
+  await test('size');
+  await test('repeated-informational', 0);
+  await test('repeated-informational');
+})().then(common.mustCall());

test/parallel/test-http2-max-session-memory-response-push.js

@@ -0,0 +1,57 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const assert = require('assert');
+const http2 = require('http2');
+const { once } = require('events');
+
+(async () => {
+  const sessions = new Set();
+  let serverRequest;
+  let serverPush;
+  const server = http2.createServer({ settings: { initialWindowSize: 0 } });
+  server.on('session', (session) => sessions.add(session));
+  server.on('stream', (stream) => {
+    serverRequest = stream;
+    stream.on('error', () => {});
+    stream.pushStream({ ':path': '/push' }, common.mustSucceed((push) => {
+      serverPush = push;
+      push.on('error', () => {});
+    }));
+  });
+  server.listen(0);
+  await once(server, 'listening');
+  const client = http2.connect(`http://localhost:${server.address().port}`, {
+    maxSessionMemory: 1,
+  });
+  await Promise.all([
+    once(client, 'remoteSettings'),
+    once(client, 'localSettings'),
+  ]);
+  const promised = once(client, 'stream');
+  const req = client.request({ ':method': 'POST' });
+  const response = once(req, 'response');
+  const [push] = await promised;
+  push.on('push', common.mustNotCall());
+  const failure = once(push, 'error');
+
+  // Accept the PUSH_PROMISE first, then exhaust the budget. Even an existing
+  // pushed stream must not exclude queued DATA when receiving its response.
+  req.end(Buffer.alloc(4 * 1024 * 1024));
+  serverPush.respond({ ':status': 200 });
+  serverPush.end();
+  const [err] = await failure;
+  assert.strictEqual(err.code, 'ERR_HTTP2_STREAM_ERROR');
+
+  serverRequest.respond({ ':status': 413 });
+  serverRequest.end();
+  const [headers] = await response;
+  assert.strictEqual(headers[':status'], 413);
+  req.close();
+  client.destroy();
+  for (const session of sessions) session.destroy();
+  server.close();
+})().then(common.mustCall());

test/parallel/test-http2-max-session-memory-response.js

@@ -0,0 +1,87 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const assert = require('assert');
+const http2 = require('http2');
+const { once } = require('events');
+
+// Queued request bodies must not prevent responses on local streams.
+async function test(bodySize, {
+  informational = 0,
+  blocked = false,
+  trailers = false,
+} = {}) {
+  const server = http2.createServer({
+    settings: { initialWindowSize: blocked ? 0 : 65535 },
+  });
+  const sessions = new Set();
+  server.on('session', (session) => sessions.add(session));
+  server.on('stream', (stream, headers) => {
+    stream.on('error', () => {});
+    if (headers[':path'] === '/reuse') {
+      stream.additionalHeaders({ ':status': 103 });
+      stream.respond({ ':status': 200 });
+      stream.end();
+    } else {
+      if (informational) stream.additionalHeaders({ ':status': informational });
+      stream.respond({ ':status': 413 }, { waitForTrailers: trailers });
+      if (trailers) {
+        stream.on('wantTrailers', () => stream.sendTrailers({ 'x-trailer': 'x' }));
+      }
+      stream.end();
+    }
+  });
+  server.listen(0);
+  await once(server, 'listening');
+
+  const client = http2.connect(`http://localhost:${server.address().port}`, {
+    maxSessionMemory: 1,
+  });
+  await Promise.all([
+    once(client, 'remoteSettings'),
+    once(client, 'localSettings'),
+  ]);
+  const req = client.request({ ':method': 'POST' });
+  const response = once(req, 'response');
+  const ended = once(req, 'end');
+  const closed = once(req, 'close');
+  req.on('headers', informational ? common.mustCall((headers) => {
+    assert.strictEqual(headers[':status'], informational);
+  }) : common.mustNotCall());
+  req.on('trailers', trailers ? common.mustCall((headers) => {
+    assert.strictEqual(headers['x-trailer'], 'x');
+  }) : common.mustNotCall());
+  req.resume();
+  // Fill the write queue before any response headers can arrive.
+  req.end(Buffer.alloc(bodySize));
+  const [[headers]] = await Promise.all([response, ended]);
+  assert.strictEqual(headers[':status'], 413);
+  req.destroy();
+  await closed;
+  // The session must remain usable after canceling the upload.
+  const reuse = client.request({ ':path': '/reuse' });
+  reuse.on('headers', common.mustCall());
+  const reusedResponse = once(reuse, 'response');
+  reuse.end();
+  const [reusedHeaders] = await reusedResponse;
+  assert.strictEqual(reusedHeaders[':status'], 200);
+  const reusedClose = once(reuse, 'close');
+  reuse.close();
+  await reusedClose;
+  client.destroy();
+  for (const session of sessions) session.destroy();
+  server.close();
+}
+
+(async () => {
+  const bodySize = 4 * 1024 * 1024;
+  await test(bodySize);
+  await test(bodySize, { blocked: true });
+  await test(bodySize, { informational: 100, blocked: true });
+  await test(bodySize, { informational: 103, blocked: true });
+  await test(bodySize, { trailers: true, blocked: true });
+  await test(1);
+})().then(common.mustCall());