diff --git a/src/js/node/http2.ts b/src/js/node/http2.ts index f41ce68605fc..60f58eb354a3 100644 --- a/src/js/node/http2.ts +++ b/src/js/node/http2.ts @@ -1907,6 +1907,11 @@ abstract class Http2Session extends EventEmitter { // run inside it so 'close' doesn't inherit the last stream's frame. [bunHTTP2AsyncContextFrame] = $getInternalField($asyncContext, 0); [kDeferWriteCallback]: typeof process.nextTick | typeof setImmediate = setImmediate; + constructor() { + super(); + // Refreshed like node's [kUpdateTimer]: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L1414-L1418 + this[kTimeout] = null; + } // The GOAWAY this side received (not one it sent), like node's Http2Session getters. get goawayCode() { return this[kGoawayCode] || NGHTTP2_NO_ERROR; @@ -2060,6 +2065,8 @@ enum StreamState { // callback). Until then no 'error' listener can exist, so stream errors must not be emitted: // node never constructs the JS stream object before a complete header block arrives. Delivered = 1 << 8, // 100000000 = 256 + // 'end' came with no consumer (END_STREAM on HEADERS, or unread data dropped at the close). node waits for the close. + EndedUnread = 1 << 9, // 1000000000 = 512 } // native.writeStream() return-value flag (mirrors WRITE_FLUSHED_WITHOUT_CALLBACK in // h2_frame_parser.rs): the chunk was handed to the socket without queueing and the engine did @@ -2077,6 +2084,36 @@ const kWriteFlushedWithoutCallback = 0x10; function deferWriteCallbackForSocket(nativeSocket) { return nativeSocket ? process.nextTick : setImmediate; } +// 0: a stream write since the last expiry. Above 0: the frames sent at the last expiry, with a write in flight. -1: none. +const kTimeoutFramesSent = Symbol("timeoutFramesSent"); +const kWriteCallback = Symbol("writeCallback"); +const kOnWriteComplete = Symbol("onWriteComplete"); +// node's onWriteComplete: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L81-L101 +function onStreamWriteComplete(this: Http2Stream, dropped?: null) { + const timer = this[bunHTTP2Session]?.[kTimeout]; + // `dropped` is null when the parser dropped or refused the write. + if (timer && dropped === undefined && !this.destroyed && !destroyedInNode(this)) timer.refresh(); + const callback = this[kWriteCallback]; + this[kWriteCallback] = undefined; + callback(); +} +// node has destroyed such a stream: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L607-L609 +function destroyedInNode(stream: Http2Stream) { + return ( + isFinalWrite(stream, stream._writableState.writelen) && + stream._readableState.ended && + stream.readableLength === 0 && + (stream.readableFlowing === true || stream.readableEnded) && + (stream[bunHTTP2StreamStatus] & StreamState.EndedUnread) === 0 + ); +} +// node's kWriteGeneric and trackWriteState: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2277 +function trackTimedWrite(session: Http2Session, stream: Http2Stream, callback: () => void) { + session[kTimeout].refresh(); + session[kTimeoutFramesSent] = 0; + stream[kWriteCallback] = callback; + return (stream[kOnWriteComplete] ??= onStreamWriteComplete.bind(stream)); +} // Whether this chunk is the writable's last: end() ran, nothing queued behind it (writableLength // includes the in-flight chunk, string chunks in code units), no trailers pending (END_STREAM // rides trailer HEADERS). Node packs END_STREAM onto the final DATA frame, not a trailing empty one. @@ -2300,10 +2337,12 @@ function destroyStreamForSessionDestroy(error: Error | undefined, rstCode: numbe } interface Http2StreamReadableState { destroyed: boolean; + ended: boolean; } interface Http2StreamWritableState { ending: boolean; destroyed: boolean; + writelen: number; } type DuplexStream = import("node:stream").Duplex; interface DuplexStateAccessors { @@ -2433,6 +2472,8 @@ class Http2Stream extends (Duplex as Http2StreamBase) { } else { headers = { ...headers }; } + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2387 + if (session[kTimeout]) session[kTimeout].refresh(); const sensitives = headers[sensitiveHeaders]; if (sensitives !== undefined && !$isArray(sensitives)) { throw $ERR_INVALID_ARG_VALUE("headers[http2.neverIndex]", sensitives); @@ -2903,7 +2944,7 @@ class Http2Stream extends (Duplex as Http2StreamBase) { } } const chunk = Buffer.concat(chunks || []); - if (session[kTimeout]) session[kTimeout].refresh(); + if (session[kTimeout]) callback = trackTimedWrite(session, this, callback); const endStream = isFinalWrite(this, batchLength); const status = native.writeStream(this.#id, chunk, undefined, endStream, callback, true); if (status & kWriteFlushedWithoutCallback) session[kDeferWriteCallback](callback); @@ -2944,7 +2985,7 @@ class Http2Stream extends (Duplex as Http2StreamBase) { wireChunk = Buffer.from(chunk, encoding); wireEncoding = undefined; } - if (session[kTimeout]) session[kTimeout].refresh(); + if (session[kTimeout]) callback = trackTimedWrite(session, this, callback); const endStream = isFinalWrite(this, chunk.length); const status = native.writeStream(this.#id, wireChunk, wireEncoding, endStream, callback, true); if (status & kWriteFlushedWithoutCallback) session[kDeferWriteCallback](callback); @@ -3014,7 +3055,7 @@ function doSendFileFD(options, fd, headers, err, stat) { if (onError) onError(err); else { - this.respond(headers, options); + this[kRespond](headers, options, false); this.destroy(streamErrorFromCode(NGHTTP2_INTERNAL_ERROR)); } return; @@ -3033,7 +3074,7 @@ function doSendFileFD(options, fd, headers, err, stat) { if (ownsFd) tryClose(fd); if (onError) onError(err); else { - this.respond(headers, options); + this[kRespond](headers, options, false); this.destroy(err); } return; @@ -3090,7 +3131,7 @@ function doSendFileFD(options, fd, headers, err, stat) { headers[HTTP2_HEADER_CONTENT_LENGTH] = statOptions.length; } try { - this.respond(headers, options); + this[kRespond](headers, options, false); } catch (err) { // respond() rejected the headers (e.g. a request pseudo-header in the response): the fd opened // for the file never reaches a read stream, so close it here before the stream is destroyed. @@ -3151,6 +3192,7 @@ function onFileStreamError(this: Http2Stream) { if (!this.destroyed && !this.closed) this.close(NGHTTP2_INTERNAL_ERROR); } const kFileResponseFinal = Symbol("fileResponseFinal"); +const kRespond = Symbol("respond"); // node processRespondWithFD: a file response closes the user-facing writable side // (`self._final = null; self.end()`), so a stream.end() issued by the user afterwards cannot cut // the transfer short. Returns the original _final, which the file sink runs once the whole file @@ -3271,6 +3313,8 @@ class ServerHttp2Stream extends Http2Stream { throw $ERR_HTTP2_PUSH_DISABLED(); } const session = this[bunHTTP2Session]; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2911 + if (session?.[kTimeout]) session[kTimeout].refresh(); const parser = session?.[bunHTTP2Native]; if (!parser) { throw $ERR_HTTP2_INVALID_STREAM(); @@ -3404,6 +3448,9 @@ class ServerHttp2Stream extends Http2Stream { if (options.statCheck !== undefined && typeof options.statCheck !== "function") { throw $ERR_INVALID_ARG_VALUE("options.statCheck", options.statCheck); } + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L3159 + const timer = this[bunHTTP2Session]?.[kTimeout]; + if (timer) timer.refresh(); this[kOwnsFd] = true; fs.open(path, "r", afterOpen.bind(this, options || {}, headers)); } @@ -3463,6 +3510,9 @@ class ServerHttp2Stream extends Http2Stream { if (options.statCheck !== undefined && typeof options.statCheck !== "function") { throw $ERR_INVALID_ARG_VALUE("options.statCheck", options.statCheck); } + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L3096 + const timer = this[bunHTTP2Session]?.[kTimeout]; + if (timer) timer.refresh(); // The caller owns this fd; clear any stale flag left by a prior respondWithFile() // on the same stream so doSendFileFD will not close it (node semantics). this[kOwnsFd] = false; @@ -3539,6 +3589,8 @@ class ServerHttp2Stream extends Http2Stream { } const session = this[bunHTTP2Session]; assertSession(session); + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L3206 + if (session[kTimeout]) session[kTimeout].refresh(); if (!this[kInfoHeaders]) { this[kInfoHeaders] = [headers]; } else { @@ -3548,6 +3600,10 @@ class ServerHttp2Stream extends Http2Stream { session[bunHTTP2Native]?.request(this.id, undefined, headers, sensitiveNames); } respond(headers?: HeadersObject | any[] | null, options?: any) { + this[kRespond](headers, options, true); + } + // respondWithFile() and respondWithFD() pass false: they refreshed the idle timer when they ran. + [kRespond](headers: HeadersObject | any[] | null | undefined, options: any, refresh: boolean) { if (this.destroyed || this.session === undefined) { throw $ERR_HTTP2_INVALID_STREAM(); } @@ -3558,6 +3614,8 @@ class ServerHttp2Stream extends Http2Stream { if (this.sentTrailers) { throw $ERR_HTTP2_TRAILERS_ALREADY_SENT(); } + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L3010 + if (refresh && session[kTimeout]) session[kTimeout].refresh(); // Raw (flat [name, value, ...] array) headers form: the pairs are encoded // on the wire in their given order; a default :status is prepended and a @@ -4065,7 +4123,10 @@ class ServerHttp2Session extends Http2Session { // the native handler_pair! wrap's parser-construction context, so a // peer-initiated stream's captured frame equals what every handler sees. frameError(self: ServerHttp2Session, stream: ServerHttp2Stream, frameType: number, errorCode: number) { - if (!self || typeof stream !== "object") return; + if (!self) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L670 + if (self[kTimeout]) self[kTimeout].refresh(); + if (typeof stream !== "object") return; // Emit the frameError event with the frame type and error code process.nextTick(emitFrameErrorEventNT, stream, frameType, errorCode); }, @@ -4098,6 +4159,10 @@ class ServerHttp2Session extends Http2Session { } if (state == 6 || state == 7) { if (stream.readable) { + // True when a HEADERS frame or an earlier END_STREAM already ended the readable. + const alreadyEnded = stream._readableState.ended; + // END_STREAM on a DATA frame: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L172 + if (self[kTimeout] && !alreadyEnded) self[kTimeout].refresh(); endInboundHalf(stream); // If the user hasn't tried to consume the stream then dump the incoming data so the @@ -4105,6 +4170,7 @@ class ServerHttp2Session extends Http2Session { // attach a tick later (e.g. a CONNECT tunnel piping once its socket connects) and // resuming with buffered data would silently discard it. At full close, dump as before. if ((state == 7 || stream.readableLength === 0) && stream.readableFlowing === null) { + if (alreadyEnded) stream[bunHTTP2StreamStatus] |= StreamState.EndedUnread; stream.resume(); } } @@ -4147,6 +4213,8 @@ class ServerHttp2Session extends Http2Session { }, streamData(self: ServerHttp2Session, stream: ServerHttp2Stream, data: Buffer) { if (!self || typeof stream !== "object" || !data) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L172 + if (self[kTimeout] && !stream.rstCode && !stream.destroyed) self[kTimeout].refresh(); pushToStream(stream, data); // Request data arriving unread after the response finished: the peer is uploading // into a receive window nothing will reopen (see maybeCloseUnreadServerStream). @@ -4158,7 +4226,10 @@ class ServerHttp2Session extends Http2Session { headersTuple: [string[], Record, string[] | undefined], flags: number, ) { - if (!self || typeof stream !== "object" || self.closed || stream.closed) return; + if (!self || typeof stream !== "object") return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L363 + if (self[kTimeout] && !stream.rstCode && !stream.destroyed) self[kTimeout].refresh(); + if (self.closed || stream.closed) return; const requestPerf = stream[kPerfState]; if (requestPerf !== undefined && requestPerf.firstHeader === 0) { requestPerf.firstHeader = performance.now() - requestPerf.start; @@ -4242,11 +4313,15 @@ class ServerHttp2Session extends Http2Session { remoteSettings(self: ServerHttp2Session, settings: Settings) { if (!self) return; self.#remoteSettings = settings; + // Only with a listener, like node: https://github.com/nodejs/node/blob/v26.3.0/src/node_http2.cc#L1757 + if (self[kTimeout] && self.listenerCount("remoteSettings") > 0) self[kTimeout].refresh(); self.emit("remoteSettings", settings); }, ping(self: ServerHttp2Session, payload: Buffer, isACK: boolean) { if (!self) return; if (!isACK) { + // Only with a listener, like node: https://github.com/nodejs/node/blob/v26.3.0/src/node_http2.cc#L1721 + if (self[kTimeout] && self.listenerCount("ping") > 0) self[kTimeout].refresh(); // node emits 'ping' only for pings initiated by the peer, not for ACKs of our own. self.emit("ping", payload); } @@ -4344,8 +4419,6 @@ class ServerHttp2Session extends Http2Session { }, }; #onRead(data: Buffer) { - // Inbound traffic refreshes the session's idle timer (node's [kUpdateTimer]). - if (this[kTimeout]) this[kTimeout].refresh(); this.#parser?.read(data); } #onClose() { @@ -4391,13 +4464,6 @@ class ServerHttp2Session extends Http2Session { } this.destroy(error); } - #onTimeout() { - const parser = this.#parser; - if (parser) { - parser.forEachStream(emitTimeout); - } - this.emit("timeout"); - } #onDrain() { const parser = this.#parser; if (parser) { @@ -4528,7 +4594,6 @@ class ServerHttp2Session extends Http2Session { }); socket.on("close", this.#onClose.bind(this)); socket.on("error", this.#onError.bind(this)); - socket.on("timeout", this.#onTimeout.bind(this)); initHttp2SessionPerf(this, "server"); socket.on("data", this.#onRead.bind(this)); socket.on("drain", this.#onDrain.bind(this)); @@ -4618,7 +4683,7 @@ class ServerHttp2Session extends Http2Session { setTimeout(msecs, callback) { // node's setStreamTimeout: the session owns its own unref'd idle timer under kTimeout // (refreshed on inbound frames and writes), it does not arm the socket's timeout. - return setSessionTimeout.$call(this, msecs, callback); + return setSessionTimeout.$call(this, msecs, callback, this.#connected); } ping(payload, callback) { @@ -4658,6 +4723,8 @@ class ServerHttp2Session extends Http2Session { } validateInteger(code, "code", 0, kMaxUint32); validateNumber(lastStreamID, "lastStreamID"); + // After connect, like node: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L917 + if (this[kTimeout] && this.#connected) this[kTimeout].refresh(); return this.#parser?.goaway(code, lastStreamID, opaqueData); } @@ -4696,6 +4763,8 @@ class ServerHttp2Session extends Http2Session { return; } this.#pendingSettingsAck = true; + // After connect, like node: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L902 + if (this[kTimeout] && this.#connected) this[kTimeout].refresh(); this.#parser?.settings(settings); // The frame is queued on the native session; flush it now (as close() does for its // GOAWAY) instead of waiting for the next unrelated write. Node schedules a session @@ -4843,21 +4912,14 @@ class ServerHttp2Session extends Http2Session { function emitTimeout(session: ClientHttp2Session) { session.emit("timeout"); } -// Outbound-progress snapshot taken the last time the session's idle timer expired while writes -// were still pending (see sessionTimerExpired / node's chunksSentSinceLastWrite). -const kTimeoutBytesSnapshot = Symbol("timeoutBytesSnapshot"); -const kTimeoutWrittenSnapshot = Symbol("timeoutWrittenSnapshot"); -let sessionHasPendingWrite = false; -function checkStreamWritePending(stream: Http2Stream) { - if (stream.writableLength > 0) sessionHasPendingWrite = true; -} // node setStreamTimeout (lib/internal/stream_base_commons.js): the session owns an unref'd idle // timer stored under kTimeout; activity (inbound frames, writes) refreshes it. -function setSessionTimeout(this: Http2Session, msecs, callback) { +function setSessionTimeout(this: Http2Session, msecs, callback, connected: boolean) { if (this.destroyed) return this; this.timeout = msecs; msecs = getTimerDuration(msecs, "msecs"); - clearTimeout(this[kTimeout]); + const previous = this[kTimeout]; + clearTimeout(previous); this[kTimeout] = null; if (msecs === 0) { if (callback !== undefined) { @@ -4865,12 +4927,11 @@ function setSessionTimeout(this: Http2Session, msecs, callback) { this.removeListener("timeout", callback); } } else { - // Snapshot the monotonic written counter at arm time so the first expiry - // only refreshes if bytes actually went out during the period (see - // sessionTimerExpired for why the native handle's counter, not the JS - // getter's drain-driven mirror). - this[kTimeoutWrittenSnapshot] = - this[bunHTTP2Socket]?._handle?.bytesWritten ?? this[bunHTTP2Socket]?.bytesWritten ?? 0; + if (!previous) { + // A write from before this call is not tracked: bytes that a connected parser holds count as that write. + const parser = this[bunHTTP2Native]; + this[kTimeoutFramesSent] = connected && parser && parser.bufferSize() > 0 ? 0 : -1; + } this[kTimeout] = setTimeout(sessionTimerExpired, msecs, this).unref(); if (callback !== undefined) { validateFunction(callback, "callback"); @@ -4894,41 +4955,37 @@ function destroySessionSocketDelayedNT(socket, error) { function endThenDestroySessionSocket(socket, error) { socket.end(() => setImmediate(destroySessionSocketDelayedNT, socket, error)); } -// node callTimeout (lib/internal/http2/core.js): when the timer expires while writes are still in -// flight and bytes have reached the wire since the previous expiry, the session is not idle — -// refresh instead of emitting 'timeout' (a large flow-controlled response trickling out to a slow -// peer must not be treated as an idle connection). +function checkFileResponse(this: { inFlight: boolean }, stream: Http2Stream) { + if (stream[kFileResponseFinal] !== undefined) this.inFlight = true; +} +// node's kState.writeQueueSize > 0. `tracked`: a stream write went to the parser and no expiry saw it complete. +function sessionHasWriteInFlight(session: Http2Session, parser, tracked: boolean) { + // The parser holds frames that wait for window or for the socket, or a JS transport holds their bytes. + if (tracked) { + if (parser.bufferSize() > 0) return true; + const socket = session[bunHTTP2Socket]; + if (socket && socket.writableLength > 0) return true; + } + // Like node, for the life of the stream: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2749-L2752 + const state = { inFlight: false }; + parser.forEachStream(checkFileResponse, state); + return state.inFlight; +} +// node's callTimeout: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L2556-L2578 function sessionTimerExpired(session: Http2Session) { if (session.destroyed) return; const parser = session[bunHTTP2Native]; if (parser) { - // Node compares a monotonic chunks-sent counter, not instantaneous buffer - // levels: a write that filled and fully drained between two expiries is - // still progress, and sampling the (empty) buffer would misread it as an - // idle session. The native handle's bytesWritten is the live cumulative - // counter (advanced by the parser's direct native writes); the JS socket's - // getter only mirrors it on drain events, which the native write path does - // not raise - reading the mirror missed every native write on Windows. - // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js (callTimeout) - const sessionSocket = session[bunHTTP2Socket]; - const bytesWritten = sessionSocket?._handle?.bytesWritten ?? sessionSocket?.bytesWritten ?? 0; - if (bytesWritten !== (session[kTimeoutWrittenSnapshot] ?? 0)) { - session[kTimeoutWrittenSnapshot] = bytesWritten; - session[kTimeout]?.refresh(); - return; - } - sessionHasPendingWrite = false; - parser.forEachStream(checkStreamWritePending); - // Bytes still queued natively (flow-control or socket backpressure). A change since the - // previous expiry means the pending writes are draining (or growing) — not an idle session. - const nativeBuffered = parser.bufferSize(); - const previousBuffered = session[kTimeoutBytesSnapshot] ?? 0; - if (sessionHasPendingWrite || nativeBuffered > 0 || previousBuffered > 0) { - if (nativeBuffered !== previousBuffered) { - session[kTimeoutBytesSnapshot] = nativeBuffered; + const snapshot = session[kTimeoutFramesSent]; + if (sessionHasWriteInFlight(session, parser, snapshot >= 0)) { + const framesSent = parser.getFrameCounters().framesSent; + if (framesSent !== snapshot) { + session[kTimeoutFramesSent] = framesSent; session[kTimeout]?.refresh(); return; } + } else { + session[kTimeoutFramesSent] = -1; } parser.forEachStream(emitTimeout); } @@ -5033,6 +5090,8 @@ class ClientHttp2Session extends Http2Session { flags: number, ) { if (!self) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L363 + if (self[kTimeout]) self[kTimeout].refresh(); if (self.#reservedStreamsCount >= self.#maxReservedRemoteStreams) { // Too many reserved (pushed) streams: refuse this one (node cancels it instead of // surfacing it). @@ -5070,7 +5129,10 @@ class ClientHttp2Session extends Http2Session { }, frameError: withStreamFrame( (self: ClientHttp2Session, stream: ClientHttp2Stream, frameType: number, errorCode: number) => { - if (!self || typeof stream !== "object") return; + if (!self) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L670 + if (self[kTimeout]) self[kTimeout].refresh(); + if (typeof stream !== "object") return; // Emit the frameError event with the frame type and error code process.nextTick(emitFrameErrorEventNT, stream, frameType, errorCode); }, @@ -5103,9 +5165,16 @@ class ClientHttp2Session extends Http2Session { } if (state == 6 || state == 7) { if (stream.readable) { + // True when a HEADERS frame with END_STREAM already ended the readable. + const alreadyEnded = stream._readableState.ended; + // END_STREAM on a DATA frame: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L172 + if (self[kTimeout] && !alreadyEnded) self[kTimeout].refresh(); // Push a null so the stream can end whenever the client consumes // it completely. endInboundHalf(stream); + if (alreadyEnded && stream.readableFlowing === null && stream.readableLength === 0) { + stream[bunHTTP2StreamStatus] |= StreamState.EndedUnread; + } stream.read(0); } } @@ -5150,6 +5219,8 @@ class ClientHttp2Session extends Http2Session { }), streamData: withStreamFrame((self: ClientHttp2Session, stream: ClientHttp2Stream, data: Buffer) => { if (!self || typeof stream !== "object" || !data) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/stream_base_commons.js#L172 + if (self[kTimeout] && !stream.rstCode && !stream.destroyed) self[kTimeout].refresh(); pushToStream(stream, data); }), streamHeaders: withStreamFrame( @@ -5160,6 +5231,8 @@ class ClientHttp2Session extends Http2Session { flags: number, ) => { if (!self || typeof stream !== "object" || stream.rstCode) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L363 + if (self[kTimeout] && !stream.destroyed) self[kTimeout].refresh(); let rawheaders = headersTuple[0]; let headers = headersTuple[1]; if (self.#strictFieldWhitespaceValidation) { @@ -5245,6 +5318,8 @@ class ClientHttp2Session extends Http2Session { remoteSettings(self: ClientHttp2Session, settings: Settings) { if (!self) return; self.#remoteSettings = settings; + // Only with a listener, like node: https://github.com/nodejs/node/blob/v26.3.0/src/node_http2.cc#L1757 + if (self[kTimeout] && self.listenerCount("remoteSettings") > 0) self[kTimeout].refresh(); self.emit("remoteSettings", settings); // The peer may have raised maxConcurrentStreams: queued requests might fit now. self.#flushPendingRequests(); @@ -5252,6 +5327,8 @@ class ClientHttp2Session extends Http2Session { ping(self: ClientHttp2Session, payload: Buffer, isACK: boolean) { if (!self) return; if (!isACK) { + // Only with a listener, like node: https://github.com/nodejs/node/blob/v26.3.0/src/node_http2.cc#L1721 + if (self[kTimeout] && self.listenerCount("ping") > 0) self[kTimeout].refresh(); // node emits 'ping' only for pings initiated by the peer, not for ACKs of our own. self.emit("ping", payload); } @@ -5359,11 +5436,15 @@ class ClientHttp2Session extends Http2Session { }, altsvc(self: ClientHttp2Session, origin: string, value: string, streamId: number) { if (!self) return; + // Only with a listener, like node: https://github.com/nodejs/node/blob/v26.3.0/src/node_http2.cc#L1651 + if (self[kTimeout] && self.listenerCount("altsvc") > 0) self[kTimeout].refresh(); // node.js emits value, origin, streamId self.emit("altsvc", value, origin, streamId); }, origin(self: ClientHttp2Session, origin: string | Array | undefined) { if (!self) return; + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L722 + if (self[kTimeout]) self[kTimeout].refresh(); const socket = self[bunHTTP2Socket]; if (self.encrypted && socket) { const originSet = initOriginSet(self, socket); @@ -5390,8 +5471,6 @@ class ClientHttp2Session extends Http2Session { }; #onRead(data: Buffer) { - // Inbound traffic refreshes the session's idle timer (node's [kUpdateTimer]). - if (this[kTimeout]) this[kTimeout].refresh(); this.#parser?.read(data); } @@ -5484,13 +5563,6 @@ class ClientHttp2Session extends Http2Session { } this.destroy(error); } - #onTimeout() { - const parser = this.#parser; - if (parser) { - parser.forEachStream(emitTimeout); - } - this.emit("timeout"); - } #onDrain() { const parser = this.#parser; if (parser) { @@ -5553,7 +5625,7 @@ class ClientHttp2Session extends Http2Session { setTimeout(msecs, callback) { // node's setStreamTimeout: the session owns its own unref'd idle timer under kTimeout // (refreshed on inbound frames and writes), it does not arm the socket's timeout. - return setSessionTimeout.$call(this, msecs, callback); + return setSessionTimeout.$call(this, msecs, callback, this.#connected); } ping(payload, callback) { if (this.destroyed) throw $ERR_HTTP2_INVALID_SESSION(); @@ -5586,6 +5658,8 @@ class ClientHttp2Session extends Http2Session { } goaway(errorCode = constants.NGHTTP2_NO_ERROR, lastStreamId = 0, opaqueData) { if (this.destroyed) throw $ERR_HTTP2_INVALID_SESSION(); + // After connect, like node: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L917 + if (this[kTimeout] && this.#connected) this[kTimeout].refresh(); return this.#parser?.goaway(errorCode, lastStreamId, opaqueData); } @@ -5624,6 +5698,8 @@ class ClientHttp2Session extends Http2Session { return; } this.#pendingSettingsAck = true; + // After connect, like node: https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L902 + if (this[kTimeout] && this.#connected) this[kTimeout].refresh(); this.#parser?.settings(settings); // The frame is queued on the native session; flush it now (as close() does for its // GOAWAY) instead of waiting for the next unrelated write. Node schedules a session @@ -5712,6 +5788,8 @@ class ClientHttp2Session extends Http2Session { return; } try { + // node submits SETTINGS when a connecting socket connects. For an open socket, inside connect(), with no timer yet. + if (!connectOnNextTick && this[kTimeout]) this[kTimeout].refresh(); this.#onConnect(arguments); listener?.$call(this, this); } catch (e) { @@ -5775,7 +5853,6 @@ class ClientHttp2Session extends Http2Session { socket.on("drain", this.#onDrain.bind(this)); socket.on("close", this.#onClose.bind(this)); socket.on("error", this.#onError.bind(this)); - socket.on("timeout", this.#onTimeout.bind(this)); initHttp2SessionPerf(this, "client"); if (connectOnNextTick) { // Queued only now that the session is fully built: the parser's construction @@ -5954,6 +6031,8 @@ class ClientHttp2Session extends Http2Session { // throws before that point must not decrement. let connectionsCounted = false; try { + // https://github.com/nodejs/node/blob/v26.3.0/lib/internal/http2/core.js#L1825 + if (this[kTimeout]) this[kTimeout].refresh(); // node validates arguments synchronously and only defers session-state failures // (destroyed/closed/GOAWAY) to the returned stream, so bad options throw even on // a destroyed session (lib/internal/http2/core.js request()). diff --git a/src/runtime/api/bun/h2_frame_parser.rs b/src/runtime/api/bun/h2_frame_parser.rs index 6fdc51e902ef..eabc8488521b 100644 --- a/src/runtime/api/bun/h2_frame_parser.rs +++ b/src/runtime/api/bun/h2_frame_parser.rs @@ -1930,7 +1930,7 @@ impl Stream { .set(client.queued_data_size.get() - len as u64); if !FINALIZING { if let Some(callback_value) = frame.callback.get() { - client.dispatch_write_callback(callback_value); + client.dispatch_dropped_write_callback(callback_value); } } drop(frame); @@ -2428,6 +2428,15 @@ impl H2FrameParser { let _ = self.handlers.get().call_write_callback(callback, &[]); } + /// For a dropped or refused write: `null` is "no error" to a Writable, and "not sent" to http2.ts. + pub(crate) fn dispatch_dropped_write_callback(&self, callback: JSValue) { + let _dispatch = self.enter_dispatch(); + let _ = self + .handlers + .get() + .call_write_callback(callback, &[JSValue::NULL]); + } + pub(crate) fn dispatch_with_extra( &self, event: JSH2FrameParser::Gc, @@ -5950,7 +5959,7 @@ impl H2FrameParser { // is borrowed. let mut stream = this.enter_stream_dispatch(stream_ptr); if !stream.can_send_data() { - this.dispatch_write_callback(callback_arg); + this.dispatch_dropped_write_callback(callback_arg); return Ok(JSValue::FALSE); } diff --git a/test/js/node/http2/http2-session-timer-refresh.fixture.js b/test/js/node/http2/http2-session-timer-refresh.fixture.js new file mode 100644 index 000000000000..bc9234fbe448 --- /dev/null +++ b/test/js/node/http2/http2-session-timer-refresh.fixture.js @@ -0,0 +1,318 @@ +// Counts how often each step of an HTTP/2 exchange refreshes the idle timer of the client session +// and the idle timer of the server session. Run as a script, this file prints the counts as one +// JSON line. Node.js and Bun must give the same counts. +const http2 = require("node:http2"); +const fs = require("node:fs"); +const { once } = require("node:events"); + +// The Timeout of session.setTimeout(). +function idleTimer(session) { + if (typeof Bun !== "undefined") return session[Symbol.for("::buntimeout::")]; + const symbols = Object.getOwnPropertySymbols(session); + return session[symbols.find(symbol => symbol.description === "kTimeout" || symbol.description === "timeout")]; +} + +// Returns a function that gives the refresh() calls on the idle timer since its last call. +function refreshCounter(session) { + const timer = idleTimer(session); + const refresh = timer.refresh; + let count = 0; + timer.refresh = function () { + count++; + return refresh.call(this); + }; + return () => { + const n = count; + count = 0; + return n; + }; +} + +// `server` is an HTTP/2 server without a session. `connect()` starts a client session to it. +async function countTimerRefreshes(server, connect) { + // The timers must not expire by themselves. + const never = 2 ** 30; + const seen = {}; + const fd = fs.openSync(__filename, "r"); + let client; + let serverSession; + let clientRefreshes; + let serverRefreshes; + const record = step => { + seen[step] = { client: clientRefreshes(), server: serverRefreshes() }; + }; + const nextStream = () => once(server, "stream").then(([stream]) => stream); + + // Each step starts to wait for its last event before it acts: with a transport that delivers + // in the same tick, the event can come before the call returns. + try { + const sessionOpened = new Promise(resolve => { + server.once("session", session => { + session.setTimeout(never); + serverRefreshes = refreshCounter(session); + serverSession = session; + resolve(); + }); + }); + client = connect(); + client.setTimeout(never); + clientRefreshes = refreshCounter(client); + await Promise.all([once(client, "connect"), sessionOpened]); + record("connect"); + + // A request with a body, informational headers, a push, a response with a body and trailers. + let pending = nextStream(); + const req = client.request({ ":method": "POST", ":path": "/" }); + const stream = await pending; + record("request()"); + + let arrived = once(stream, "data"); + await new Promise(resolve => req.write("ping", resolve)); + await arrived; + record("write() on the client"); + + arrived = once(req, "headers"); + stream.additionalHeaders({ ":status": 103 }); + await arrived; + record("additionalHeaders()"); + + // ORIGIN refreshes the receiver. ALTSVC refreshes it only while it listens for the event. + // The informational headers that follow a frame tell when the client has read that frame. + arrived = once(req, "headers"); + serverSession.origin("https://example.org"); + stream.additionalHeaders({ ":status": 103 }); + await arrived; + record("origin() and additionalHeaders()"); + arrived = once(req, "headers"); + serverSession.altsvc('h2=":8000"', "https://example.org:8000"); + stream.additionalHeaders({ ":status": 103 }); + await arrived; + record("altsvc() and additionalHeaders(), no 'altsvc' listener"); + arrived = once(client, "altsvc"); + serverSession.altsvc('h2=":8000"', "https://example.org:8000"); + await arrived; + record("altsvc(), 'altsvc' listener"); + + const pushReceived = new Promise(resolve => { + client.once("stream", pushed => { + pushed.once("push", resolve); + pushed.resume(); + }); + }); + await new Promise((resolve, reject) => { + stream.pushStream({ ":path": "/pushed" }, (err, pushed) => { + if (err) return reject(err); + pushed.respond({ ":status": 200 }, { endStream: true }); + resolve(); + }); + }); + await pushReceived; + record("pushStream() and respond() on the pushed stream"); + + arrived = once(req, "response"); + stream.respond({ ":status": 200 }, { waitForTrailers: true }); + await arrived; + record("respond()"); + + arrived = once(req, "data"); + await new Promise(resolve => stream.write("pong", resolve)); + await arrived; + record("write() on the server"); + + arrived = once(req, "trailers"); + stream.once("wantTrailers", () => stream.sendTrailers({ "x-trailer": "1" })); + stream.end(); + await arrived; + record("end() and sendTrailers()"); + + arrived = once(stream, "end"); + stream.resume(); + req.end(); + await arrived; + record("end() on the client"); + + // A response that ends with data. The server has not read the request, and it responds one + // turn of the event loop later. + pending = nextStream(); + const unreadRequest = client.request({ ":path": "/" }); + arrived = once(unreadRequest, "end"); + unreadRequest.resume(); + const unread = await pending; + await new Promise(resolve => setImmediate(resolve)); + unread.respond({ ":status": 200 }); + await new Promise(resolve => unread.end("pong", resolve)); + await arrived; + record("end() with data, request not read"); + + // The same, after the server has read the request to its end. Node.js has destroyed that + // stream when the write completes, and a destroyed stream does not refresh the timer. + pending = nextStream(); + const readRequest = client.request({ ":method": "POST", ":path": "/" }); + arrived = once(readRequest, "end"); + readRequest.resume(); + const sent = new Promise(resolve => readRequest.end("ping", resolve)); + const read = await pending; + read.resume(); + await once(read, "end"); + await sent; + read.respond({ ":status": 200 }); + await new Promise(resolve => read.end("pong", resolve)); + await arrived; + record("end() with data, request read to its end"); + + // The same for a request without a body, when a consumer is attached before the response. + pending = new Promise(resolve => { + server.once("stream", consumed => { + consumed.resume(); + consumed.respond({ ":status": 200 }); + consumed.end("pong", resolve); + }); + }); + const consumedRequest = client.request({ ":path": "/" }); + arrived = once(consumedRequest, "end"); + consumedRequest.resume(); + await pending; + await arrived; + record("resume(), respond() and end() with data in the 'stream' event"); + + // END_STREAM on a DATA frame with no data ends the request, which the server does not read. + // Node.js emits 'end' for it with no consumer, so that stream is destroyed too. + pending = nextStream(); + const emptyRequest = client.request({ ":method": "POST", ":path": "/" }); + arrived = once(emptyRequest, "end"); + emptyRequest.resume(); + const emptied = await pending; + let ended = once(emptied, "end"); + emptyRequest.end(); + await ended; + emptied.respond({ ":status": 200 }); + await new Promise(resolve => emptied.end("pong", resolve)); + await arrived; + record("end() with data, request ended by an empty DATA frame and not read"); + + // The same on the client: the response ends that way and the client does not read it. + pending = nextStream(); + const lateRequest = client.request({ ":method": "POST", ":path": "/" }); + ended = once(lateRequest, "end"); + const late = await pending; + arrived = once(late, "end"); + late.resume(); + late.respond({ ":status": 200 }); + late.end(); + await ended; + await new Promise(resolve => lateRequest.end("ping", resolve)); + await arrived; + record("end() with data on the client, response ended by an empty DATA frame and not read"); + + // File responses. The call refreshes, also when the file is not sent. + for (const method of ["respondWithFD", "respondWithFile"]) { + pending = nextStream(); + const fileRequest = client.request({ ":path": "/file" }); + arrived = once(fileRequest, "end"); + fileRequest.resume(); + const fileStream = await pending; + fileStream[method](method === "respondWithFD" ? fd : __filename, {}, { length: 100 }); + await arrived; + record(`${method}()`); + } + pending = nextStream(); + const uncheckedRequest = client.request({ ":path": "/file" }); + arrived = once(uncheckedRequest, "response"); + const unchecked = await pending; + unchecked.respondWithFD( + fd, + {}, + { + statCheck() { + unchecked.respond({ ":status": 304 }, { endStream: true }); + return false; + }, + }, + ); + await arrived; + record("respondWithFD(), statCheck responds and returns false"); + pending = nextStream(); + const missingRequest = client.request({ ":path": "/file" }); + arrived = once(missingRequest, "response"); + const missing = await pending; + missing.respondWithFile( + `${__filename}.missing`, + {}, + { onError: () => missing.respond({ ":status": 404 }, { endStream: true }) }, + ); + await arrived; + record("respondWithFile() of a missing file, onError responds"); + + // A write that waits for flow-control window, on a stream that the server then resets. The + // write is dropped, not completed. + ended = once(client, "localSettings"); + client.settings({ initialWindowSize: 0 }); + await ended; + pending = nextStream(); + const blockedRequest = client.request({ ":path": "/" }); + blockedRequest.on("error", () => {}); + arrived = once(blockedRequest, "response"); + const blocked = await pending; + blocked.on("error", () => {}); + blocked.respond({ ":status": 200 }); + blocked.write("pong"); + await arrived; + arrived = once(blockedRequest, "close"); + blocked.close(http2.constants.NGHTTP2_CANCEL); + await arrived; + ended = once(client, "localSettings"); + client.settings({ initialWindowSize: 65535 }); + await ended; + record("close() with a code, on a stream with a write that waits for window"); + + // SETTINGS and PING refresh the receiver only while it listens for the event. + for (const [sender, from, to] of [ + ["server", serverSession, client], + ["client", client, serverSession], + ]) { + const settingsAcked = () => { + const acked = once(from, "localSettings"); + from.settings({ maxConcurrentStreams: 50 }); + return acked; + }; + const pingAcked = () => new Promise(resolve => from.ping(resolve)); + await settingsAcked(); + record(`settings() on the ${sender}, no 'remoteSettings' listener`); + to.on("remoteSettings", () => {}); + await settingsAcked(); + record(`settings() on the ${sender}, 'remoteSettings' listener`); + await pingAcked(); + record(`ping() on the ${sender}, no 'ping' listener`); + to.on("ping", () => {}); + await pingAcked(); + record(`ping() on the ${sender}, 'ping' listener`); + } + + // close() sends a GOAWAY. The server answers that frame with its own close(). + arrived = once(serverSession, "goaway"); + client.close(); + await arrived; + record("close()"); + return seen; + } finally { + fs.closeSync(fd); + client?.destroy(); + serverSession?.destroy(); + } +} + +module.exports = { countTimerRefreshes }; + +if (require.main === module) { + const server = http2.createServer(); + server.listen(0, "127.0.0.1", async () => { + try { + const connect = () => http2.connect(`http://127.0.0.1:${server.address().port}`); + console.log(JSON.stringify(await countTimerRefreshes(server, connect))); + process.exit(0); + } catch (err) { + console.error(err); + process.exit(1); + } + }); +} diff --git a/test/js/node/http2/node-http2-session-timeout.test.ts b/test/js/node/http2/node-http2-session-timeout.test.ts new file mode 100644 index 000000000000..d68a961c8c35 --- /dev/null +++ b/test/js/node/http2/node-http2-session-timeout.test.ts @@ -0,0 +1,298 @@ +import { describe, expect, it } from "bun:test"; +import { bunEnv, nodeExe, tls as tlsCert } from "harness"; +import { once } from "node:events"; +import http2 from "node:http2"; +import net, { type AddressInfo } from "node:net"; +import path from "node:path"; +import { duplexPair } from "node:stream"; +import { countTimerRefreshes } from "./http2-session-timer-refresh.fixture.js"; + +describe.concurrent("http2 session idle timer", () => { + // A session keeps the timer of setTimeout() under this symbol, like net.Socket does. + const kTimeout = Symbol.for("::buntimeout::"); + const period = 50; + // For a timer that must not expire by itself. + const never = 2 ** 30; + + // Counts the runs of the expiry handler of the idle timer of `session`. + function watchIdleTimer(session: http2.Http2Session) { + const timer = session[kTimeout]; + const counts = { expiries: 0 }; + const { _onTimeout } = timer; + timer._onTimeout = function (...args) { + counts.expiries++; + return _onTimeout.apply(this, args); + }; + return counts; + } + + // Rejects a promise that waits for the 'timeout' event of `session` when the session fails or closes first. + function rejectOnFailure(session: http2.Http2Session, reject: (reason: Error) => void) { + session.on("error", reject); + session.on("close", () => reject(new Error("the session closed before its 'timeout' event"))); + } + + // Arms the idle timer of `session`. Resolves with the count of expiries at its 'timeout' event. + function expiriesAtTimeout(session: http2.Http2Session) { + const { promise, resolve, reject } = Promise.withResolvers(); + session.setTimeout(period, () => resolve(counts.expiries)); + const counts = watchIdleTimer(session); + rejectOnFailure(session, reject); + return promise; + } + + // The transports of a session: a native TCP socket, a native TLS socket, a JS stream. + type Transport = "tcp" | "tls" | "duplexPair"; + const transports: Transport[] = ["tcp", "tls", "duplexPair"]; + const createServer = (transport: Transport): http2.Http2Server => + transport === "tls" ? http2.createSecureServer(tlsCert) : http2.createServer(); + + // Resolves with a function that starts a client session to `server`. + async function dialer(server: net.Server, transport: Transport = "tcp") { + if (transport === "duplexPair") { + return () => { + const [clientSide, serverSide] = duplexPair(); + server.emit("connection", serverSide); + return http2.connect("http://localhost", { createConnection: () => clientSide }); + }; + } + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const { port } = server.address() as AddressInfo; + return transport === "tls" + ? () => http2.connect(`https://127.0.0.1:${port}`, { ca: tlsCert.cert }) + : () => http2.connect(`http://127.0.0.1:${port}`); + } + + // Resolves with a client session that has connected to `server`. + async function connectTo(server: net.Server, transport?: Transport) { + const client = (await dialer(server, transport))(); + client.on("error", () => {}); + await once(client, "connect"); + return client; + } + + for (const scheme of ["https", "http"]) { + it(`emits 'timeout' at the first expiry when the peer never answers (${scheme})`, async () => { + // The peer accepts the TCP connection. It reads nothing and writes nothing. + const sockets: net.Socket[] = []; + const peer = net.createServer(socket => { + sockets.push(socket); + socket.on("error", () => {}); + }); + await new Promise(resolve => peer.listen(0, "127.0.0.1", resolve)); + const { port } = peer.address() as AddressInfo; + const client = http2.connect(`${scheme}://127.0.0.1:${port}`, { rejectUnauthorized: false }); + try { + expect(await expiriesAtTimeout(client)).toBe(1); + } finally { + client.destroy(); + for (const socket of sockets) socket.destroy(); + peer.close(); + } + }); + } + + for (const transport of transports) { + it(`emits 'timeout' at the first expiry of an idle client session (${transport})`, async () => { + const server = createServer(transport); + const client = await connectTo(server, transport); + try { + expect(await expiriesAtTimeout(client)).toBe(1); + } finally { + client.destroy(); + server.close(); + } + }); + + it(`emits 'timeout' at the first expiry of an idle server session (${transport})`, async () => { + const server = createServer(transport); + const { promise, resolve, reject } = Promise.withResolvers(); + let counts: { expiries: number }; + server.on("session", session => { + counts = watchIdleTimer(session); + rejectOnFailure(session, reject); + }); + server.setTimeout(period, () => resolve(counts.expiries)); + const client = await connectTo(server, transport); + try { + expect(await promise).toBe(1); + } finally { + client.destroy(); + server.close(); + } + }); + } + + it("destroys an idle server session at the first expiry when the server has no 'timeout' listener", async () => { + const server = http2.createServer(); + const { promise, resolve, reject } = Promise.withResolvers(); + server.setTimeout(period); + server.on("session", session => { + const counts = watchIdleTimer(session); + session.on("error", reject); + session.on("close", () => resolve(counts.expiries)); + }); + const client = await connectTo(server); + try { + expect(await promise).toBe(1); + } finally { + client.destroy(); + server.close(); + } + }); + + it("holds an expiry back only while a stream write is in flight and frames leave", async () => { + const server = http2.createServer(); + const { promise: written, resolve: onWritten } = Promise.withResolvers(); + server.on("stream", (stream: http2.ServerHttp2Stream) => { + const session = stream.session!; + stream.on("error", () => {}); + session.setTimeout(never); + stream.respond({ ":status": 200 }); + // The client reads nothing at first, so flow control stops this write after one window. + stream.write(Buffer.alloc(1024 * 1024, "a")); + onWritten(session); + }); + const client = await connectTo(server); + try { + const req = client.request({ ":path": "/" }); + req.on("error", () => {}); + const session = await written; + let timeouts = 0; + session.on("timeout", () => timeouts++); + // Runs the expiry handler of the timer. The test then does not depend on the transfer speed. + const expire = () => { + session[kTimeout]._onTimeout(session); + return timeouts; + }; + + // Frames left since the write went out. + expect(expire()).toBe(0); + // No frame left since the last expiry. + expect(expire()).toBe(1); + + // More than one window of data at the client means that frames left after that expiry. + const window = client.localSettings.initialWindowSize!; + for (let received = 0; received <= window;) { + const chunk = req.read(); + if (chunk === null) await once(req, "readable"); + else received += chunk.length; + } + expect(expire()).toBe(1); + } finally { + client.destroy(); + server.close(); + } + }); + + // http2-session-timer-refresh.fixture.js takes the same steps in Node.js and in Bun. For each + // step it counts the refresh() calls on the idle timer of the client session and of the server + // session. These are the counts of Node.js v18.20.8 to v26.10.0 on a TCP or TLS socket. + const refreshesInNode = { + "connect": { client: 1, server: 0 }, + "request()": { client: 1, server: 1 }, + "write() on the client": { client: 2, server: 1 }, + "additionalHeaders()": { client: 1, server: 1 }, + "origin() and additionalHeaders()": { client: 2, server: 1 }, + "altsvc() and additionalHeaders(), no 'altsvc' listener": { client: 1, server: 1 }, + "altsvc(), 'altsvc' listener": { client: 1, server: 0 }, + "pushStream() and respond() on the pushed stream": { client: 2, server: 2 }, + "respond()": { client: 1, server: 1 }, + "write() on the server": { client: 1, server: 2 }, + "end() and sendTrailers()": { client: 1, server: 1 }, + "end() on the client": { client: 0, server: 1 }, + "end() with data, request not read": { client: 4, server: 4 }, + "end() with data, request read to its end": { client: 6, server: 5 }, + "resume(), respond() and end() with data in the 'stream' event": { client: 4, server: 3 }, + "end() with data, request ended by an empty DATA frame and not read": { client: 4, server: 4 }, + "end() with data on the client, response ended by an empty DATA frame and not read": { client: 4, server: 4 }, + "respondWithFD()": { client: 4, server: 2 }, + "respondWithFile()": { client: 4, server: 2 }, + "respondWithFD(), statCheck responds and returns false": { client: 2, server: 3 }, + "respondWithFile() of a missing file, onError responds": { client: 2, server: 3 }, + "close() with a code, on a stream with a write that waits for window": { client: 4, server: 3 }, + "settings() on the server, no 'remoteSettings' listener": { client: 0, server: 1 }, + "settings() on the server, 'remoteSettings' listener": { client: 1, server: 1 }, + "ping() on the server, no 'ping' listener": { client: 0, server: 0 }, + "ping() on the server, 'ping' listener": { client: 1, server: 0 }, + "settings() on the client, no 'remoteSettings' listener": { client: 1, server: 0 }, + "settings() on the client, 'remoteSettings' listener": { client: 1, server: 1 }, + "ping() on the client, no 'ping' listener": { client: 0, server: 0 }, + "ping() on the client, 'ping' listener": { client: 0, server: 1 }, + "close()": { client: 1, server: 1 }, + }; + + for (const transport of transports) { + // The fixture takes 31 steps on the one JS thread. Each run has the thread to itself, so + // that it stays far from the default time limit in a debug build. + it.serial(`refreshes the timers as often as node for each call and frame (${transport})`, async () => { + const server = createServer(transport); + try { + const counts = await countTimerRefreshes(server, await dialer(server, transport)); + // node sets up a session on a stream that has connected already inside connect(). The + // client has no timer at that time, so node counts no refresh for the connect. + const connect = transport === "duplexPair" ? { client: 0, server: 0 } : refreshesInNode.connect; + expect(counts).toEqual({ ...refreshesInNode, connect }); + } finally { + server.close(); + } + }); + } + + const node = nodeExe(); + it.skipIf(!node)("node gives the refresh counts that the tests above expect", async () => { + const fixture = path.join(import.meta.dir, "http2-session-timer-refresh.fixture.js"); + await using proc = Bun.spawn({ cmd: [node!, fixture], env: bunEnv, stdout: "pipe", stderr: "pipe" }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual(refreshesInNode); + expect(exitCode).toBe(0); + }); + + it("does not emit 'timeout' for the timer of its socket", async () => { + const server = http2.createServer(); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + const socket = net.connect((server.address() as AddressInfo).port, "127.0.0.1"); + socket.setTimeout(period); + const client = http2.connect("http://127.0.0.1", { createConnection: () => socket }); + client.on("error", () => {}); + try { + let timeouts = 0; + client.on("timeout", () => timeouts++); + await once(socket, "timeout"); + expect(timeouts).toBe(0); + } finally { + client.destroy(); + server.close(); + } + }); + + it("arms a timer that has fired again on a request without a body", async () => { + const server = http2.createServer(); + server.on("stream", (stream: http2.ServerHttp2Stream) => { + stream.respond({ ":status": 204 }, { endStream: true }); + }); + const client = await connectTo(server); + try { + client.setTimeout(period); + await once(client, "timeout"); + + const { promise, resolve } = Promise.withResolvers(); + client.once("timeout", () => resolve("timeout")); + const req = client.request({ ":path": "/" }); + req.resume(); + await once(req, "close"); + // The response was the last activity. The idle timer, if the request armed it, is due + // before this timer, which starts later and runs longer. + const reference = setTimeout(resolve, period * 4, "the request did not arm the idle timer"); + try { + expect(await promise).toBe("timeout"); + } finally { + clearTimeout(reference); + } + } finally { + client.destroy(); + server.close(); + } + }); +});