http2: add session tracking and graceful server shutdown of http2 server

lib/internal/http2/core.js

@@ -251,6 +251,7 @@ const kServer = Symbol('server');
 const kState = Symbol('state');
 const kType = Symbol('type');
 const kWriteGeneric = Symbol('write-generic');
+const kSessions = Symbol('sessions');
 
 const {
   kBitfield,
@@ -1125,9 +1126,13 @@ function emitClose(self, error) {
 function cleanupSession(session) {
   const socket = session[kSocket];
   const handle = session[kHandle];
+  const server = session[kServer];
   session[kProxySocket] = undefined;
   session[kSocket] = undefined;
   session[kHandle] = undefined;
+  if (server) {
+    server[kSessions].delete(session);
+  }
   session[kNativeFields] = trackAssignmentsTypedArray(
     new Uint8Array(kSessionUint8FieldCount));
   if (handle)
@@ -1644,6 +1649,9 @@ class ServerHttp2Session extends Http2Session {
   constructor(options, socket, server) {
     super(NGHTTP2_SESSION_SERVER, options, socket);
     this[kServer] = server;
+    if (server) {
+      server[kSessions].add(this);
+    }
     // This is a bit inaccurate because it does not reflect changes to
     // number of listeners made after the session was created. This should
     // not be an issue in practice. Additionally, the 'priority' event on
@@ -3168,11 +3176,25 @@ function onErrorSecureServerSession(err, socket) {
     socket.destroy(err);
 }
 
+/**
+ * This function closes all active sessions gracefully.
+ * @param {*} server the underlying server whose sessions to be closed
+ */
+function closeAllSessions(server) {
+  const sessions = server[kSessions];
+  if (sessions.size > 0) {
+    for (const session of sessions) {
+      session.close();
+    }
+  }
+}
+
 class Http2SecureServer extends TLSServer {
   constructor(options, requestListener) {
     options = initializeTLSOptions(options);
     super(options, connectionListener);
     this[kOptions] = options;
+    this[kSessions] = new SafeSet();
     this.timeout = 0;
     this.on('newListener', setupCompat);
     if (options.allowHTTP1 === true) {
@@ -3202,10 +3224,11 @@ class Http2SecureServer extends TLSServer {
   }
 
   close() {
+    ReflectApply(TLSServer.prototype.close, this, arguments);
     if (this[kOptions].allowHTTP1 === true) {
       httpServerPreClose(this);
     }
-    ReflectApply(TLSServer.prototype.close, this, arguments);
+    closeAllSessions(this);
   }
 
   closeIdleConnections() {
@@ -3220,6 +3243,7 @@ class Http2Server extends NETServer {
     options = initializeOptions(options);
     super(options, connectionListener);
     this[kOptions] = options;
+    this[kSessions] = new SafeSet();
     this.timeout = 0;
     this.on('newListener', setupCompat);
     if (typeof requestListener === 'function')
@@ -3241,6 +3265,11 @@ class Http2Server extends NETServer {
     this[kOptions].settings = { ...this[kOptions].settings, ...settings };
   }
 
+  close() {
+    ReflectApply(NETServer.prototype.close, this, arguments);
+    closeAllSessions(this);
+  }
+
   async [SymbolAsyncDispose]() {
     return promisify(super.close).call(this);
   }

test/parallel/test-http2-capture-rejection.js

@@ -108,8 +108,6 @@ events.captureRejections = true;
   server.on('stream', common.mustCall(async (stream) => {
     const { port } = server.address();
 
-    server.close();
-
     stream.pushStream({
       ':scheme': 'http',
       ':path': '/foobar',
@@ -127,6 +125,8 @@ events.captureRejections = true;
     stream.respond({
       ':status': 200
     });
+
+    server.close();
   }));
 
   server.listen(0, common.mustCall(() => {

test/parallel/test-http2-compat-serverresponse-statusmessage-property-set.js

@@ -24,7 +24,6 @@ server.listen(0, common.mustCall(function() {
       response.statusMessage = 'test';
       response.statusMessage = 'test'; // only warn once
       assert.strictEqual(response.statusMessage, ''); // no change
-      server.close();
     }));
     response.end();
   }));
@@ -44,6 +43,9 @@ server.listen(0, common.mustCall(function() {
     request.on('end', common.mustCall(function() {
       client.close();
     }));
+    request.on('close', common.mustCall(function() {
+      server.close();
+    }));
     request.end();
     request.resume();
   }));

test/parallel/test-http2-compat-serverresponse-statusmessage-property.js

@@ -23,7 +23,6 @@ server.listen(0, common.mustCall(function() {
     response.on('finish', common.mustCall(function() {
       assert.strictEqual(response.statusMessage, '');
       assert.strictEqual(response.statusMessage, ''); // only warn once
-      server.close();
     }));
     response.end();
   }));
@@ -43,6 +42,9 @@ server.listen(0, common.mustCall(function() {
     request.on('end', common.mustCall(function() {
       client.close();
     }));
+    request.on('close', common.mustCall(function() {
+      server.close();
+    }));
     request.end();
     request.resume();
   }));

test/parallel/test-http2-graceful-close.js

@@ -0,0 +1,67 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+
+// This test verifies that the server waits for active connections/sessions
+// to complete and all data to be sent before fully closing
+
+// Create a server that will send a large response in chunks
+const server = http2.createServer();
+
+// Track server events
+server.on('stream', common.mustCall((stream, headers) => {
+
+  // Initiate the server close before client data is sent, this will
+  // test if the server properly waits for the stream to finish
+  server.close();
+  setImmediate(() => {
+    stream.respond({
+      'content-type': 'text/plain',
+      ':status': 200
+    });
+
+    // Create a large response (1MB) to ensure it takes some time to send
+    const chunk = Buffer.alloc(64 * 1024, 'x');
+
+    // Send 16 chunks (1MB total) to simulate a large response
+    for (let i = 0; i < 16; i++) {
+      stream.write(chunk);
+    }
+
+    // Stream end should happen after data is written
+    stream.end();
+  });
+
+  stream.on('close', common.mustCall(() => {
+    assert.strictEqual(stream.readableEnded, true);
+    assert.strictEqual(stream.writableFinished, true);
+  }));
+}));
+
+// Start the server
+server.listen(0, common.mustCall(() => {
+  // Create client and request
+  const client = http2.connect(`http://localhost:${server.address().port}`);
+  const req = client.request({ ':path': '/' });
+
+  // Track received data
+  let receivedData = 0;
+  req.on('data', (chunk) => {
+    receivedData += chunk.length;
+  });
+
+  // When request closes
+  req.on('close', common.mustCall(() => {
+    // Should receive all data
+    assert.strictEqual(req.readableEnded, true);
+    assert.strictEqual(receivedData, 64 * 1024 * 16);
+    assert.strictEqual(req.writableFinished, true);
+  }));
+
+  // Start the request
+  req.end();
+}));

test/parallel/test-http2-server-close-idle-connection.js

@@ -0,0 +1,58 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+
+// This test verifies that the server closes idle connections
+
+const server = http2.createServer();
+
+server.on('session', common.mustCall((session) => {
+  session.on('stream', common.mustCall((stream) => {
+    // Respond to the request with a small payload
+    stream.respond({
+      'content-type': 'text/plain',
+      ':status': 200
+    });
+    stream.end('hello');
+    stream.on('close', common.mustCall(() => {
+      assert.strictEqual(stream.writableFinished, true);
+    }));
+  }));
+  session.on('close', common.mustCall());
+}));
+
+// Start the server
+server.listen(0, common.mustCall(() => {
+  // Create client and initial request
+  const client = http2.connect(`http://localhost:${server.address().port}`);
+
+  // This will ensure that server closed the idle connection
+  client.on('close', common.mustCall());
+
+  // Make an initial request
+  const req = client.request({ ':path': '/' });
+
+  // Process the response data
+  req.setEncoding('utf8');
+  let data = '';
+  req.on('data', (chunk) => {
+    data += chunk;
+  });
+
+  // When initial request ends
+  req.on('end', common.mustCall(() => {
+    assert.strictEqual(data, 'hello');
+    // Close the server as client is idle now
+    setImmediate(() => {
+      server.close();
+    });
+  }));
+
+  // Don't explicitly close the client, we want to keep it
+  // idle and check if the server closes the idle connection.
+  // As this is what we want to test here.
+}));

test/parallel/test-http2-server-http1-client.js

@@ -23,5 +23,5 @@ server.on('session', common.mustCall((session) => {
 
 server.listen(0, common.mustCall(() => {
   const req = http.get(`http://localhost:${server.address().port}`);
-  req.on('error', (error) => server.close());
+  req.on('error', (error) => setImmediate(() => server.close()));
 }));