From dffda21145f5f5cbcb2d3962538d61c90eb236e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Sun, 13 Sep 2026 18:21:09 +0200 Subject: [PATCH 1/2] fix(grpc-js): avoid redundant end() calls on completed HTTP/2 streams When a call completes, destroyHttp2Stream() invokes http2Stream.end() if the server ended the call. However, for unary and server-streaming calls (as well as client streams where writing already finished), halfClose() has already ended the stream. In Node.js, calling .end() on an already finished stream without a callback causes Node's stream internals to construct an ERR_STREAM_ALREADY_FINISHED Error with a full V8 stack trace, only to immediately discard it. Under high throughput, this introduces significant CPU overhead and garbage collection pressure. This change checks writableEnded before calling http2Stream.end() in destroyHttp2Stream() and halfClose(), ensuring .end() is only called when the client side of the stream has not yet been closed. --- packages/grpc-js/src/subchannel-call.ts | 21 ++- packages/grpc-js/test/test-client.ts | 175 ++++++++++++++++++++++++ 2 files changed, 195 insertions(+), 1 deletion(-) diff --git a/packages/grpc-js/src/subchannel-call.ts b/packages/grpc-js/src/subchannel-call.ts index 207b781c7..ae1bdb83c 100644 --- a/packages/grpc-js/src/subchannel-call.ts +++ b/packages/grpc-js/src/subchannel-call.ts @@ -521,9 +521,20 @@ export class Http2SubchannelCall implements SubchannelCall { } /* If the server ended the call, sending an RST_STREAM is redundant, so we * just half close on the client side instead to finish closing the stream. + * + * Only call end() if writableEnded is false. For unary and server-streaming + * calls (and client streams where the client already finished sending), + * halfClose() has already called http2Stream.end(). Calling end() again on + * an already finished Node stream causes Node core stream internals to + * construct an ERR_STREAM_ALREADY_FINISHED Error (which synchronously captures + * a full native V8 stack trace) and immediately discard it because no callback + * is passed. On high-throughput workloads, this causes unnecessary CPU + * overhead and garbage collection pressure. */ if (this.serverEndedCall) { - this.http2Stream.end(); + if (!this.http2Stream.writableEnded) { + this.http2Stream.end(); + } } else { /* If the call has ended with an OK status, communicate that when closing * the stream, partly to avoid a situation in which we detect an error @@ -616,6 +627,14 @@ export class Http2SubchannelCall implements SubchannelCall { halfClose() { this.trace('end() called'); + /* Calling end() on a stream that is already ended or destroyed causes Node + * core stream internals to construct ERR_STREAM_ALREADY_FINISHED or + * ERR_STREAM_DESTROYED Error instances with synchronous native V8 stack traces, + * which are immediately discarded when no callback is passed. + */ + if (this.http2Stream.destroyed || this.http2Stream.writableEnded) { + return; + } this.trace('calling end() on HTTP/2 stream'); this.http2Stream.end(); } diff --git a/packages/grpc-js/test/test-client.ts b/packages/grpc-js/test/test-client.ts index bbb3f063f..082f511e7 100644 --- a/packages/grpc-js/test/test-client.ts +++ b/packages/grpc-js/test/test-client.ts @@ -16,11 +16,14 @@ */ import * as assert from 'assert'; +import { EventEmitter } from 'events'; +import * as http2 from 'http2'; import * as grpc from '../src'; import { Server, ServerCredentials } from '../src'; import { Client } from '../src'; import { ConnectivityState } from '../src/connectivity-state'; +import { Http2SubchannelCall } from '../src/subchannel-call'; const clientInsecureCreds = grpc.credentials.createInsecure(); const serverInsecureCreds = ServerCredentials.createInsecure(); @@ -63,6 +66,178 @@ describe('Client', () => { }); }); +describe('Client HTTP/2 stream lifecycle', () => { + let originalConnect: typeof http2.connect; + let testServer: Server | null = null; + let testClient: Client | null = null; + + beforeEach(() => { + originalConnect = http2.connect; + }); + + afterEach(done => { + (http2 as any).connect = originalConnect; + testClient?.close(); + testClient = null; + if (testServer) { + testServer.forceShutdown(); + testServer = null; + } + done(); + }); + + it('should not redundantly call http2Stream.end when stream is already ended', done => { + let streamEndCount = 0; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (http2 as any).connect = function ( + this: unknown, + ...connectArguments: any[] + ) { + const session = Reflect.apply(originalConnect, this, connectArguments); + const originalRequest = session.request; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + session.request = function (this: unknown, ...requestArguments: any[]) { + const stream = Reflect.apply(originalRequest, this, requestArguments); + const originalEnd = stream.end; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + stream.end = function (this: unknown, ...endArguments: any[]) { + streamEndCount++; + return Reflect.apply(originalEnd, this, endArguments); + }; + return stream; + }; + return session; + }; + testServer = new Server(); + testServer.bindAsync( + 'localhost:0', + serverInsecureCreds, + (bindError, port) => { + assert.ifError(bindError); + testClient = new Client( + `localhost:${port}`, + clientInsecureCreds + ); + testServer!.start(); + testClient.makeUnaryRequest( + '/service/method', + message => message, + message => message, + Buffer.from([]), + () => { + assert.strictEqual(streamEndCount, 1); + done(); + } + ); + } + ); + }); + + it('should call http2Stream.end when server ends call early on an un-ended client stream', done => { + let streamEndCount = 0; + let writableEndedBeforeEndCall: boolean | undefined; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (http2 as any).connect = function ( + this: unknown, + ...connectArguments: any[] + ) { + const session = Reflect.apply(originalConnect, this, connectArguments); + const originalRequest = session.request; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + session.request = function (this: unknown, ...requestArguments: any[]) { + const stream = Reflect.apply(originalRequest, this, requestArguments); + const originalEnd = stream.end; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + stream.end = function (this: unknown, ...endArguments: any[]) { + streamEndCount++; + writableEndedBeforeEndCall = (this as {writableEnded?: boolean}) + .writableEnded; + return Reflect.apply(originalEnd, this, endArguments); + }; + return stream; + }; + return session; + }; + testServer = new Server(); + testServer.bindAsync( + 'localhost:0', + serverInsecureCreds, + (bindError, port) => { + assert.ifError(bindError); + testClient = new Client( + `localhost:${port}`, + clientInsecureCreds + ); + testServer!.start(); + const clientStream = testClient.makeClientStreamRequest( + '/service/method', + message => message, + message => message, + callError => { + assert(callError); + assert.strictEqual(streamEndCount, 1); + assert.strictEqual(writableEndedBeforeEndCall, false); + done(); + } + ); + // Write data without ending the client stream so that writableEnded remains false + clientStream.write(Buffer.from('hello')); + } + ); + }); + + it('should not call http2Stream.end when halfClose is called on an ended or destroyed stream', () => { + let streamEndCount = 0; + const mockHttp2Stream = Object.assign(new EventEmitter(), { + destroyed: false, + writableEnded: false, + end() { + streamEndCount++; + this.writableEnded = true; + }, + rstCode: 0, + }); + const mockTransport = { + getOptions: () => ({}), + getPeerName: () => 'localhost', + getAuthContext: () => null, + }; + const mockTracker = { + addMessageReceived: () => {}, + addMessageSent: () => {}, + onStreamEnd: () => {}, + }; + const mockListener = { + onReceiveMetadata: () => {}, + onReceiveMessage: () => {}, + onReceiveStatus: () => {}, + }; + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const subchannelCall = new Http2SubchannelCall( + mockHttp2Stream as any, + mockTracker as any, + mockListener as any, + mockTransport as any, + 1 + ); + + // Initial halfClose calls http2Stream.end() + subchannelCall.halfClose(); + assert.strictEqual(streamEndCount, 1); + + // Subsequent halfClose when writableEnded is true is a no-op + subchannelCall.halfClose(); + assert.strictEqual(streamEndCount, 1); + + // halfClose when destroyed is true is a no-op even if writableEnded were false + mockHttp2Stream.destroyed = true; + mockHttp2Stream.writableEnded = false; + subchannelCall.halfClose(); + assert.strictEqual(streamEndCount, 1); + }); +}); + describe('Client without a server', () => { let client: Client; before(() => { From b487d4dbf7bbb1cbe79052e3abb601878f369043 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Tue, 15 Sep 2026 09:34:37 +0200 Subject: [PATCH 2/2] test(grpc-js): fix type errors in client stream lifecycle tests Add explicit type cast for the intercepted HTTP/2 stream and type arguments for makeClientStreamRequest. --- packages/grpc-js/test/test-client.ts | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/packages/grpc-js/test/test-client.ts b/packages/grpc-js/test/test-client.ts index 082f511e7..e19c87bb6 100644 --- a/packages/grpc-js/test/test-client.ts +++ b/packages/grpc-js/test/test-client.ts @@ -97,7 +97,11 @@ describe('Client HTTP/2 stream lifecycle', () => { const originalRequest = session.request; // eslint-disable-next-line @typescript-eslint/no-explicit-any session.request = function (this: unknown, ...requestArguments: any[]) { - const stream = Reflect.apply(originalRequest, this, requestArguments); + const stream = Reflect.apply( + originalRequest, + this, + requestArguments + ) as http2.ClientHttp2Stream; const originalEnd = stream.end; // eslint-disable-next-line @typescript-eslint/no-explicit-any stream.end = function (this: unknown, ...endArguments: any[]) { @@ -114,10 +118,7 @@ describe('Client HTTP/2 stream lifecycle', () => { serverInsecureCreds, (bindError, port) => { assert.ifError(bindError); - testClient = new Client( - `localhost:${port}`, - clientInsecureCreds - ); + testClient = new Client(`localhost:${port}`, clientInsecureCreds); testServer!.start(); testClient.makeUnaryRequest( '/service/method', @@ -145,12 +146,16 @@ describe('Client HTTP/2 stream lifecycle', () => { const originalRequest = session.request; // eslint-disable-next-line @typescript-eslint/no-explicit-any session.request = function (this: unknown, ...requestArguments: any[]) { - const stream = Reflect.apply(originalRequest, this, requestArguments); + const stream = Reflect.apply( + originalRequest, + this, + requestArguments + ) as http2.ClientHttp2Stream; const originalEnd = stream.end; // eslint-disable-next-line @typescript-eslint/no-explicit-any stream.end = function (this: unknown, ...endArguments: any[]) { streamEndCount++; - writableEndedBeforeEndCall = (this as {writableEnded?: boolean}) + writableEndedBeforeEndCall = (this as { writableEnded?: boolean }) .writableEnded; return Reflect.apply(originalEnd, this, endArguments); }; @@ -164,12 +169,9 @@ describe('Client HTTP/2 stream lifecycle', () => { serverInsecureCreds, (bindError, port) => { assert.ifError(bindError); - testClient = new Client( - `localhost:${port}`, - clientInsecureCreds - ); + testClient = new Client(`localhost:${port}`, clientInsecureCreds); testServer!.start(); - const clientStream = testClient.makeClientStreamRequest( + const clientStream = testClient.makeClientStreamRequest( '/service/method', message => message, message => message,