diff --git a/doc/api/quic.md b/doc/api/quic.md index a9e7414f4a3e..a4b1d9bdb937 100644 --- a/doc/api/quic.md +++ b/doc/api/quic.md @@ -3425,6 +3425,30 @@ value, PING frames will be sent automatically to keep the connection alive before the idle timeout fires. The value should be less than the effective idle timeout (`maxIdleTimeout` transport parameter) to be useful. +#### `sessionOptions.truncatedReads` + +* Type: {string} One of `'error'` or `'allow'`. +* **Default:** `'error'` + +Controls how reading a stream reports a truncated read. A stream's read side +can end without receiving a QUIC FIN, meaning the peer never signalled that +the whole stream had been sent and the data received may be incomplete. This +selects how the stream's async iterator reports this: + +* `'error'` - The default. Peers are expected to always send a FIN to end + their data explicitly, and so any truncation is an error. The iterator yields + the data that did arrive and then throws, so an incomplete stream can never + be mistaken for a complete one. Incomplete streams will either throw a + `ERR_QUIC_STREAM_RESET` carrying the peer's error code, a connection error, + or `ERR_QUIC_STREAM_ABORTED` for other cases. + +* `'allow'` - Truncated reads are allowed: only a stream or connection error + is reported, and any clean abort/cancellation or similar simply ends the + stream. A non-zero peer reset, non-zero local stop-sending or connection + error still fails, but a truncation with no error at all (an idle timeout, + a graceful close, or a plain `stopSending()`) ends the read cleanly with the + data received. This matches `stream.closed`, which rejects only on an error. + #### `sessionOptions.verifyPeer` (client only) * Type: {string} One of `'strict'`, `'auto'`, or `'manual'`. diff --git a/lib/internal/blob.js b/lib/internal/blob.js index 8e26e40b9367..6d359c5adf7e 100644 --- a/lib/internal/blob.js +++ b/lib/internal/blob.js @@ -631,7 +631,7 @@ const kMaxBatchChunks = 16; const kDefaultMaxBatchBytes = 65536; async function* createBlobReaderIterable(reader, options = kEmptyObject) { - const { getReadError, maxBatchBytes = kDefaultMaxBatchBytes } = options; + const { maxBatchBytes = kDefaultMaxBatchBytes } = options; let wakeup = PromiseWithResolvers(); let immediate; reader.setWakeup(() => { @@ -664,9 +664,7 @@ async function* createBlobReaderIterable(reader, options = kEmptyObject) { break; } if (pullResult.status < 0) { - error = typeof getReadError === 'function' ? - getReadError(pullResult.status) : - new ERR_INVALID_STATE('The reader is not readable'); + error = new ERR_INVALID_STATE('The reader is not readable'); break; } if (pullResult.status === 2) { diff --git a/lib/internal/errors.js b/lib/internal/errors.js index 221a40ecdf86..69bf0792ffed 100644 --- a/lib/internal/errors.js +++ b/lib/internal/errors.js @@ -1712,9 +1712,6 @@ E('ERR_PROXY_TUNNEL', '%s', Error); E('ERR_QUIC_CONNECTION_FAILED', 'QUIC connection failed', Error); E('ERR_QUIC_ENDPOINT_CLOSED', 'QUIC endpoint closed: %s (%d)', Error); E('ERR_QUIC_OPEN_STREAM_FAILED', 'Failed to open QUIC stream', Error); -E('ERR_QUIC_STREAM_ABORTED', '%s', Error); -E('ERR_QUIC_STREAM_RESET', - 'The QUIC stream was reset by the peer with error code %d', Error); E('ERR_QUIC_VERSION_NEGOTIATION_ERROR', 'The QUIC session requires version negotiation', Error); E('ERR_REQUIRE_ASYNC_MODULE', function(filename, parent, locations) { let message = 'require() cannot be used on an ESM graph with top-level await. Use import() instead.'; diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index d55603daf1f4..10b6e32f91a4 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -13,7 +13,6 @@ const { ErrorCaptureStackTrace, FunctionPrototypeBind, FunctionPrototypeCall, - Number, ObjectDefineProperties, ObjectKeys, PromisePrototypeThen, @@ -114,8 +113,6 @@ const { ERR_QUIC_CONNECTION_FAILED, ERR_QUIC_ENDPOINT_CLOSED, ERR_QUIC_OPEN_STREAM_FAILED, - ERR_QUIC_STREAM_ABORTED, - ERR_QUIC_STREAM_RESET, ERR_QUIC_VERSION_NEGOTIATION_ERROR, }, } = require('internal/errors'); @@ -398,6 +395,7 @@ const endpointRegistry = new SafeSet(); * @property {number} [minVersion] The minimum acceptable QUIC version * @property {'use'|'ignore'|'default'} [preferredAddressPolicy] The preferred address policy * @property {'strict'|'auto'|'manual'} [verifyPeer='auto'] Peer certificate verification policy (client only) + * @property {'error'|'allow'} [truncatedReads] Truncated read policy * @property {ApplicationOptions} [application] The application options * @property {TransportParams} [transportParams] The transport parameters * @property {string} [servername] The server name identifier (client only) @@ -1580,6 +1578,9 @@ class QuicStream { state: undefined, stats: undefined, pendingClose: undefined, + destroyError: undefined, + stopSendingCode: undefined, + truncatedReads: undefined, reader: undefined, destroying: false, iteratorLocked: false, @@ -1627,9 +1628,10 @@ class QuicStream { * @param {object} handle * @param {QuicSession} session * @param {number} direction - * @param {boolean} [isLocal] + * @param {boolean} isLocal + * @param {'error'|'allow'} truncatedReads */ - constructor(privateSymbol, handle, session, direction, isLocal) { + constructor(privateSymbol, handle, session, direction, isLocal, truncatedReads) { assertPrivateSymbol(privateSymbol); this.#handle = handle; @@ -1638,6 +1640,7 @@ class QuicStream { inner.session = session; inner.direction = direction; inner.isLocal = isLocal; + inner.truncatedReads = truncatedReads; inner.state = new QuicStreamState( kPrivateConstructor, handle.state, handle.stateByteOffset); @@ -1665,34 +1668,49 @@ class QuicStream { inner.iteratorLocked = true; inner.reader ??= this.#handle?.getReader(); - // Non-readable stream (outbound-only unidirectional, or closed) - if (!inner.reader) return; - - yield* createBlobReaderIterable(inner.reader, { - getReadError: () => { - // The read side ends for one of three reasons: - // * Clean FIN received from the peer (state.finReceived - // === true). The iterator stops without calling this; - // fall through to the generic state error if it does. - // * Peer sent us a RESET_STREAM. The C++ side records the - // code in state.resetCode regardless of whether the JS - // onreset handler was attached. state.finReceived stays - // false because no FIN was seen. - // * We aborted locally via stream.resetStream() or - // stream.stopSending(). Both paths run EndReadable in - // C++, setting state.readEnded without setting - // state.finReceived. There is no peer code to surface. - if (inner.state.readEnded && !inner.state.finReceived) { - const peerResetCode = inner.state.resetCode; - if (peerResetCode !== undefined && peerResetCode > 0n) { - return new ERR_QUIC_STREAM_RESET(Number(peerResetCode)); - } - return new ERR_QUIC_STREAM_ABORTED( - 'Stream aborted before FIN was received'); - } - return new ERR_INVALID_STATE('The stream is not readable'); - }, - }); + // No reader means either an outbound-only unidirectional stream, or a + // stream already destroyed (data gone, but truncation must still be + // checked below). + if (inner.reader) { + yield* createBlobReaderIterable(inner.reader); + } + + if (inner.state.readEnded && !inner.state.finReceived) { + // The readable has been truncated - ended with no clean FIN. We expose + // this in different ways depending on the truncatedReads option. + + // If we cancelled ourselves, check our own stop-sending code (not the + // result mirrored by the remote peer) + if (inner.stopSendingCode > 0n) { + throw new QuicError('Stream aborted before FIN was received', + { __proto__: null, + errorCode: inner.stopSendingCode }); + } + + // Non-zero reset is always an error: + const peerResetCode = inner.state.resetCode; + if (peerResetCode > 0n) { + throw new QuicError( + `The QUIC stream was reset by the peer with error code ${peerResetCode}`, + { __proto__: null, + code: 'ERR_QUIC_STREAM_RESET', + errorCode: peerResetCode }); + } + + // If stream teardown has started (stats.destroyedAt is set) then a + // close event confirming a final error/clean close will settle + // imminently (might be settled already). We await here to rethrow + // any errors if the connection has failed. + if (this.destroyed || this.stats.destroyedAt !== 0n) { + await this.closed; + } + + // Clean abort is truncation, but not necessarily an error: + if (inner.truncatedReads === 'error') { + throw new QuicError('Stream aborted before FIN was received', + { __proto__: null, errorCode: peerResetCode ?? 0n }); + } + } } /** @@ -2063,9 +2081,11 @@ class QuicStream { if (error !== undefined && typeof inner.onerror === 'function') { invokeOnerror(inner.onerror, error); } - const handle = this.#handle; - this[kFinishClose](error); - handle.destroy(); + // handle.destroy() kicks off all the cleanup internals, eventually + // including [kFinishClose] which needs this original destroy error: + inner.destroyError = error; + this.#handle.destroy(); + this[kFinishClose](error); // A no-op, unless destroy() failed } /** @@ -2468,7 +2488,9 @@ class QuicStream { stopSending(code = 0n) { assertIsQuicStream(this); if (this.destroyed) return; - this.#handle.stopSending(BigInt(code)); + const abortCode = BigInt(code); + this.#inner.stopSendingCode = abortCode; + this.#handle.stopSending(abortCode); } /** @@ -2561,6 +2583,9 @@ class QuicStream { if (this.destroyed) { return inner.pendingClose.promise; } + // Prefer an error staged by destroy() (the original object the caller + // passed) over the error delivered by the native close callback. + error = inner.destroyError ?? error; if (error !== undefined) { inner.pendingClose.reject(error); } else { @@ -2591,6 +2616,7 @@ class QuicStream { inner.session = undefined; inner.pendingClose.reject = undefined; inner.pendingClose.resolve = undefined; + inner.destroyError = undefined; inner.onblocked = undefined; inner.onreset = undefined; inner.onstopsending = undefined; @@ -2783,6 +2809,7 @@ class QuicSession { // because server-side cert validation is handled by rejectUnauthorized // at the C++ level. verifyPeer: 'manual', + truncatedReads: 'error', handshakeInfo: undefined, /** @type {QuicSessionPath|undefined} */ path: undefined, @@ -2816,8 +2843,9 @@ class QuicSession { * @param {symbol} privateSymbol * @param {object} handle * @param {QuicEndpoint} endpoint + * @param {{ truncatedReads?: 'error'|'allow' }} [options] */ - constructor(privateSymbol, handle, endpoint) { + constructor(privateSymbol, handle, endpoint, options = kEmptyObject) { // Instances of QuicSession can only be created internally. assertPrivateSymbol(privateSymbol); @@ -2826,6 +2854,8 @@ class QuicSession { const inner = this.#inner; inner.endpoint = endpoint; + const { truncatedReads } = options; + if (truncatedReads !== undefined) inner.truncatedReads = truncatedReads; // Move any qlog entries that arrived before the wrapper existed. if (handle._pendingQlog !== undefined) { inner.pendingQlog = handle._pendingQlog; @@ -3371,7 +3401,8 @@ class QuicSession { } const stream = new QuicStream( - kPrivateConstructor, handle, this, direction, true /* isLocal */); + kPrivateConstructor, handle, this, direction, true /* isLocal */, + inner.truncatedReads); inner.streams.add(stream); if (typeof this.#inner.onerror === 'function') { markPromiseAsHandled(stream.closed); @@ -4156,7 +4187,7 @@ class QuicSession { [kNewStream](handle, direction) { const inner = this.#inner; const stream = new QuicStream(kPrivateConstructor, handle, this, direction, - false /* isLocal */); + false /* isLocal */, inner.truncatedReads); // Set the default byte budget for received streams. stream.budget = kDefaultBudget; @@ -4273,6 +4304,7 @@ class QuicEndpoint { sessions: new SafeSet(), stat: undefined, stats: undefined, + truncatedReads: undefined, onsession: undefined, sessionCallbacks: undefined, }; @@ -4405,8 +4437,8 @@ class QuicEndpoint { }; } - #newSession(handle) { - const session = new QuicSession(kPrivateConstructor, handle, this); + #newSession(handle, options) { + const session = new QuicSession(kPrivateConstructor, handle, this, options); this.#inner.sessions.add(session); // Set default pending datagram queue size. session.maxPendingDatagrams = kDefaultMaxPendingDatagrams; @@ -4580,9 +4612,13 @@ class QuicEndpoint { ontrailers, oninfo, onwanttrailers, + // Stored on the endpoint and applied to each incoming session. + truncatedReads, ...rest } = options; + inner.truncatedReads = truncatedReads; + // Store session and stream callbacks to apply to each new incoming session. inner.sessionCallbacks = { __proto__: null, @@ -4624,6 +4660,7 @@ class QuicEndpoint { validateObject(options, 'options'); const { sessionTicket, + truncatedReads, ...rest } = options; @@ -4632,7 +4669,7 @@ class QuicEndpoint { if (handle === undefined) { throw new ERR_QUIC_CONNECTION_FAILED(); } - const session = this.#newSession(handle); + const session = this.#newSession(handle, { __proto__: null, truncatedReads }); // Set callbacks before any async work to avoid missing events // that fire during or immediately after the handshake. applyCallbacks(session, options); @@ -4866,7 +4903,8 @@ class QuicEndpoint { const inner = this.#inner; assert(typeof inner.onsession === 'function', 'onsession callback not specified'); - const session = this.#newSession(handle); + const session = this.#newSession(handle, + { __proto__: null, truncatedReads: inner.truncatedReads }); // Apply session callbacks stored at listen time before notifying // the onsession callback, to avoid missing events that fire // during or immediately after the handshake. @@ -5358,6 +5396,7 @@ function processSessionOptions(options, config = kEmptyObject) { maxDatagramSendAttempts = 5, streamIdleTimeout, verifyPeer = 'auto', + truncatedReads = 'error', // HTTP/3 application-specific options. Nested under `application` // to separate protocol-specific settings from transport-level ones. application = kEmptyObject, @@ -5409,6 +5448,9 @@ function processSessionOptions(options, config = kEmptyObject) { validateOneOf(verifyPeer, 'options.verifyPeer', ['strict', 'auto', 'manual']); + validateOneOf(truncatedReads, 'options.truncatedReads', + ['error', 'allow']); + validateInteger(drainingPeriodMultiplier, 'options.drainingPeriodMultiplier', 3, 255); @@ -5471,6 +5513,7 @@ function processSessionOptions(options, config = kEmptyObject) { verifyHostname: verifyPeer !== 'manual', }, verifyPeer, + truncatedReads, qlog, maxPayloadSize, unacknowledgedPacketThreshold, diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js index 769fee687498..874349ffd9b2 100644 --- a/lib/internal/quic/state.js +++ b/lib/internal/quic/state.js @@ -8,7 +8,9 @@ const { DataView, DataViewPrototypeGetBigInt64, DataViewPrototypeGetBigUint64, + DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, + DataViewPrototypeGetByteOffset, DataViewPrototypeGetUint16, DataViewPrototypeGetUint32, DataViewPrototypeGetUint8, @@ -17,6 +19,9 @@ const { DataViewPrototypeSetUint8, JSONStringify, Number, + TypedArrayPrototypeGetBuffer, + TypedArrayPrototypeSlice, + Uint8Array, } = primordials; const { @@ -109,6 +114,7 @@ const { IDX_STATE_STREAM_WRITE_DESIRED_SIZE, IDX_STATE_STREAM_BUDGET, IDX_STATE_STREAM_RESET_CODE, + IDX_STATE_STREAM_SIZE, } = internalBinding('quic'); assert(IDX_STATE_SESSION_LISTENER_FLAGS !== undefined); @@ -727,8 +733,8 @@ class QuicStreamState { #handle; /** @type {number} */ #offset = 0; - /** @type {bigint|undefined} */ - #id = undefined; + /** @type {boolean} */ + #disconnected = false; /** * @param {symbol} privateSymbol @@ -750,7 +756,7 @@ class QuicStreamState { /** @type {bigint} */ get id() { const handle = this.#handle; - if (handle === undefined) return this.#id; + if (handle === undefined) return undefined; return DataViewPrototypeGetBigInt64(handle, this.#offset + IDX_STATE_STREAM_ID, kIsLittleEndian); } @@ -979,7 +985,7 @@ class QuicStreamState { } [kInspect](depth, options) { - if (this.#handle === undefined || + if (this.#disconnected || this.#handle === undefined || DataViewPrototypeGetByteLength(this.#handle) === 0) { return 'QuicStreamState { }'; } @@ -1038,9 +1044,24 @@ class QuicStreamState { } [kFinishClose]() { - // Cache the stream ID since the buffer will be zeroed out and the ID will be lost. - this.#id = this.id; - this.#handle = undefined; + // Copy the final values out of the shared buffer (zeroed when the + // underlying stream goes away) so getters keep reporting the final + // state, the same way QuicStreamStats does. + const handle = this.#handle; + if (handle === undefined || + DataViewPrototypeGetByteLength(handle) < + this.#offset + IDX_STATE_STREAM_SIZE) { + this.#handle = undefined; + this.#disconnected = true; + return; + } + const copy = TypedArrayPrototypeSlice(new Uint8Array( + DataViewPrototypeGetBuffer(handle), + DataViewPrototypeGetByteOffset(handle) + this.#offset, + IDX_STATE_STREAM_SIZE)); + this.#handle = new DataView(TypedArrayPrototypeGetBuffer(copy)); + this.#offset = 0; + this.#disconnected = true; } } diff --git a/src/quic/streams.cc b/src/quic/streams.cc index 1f8761c75049..556649f1511a 100644 --- a/src/quic/streams.cc +++ b/src/quic/streams.cc @@ -1058,6 +1058,9 @@ void Stream::InitPerContext(Realm* realm, Local target) { NODE_DEFINE_CONSTANT(target, IDX_STATS_STREAM_COUNT); + constexpr auto IDX_STATE_STREAM_SIZE = sizeof(Stream::State); + NODE_DEFINE_CONSTANT(target, IDX_STATE_STREAM_SIZE); + constexpr int QUIC_STREAM_HEADERS_KIND_HINTS = static_cast(HeadersKind::HINTS); constexpr int QUIC_STREAM_HEADERS_KIND_INITIAL = @@ -1404,13 +1407,6 @@ BaseObjectPtr Stream::get_reader() { return reader; } -void Stream::set_final_size(uint64_t final_size) { - DCHECK_IMPLIES(state()->fin_received == 1, - final_size <= STAT_GET(Stats, final_size)); - state()->fin_received = 1; - STAT_SET(Stats, final_size, final_size); -} - void Stream::set_outbound(std::shared_ptr source) { if (!source || !is_writable()) return; Debug(this, "Setting the outbound data source"); @@ -1554,7 +1550,7 @@ void Stream::FlushAccumulation() { return; } // Should be unreachable: append() only fails once the queue has been capped, - // EndReadable() flushes before capping, and ReceiveData() accumulates + // CapReadable() flushes before capping, and ReceiveData() accumulates // nothing once read_ended is set. Reaching here means received stream data // is being dropped on the floor, so say so and at least do not also leak // the flow control credit for it. @@ -1641,13 +1637,26 @@ void Stream::EndWritable() { state()->write_ended = 1; } -void Stream::EndReadable(std::optional maybe_final_size) { +void Stream::FinishReadable() { + if (!is_readable()) return; + state()->fin_received = 1; + CapReadable(std::nullopt); +} + +void Stream::TruncateReadable(std::optional maybe_final_size) { if (!is_readable()) return; + CapReadable(maybe_final_size); +} + +void Stream::CapReadable(std::optional maybe_final_size) { + DCHECK(is_readable()); state()->read_ended = 1; // Flush any accumulated data before capping so the reader can see it. FlushAccumulation(); - set_final_size(maybe_final_size.value_or(STAT_GET(Stats, bytes_received))); - inbound_->cap(STAT_GET(Stats, final_size)); + const uint64_t final_size = + maybe_final_size.value_or(STAT_GET(Stats, bytes_received)); + STAT_SET(Stats, final_size, final_size); + inbound_->cap(final_size); // Notify the JS reader so it can see EOS. The subsequent pull observes // the now-capped DataQueue and returns EOS. if (reader_) reader_->NotifyPull(); @@ -1675,13 +1684,14 @@ void Stream::Destroy(QuicError error) { // End the writable before marking as destroyed. EndWritable(); - // Also end the readable side if it isn't already. - EndReadable(); + // Also end the readable side if it isn't already. If not already ended, + // this will eventually surface as an error, since the data is truncated. + TruncateReadable(); // We are going to release our reference to the outbound_ queue here. outbound_.reset(); - // EndReadable() above already flushed accumulated data. Just release + // TruncateReadable() above already flushed accumulated data. Just release // the ring buffer memory. recv_accumulator_.reset(); @@ -1736,7 +1746,7 @@ void Stream::ReceiveData(const uint8_t* data, // end the readable side if this is the last bit of data we've received. Debug(this, "Receiving %zu bytes of data", len); if (state()->read_ended == 1 || len == 0) { - if (flags.fin) EndReadable(); + if (flags.fin) FinishReadable(); // Nothing will ever consume these bytes, so return the connection-level // credit ngtcp2 charged for them. The stream window is deliberately left // alone: there is no point inviting more data onto a stream we have @@ -1819,7 +1829,7 @@ void Stream::ReceiveData(const uint8_t* data, if (flags.fin) { FlushAccumulation(); - EndReadable(); + FinishReadable(); } else if (reader_ && was_empty) { // Notify the reader once when the accumulator transitions from empty // to non-empty. This wakes the reader exactly once per accumulation @@ -1850,7 +1860,7 @@ void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) { final_size, error); state()->reset_code = error.code(); - EndReadable(final_size); + TruncateReadable(final_size); EmitReset(error); } @@ -1873,7 +1883,7 @@ void Stream::DoStreamReset(error_code code) { } void Stream::SendStopSending(error_code code) { - EndReadable(); + TruncateReadable(); if (!is_pending()) { // If the stream is a local unidirectional there's nothing to do here. diff --git a/src/quic/streams.h b/src/quic/streams.h index cd4849aae277..c40d2fb2596a 100644 --- a/src/quic/streams.h +++ b/src/quic/streams.h @@ -309,7 +309,12 @@ class Stream final : public AsyncWrap, void Commit(size_t datalen, bool fin = false); void EndWritable(); - void EndReadable(std::optional maybe_final_size = std::nullopt); + // The read side ended cleanly with a peer FIN: the content is complete. + void FinishReadable(); + // The read side ended without a FIN (a reset, a local abort, or the session + // being torn down) so the content is truncated. + void TruncateReadable( + std::optional maybe_final_size = std::nullopt); void EntryRead(size_t amount) override; void BeforePull() override; @@ -411,10 +416,13 @@ class Stream final : public AsyncWrap, // consumer or dropped before reaching one. void CreditConsumedBytes(uint64_t amount); + // Common tail of FinishReadable()/TruncateReadable(): marks the read side + // ended, caps it at the final size, and notifies the reader. + void CapReadable(std::optional maybe_final_size); + // Gets a reader for the data received for this stream from the peer, BaseObjectPtr get_reader(); - void set_final_size(uint64_t amount); void set_outbound(std::shared_ptr source); // Streaming outbound support diff --git a/test/common/quic.mjs b/test/common/quic.mjs index dc4b094cf900..bb6db9ed63de 100644 --- a/test/common/quic.mjs +++ b/test/common/quic.mjs @@ -7,6 +7,7 @@ // listen/connect that apply default options suitable for most tests. import * as fixtures from '../common/fixtures.mjs'; +import { setTimeout } from 'node:timers/promises'; const { createPrivateKey } = await import('node:crypto'); const quic = await import('node:quic'); @@ -98,6 +99,82 @@ function hashBytes(buf) { return h >>> 0; } +/** + * Write `size` bytes to the stream and wait until the peer has + * acknowledged them. + */ +async function writeAndAwaitAck(stream, size) { + stream.writer.write(new Uint8Array(size).fill(7)); + while (stream.stats.maxOffsetAcknowledged < BigInt(size)) await setTimeout(5); +} + +/** + * Send `size` bytes and then stall forever without a FIN + * @yields {Uint8Array} + */ +async function* stallingBody(size) { + yield new Uint8Array(size).fill(7); + await new Promise(() => {}); +} + +/** + * Do a full single stream-read scenario, with the given client options and + * server body, and two hooks available to tweak behaviour, returning the + * details of the result after completion. + * @param {Function} serverBody + * @param {object} [options] + * @param {object} [options.clientOptions] Options forwarded to connect(). + * @param {Function} [options.beforeIterate] + * @param {Function} [options.onFirstChunk] + * @returns {Promise<{received: number, threw: any, closedError: any}>} + * Bytes received, the iteration error if any, and the client stream's + * closed rejection if any. + */ +async function readStream(serverBody, options = {}) { + const { clientOptions, beforeIterate, onFirstChunk } = options; + + const serverEndpoint = await listen((session) => { + session.closed.catch(() => {}); + session.onstream = (stream) => { + stream.closed.catch(() => {}); + serverBody(stream, session); + }; + }); + + const session = await connect(serverEndpoint.address, clientOptions); + await session.opened; + session.closed.catch(() => {}); + + const stream = await session.createBidirectionalStream(); + await stream.writer.write(new Uint8Array([1])); + + let closedError; + const closedSettled = + stream.closed.then(() => {}, (err) => { closedError = err; }); + + await beforeIterate?.({ stream, session }); + + let received = 0; + let threw; + let firstChunk = true; + try { + for await (const chunk of stream) { + for (const c of chunk) received += c.byteLength; + if (firstChunk) { + firstChunk = false; + await onFirstChunk?.({ stream, session }); + } + } + } catch (err) { + threw = err; + } + + session.close(); + await closedSettled; + await serverEndpoint.close(); + return { received, threw, closedError }; +} + export { key, cert, @@ -105,4 +182,7 @@ export { connect, makePayload, hashBytes, + readStream, + stallingBody, + writeAndAwaitAck, }; diff --git a/test/parallel/test-quic-stream-iteration-destroyed.mjs b/test/parallel/test-quic-stream-iteration-destroyed.mjs deleted file mode 100644 index 585be6d4a813..000000000000 --- a/test/parallel/test-quic-stream-iteration-destroyed.mjs +++ /dev/null @@ -1,37 +0,0 @@ -// Flags: --experimental-quic --no-warnings - -// Test: destroyed stream returns finished iterator. - -import { hasQuic, skip, mustCall } from '../common/index.mjs'; -import * as assert from 'node:assert'; - -if (!hasQuic) { - skip('QUIC is not enabled'); -} - -const { listen, connect } = await import('../common/quic.mjs'); - -const encoder = new TextEncoder(); - -const serverEndpoint = await listen(mustCall(async (serverSession) => { - await serverSession.closed; -})); - -const clientSession = await connect(serverEndpoint.address); -await clientSession.opened; - -const stream = await clientSession.createBidirectionalStream({ - body: encoder.encode('destroy test'), -}); - -// Destroy the stream immediately. -stream.destroy(); - -// Iterating a destroyed stream should immediately finish. -const iter = stream[Symbol.asyncIterator](); -const { done } = await iter.next(); -assert.strictEqual(done, true); - -await stream.closed; -await clientSession.close(); -await serverEndpoint.destroy(); diff --git a/test/parallel/test-quic-stream-iteration-reset.mjs b/test/parallel/test-quic-stream-iteration-reset.mjs deleted file mode 100644 index 6df571bd9ea7..000000000000 --- a/test/parallel/test-quic-stream-iteration-reset.mjs +++ /dev/null @@ -1,64 +0,0 @@ -// Flags: --experimental-quic --experimental-stream-iter --no-warnings - -// Test: peer RESET_STREAM causes iterator to error. -// When the server resets the stream, the client's async iterator -// should throw or return early. - -import { hasQuic, skip, mustCall } from '../common/index.mjs'; -import * as assert from 'node:assert'; - -if (!hasQuic) { - skip('QUIC is not enabled'); -} - -const { listen, connect } = await import('../common/quic.mjs'); - -const encoder = new TextEncoder(); - -const serverReady = Promise.withResolvers(); - -const serverEndpoint = await listen(mustCall((serverSession) => { - serverSession.onstream = mustCall(async (stream) => { - // Reset the stream from the server side. - stream.resetStream(42n); - await assert.rejects(stream.closed, mustCall((err) => { - assert.ok(err); - return true; - })); - serverReady.resolve(); - await serverSession.closed; - }); -}), { transportParams: { maxIdleTimeout: 1 } }); - -const clientSession = await connect(serverEndpoint.address, { - transportParams: { maxIdleTimeout: 1 }, -}); -await clientSession.opened; - -const stream = await clientSession.createBidirectionalStream({ - body: encoder.encode('will be reset by server'), -}); - -// Set up the closed handler before the reset to avoid unhandled rejection. -const closedPromise = assert.rejects(stream.closed, mustCall((err) => { - assert.ok(err); - return true; -})); - -await serverReady.promise; - -// The async iterator should either throw or return early when the -// peer resets the readable side. -try { - for await (const batch of stream) { - // May receive some data before the reset arrives. - assert.ok(Array.isArray(batch)); - } -} catch { - // The iterator may throw when the reset arrives mid-iteration. -} - -// Either way, the stream should close. -await closedPromise; -await clientSession.closed; -await serverEndpoint.close(); diff --git a/test/parallel/test-quic-stream-setbody-errors.mjs b/test/parallel/test-quic-stream-setbody-errors.mjs index a15f2347171e..da045d764c2f 100644 --- a/test/parallel/test-quic-stream-setbody-errors.mjs +++ b/test/parallel/test-quic-stream-setbody-errors.mjs @@ -53,7 +53,13 @@ await clientSession.opened; message: /writer already accessed/, }); - for await (const _ of stream) { /* drain */ } // eslint-disable-line no-unused-vars + // The server handles only the first stream and then closes its session, so + // this stream is never answered and never receives a FIN. Reading it + // therefore surfaces the truncation rather than ending cleanly. + await assert.rejects((async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of stream) { /* drain */ } + })(), { code: 'ERR_QUIC_STREAM_ABORTED' }); await stream.closed; } diff --git a/test/parallel/test-quic-stream-truncated-reads-destroy.mjs b/test/parallel/test-quic-stream-truncated-reads-destroy.mjs new file mode 100644 index 000000000000..01eddd3b0f0f --- /dev/null +++ b/test/parallel/test-quic-stream-truncated-reads-destroy.mjs @@ -0,0 +1,72 @@ +// Flags: --experimental-quic --no-warnings + +// Test: truncation is still reported when the stream is torn down locally +// before the iterator finishes draining. The stream's final read state and +// close error are persisted at close, so a consumer slower than the teardown +// (or one that only starts reading after it) still observes the truncation +// rather than a clean end that would make an incomplete stream look complete. + +import { hasQuic, skip } from '../common/index.mjs'; +import { setTimeout } from 'node:timers/promises'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { readStream, stallingBody } = await import('../common/quic.mjs'); + +// The server sends 1000 bytes then stalls without a FIN, so the client's +// read side can only end by truncation. +const serve = (stream) => { stream.setBody(stallingBody(1000)); }; + +// The session is destroyed with an error while the consumer is mid-iteration: +// the iterator delivers the data that arrived, then throws that same error +// object. +{ + const boom = new Error('boom'); + const { received, threw } = await readStream(serve, { + onFirstChunk: ({ session }) => session.destroy(boom), + }); + assert.ok(received > 0); + assert.strictEqual(threw, boom); +} + +// The session is destroyed without an error mid-iteration: an errorless +// truncation, reported as aborted under the default policy. +{ + const { received, threw } = await readStream(serve, { + onFirstChunk: ({ session }) => session.destroy(), + }); + assert.ok(received > 0); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(threw.errorCode, 0n); +} + +// The session is destroyed before iteration even starts: the buffered data +// is gone, but the truncation is still reported rather than a clean empty +// end. +{ + const { received, threw } = await readStream(serve, { + beforeIterate: async ({ stream, session }) => { + while (stream.stats.bytesReceived === 0n) await setTimeout(5); + session.destroy(); + }, + }); + assert.strictEqual(received, 0); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); +} + +// Destroying the stream itself (rather than the session) before iterating +// reports the truncation the same way, with closed resolving cleanly. +{ + const { received, threw, closedError } = await readStream(serve, { + beforeIterate: async ({ stream }) => { + while (stream.stats.bytesReceived === 0n) await setTimeout(5); + stream.destroy(); + }, + }); + assert.strictEqual(received, 0); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(closedError, undefined); +} diff --git a/test/parallel/test-quic-stream-truncated-reads-timeout.mjs b/test/parallel/test-quic-stream-truncated-reads-timeout.mjs new file mode 100644 index 000000000000..5119d348537b --- /dev/null +++ b/test/parallel/test-quic-stream-truncated-reads-timeout.mjs @@ -0,0 +1,35 @@ +// Flags: --experimental-quic --no-warnings + +// Test: a readable truncated by the connection idle timeout delivers the +// data it received, then ends per the truncatedReads policy: an error under +// the default (so an incomplete stream can never look complete), and a clean +// end under 'allow' (an idle timeout carries no error). + +import { hasQuic, skip } from '../common/index.mjs'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { readStream, stallingBody } = await import('../common/quic.mjs'); + +// The server sends 1000 bytes then stalls without a FIN, so the short (1s) +// connection idle timeout is the only thing that ends the read side. +const serve = (stream) => { stream.setBody(stallingBody(1000)); }; +const transportParams = { maxIdleTimeout: 1 }; + +{ + const { received, threw } = + await readStream(serve, { clientOptions: { transportParams } }); + assert.strictEqual(received, 1000); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(threw.errorCode, 0n); +} +{ + const { received, threw } = await readStream(serve, { + clientOptions: { transportParams, truncatedReads: 'allow' }, + }); + assert.strictEqual(received, 1000); + assert.strictEqual(threw, undefined); +} diff --git a/test/parallel/test-quic-stream-truncated-reads.mjs b/test/parallel/test-quic-stream-truncated-reads.mjs new file mode 100644 index 000000000000..25cc3f82755e --- /dev/null +++ b/test/parallel/test-quic-stream-truncated-reads.mjs @@ -0,0 +1,184 @@ +// Flags: --experimental-quic --no-warnings + +// Test: a stream read that ends without a FIN is a truncation. What the read +// reports is decided in this order: +// +// 1. the peer reset the stream with a non-zero code: ERR_QUIC_STREAM_RESET +// carrying that code, under either policy; +// 2. the stream was already being torn down and its closed promise rejected: +// that same error, under either policy; +// 3. otherwise the truncation carries no error of its own and the +// truncatedReads policy decides - 'error' (the default) throws +// ERR_QUIC_STREAM_ABORTED so an incomplete stream can never look +// complete, 'allow' ends the read cleanly. +// +// This file covers rules 1 and 3 on a live connection: peer resets and local +// aborts. Rule 2 needs the stream to already be tearing down when the reader +// resumes, which is covered by the sibling +// test-quic-stream-truncated-reads-destroy test, and truncation by idle +// timeout by test-quic-stream-truncated-reads-timeout. + +import { hasQuic, skip } from '../common/index.mjs'; +import assert from 'node:assert'; + +if (!hasQuic) { + skip('QUIC is not enabled'); +} + +const { connect, listen, readStream, stallingBody, writeAndAwaitAck } = + await import('../common/quic.mjs'); + +// Sends 1000 bytes, waits for them to land, then resets with the given code. +const resetWith = (code) => async (stream) => { + await writeAndAwaitAck(stream, 1000); + stream.resetStream(code); +}; + +// Sends 1000 bytes and then stalls without a FIN, so only the client's own +// abort can end the read. +const stall = (stream) => { stream.setBody(stallingBody(1000)); }; + +// A peer reset with code 0 is a clean abort: a truncation, but not an error. +// Rule 3 - the default reports it, 'allow' treats it as a clean end. +{ + const { received, threw } = await readStream(resetWith(0n)); + assert.strictEqual(received, 1000); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(threw.errorCode, 0n); +} +{ + const { received, threw } = + await readStream(resetWith(0n), { clientOptions: { truncatedReads: 'allow' } }); + assert.strictEqual(received, 1000); + assert.strictEqual(threw, undefined); +} + +// A peer reset with a nonzero code is rule 1: an error under either policy, +// carrying the peer's code. It also rejects the closed promise on both sides. +{ + const serverClosed = Promise.withResolvers(); + const { received, threw, closedError } = await readStream(async (stream) => { + await writeAndAwaitAck(stream, 1000); + stream.resetStream(42n); + // Our own reset closes the server-side stream with an error too. + serverClosed.resolve(stream.closed.then(() => undefined, (err) => err)); + }); + assert.strictEqual(received, 1000); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_RESET'); + assert.strictEqual(threw.errorCode, 42n); + assert.strictEqual(threw.message, + 'The QUIC stream was reset by the peer with error code 42'); + assert.strictEqual(closedError?.code, 'ERR_QUIC_APPLICATION_ERROR'); + assert.strictEqual(closedError.errorCode, 42n); + const serverClosedError = await serverClosed.promise; + assert.strictEqual(serverClosedError?.code, 'ERR_QUIC_APPLICATION_ERROR'); + assert.strictEqual(serverClosedError.errorCode, 42n); +} +{ + const { received, threw } = + await readStream(resetWith(42n), { clientOptions: { truncatedReads: 'allow' } }); + assert.strictEqual(received, 1000); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_RESET'); + assert.strictEqual(threw.errorCode, 42n); +} + +// Aborting our own read with stopSending() is rule 3, not rule 1: we asked for +// the truncation, so it is not an error the peer inflicted on us, and the code +// we send does not come back as one. The read still stops short, so the +// default policy reports it and 'allow' does not. +{ + const { received, threw, closedError } = await readStream(stall, { + onFirstChunk: ({ stream }) => stream.stopSending(0n), + }); + assert.ok(received > 0); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(threw.errorCode, 0n); + assert.strictEqual(closedError, undefined); +} +{ + const { received, threw } = await readStream(stall, { + clientOptions: { truncatedReads: 'allow' }, + onFirstChunk: ({ stream }) => stream.stopSending(0n), + }); + assert.ok(received > 0); + assert.strictEqual(threw, undefined); +} + +// A nonzero stopSending code is an error this end raised, so it is reported +// under either policy and carries that code. The peer answers our +// STOP_SENDING with a RESET_STREAM echoing it, which is what rejects closed - +// but the read reports our own abort rather than attributing it to the peer, +// and does so whether or not that answer has arrived yet. +for (const truncatedReads of ['error', 'allow']) { + for (const awaitEcho of [false, true]) { + const peerReset = Promise.withResolvers(); + const { received, threw, closedError } = await readStream(stall, { + clientOptions: { truncatedReads }, + beforeIterate: ({ stream }) => { stream.onreset = () => peerReset.resolve(); }, + onFirstChunk: async ({ stream }) => { + stream.stopSending(7n); + // For the 2nd pass, wait until we receive the corresponding reset: + if (awaitEcho) await peerReset.promise; + }, + }); + assert.ok(received > 0); + assert.strictEqual(threw?.code, 'ERR_QUIC_STREAM_ABORTED'); + assert.strictEqual(threw.errorCode, 7n); + assert.strictEqual(closedError?.code, 'ERR_QUIC_APPLICATION_ERROR'); + assert.strictEqual(closedError.errorCode, 7n); + } +} + +// A peer abruptly destroying its session truncates the read too. That +// currently reaches us as an implicit reset, so which rule applies depends on +// the code the peer's teardown puts on the wire - not part of this contract, +// and deliberately not pinned here. What must hold either way is the default +// policy's promise: the incomplete stream never looks complete. +{ + const { received, threw } = await readStream(async (stream, session) => { + await writeAndAwaitAck(stream, 1000); + session.destroy(new Error('connection boom')); + }); + assert.strictEqual(received, 1000); + assert.ok(threw); +} + +// Check the option in the server case as well: +{ + const serverRead = Promise.withResolvers(); + const serverEndpoint = await listen((session) => { + session.closed.catch(() => {}); + session.onstream = async (stream) => { + stream.closed.catch(() => {}); + let received = 0; + try { + for await (const chunk of stream) { + for (const c of chunk) received += c.byteLength; + } + serverRead.resolve({ received, threw: undefined }); + } catch (err) { + serverRead.resolve({ received, threw: err }); + } + }; + }, { truncatedReads: 'allow' }); + + const session = await connect(serverEndpoint.address); + await session.opened; + session.closed.catch(() => {}); + + const stream = await session.createBidirectionalStream(); + stream.closed.catch(() => {}); + await writeAndAwaitAck(stream, 100); + stream.resetStream(0n); + + const { threw } = await serverRead.promise; + assert.strictEqual(threw, undefined); + + session.close(); + await serverEndpoint.close(); +} + +// The option is validated. +await assert.rejects(connect('127.0.0.1:1234', { truncatedReads: 'nope' }), { + code: 'ERR_INVALID_ARG_VALUE', +});