From bcf286e07acc75ef9ee9949554164cc3731201bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Sun, 13 Sep 2026 18:47:02 +0200 Subject: [PATCH] perf(spanner): avoid redundant end() calls on completed HTTP/2 streams In @grpc/grpc-js, destroyHttp2Stream() calls http2Stream.end() when the server finishes a call, even if the stream was already ended by halfClose(). Calling .end() on an already ended or destroyed Node stream without a callback causes Node core to construct an ERR_STREAM_ALREADY_FINISHED or ERR_STREAM_DESTROYED error with a full V8 stack capture and immediately discard it, creating unnecessary CPU overhead on high-throughput workloads. This adds a workaround on stream.Duplex.prototype.end scoped to HTTP/2 streams that returns early when called on an already ended or destroyed stream without data or a callback. The workaround can be disabled with SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND=true. This is a temporary client-side workaround for an upstream issue in @grpc/grpc-js, tracked in https://github.com/grpc/grpc-node/pull/3082. It can be removed once Spanner requires a grpc-js release that includes that fix. --- handwritten/spanner/src/http2-workaround.ts | 162 ++++++++ handwritten/spanner/src/index.ts | 4 + handwritten/spanner/test/http2-workaround.ts | 369 +++++++++++++++++++ 3 files changed, 535 insertions(+) create mode 100644 handwritten/spanner/src/http2-workaround.ts create mode 100644 handwritten/spanner/test/http2-workaround.ts diff --git a/handwritten/spanner/src/http2-workaround.ts b/handwritten/spanner/src/http2-workaround.ts new file mode 100644 index 000000000000..a4b5bb2c805c --- /dev/null +++ b/handwritten/spanner/src/http2-workaround.ts @@ -0,0 +1,162 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +const WORKAROUND_APPLIED_SYMBOL = Symbol.for( + '@google-cloud/spanner.http2SubchannelCallWorkaround', +); + +let isInstallationAttempted = false; +// eslint-disable-next-line @typescript-eslint/ban-types +let savedOriginalDestroyHttp2Stream: Function | undefined; +// eslint-disable-next-line @typescript-eslint/ban-types +let savedOriginalHalfClose: Function | undefined; + +/** + * Installs a targeted workaround for redundant stream `.end()` calls in `@grpc/grpc-js`. + * + * In `@grpc/grpc-js` (inside `Http2SubchannelCall.prototype.destroyHttp2Stream`), + * when a server ends a call by sending trailers, `this.http2Stream.end()` is invoked. + * However, for unary and server-streaming RPCs, the client stream was already + * half-closed when sending the request payload (`halfClose()`), leaving `writableEnded = true`. + * + * In Node.js streams (`node:internal/streams/writable:805`), calling `.end()` on an + * already-finished stream without a callback causes Node to eagerly construct + * `new ERR_STREAM_ALREADY_FINISHED('end')` with a full native V8 stack trace capture, and + * then immediately discard it because no callback was provided. On high-throughput + * workloads, this generates an unused `NodeError` and stack trace on every single RPC, + * consuming ~4% of total process CPU and driving significant GC allocation pressure. + * + * This function targets the exact source of the bug by patching + * `Http2SubchannelCall.prototype.destroyHttp2Stream` and + * `Http2SubchannelCall.prototype.halfClose` in `@grpc/grpc-js` to return early when + * called on an already-ended or destroyed HTTP/2 stream. + * + * Crucially, unlike patching `stream.Duplex.prototype.end`, this targeted approach: + * - Leaves `stream.Duplex.prototype` 100% untouched. + * - Adds ZERO overhead to `net.Socket` (TCP), `tls.TLSSocket`, `PassThrough`, or `Transform` streams. + * - Does not deoptimize V8 inline caches for stream operations. + * + * Tracking upstream fix: https://github.com/grpc/grpc-node/pull/3082 + * Once `@grpc/grpc-js` merges and releases PR #3082, and this package adopts that + * version as a minimum dependency, this workaround file and its invocation in + * `Spanner` constructor can be safely removed. + */ +export function installHttp2StreamEndWorkaround(): void { + if (isInstallationAttempted) { + return; + } + isInstallationAttempted = true; + + if ( + process.env[ + 'SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND' + ]?.toLowerCase() === 'true' + ) { + return; + } + + try { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { + Http2SubchannelCall, + } = require('@grpc/grpc-js/build/src/subchannel-call'); + if (!Http2SubchannelCall?.prototype) { + return; + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + if ((Http2SubchannelCall.prototype as any)[WORKAROUND_APPLIED_SYMBOL]) { + return; + } + + const originalDestroyHttp2Stream = + Http2SubchannelCall.prototype.destroyHttp2Stream; + savedOriginalDestroyHttp2Stream = originalDestroyHttp2Stream; + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Http2SubchannelCall.prototype.destroyHttp2Stream = function (this: any) { + if (this.http2Stream?.destroyed) { + return; + } + if (this.serverEndedCall && this.http2Stream?.writableEnded) { + return; + } + // eslint-disable-next-line prefer-rest-params + return Reflect.apply(originalDestroyHttp2Stream, this, arguments); + }; + + const originalHalfClose = Http2SubchannelCall.prototype.halfClose; + savedOriginalHalfClose = originalHalfClose; + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Http2SubchannelCall.prototype.halfClose = function (this: any) { + if (this.http2Stream?.destroyed || this.http2Stream?.writableEnded) { + return; + } + // eslint-disable-next-line prefer-rest-params + return Reflect.apply(originalHalfClose, this, arguments); + }; + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (Http2SubchannelCall.prototype as any)[WORKAROUND_APPLIED_SYMBOL] = true; + } catch { + // Graceful fallback if @grpc/grpc-js internal structure changes + } +} + +/** + * Resets and restores the original methods (for unit testing). + */ +export function _resetHttp2StreamEndWorkaroundForTest(): void { + try { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { + Http2SubchannelCall, + } = require('@grpc/grpc-js/build/src/subchannel-call'); + if (Http2SubchannelCall?.prototype) { + if (savedOriginalDestroyHttp2Stream) { + Http2SubchannelCall.prototype.destroyHttp2Stream = + savedOriginalDestroyHttp2Stream; + savedOriginalDestroyHttp2Stream = undefined; + } + if (savedOriginalHalfClose) { + Http2SubchannelCall.prototype.halfClose = savedOriginalHalfClose; + savedOriginalHalfClose = undefined; + } + // eslint-disable-next-line @typescript-eslint/no-explicit-any + delete (Http2SubchannelCall.prototype as any)[WORKAROUND_APPLIED_SYMBOL]; + } + } catch { + // Ignore cleanup error + } + isInstallationAttempted = false; +} + +/** + * Checks whether the workaround has been marked as installed (for unit testing). + */ +export function _isHttp2StreamEndWorkaroundInstalled(): boolean { + try { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { + Http2SubchannelCall, + } = require('@grpc/grpc-js/build/src/subchannel-call'); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return !!(Http2SubchannelCall?.prototype as any)?.[ + WORKAROUND_APPLIED_SYMBOL + ]; + } catch { + return false; + } +} diff --git a/handwritten/spanner/src/index.ts b/handwritten/spanner/src/index.ts index 17287fba98bf..8c8f7cb32e5b 100644 --- a/handwritten/spanner/src/index.ts +++ b/handwritten/spanner/src/index.ts @@ -14,6 +14,8 @@ * limitations under the License. */ +/* eslint-disable import/namespace, promise/catch-or-return, promise/always-return */ + import {GrpcService, GrpcServiceConfig} from './common-grpc/service'; import {PreciseDate} from '@google-cloud/precise-date'; import {replaceProjectIdToken} from './helper'; @@ -103,6 +105,7 @@ import {MetricInterceptor} from './metrics/interceptor'; import {CloudMonitoringMetricsExporter} from './metrics/spanner-metrics-exporter'; import {MetricsTracerFactory} from './metrics/metrics-tracer-factory'; import {MetricsTracer} from './metrics/metrics-tracer'; +import {installHttp2StreamEndWorkaround} from './http2-workaround'; // eslint-disable-next-line @typescript-eslint/no-var-requires const gcpApiConfig = require('./spanner_grpc_config.json'); @@ -510,6 +513,7 @@ class Spanner extends GrpcService { this._universeDomain = universeEndpoint; this.projectId_ = options.projectId; this.configureMetrics_(options.disableBuiltInMetrics); + installHttp2StreamEndWorkaround(); } get universeDomain() { diff --git a/handwritten/spanner/test/http2-workaround.ts b/handwritten/spanner/test/http2-workaround.ts new file mode 100644 index 000000000000..03fc5050716b --- /dev/null +++ b/handwritten/spanner/test/http2-workaround.ts @@ -0,0 +1,369 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +import * as assert from 'assert'; +import * as http2 from 'http2'; +import * as sinon from 'sinon'; +import * as stream from 'stream'; +import { + installHttp2StreamEndWorkaround, + _resetHttp2StreamEndWorkaroundForTest, + _isHttp2StreamEndWorkaroundInstalled, +} from '../src/http2-workaround'; + +// eslint-disable-next-line @typescript-eslint/no-var-requires +const { + Http2SubchannelCall, +} = require('@grpc/grpc-js/build/src/subchannel-call'); + +describe('http2-workaround (targeted Http2SubchannelCall patch)', () => { + let sandbox: sinon.SinonSandbox; + + beforeEach(() => { + sandbox = sinon.createSandbox(); + _resetHttp2StreamEndWorkaroundForTest(); + }); + + afterEach(() => { + sandbox.restore(); + _resetHttp2StreamEndWorkaroundForTest(); + }); + + it('should not install when disabled via environment variable', () => { + const originalEnvironmentVariable = + process.env['SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND']; + try { + process.env['SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND'] = 'true'; + installHttp2StreamEndWorkaround(); + assert.strictEqual(_isHttp2StreamEndWorkaroundInstalled(), false); + } finally { + if (originalEnvironmentVariable !== undefined) { + process.env['SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND'] = + originalEnvironmentVariable; + } else { + delete process.env['SPANNER_DISABLE_HTTP2_STREAM_END_WORKAROUND']; + } + } + }); + + it('should install and mark workaround as installed on Http2SubchannelCall', () => { + installHttp2StreamEndWorkaround(); + assert.strictEqual(_isHttp2StreamEndWorkaroundInstalled(), true); + }); + + it('should not double-install if symbol is already present', () => { + installHttp2StreamEndWorkaround(); + assert.strictEqual(_isHttp2StreamEndWorkaroundInstalled(), true); + + // Call again to verify idempotency + installHttp2StreamEndWorkaround(); + assert.strictEqual(_isHttp2StreamEndWorkaroundInstalled(), true); + }); + + it('should restore original methods when reset function is called', () => { + const originalDestroyHttp2Stream = + Http2SubchannelCall.prototype.destroyHttp2Stream; + const originalHalfClose = Http2SubchannelCall.prototype.halfClose; + + installHttp2StreamEndWorkaround(); + assert.notStrictEqual( + Http2SubchannelCall.prototype.destroyHttp2Stream, + originalDestroyHttp2Stream, + ); + assert.notStrictEqual( + Http2SubchannelCall.prototype.halfClose, + originalHalfClose, + ); + + _resetHttp2StreamEndWorkaroundForTest(); + assert.strictEqual( + Http2SubchannelCall.prototype.destroyHttp2Stream, + originalDestroyHttp2Stream, + ); + assert.strictEqual( + Http2SubchannelCall.prototype.halfClose, + originalHalfClose, + ); + assert.strictEqual(_isHttp2StreamEndWorkaroundInstalled(), false); + }); + + it('should leave stream.Duplex.prototype completely untouched', () => { + const originalDuplexEnd = stream.Duplex.prototype.end; + + installHttp2StreamEndWorkaround(); + + assert.strictEqual( + stream.Duplex.prototype.end, + originalDuplexEnd, + 'Duplex.prototype.end must not be modified by targeted workaround', + ); + + const symbolList = Object.getOwnPropertySymbols(stream.Duplex.prototype); + const workaroundSymbols = symbolList.filter(s => + s.toString().includes('http2'), + ); + assert.strictEqual( + workaroundSymbols.length, + 0, + 'Duplex.prototype must not have any workaround symbols attached', + ); + }); + + it('should not affect standard Node.js duplex streams', async () => { + installHttp2StreamEndWorkaround(); + + const passThrough = new stream.PassThrough(); + await new Promise((resolve, reject) => { + passThrough.on('finish', () => { + passThrough.end((error: Error & {code?: string}) => { + try { + assert.ok( + error, + 'Standard duplex stream should receive error on callback', + ); + assert.strictEqual(error.code, 'ERR_STREAM_ALREADY_FINISHED'); + resolve(); + } catch (assertionError) { + reject(assertionError); + } + }); + }); + passThrough.end(); + }); + }); + + describe('destroyHttp2Stream', () => { + it('should skip http2Stream.end() when serverEndedCall is true and stream is writableEnded', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.serverEndedCall = true; + mockCall.http2Stream = { + writableEnded: true, + destroyed: false, + end: endSpy, + }; + + mockCall.destroyHttp2Stream(); + + assert.strictEqual( + endSpy.called, + false, + 'http2Stream.end() should have been skipped', + ); + }); + + it('should call http2Stream.end() when serverEndedCall is true but stream is not writableEnded', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.serverEndedCall = true; + mockCall.http2Stream = { + writableEnded: false, + destroyed: false, + end: endSpy, + }; + + mockCall.destroyHttp2Stream(); + + assert.strictEqual( + endSpy.calledOnce, + true, + 'http2Stream.end() should have been called', + ); + }); + + it('should return early when http2Stream is destroyed', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const closeSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.serverEndedCall = true; + mockCall.http2Stream = { + writableEnded: false, + destroyed: true, + end: endSpy, + close: closeSpy, + }; + + mockCall.destroyHttp2Stream(); + + assert.strictEqual(endSpy.called, false); + assert.strictEqual(closeSpy.called, false); + }); + + it('should delegate to original destroyHttp2Stream when serverEndedCall is false', () => { + installHttp2StreamEndWorkaround(); + + const closeSpy = sandbox.spy(); + const endSpy = sandbox.spy(); + const traceSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.serverEndedCall = false; + mockCall.finalStatus = {code: 0}; + mockCall.trace = traceSpy; + mockCall.http2Stream = { + destroyed: false, + writableEnded: false, + close: closeSpy, + end: endSpy, + }; + + mockCall.destroyHttp2Stream(); + + assert.strictEqual( + closeSpy.calledOnce, + true, + 'close() should be called when serverEndedCall is false', + ); + assert.strictEqual( + endSpy.called, + false, + 'end() should not be called when serverEndedCall is false', + ); + }); + }); + + describe('halfClose', () => { + it('should skip http2Stream.end() when stream is already writableEnded', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const traceSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.trace = traceSpy; + mockCall.http2Stream = { + writableEnded: true, + destroyed: false, + end: endSpy, + }; + + mockCall.halfClose(); + + assert.strictEqual(endSpy.called, false); + }); + + it('should skip http2Stream.end() when stream is destroyed', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const traceSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.trace = traceSpy; + mockCall.http2Stream = { + writableEnded: false, + destroyed: true, + end: endSpy, + }; + + mockCall.halfClose(); + + assert.strictEqual(endSpy.called, false); + }); + + it('should call original halfClose when stream is neither writableEnded nor destroyed', () => { + installHttp2StreamEndWorkaround(); + + const endSpy = sandbox.spy(); + const traceSpy = sandbox.spy(); + const mockCall = Object.create(Http2SubchannelCall.prototype); + mockCall.trace = traceSpy; + mockCall.http2Stream = { + writableEnded: false, + destroyed: false, + end: endSpy, + }; + + mockCall.halfClose(); + + assert.strictEqual(endSpy.calledOnce, true); + }); + }); + + describe('real HTTP/2 stream integration', () => { + it('should skip redundant end() during full call lifecycle', async () => { + installHttp2StreamEndWorkaround(); + + const server = http2.createServer(); + server.on('stream', (serverStream: http2.ServerHttp2Stream) => { + serverStream.respond({':status': 200}); + serverStream.end('ok'); + }); + + await new Promise((resolve, reject) => { + server.listen(0, () => { + const address = server.address(); + if (!address || typeof address === 'string') { + reject(new Error('Invalid server address')); + return; + } + + const clientSession = http2.connect( + `http://localhost:${address.port}`, + ); + const requestStream = clientSession.request({ + ':method': 'POST', + ':path': '/', + }); + + // Track calls to requestStream.end + const endSpy = sandbox.spy(requestStream, 'end'); + + const call = Object.create(Http2SubchannelCall.prototype); + call.http2Stream = requestStream; + call.trace = () => {}; + + requestStream.on('response', () => {}); + requestStream.on('data', () => {}); + requestStream.on('end', () => { + try { + // 1. In gRPC unary RPC, client half-closes request stream + call.halfClose(); + assert.strictEqual( + requestStream.writableEnded, + true, + 'Request stream should be writableEnded after halfClose', + ); + assert.strictEqual( + endSpy.callCount, + 1, + 'requestStream.end() should be called once by halfClose()', + ); + + // 2. Server trailers arrive and server ends the call + call.serverEndedCall = true; + call.destroyHttp2Stream(); + + // With targeted workaround, redundant end() must be skipped! + assert.strictEqual( + endSpy.callCount, + 1, + 'destroyHttp2Stream() must not call end() again when stream is already ended', + ); + + clientSession.close(); + server.close(() => resolve()); + } catch (assertionError) { + clientSession.close(); + server.close(() => reject(assertionError)); + } + }); + }); + }); + }); + }); +});