quic: address pimterry review comments

Author: jasnell

lib/internal/quic/quic.js

@@ -158,6 +158,7 @@ const {
 } = require('internal/fs/promises');
 
 const {
+  validateAbortSignal,
   validateBoolean,
   validateFunction,
   validateInteger,
@@ -859,7 +860,11 @@ function parseHeaderPairs(pairs) {
   const block = { __proto__: null };
   for (let n = 0; n + 1 < pairs.length; n += 2) {
     if (block[pairs[n]] !== undefined) {
-      block[pairs[n]] = [block[pairs[n]], pairs[n + 1]];
+      if (ArrayIsArray(block[pairs[n]])) {
+        ArrayPrototypePush(block[pairs[n]], pairs[n + 1]);
+      } else {
+        block[pairs[n]] = [block[pairs[n]], pairs[n + 1]];
+      }
     } else {
       block[pairs[n]] = pairs[n + 1];
     }
@@ -1045,9 +1050,13 @@ async function consumeSyncSource(handle, stream, source) {
       if (await writeBatchWithDrain(handle, stream, batch)) return;
     }
     handle.endWrite();
-  } catch {
+  } catch (err) {
     if (!stream.destroyed) {
-      handle.resetStream(0n);
+      stream.destroy(err);
+    } else {
+      // If the stream is already destroyed, rethrow the error to avoid
+      // silently swallowing it. Tho in practice this shouldn't happen.
+      throw err;
     }
   }
 }
@@ -1666,29 +1675,69 @@ class QuicStream {
     }
 
     function endSync() {
+      // Per the streams/iter spec, endSync and end follow a try-fallback
+      // pattern. That is, callers should try endSync first and if it returns
+      // -1, then they should call and await end(). This is a signal that sync
+      // end is not currently possible. However, we always support sync end
+      // here unless the stream is already errored.
       if (errored) return -1;
+
+      // If we're already closed, just return the total bytes written.
       if (closed) return totalBytesWritten;
+
+      // If we are waiting for drain to complete, we cannot end synchronously.
+      if (drainWakeup != null) return -1;
+
+      // Fantastic, we can end synchronously!
       handle.endWrite();
       closed = true;
       return totalBytesWritten;
     }
 
-    async function end(options) {
+    async function end(options = kEmptyObject) {
+      validateObject(options, 'options');
+      const { signal } = options;
+      if (signal !== undefined) {
+        validateAbortSignal(signal, 'options.signal');
+        signal.throwIfAborted();
+        // TODO(@jasnell): The stream/iter spec allows individual sync end
+        // calls to be canceled via an AbortSignal. We currently do not support
+        // this, but we can add before the impl is graduated from experimental.
+        // At most we do here is check for signal abort at the start of the call.
+      }
+
+      // Per the streams/iter spec, endSync and end follow a try-fallback
+      // pattern. That is, callers should try endSync first and if it returns
+      // -1, then they should call and await end(). This is a signal that sync
+      // end is not currently possible. However, we always support sync end
+      // here unless the stream is already errored.
+      // While the user should have already called endSync, we call it again
+      // here to actually process the end request. At worst it's called twice.
       const n = endSync();
+
+      // A return value of -1 indicates that endSync was not yet able to
+      // process the end request, either because we are errored or because we
+      // are awaiting drain. If we're errored, throw the error. If we're waiting
+      // for drain, await it and then try ending again.
+
       if (n >= 0) return n;
       if (errored) throw error;
-      drainWakeup = PromiseWithResolvers();
-      await drainWakeup.promise;
-      drainWakeup = null;
+
+      drainWakeup ??= PromiseWithResolvers();
+      try {
+        await drainWakeup.promise;
+      } finally {
+        drainWakeup = null;
+      }
       return endSync();
     }
 
     function fail(reason) {
       if (closed || errored) return;
       errored = true;
-      error = reason;
+      error = reason ?? new ERR_INVALID_STATE('Failed');
       handle.resetStream(0n);
-      if (drainWakeup) {
+      if (drainWakeup != null) {
         drainWakeup.reject(reason);
         drainWakeup = null;
       }
@@ -1710,7 +1759,7 @@ class QuicStream {
       [drainableProtocol]() {
         if (closed || errored) return null;
         if (stream.#state.writeDesiredSize > 0) return null;
-        drainWakeup = PromiseWithResolvers();
+        drainWakeup ??= PromiseWithResolvers();
         return drainWakeup.promise;
       },
       [SymbolAsyncDispose]() {
@@ -1912,6 +1961,9 @@ class QuicStream {
     this.#stats[kFinishClose]();
     this.#state[kFinishClose]();
     this.#session[kRemoveStream](this);
+    if (this.#writer !== undefined) {
+      this.#writer.fail(error);
+    }
     this.#session = undefined;
     this.#pendingClose.reject = undefined;
     this.#pendingClose.resolve = undefined;