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];
     }
@@ -990,6 +995,23 @@ function configureOutbound(handle, stream, body, depth = 0) {
   );
 }
 
+// Sets the high water mark and initial writeDesiredSize for a streaming
+// outbound source. Called after handle.initStreamingSource() for both
+// body-source and writer paths. One-shot body sources (string, Uint8Array,
+// Blob, FileHandle, etc.) do not use this -- they go through attachSource
+// and are not subject to backpressure.
+function initStreamingBackpressure(stream) {
+  const state = getQuicStreamState(stream);
+  // Only set defaults if the user hasn't already configured them
+  // (e.g., via createBidirectionalStream({ highWaterMark: N })).
+  if (state.highWaterMark === 0) {
+    state.highWaterMark = kDefaultHighWaterMark;
+  }
+  if (state.writeDesiredSize === 0) {
+    state.writeDesiredSize = state.highWaterMark;
+  }
+}
+
 // Waits for the stream's drain callback to fire, indicating the
 // outbound has capacity for more data.
 function waitForDrain(stream) {
@@ -1004,18 +1026,39 @@ function waitForDrain(stream) {
 
 // Writes a batch to the handle, awaiting drain if backpressured.
 // Returns true if the stream was destroyed during the wait.
+// Checks writeDesiredSize before writing to enforce backpressure
+// against the outbound DataQueue's uncommitted bytes.
 async function writeBatchWithDrain(handle, stream, batch) {
+  const state = getQuicStreamState(stream);
+
+  // Calculate total batch size for the capacity check.
+  let len = 0;
+  for (const chunk of batch) len += TypedArrayPrototypeGetByteLength(chunk);
+
+  // If insufficient capacity, wait for the C++ drain signal which
+  // fires when writeDesiredSize transitions from 0 to > 0 (i.e.,
+  // ngtcp2 has consumed data from the outbound DataQueue).
+  if (len > state.writeDesiredSize) {
+    await waitForDrain(stream);
+    if (stream.destroyed) return true;
+  }
+
+  // Write the batch. The return value is the total queued byte count
+  // on success, or undefined on failure (e.g., DataQueue append
+  // rejected). Guard against silent data loss.
   const result = handle.write(batch);
-  if (result !== undefined) return false;
-  // Write rejected (flow control) - wait for drain
-  await waitForDrain(stream);
-  if (stream.destroyed) return true;
-  handle.write(batch);
+  if (result === undefined) {
+    if (!stream.destroyed) {
+      stream.destroy(new ERR_INVALID_STATE('Stream write failed'));
+    }
+    return true;
+  }
   return false;
 }
 
 async function consumeAsyncSource(handle, stream, source) {
   handle.initStreamingSource();
+  initStreamingBackpressure(stream);
   try {
     // Normalize to AsyncIterable<Uint8Array[]>
     const normalized = streamFrom(source);
@@ -1033,6 +1076,7 @@ async function consumeAsyncSource(handle, stream, source) {
 
 async function consumeSyncSource(handle, stream, source) {
   handle.initStreamingSource();
+  initStreamingBackpressure(stream);
   // Normalize to Iterable<Uint8Array[]>. Manually iterate so we can
   // pause between next() calls when backpressure hits.
   const normalized = streamFromSync(source);
@@ -1045,9 +1089,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;
     }
   }
 }
@@ -1598,10 +1646,22 @@ class QuicStream {
       }
     };
 
+    // A note on backpressure handling: per the stream/iter spec, the default
+    // backpressure policy for writers is strict, meaning that if the stream
+    // signals backpressure additional writes are rejected until the buffer has
+    // capacity again.
+
     function writeSync(chunk) {
-      if (closed || errored || stream.#state.writeEnded) return false;
+      // If the stream is closed, errored, or write-ended, we cannot accept
+      // more data. Refuse the sync write.
+      // If a drain is already pending, another operation is waiting
+      // for capacity. Refuse the sync write.
+      if (closed || errored || stream.#state.writeEnded || drainWakeup != null) {
+        return false;
+      }
       chunk = toUint8Array(chunk);
-      const len = chunk.byteLength;
+      const len = TypedArrayPrototypeGetByteLength(chunk);
+      if (len === 0) return true;
       // Refuse the write if the chunk doesn't fit in the available
       // buffer capacity. The caller should wait for drain and retry.
       if (len > stream.#state.writeDesiredSize) return false;
@@ -1611,85 +1671,136 @@ class QuicStream {
       return true;
     }
 
-    async function write(chunk, options) {
-      if (options?.signal?.aborted) {
-        throw options.signal.reason;
+    async function write(chunk, options = kEmptyObject) {
+      validateObject(options, 'options');
+      const { signal } = options;
+      if (signal !== undefined) {
+        validateAbortSignal(signal, 'options.signal');
+        signal.throwIfAborted();
       }
       if (errored) throw error;
       if (closed || stream.#state.writeEnded) {
         throw new ERR_INVALID_STATE('Writer is closed');
       }
-      chunk = toUint8Array(chunk);
-      const len = chunk.byteLength;
-      // Reject if the chunk doesn't fit in the available buffer capacity.
-      if (len > stream.#state.writeDesiredSize) {
+      // If a drain is already pending, another operation is waiting
+      // for capacity. Under strict policy, reject immediately.
+      // Later, if we add support for other backpressure policies,
+      // we could instead await the existing drain before proceeding.
+      if (drainWakeup != null) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
-      const result = handle.write([chunk]);
-      if (result === undefined) {
+
+      if (!writeSync(chunk)) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
-      totalBytesWritten += len;
     }
 
     function writevSync(chunks) {
-      if (closed || errored || stream.#state.writeEnded) return false;
+      if (closed || errored || stream.#state.writeEnded || drainWakeup != null) {
+        return false;
+      }
       chunks = convertChunks(chunks);
       let len = 0;
       for (const c of chunks) len += TypedArrayPrototypeGetByteLength(c);
+      if (len === 0) return true;
       if (len > stream.#state.writeDesiredSize) return false;
       const result = handle.write(chunks);
       if (result === undefined) return false;
       totalBytesWritten += len;
       return true;
     }
 
-    async function writev(chunks, options) {
-      if (options?.signal?.aborted) {
-        throw options.signal.reason;
+    async function writev(chunks, options = kEmptyObject) {
+      validateObject(options, 'options');
+      const { signal } = options;
+      if (signal !== undefined) {
+        validateAbortSignal(signal, 'options.signal');
+        signal.throwIfAborted();
       }
+
       if (errored) throw error;
       if (closed || stream.#state.writeEnded) {
         throw new ERR_INVALID_STATE('Writer is closed');
       }
-      chunks = convertChunks(chunks);
-      let len = 0;
-      for (const c of chunks) len += TypedArrayPrototypeGetByteLength(c);
-      if (len > stream.#state.writeDesiredSize) {
+
+      // If a drain is already pending, another operation is waiting
+      // for capacity. Under strict policy, reject immediately.
+      // Later, if we add support for other backpressure policies,
+      // we could instead await the existing drain before proceeding.
+      if (drainWakeup != null) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
-      const result = handle.write(chunks);
-      if (result === undefined) {
+
+      if (!writevSync(chunks)) {
         throw new ERR_INVALID_STATE('Stream write buffer is full');
       }
-      totalBytesWritten += len;
     }
 
     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) {
-        drainWakeup.reject(reason);
+      if (drainWakeup != null) {
+        drainWakeup.reject(error);
         drainWakeup = null;
       }
     }
@@ -1709,6 +1820,8 @@ class QuicStream {
       fail,
       [drainableProtocol]() {
         if (closed || errored) return null;
+        // If a drain is already pending, return the existing promise.
+        if (drainWakeup != null) return drainWakeup.promise;
         if (stream.#state.writeDesiredSize > 0) return null;
         drainWakeup = PromiseWithResolvers();
         return drainWakeup.promise;
@@ -1734,6 +1847,7 @@ class QuicStream {
 
     // Initialize the outbound DataQueue for streaming writes
     handle.initStreamingSource();
+    initStreamingBackpressure(this);
 
     this.#writer = writer;
     return this.#writer;
@@ -1912,6 +2026,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;