http2: do not end compat requests reset before END_STREAM

When the peer reset a stream before sending END_STREAM, the compat
`Http2ServerRequest` still pushed EOF from its 'close' handler, so it
emitted 'end' for a truncated body (on main even after 'error' and
'close'). Unlike http1, where an aborted `IncomingMessage` is destroyed
and never emits 'end', applications listening for 'end' then processed
the partial body as if it were complete.

Record in `onStreamClose()` that the peer closed the stream before
END_STREAM, since `destroy()` ends the readable side afterwards, and
destroy the request instead of ending it in that case. Requests whose
body was complete before the reset still end normally.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/http2/compat.js

@@ -57,6 +57,7 @@ const {
   kSocket,
   kRequest,
   kProxySocket,
+  kIncomplete,
   assertValidPseudoHeader,
   getAuthority,
 } = require('internal/http2/util');
@@ -291,15 +292,24 @@ function onStreamCloseRequest() {
   const state = req[kState];
   state.closed = true;
 
+  this[kProxySocket] = null;
+  this[kRequest] = undefined;
+
+  // The peer closed the stream before END_STREAM, so the request body is
+  // incomplete. Like http1's IncomingMessage on abort, destroy the request
+  // instead of ending it, so 'end' is not emitted for a truncated body.
+  // destroy() emits 'close'.
+  if (this[kIncomplete]) {
+    req.destroy();
+    return;
+  }
+
   req.push(null);
   // If the user didn't interact with incoming data and didn't pipe it,
   // dump it for compatibility with http1
   if (!state.didRead && !req._readableState.resumeScheduled)
     req.resume();
 
-  this[kProxySocket] = null;
-  this[kRequest] = undefined;
-
   req.emit('close');
 }
 

lib/internal/http2/core.js

@@ -156,6 +156,7 @@ const {
   kStrictSingleValueFields,
   kSocket,
   kRequest,
+  kIncomplete,
   kProtocol,
   kProxySocket,
   MAX_ADDITIONAL_SETTINGS,
@@ -652,6 +653,9 @@ function onStreamClose(code, peerReset) {
   const cleanCode = code === NGHTTP2_NO_ERROR || code === NGHTTP2_CANCEL;
 
   if (!peerReadableEnded) {
+    // The peer closed the stream before END_STREAM, so the incoming message
+    // is incomplete. destroy() ends the readable side, so record it here.
+    stream[kIncomplete] = true;
     stream.destroy(cleanCode ?
       new ERR_HTTP2_STREAM_ABORTED() :
       new ERR_HTTP2_STREAM_ERROR(nameForErrorCode[code] || code));

lib/internal/http2/util.js

@@ -46,6 +46,7 @@ const kSocket = Symbol('socket');
 const kProtocol = Symbol('protocol');
 const kProxySocket = Symbol('proxySocket');
 const kRequest = Symbol('request');
+const kIncomplete = Symbol('incomplete');
 
 const {
   NGHTTP2_NV_FLAG_NONE,
@@ -1005,6 +1006,7 @@ module.exports = {
   kStrictSingleValueFields,
   kSocket,
   kProtocol,
+  kIncomplete,
   kProxySocket,
   kRequest,
   MAX_ADDITIONAL_SETTINGS,

test/parallel/test-http2-compat-serverrequest-end-after-abort.js

@@ -0,0 +1,83 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+// A request whose stream is reset by the peer before END_STREAM has an
+// incomplete body. Like http1, the compat request must not emit 'end' for it.
+
+const assert = require('assert');
+const http2 = require('http2');
+const { finished } = require('stream');
+
+const codes = [
+  http2.constants.NGHTTP2_NO_ERROR,
+  http2.constants.NGHTTP2_CANCEL,
+];
+
+const server = http2.createServer(common.mustCall((req, res) => {
+  // Tell the client the partial body arrived, so it can reset the stream.
+  req.once('data', common.mustCall(() => res.writeHead(200).flushHeaders()));
+  req.on('end', common.mustNotCall());
+  req.on('aborted', common.mustCall());
+  req.on('error', common.expectsError({ code: 'ERR_HTTP2_STREAM_ABORTED' }));
+  req.on('close', common.mustCall(() => {
+    assert.strictEqual(req.aborted, true);
+    assert.strictEqual(req.readableEnded, false);
+  }));
+  finished(req, common.mustCall((err) => {
+    assert.ok(err);
+  }));
+}, codes.length));
+
+server.listen(0, common.mustCall(() => {
+  const client = http2.connect(`http://localhost:${server.address().port}`);
+  let pending = codes.length;
+
+  for (const code of codes) {
+    const req = client.request({
+      ':method': 'POST',
+      'content-length': '1000',
+    });
+    req.on('error', () => {});
+    req.on('close', common.mustCall(() => {
+      if (--pending === 0) {
+        client.close();
+        server.close();
+      }
+    }));
+    // Send part of the body, then reset the stream without END_STREAM once
+    // the server has seen it.
+    req.write('{"action":"partial"}');
+    req.on('response', common.mustCall(() => req.close(code)));
+  }
+}));
+
+// A complete body followed by a reset still ends the request normally.
+{
+  const server = http2.createServer(common.mustCall((req, res) => {
+    let body = '';
+    req.setEncoding('utf8');
+    req.on('data', (chunk) => body += chunk);
+    req.on('end', common.mustCall(() => {
+      assert.strictEqual(body, 'complete');
+      res.writeHead(200).flushHeaders();
+    }));
+    req.on('error', common.mustNotCall());
+  }));
+
+  server.listen(0, common.mustCall(() => {
+    const client = http2.connect(`http://localhost:${server.address().port}`);
+    const req = client.request({ ':method': 'POST' });
+    req.on('error', () => {});
+    req.on('response', common.mustCall(() => {
+      req.close(http2.constants.NGHTTP2_CANCEL);
+    }));
+    req.on('close', common.mustCall(() => {
+      client.close();
+      server.close();
+    }));
+    req.end('complete');
+  }));
+}