http2: finish pending writes after stream close

lib/internal/http2/core.js

@@ -2262,6 +2262,11 @@ class Http2Stream extends Duplex {
     if (this.destroyed)
       return;
 
+    if (this.closed) {
+      cb();
+      return;
+    }
+
     this[kUpdateTimer]();
     if (!this.headersSent)
       this[kProceed]();

src/node_http2.cc

@@ -1316,7 +1316,11 @@ int Http2Session::OnStreamClose(nghttp2_session* handle,
     if (answer.IsEmpty() || answer.ToLocalChecked()->IsFalse()) {
       // Skip to destroy
       stream->Destroy();
+    } else if (!stream->is_destroyed()) {
+      stream->FlushPendingWrites(0);
     }
+  } else {
+    stream->FlushPendingWrites(0);
   }
   return 0;
 }
@@ -2376,12 +2380,7 @@ void Http2Stream::Destroy() {
       // Free any remaining outgoing data chunks here. This should be done
       // here because it's possible for destroy to have been called while
       // we still have queued outbound writes.
-      while (!queue_.empty()) {
-        NgHttp2StreamWrite& head = queue_.front();
-        if (head.req_wrap)
-          WriteWrap::FromObject(head.req_wrap)->Done(UV_ECANCELED);
-        queue_.pop();
-      }
+      FlushPendingWrites(UV_ECANCELED);
 
       // We can destroy the stream now if there are no writes for it
       // already on the socket. Otherwise, we'll wait for the garbage collector
@@ -2401,6 +2400,14 @@ void Http2Stream::Destroy() {
   EmitStatistics();
 }
 
+void Http2Stream::FlushPendingWrites(int status) {
+  while (!queue_.empty()) {
+    NgHttp2StreamWrite& head = queue_.front();
+    DecrementAvailableOutboundLength(head.buf.len);
+    if (head.req_wrap) WriteWrap::FromObject(head.req_wrap)->Done(status);
+    queue_.pop();
+  }
+}
 
 // Initiates a response on the Http2Stream using data provided via the
 // StreamBase Streams API.
@@ -2618,7 +2625,7 @@ int Http2Stream::DoWrite(WriteWrap* req_wrap,
                          uv_stream_t* send_handle) {
   CHECK_NULL(send_handle);
   Http2Scope h2scope(this);
-  if (!is_writable() || is_destroyed()) {
+  if (!is_writable() || is_destroyed() || is_closed()) {
     return UV_EOF;
   }
   Debug(this, "queuing %d buffers to send", nbufs);
@@ -2789,7 +2796,6 @@ void HttpErrorString(const FunctionCallbackInfo<Value>& args) {
   }
 }
 
-
 // Serializes the settings object into a Buffer instance that
 // would be suitable, for instance, for creating the Base64
 // output for an HTTP2-Settings header field.

src/node_http2.h

@@ -330,6 +330,7 @@ class Http2Stream : public AsyncWrap,
 
   // Destroy this stream instance and free all held memory.
   void Destroy();
+  void FlushPendingWrites(int status);
 
   bool is_destroyed() const {
     return flags_ & kStreamStateDestroyed;

test/parallel/test-http2-close-while-writing.js

@@ -2,6 +2,7 @@
 // https://github.com/nodejs/node/issues/33156
 const common = require('../common');
 const fixtures = require('../common/fixtures');
+const assert = require('assert');
 
 if (!common.hasCrypto) {
   common.skip('missing crypto');
@@ -23,11 +24,20 @@ let client_stream;
 
 server.on('session', common.mustCall(function(session) {
   session.on('stream', common.mustCall(function(stream) {
+    let writes = 0;
+    let writeCallbacks = 0;
+
     stream.resume();
-    stream.on('data', function() {
-      this.write(Buffer.alloc(1));
+    stream.on('data', common.mustCallAtLeast(function() {
+      writes++;
+      this.write(Buffer.alloc(1), common.mustCall(() => {
+        writeCallbacks++;
+      }));
       process.nextTick(() => client_stream.destroy());
-    });
+    }));
+    stream.on('close', common.mustCall(() => {
+      assert.strictEqual(writeCallbacks, writes);
+    }));
   }));
 }));