http2: refactor how trailers are done

Rather than an option, introduce a method and an event...

```js
server.on('stream', (stream) => {
  stream.respond(undefined, { waitForTrailers: true });
  stream.on('wantTrailers', () => {
    stream.sendTrailers({ abc: 'xyz'});
  });
  stream.end('hello world');
});
```

This is a breaking change in the API such that the prior
`options.getTrailers` is no longer supported at all.
Ordinarily this would be semver-major and require a
deprecation but the http2 stuff is still experimental.

PR-URL: #19959
Reviewed-By: Yuta Hiroto <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: Matteo Collina <[email protected]>

Author: jasnell

doc/api/errors.md

@@ -1017,6 +1017,19 @@ When setting the priority for an HTTP/2 stream, the stream may be marked as
 a dependency for a parent stream. This error code is used when an attempt is
 made to mark a stream and dependent of itself.
 
+<a id="ERR_HTTP2_TRAILERS_ALREADY_SENT"></a>
+### ERR_HTTP2_TRAILERS_ALREADY_SENT
+
+Trailing headers have already been sent on the `Http2Stream`.
+
+<a id="ERR_HTTP2_TRAILERS_NOT_READY"></a>
+### ERR_HTTP2_TRAILERS_NOT_READY
+
+The `http2stream.sendTrailers()` method cannot be called until after the
+`'wantTrailers'` event is emitted on an `Http2Stream` object. The
+`'wantTrailers'` event will only be emitted if the `waitForTrailers` option
+is set for the `Http2Stream`.
+
 <a id="ERR_HTTP2_UNSUPPORTED_PROTOCOL"></a>
 ### ERR_HTTP2_UNSUPPORTED_PROTOCOL
 

doc/api/http2.md

@@ -842,6 +842,11 @@ E('ERR_HTTP2_STREAM_CANCEL', 'The pending stream has been canceled', Error);
 E('ERR_HTTP2_STREAM_ERROR', 'Stream closed with error code %s', Error);
 E('ERR_HTTP2_STREAM_SELF_DEPENDENCY',
   'A stream cannot depend on itself', Error);
+E('ERR_HTTP2_TRAILERS_ALREADY_SENT',
+  'Trailing headers have already been sent', Error);
+E('ERR_HTTP2_TRAILERS_NOT_READY',
+  'Trailing headers cannot be sent until after the wantTrailers event is ' +
+  'emitted', Error);
 E('ERR_HTTP2_UNSUPPORTED_PROTOCOL', 'protocol "%s" is unsupported.', Error);
 E('ERR_HTTP_HEADERS_SENT',
   'Cannot %s headers after they are sent to the client', Error);

lib/internal/errors.js

@@ -358,6 +358,10 @@ class Http2ServerRequest extends Readable {
   }
 }
 
+function onStreamTrailersReady() {
+  this[kStream].sendTrailers(this[kTrailers]);
+}
+
 class Http2ServerResponse extends Stream {
   constructor(stream, options) {
     super(options);
@@ -377,6 +381,7 @@ class Http2ServerResponse extends Stream {
     stream.on('drain', onStreamDrain);
     stream.on('aborted', onStreamAbortedResponse);
     stream.on('close', this[kFinish].bind(this));
+    stream.on('wantTrailers', onStreamTrailersReady.bind(this));
   }
 
   // User land modules such as finalhandler just check truthiness of this
@@ -648,7 +653,7 @@ class Http2ServerResponse extends Stream {
     headers[HTTP2_HEADER_STATUS] = state.statusCode;
     const options = {
       endStream: state.ending,
-      getTrailers: (trailers) => Object.assign(trailers, this[kTrailers])
+      waitForTrailers: true,
     };
     this[kStream].respond(headers, options);
   }

lib/internal/http2/compat.js

@@ -49,6 +49,8 @@ const {
     ERR_HTTP2_STREAM_CANCEL,
     ERR_HTTP2_STREAM_ERROR,
     ERR_HTTP2_STREAM_SELF_DEPENDENCY,
+    ERR_HTTP2_TRAILERS_ALREADY_SENT,
+    ERR_HTTP2_TRAILERS_NOT_READY,
     ERR_HTTP2_UNSUPPORTED_PROTOCOL,
     ERR_INVALID_ARG_TYPE,
     ERR_INVALID_CALLBACK,
@@ -295,25 +297,18 @@ function tryClose(fd) {
   fs.close(fd, (err) => assert.ifError(err));
 }
 
-// Called to determine if there are trailers to be sent at the end of a
-// Stream. The 'getTrailers' callback is invoked and passed a holder object.
-// The trailers to return are set on that object by the handler. Once the
-// event handler returns, those are sent off for processing. Note that this
-// is a necessarily synchronous operation. We need to know immediately if
-// there are trailing headers to send.
+// Called when the Http2Stream has finished sending data and is ready for
+// trailers to be sent. This will only be called if the { hasOptions: true }
+// option is set.
 function onStreamTrailers() {
   const stream = this[kOwner];
+  stream[kState].trailersReady = true;
   if (stream.destroyed)
-    return [];
-  const trailers = Object.create(null);
-  stream[kState].getTrailers.call(stream, trailers);
-  const headersList = mapToHeaders(trailers, assertValidPseudoHeaderTrailer);
-  if (!Array.isArray(headersList)) {
-    stream.destroy(headersList);
-    return [];
+    return;
+  if (!stream.emit('wantTrailers')) {
+    // There are no listeners, send empty trailing HEADERS frame and close.
+    stream.sendTrailers({});
   }
-  stream[kSentTrailers] = trailers;
-  return headersList;
 }
 
 // Submit an RST-STREAM frame to be sent to the remote peer.
@@ -527,10 +522,8 @@ function requestOnConnect(headers, options) {
   if (options.endStream)
     streamOptions |= STREAM_OPTION_EMPTY_PAYLOAD;
 
-  if (typeof options.getTrailers === 'function') {
+  if (options.waitForTrailers)
     streamOptions |= STREAM_OPTION_GET_TRAILERS;
-    this[kState].getTrailers = options.getTrailers;
-  }
 
   // ret will be either the reserved stream ID (if positive)
   // or an error code (if negative)
@@ -1408,11 +1401,6 @@ class ClientHttp2Session extends Http2Session {
       throw new ERR_INVALID_OPT_VALUE('endStream', options.endStream);
     }
 
-    if (options.getTrailers !== undefined &&
-      typeof options.getTrailers !== 'function') {
-      throw new ERR_INVALID_OPT_VALUE('getTrailers', options.getTrailers);
-    }
-
     const headersList = mapToHeaders(headers);
     if (!Array.isArray(headersList))
       throw headersList;
@@ -1504,7 +1492,8 @@ class Http2Stream extends Duplex {
     this[kState] = {
       flags: STREAM_FLAGS_PENDING,
       rstCode: NGHTTP2_NO_ERROR,
-      writeQueueSize: 0
+      writeQueueSize: 0,
+      trailersReady: false
     };
 
     this.on('resume', streamOnResume);
@@ -1745,6 +1734,33 @@ class Http2Stream extends Duplex {
     priorityFn();
   }
 
+  sendTrailers(headers) {
+    if (this.destroyed || this.closed)
+      throw new ERR_HTTP2_INVALID_STREAM();
+    if (this[kSentTrailers])
+      throw new ERR_HTTP2_TRAILERS_ALREADY_SENT();
+    if (!this[kState].trailersReady)
+      throw new ERR_HTTP2_TRAILERS_NOT_READY();
+
+    assertIsObject(headers, 'headers');
+    headers = Object.assign(Object.create(null), headers);
+
+    const session = this[kSession];
+    debug(`Http2Stream ${this[kID]} [Http2Session ` +
+    `${sessionName(session[kType])}]: sending trailers`);
+
+    this[kUpdateTimer]();
+
+    const headersList = mapToHeaders(headers, assertValidPseudoHeaderTrailer);
+    if (!Array.isArray(headersList))
+      throw headersList;
+    this[kSentTrailers] = headers;
+
+    const ret = this[kHandle].trailers(headersList);
+    if (ret < 0)
+      this.destroy(new NghttpError(ret));
+  }
+
   get closed() {
     return !!(this[kState].flags & STREAM_FLAGS_CLOSED);
   }
@@ -2208,13 +2224,8 @@ class ServerHttp2Stream extends Http2Stream {
     if (options.endStream)
       streamOptions |= STREAM_OPTION_EMPTY_PAYLOAD;
 
-    if (options.getTrailers !== undefined) {
-      if (typeof options.getTrailers !== 'function') {
-        throw new ERR_INVALID_OPT_VALUE('getTrailers', options.getTrailers);
-      }
+    if (options.waitForTrailers)
       streamOptions |= STREAM_OPTION_GET_TRAILERS;
-      state.getTrailers = options.getTrailers;
-    }
 
     headers = processHeaders(headers);
     const statusCode = headers[HTTP2_HEADER_STATUS] |= 0;
@@ -2274,13 +2285,8 @@ class ServerHttp2Stream extends Http2Stream {
     }
 
     let streamOptions = 0;
-    if (options.getTrailers !== undefined) {
-      if (typeof options.getTrailers !== 'function') {
-        throw new ERR_INVALID_OPT_VALUE('getTrailers', options.getTrailers);
-      }
+    if (options.waitForTrailers)
       streamOptions |= STREAM_OPTION_GET_TRAILERS;
-      this[kState].getTrailers = options.getTrailers;
-    }
 
     if (typeof fd !== 'number')
       throw new ERR_INVALID_ARG_TYPE('fd', 'number', fd);
@@ -2340,13 +2346,8 @@ class ServerHttp2Stream extends Http2Stream {
     }
 
     let streamOptions = 0;
-    if (options.getTrailers !== undefined) {
-      if (typeof options.getTrailers !== 'function') {
-        throw new ERR_INVALID_OPT_VALUE('getTrailers', options.getTrailers);
-      }
+    if (options.waitForTrailers)
       streamOptions |= STREAM_OPTION_GET_TRAILERS;
-      this[kState].getTrailers = options.getTrailers;
-    }
 
     const session = this[kSession];
     debug(`Http2Stream ${this[kID]} [Http2Session ` +

lib/internal/http2/core.js

@@ -1066,16 +1066,6 @@ int Http2Session::OnNghttpError(nghttp2_session* handle,
   return 0;
 }
 
-// Once all of the DATA frames for a Stream have been sent, the GetTrailers
-// method calls out to JavaScript to fetch the trailing headers that need
-// to be sent.
-void Http2Session::GetTrailers(Http2Stream* stream, uint32_t* flags) {
-  if (!stream->IsDestroyed() && stream->HasTrailers()) {
-    Http2Stream::SubmitTrailers submit_trailers{this, stream, flags};
-    stream->OnTrailers(submit_trailers);
-  }
-}
-
 uv_buf_t Http2StreamListener::OnStreamAlloc(size_t size) {
   // See the comments in Http2Session::OnDataChunkReceived
   // (which is the only possible call site for this method).
@@ -1111,25 +1101,6 @@ void Http2StreamListener::OnStreamRead(ssize_t nread, const uv_buf_t& buf) {
   stream->CallJSOnreadMethod(nread, buffer);
 }
 
-Http2Stream::SubmitTrailers::SubmitTrailers(
-    Http2Session* session,
-    Http2Stream* stream,
-    uint32_t* flags)
-  : session_(session), stream_(stream), flags_(flags) { }
-
-
-void Http2Stream::SubmitTrailers::Submit(nghttp2_nv* trailers,
-                                         size_t length) const {
-  Http2Scope h2scope(session_);
-  if (length == 0)
-    return;
-  DEBUG_HTTP2SESSION2(session_, "sending trailers for stream %d, count: %d",
-                      stream_->id(), length);
-  *flags_ |= NGHTTP2_DATA_FLAG_NO_END_STREAM;
-  CHECK_EQ(
-      nghttp2_submit_trailer(**session_, stream_->id(), trailers, length), 0);
-}
-
 
 // Called by OnFrameReceived to notify JavaScript land that a complete
 // HEADERS frame has been received and processed. This method converts the
@@ -1725,30 +1696,6 @@ nghttp2_stream* Http2Stream::operator*() {
   return nghttp2_session_find_stream(**session_, id_);
 }
 
-
-// Calls out to JavaScript land to fetch the actual trailer headers to send
-// for this stream.
-void Http2Stream::OnTrailers(const SubmitTrailers& submit_trailers) {
-  DEBUG_HTTP2STREAM(this, "prompting for trailers");
-  CHECK(!this->IsDestroyed());
-  Isolate* isolate = env()->isolate();
-  HandleScope scope(isolate);
-  Local<Context> context = env()->context();
-  Context::Scope context_scope(context);
-
-  Local<Value> ret =
-      MakeCallback(env()->ontrailers_string(), 0, nullptr).ToLocalChecked();
-  if (!ret.IsEmpty() && !IsDestroyed()) {
-    if (ret->IsArray()) {
-      Local<Array> headers = ret.As<Array>();
-      if (headers->Length() > 0) {
-        Headers trailers(isolate, context, headers);
-        submit_trailers.Submit(*trailers, trailers.length());
-      }
-    }
-  }
-}
-
 void Http2Stream::Close(int32_t code) {
   CHECK(!this->IsDestroyed());
   flags_ |= NGHTTP2_STREAM_FLAG_CLOSED;
@@ -1843,6 +1790,26 @@ int Http2Stream::SubmitInfo(nghttp2_nv* nva, size_t len) {
   return ret;
 }
 
+void Http2Stream::OnTrailers() {
+  DEBUG_HTTP2STREAM(this, "let javascript know we are ready for trailers");
+  CHECK(!this->IsDestroyed());
+  Isolate* isolate = env()->isolate();
+  HandleScope scope(isolate);
+  Local<Context> context = env()->context();
+  Context::Scope context_scope(context);
+  MakeCallback(env()->ontrailers_string(), 0, nullptr);
+}
+
+// Submit informational headers for a stream.
+int Http2Stream::SubmitTrailers(nghttp2_nv* nva, size_t len) {
+  CHECK(!this->IsDestroyed());
+  Http2Scope h2scope(this);
+  DEBUG_HTTP2STREAM2(this, "sending %d trailers", len);
+  int ret = nghttp2_submit_trailer(**session_, id_, nva, len);
+  CHECK_NE(ret, NGHTTP2_ERR_NOMEM);
+  return ret;
+}
+
 // Submit a PRIORITY frame to the connected peer.
 int Http2Stream::SubmitPriority(nghttp2_priority_spec* prispec,
                                 bool silent) {
@@ -2068,13 +2035,10 @@ ssize_t Http2Stream::Provider::Stream::OnRead(nghttp2_session* handle,
   if (stream->queue_.empty() && !stream->IsWritable()) {
     DEBUG_HTTP2SESSION2(session, "no more data for stream %d", id);
     *flags |= NGHTTP2_DATA_FLAG_EOF;
-    session->GetTrailers(stream, flags);
-    // If the stream or session gets destroyed during the GetTrailers
-    // callback, check that here and close down the stream
-    if (stream->IsDestroyed())
-      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
-    if (session->IsDestroyed())
-      return NGHTTP2_ERR_CALLBACK_FAILURE;
+    if (stream->HasTrailers()) {
+      *flags |= NGHTTP2_DATA_FLAG_NO_END_STREAM;
+      stream->OnTrailers();
+    }
   }
 
   stream->statistics_.sent_bytes += amount;
@@ -2361,6 +2325,21 @@ void Http2Stream::Info(const FunctionCallbackInfo<Value>& args) {
                      headers->Length());
 }
 
+// Submits trailing headers on the Http2Stream
+void Http2Stream::Trailers(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  Local<Context> context = env->context();
+  Isolate* isolate = env->isolate();
+  Http2Stream* stream;
+  ASSIGN_OR_RETURN_UNWRAP(&stream, args.Holder());
+
+  Local<Array> headers = args[0].As<Array>();
+
+  Headers list(isolate, context, headers);
+  args.GetReturnValue().Set(stream->SubmitTrailers(*list, list.length()));
+  DEBUG_HTTP2STREAM2(stream, "%d trailing headers sent", headers->Length());
+}
+
 // Grab the numeric id of the Http2Stream
 void Http2Stream::GetID(const FunctionCallbackInfo<Value>& args) {
   Http2Stream* stream;
@@ -2706,6 +2685,7 @@ void Initialize(Local<Object> target,
   env->SetProtoMethod(stream, "priority", Http2Stream::Priority);
   env->SetProtoMethod(stream, "pushPromise", Http2Stream::PushPromise);
   env->SetProtoMethod(stream, "info", Http2Stream::Info);
+  env->SetProtoMethod(stream, "trailers", Http2Stream::Trailers);
   env->SetProtoMethod(stream, "respond", Http2Stream::Respond);
   env->SetProtoMethod(stream, "rstStream", Http2Stream::RstStream);
   env->SetProtoMethod(stream, "refreshState", Http2Stream::RefreshState);