From 152eebdcd5bbb1548b7779d14e0f5046784c21c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Mon, 14 Sep 2026 16:09:58 +0200 Subject: [PATCH 1/3] chore(grpc-js): unify call numbers and avoid disabled trace allocations Previously, each layer of an RPC (ResolvingCall, RetryingCall, LoadBalancingCall, and Http2SubchannelCall) allocated its own separate call number via getNextCallNumber(). This made correlating logs across layers difficult and consumed multiple IDs per logical RPC. Additionally, dynamic trace arguments such as string concatenations, JSON.stringify(), and deadline formatting were evaluated eagerly at call sites even when tracing was disabled, adding unnecessary garbage collection pressure on the fast path. This change: 1. Passes the initial callNumber from ResolvingCall down through RetryingCall, LoadBalancingCall, and Http2SubchannelCall so that all layers of an attempt share the same call number. Subsequent retries and hedged attempts allocate a new call number per attempt. 2. Short-circuits isTracerEnabled() when no tracers are active and guards dynamic trace argument evaluation behind traceEnabled checks so that disabled tracers incur zero string or formatting allocations. --- packages/grpc-js/src/backoff-timeout.ts | 33 +- packages/grpc-js/src/internal-channel.ts | 114 ++-- packages/grpc-js/src/load-balancing-call.ts | 175 +++--- packages/grpc-js/src/logging.ts | 7 + packages/grpc-js/src/resolving-call.ts | 69 ++- packages/grpc-js/src/retrying-call.ts | 150 +++-- .../grpc-js/src/single-subchannel-channel.ts | 8 +- packages/grpc-js/src/subchannel-call.ts | 136 +++-- packages/grpc-js/src/subchannel.ts | 107 ++-- packages/grpc-js/src/transport.ts | 220 ++++--- packages/grpc-js/test/test-call-number.ts | 574 ++++++++++++++++++ packages/grpc-js/test/test-logging.ts | 165 +++++ 12 files changed, 1383 insertions(+), 375 deletions(-) create mode 100644 packages/grpc-js/test/test-call-number.ts diff --git a/packages/grpc-js/src/backoff-timeout.ts b/packages/grpc-js/src/backoff-timeout.ts index 8be560da1..8bb9fa8fd 100644 --- a/packages/grpc-js/src/backoff-timeout.ts +++ b/packages/grpc-js/src/backoff-timeout.ts @@ -109,7 +109,18 @@ export class BackoffTimeout { this.maxDelay = options.maxDelay; } } - this.trace('constructed initialDelay=' + this.initialDelay + ' multiplier=' + this.multiplier + ' jitter=' + this.jitter + ' maxDelay=' + this.maxDelay); + if (this.traceEnabled) { + this.trace( + 'constructed initialDelay=' + + this.initialDelay + + ' multiplier=' + + this.multiplier + + ' jitter=' + + this.jitter + + ' maxDelay=' + + this.maxDelay + ); + } this.nextDelay = this.initialDelay; this.timerId = setTimeout(() => {}, 0); clearTimeout(this.timerId); @@ -119,12 +130,24 @@ export class BackoffTimeout { return this.nextId++; } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string) { - logging.trace(LogVerbosity.DEBUG, TRACER_NAME, '{' + this.id + '} ' + text); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '{' + this.id + '} ' + text + ); + } } private runTimer(delay: number) { - this.trace('runTimer(delay=' + delay + ')'); + if (this.traceEnabled) { + this.trace('runTimer(delay=' + delay + ')'); + } this.endTime = this.startTime; this.endTime.setMilliseconds( this.endTime.getMilliseconds() + delay @@ -172,7 +195,9 @@ export class BackoffTimeout { * retroactively apply that reset to the current timer. */ reset() { - this.trace('reset() running=' + this.running); + if (this.traceEnabled) { + this.trace('reset() running=' + this.running); + } this.nextDelay = this.initialDelay; if (this.running) { const now = new Date(); diff --git a/packages/grpc-js/src/internal-channel.ts b/packages/grpc-js/src/internal-channel.ts index 942d7a449..9c05b91eb 100644 --- a/packages/grpc-js/src/internal-channel.ts +++ b/packages/grpc-js/src/internal-channel.ts @@ -443,12 +443,14 @@ export class InternalChannel { this.filterStackFactory = new FilterStackFactory([ new CompressionFilterFactory(this, this.options), ]); - this.trace( - 'Channel constructed with options ' + - JSON.stringify(options, undefined, 2) - ); - const error = new Error(); - if (isTracerEnabled('channel_stacktrace')){ + if (this.traceEnabled) { + this.trace( + 'Channel constructed with options ' + + JSON.stringify(options, undefined, 2) + ); + } + if (isTracerEnabled('channel_stacktrace')) { + const error = new Error(); trace( LogVerbosity.DEBUG, 'channel_stacktrace', @@ -462,12 +464,18 @@ export class InternalChannel { this.lastActivityTimestamp = new Date(); } + private get traceEnabled(): boolean { + return isTracerEnabled('channel'); + } + private trace(text: string, verbosityOverride?: LogVerbosity) { - trace( - verbosityOverride ?? LogVerbosity.DEBUG, - 'channel', - '(' + this.channelzRef.id + ') ' + uriToString(this.target) + ' ' + text - ); + if (this.traceEnabled) { + trace( + verbosityOverride ?? LogVerbosity.DEBUG, + 'channel', + '(' + this.channelzRef.id + ') ' + uriToString(this.target) + ' ' + text + ); + } } private callRefTimerRef() { @@ -476,12 +484,14 @@ export class InternalChannel { } // If the hasRef function does not exist, always run the code if (!this.callRefTimer.hasRef?.()) { - this.trace( - 'callRefTimer.ref | configSelectionQueue.length=' + - this.configSelectionQueue.length + - ' pickQueue.size=' + - this.pickQueue.size - ); + if (this.traceEnabled) { + this.trace( + 'callRefTimer.ref | configSelectionQueue.length=' + + this.configSelectionQueue.length + + ' pickQueue.length=' + + this.pickQueue.size + ); + } this.callRefTimer.ref?.(); } } @@ -489,12 +499,14 @@ export class InternalChannel { private callRefTimerUnref() { // If the timer or the hasRef function does not exist, always run the code if (!this.callRefTimer?.hasRef || this.callRefTimer.hasRef()) { - this.trace( - 'callRefTimer.unref | configSelectionQueue.length=' + - this.configSelectionQueue.length + - ' pickQueue.size=' + - this.pickQueue.size - ); + if (this.traceEnabled) { + this.trace( + 'callRefTimer.unref | configSelectionQueue.length=' + + this.configSelectionQueue.length + + ' pickQueue.length=' + + this.pickQueue.size + ); + } this.callRefTimer?.unref?.(); } } @@ -688,12 +700,19 @@ export class InternalChannel { method: string, host: string, credentials: CallCredentials, - deadline: Deadline + deadline: Deadline, + callNumber?: number ): LoadBalancingCall { - const callNumber = getNextCallNumber(); - this.trace( - 'createLoadBalancingCall [' + callNumber + '] method="' + method + '"' - ); + const finalCallNumber = callNumber ?? getNextCallNumber(); + if (this.traceEnabled) { + this.trace( + 'createLoadBalancingCall [' + + finalCallNumber + + '] method="' + + method + + '"' + ); + } return new LoadBalancingCall( this, callConfig, @@ -701,7 +720,7 @@ export class InternalChannel { host, credentials, deadline, - callNumber + finalCallNumber ); } @@ -710,12 +729,19 @@ export class InternalChannel { method: string, host: string, credentials: CallCredentials, - deadline: Deadline + deadline: Deadline, + callNumber?: number ): RetryingCall { - const callNumber = getNextCallNumber(); - this.trace( - 'createRetryingCall [' + callNumber + '] method="' + method + '"' - ); + const finalCallNumber = callNumber ?? getNextCallNumber(); + if (this.traceEnabled) { + this.trace( + 'createRetryingCall [' + + finalCallNumber + + '] method="' + + method + + '"' + ); + } return new RetryingCall( this, callConfig, @@ -723,7 +749,7 @@ export class InternalChannel { host, credentials, deadline, - callNumber, + finalCallNumber, this.retryBufferTracker, RETRY_THROTTLER_MAP.get(this.getTarget()) ); @@ -737,14 +763,16 @@ export class InternalChannel { propagateFlags: number | null | undefined ): ResolvingCall { const callNumber = getNextCallNumber(); - this.trace( - 'createResolvingCall [' + - callNumber + - '] method="' + - method + - '", deadline=' + - deadlineToString(deadline) - ); + if (this.traceEnabled) { + this.trace( + 'createResolvingCall [' + + callNumber + + '] method="' + + method + + '", deadline=' + + deadlineToString(deadline) + ); + } const finalOptions: CallStreamOptions = { deadline: deadline, flags: propagateFlags ?? Propagate.DEFAULTS, diff --git a/packages/grpc-js/src/load-balancing-call.ts b/packages/grpc-js/src/load-balancing-call.ts index 36014ee4f..a3a7495da 100644 --- a/packages/grpc-js/src/load-balancing-call.ts +++ b/packages/grpc-js/src/load-balancing-call.ts @@ -36,6 +36,7 @@ import * as logging from './logging'; import { restrictControlPlaneStatusCode } from './control-plane-status'; import * as http2 from 'http2'; import { AuthContext } from './auth-context'; +import { SubchannelInterface } from './subchannel-interface'; const TRACER_NAME = 'load_balancing_call'; @@ -106,25 +107,44 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { return deadlineInfo; } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '[' + this.callNumber + '] ' + text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '[' + this.callNumber + '] ' + text + ); + } + } + + private getSubchannelString( + subchannel: SubchannelInterface | null | undefined + ): string { + return subchannel + ? '(' + + subchannel.getChannelzRef().id + + ') ' + + subchannel.getAddress() + : '' + subchannel; } private outputStatus(status: StatusObject, progress: RpcProgress) { if (!this.ended) { this.ended = true; - this.trace( - 'ended with status: code=' + - status.code + - ' details="' + - status.details + - '" start time=' + - this.startTime.toISOString() - ); + if (this.traceEnabled) { + this.trace( + 'ended with status: code=' + + status.code + + ' details="' + + status.details + + '" start time=' + + this.startTime.toISOString() + ); + } const finalStatus = { ...status, progress }; this.listener?.onReceiveStatus(finalStatus); this.onCallEnded?.(finalStatus.code, finalStatus.details, finalStatus.metadata); @@ -145,22 +165,19 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { finalMetadata, this.callConfig.pickInformation ); - const subchannelString = pickResult.subchannel - ? '(' + - pickResult.subchannel.getChannelzRef().id + - ') ' + - pickResult.subchannel.getAddress() - : '' + pickResult.subchannel; - this.trace( - 'Pick result: ' + - PickResultType[pickResult.pickResultType] + - ' subchannel: ' + - subchannelString + - ' status: ' + - pickResult.status?.code + - ' ' + - pickResult.status?.details - ); + + if (this.traceEnabled) { + this.trace( + 'Pick result: ' + + PickResultType[pickResult.pickResultType] + + ' subchannel: ' + + this.getSubchannelString(pickResult.subchannel) + + ' status: ' + + pickResult.status?.code + + ' ' + + pickResult.status?.details + ); + } switch (pickResult.pickResultType) { case PickResultType.COMPLETE: const combinedCallCredentials = this.credentials.compose(pickResult.subchannel!.getCallCredentials()); @@ -193,15 +210,17 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { pickResult.subchannel!.getConnectivityState() !== ConnectivityState.READY ) { - this.trace( - 'Picked subchannel ' + - subchannelString + - ' has state ' + - ConnectivityState[ - pickResult.subchannel!.getConnectivityState() - ] + - ' after getting credentials metadata. Retrying pick' - ); + if (this.traceEnabled) { + this.trace( + 'Picked subchannel ' + + this.getSubchannelString(pickResult.subchannel) + + ' has state ' + + ConnectivityState[ + pickResult.subchannel!.getConnectivityState() + ] + + ' after getting credentials metadata. Retrying pick' + ); + } this.doPick(); return; } @@ -215,35 +234,43 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { try { this.child = pickResult .subchannel!.getRealSubchannel() - .createCall(finalMetadata, this.host, this.methodName, { - onReceiveMetadata: metadata => { - this.trace('Received metadata'); - this.listener!.onReceiveMetadata(metadata); - }, - onReceiveMessage: message => { - this.trace('Received message'); - this.listener!.onReceiveMessage(message); + .createCall( + finalMetadata, + this.host, + this.methodName, + { + onReceiveMetadata: metadata => { + this.trace('Received metadata'); + this.listener!.onReceiveMetadata(metadata); + }, + onReceiveMessage: message => { + this.trace('Received message'); + this.listener!.onReceiveMessage(message); + }, + onReceiveStatus: status => { + this.trace('Received status'); + if ( + status.rstCode === + http2.constants.NGHTTP2_REFUSED_STREAM + ) { + this.outputStatus(status, 'REFUSED'); + } else { + this.outputStatus(status, 'PROCESSED'); + } + }, }, - onReceiveStatus: status => { - this.trace('Received status'); - if ( - status.rstCode === - http2.constants.NGHTTP2_REFUSED_STREAM - ) { - this.outputStatus(status, 'REFUSED'); - } else { - this.outputStatus(status, 'PROCESSED'); - } - }, - }); + this.callNumber + ); this.childStartTime = new Date(); } catch (error) { - this.trace( - 'Failed to start call on picked subchannel ' + - subchannelString + - ' with error ' + - (error as Error).message - ); + if (this.traceEnabled) { + this.trace( + 'Failed to start call on picked subchannel ' + + this.getSubchannelString(pickResult.subchannel) + + ' with error ' + + (error as Error).message + ); + } this.outputStatus( { code: Status.INTERNAL, @@ -258,9 +285,11 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { } pickResult.onCallStarted?.(); this.onCallEnded = pickResult.onCallEnded; - this.trace( - 'Created child call [' + this.child.getCallNumber() + ']' - ); + if (this.traceEnabled) { + this.trace( + 'Created child call [' + this.child.getCallNumber() + ']' + ); + } if (this.readPending) { this.child.startRead(); } @@ -325,9 +354,11 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { } cancelWithStatus(status: Status, details: string): void { - this.trace( - 'cancelWithStatus code: ' + status + ' details: "' + details + '"' - ); + if (this.traceEnabled) { + this.trace( + 'cancelWithStatus code: ' + status + ' details: "' + details + '"' + ); + } this.child?.cancelWithStatus(status, details); this.outputStatus( { code: status, details: details, metadata: new Metadata() }, @@ -347,7 +378,9 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { this.doPick(); } sendMessageWithContext(context: MessageContext, message: Buffer): void { - this.trace('write() called with message of length ' + message.length); + if (this.traceEnabled) { + this.trace('write() called with message of length ' + message.length); + } if (this.child) { this.child.sendMessageWithContext(context, message); } else { diff --git a/packages/grpc-js/src/logging.ts b/packages/grpc-js/src/logging.ts index 2279d3b65..151ebb03b 100644 --- a/packages/grpc-js/src/logging.ts +++ b/packages/grpc-js/src/logging.ts @@ -98,6 +98,9 @@ const tracersString = const enabledTracers = new Set(); const disabledTracers = new Set(); for (const tracerName of tracersString.split(',')) { + if (tracerName.length === 0) { + continue; + } if (tracerName.startsWith('-')) { disabledTracers.add(tracerName.substring(1)); } else { @@ -105,6 +108,7 @@ for (const tracerName of tracersString.split(',')) { } } const allEnabled = enabledTracers.has('all'); +const anyTracerEnabled = allEnabled || enabledTracers.size > 0; export function trace( severity: LogVerbosity, @@ -128,6 +132,9 @@ export function trace( } export function isTracerEnabled(tracer: string): boolean { + if (!anyTracerEnabled) { + return false; + } return ( !disabledTracers.has(tracer) && (allEnabled || enabledTracers.has(tracer)) ); diff --git a/packages/grpc-js/src/resolving-call.ts b/packages/grpc-js/src/resolving-call.ts index 2a4f5edc4..d7ec6ddcd 100644 --- a/packages/grpc-js/src/resolving-call.ts +++ b/packages/grpc-js/src/resolving-call.ts @@ -86,10 +86,12 @@ export class ResolvingCall implements Call { }); } if (options.flags & Propagate.DEADLINE) { - this.trace( - 'Propagating deadline from parent: ' + - options.parentCall.getDeadline() - ); + if (this.traceEnabled) { + this.trace( + 'Propagating deadline from parent: ' + + options.parentCall.getDeadline() + ); + } this.deadline = minDeadline( this.deadline, options.parentCall.getDeadline() @@ -100,21 +102,31 @@ export class ResolvingCall implements Call { this.runDeadlineTimer(); } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '[' + this.callNumber + '] ' + text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '[' + this.callNumber + '] ' + text + ); + } } private runDeadlineTimer() { clearTimeout(this.deadlineTimer); this.deadlineStartTime = new Date(); - this.trace('Deadline: ' + deadlineToString(this.deadline)); + if (this.traceEnabled) { + this.trace('Deadline: ' + deadlineToString(this.deadline)); + } const timeout = getRelativeTimeout(this.deadline); if (timeout !== Infinity) { - this.trace('Deadline will be reached in ' + timeout + 'ms'); + if (this.traceEnabled) { + this.trace('Deadline will be reached in ' + timeout + 'ms'); + } const handleDeadline = () => { if (!this.deadlineStartTime) { this.cancelWithStatus(Status.DEADLINE_EXCEEDED, 'Deadline exceeded'); @@ -158,13 +170,15 @@ export class ResolvingCall implements Call { } clearTimeout(this.deadlineTimer); const filteredStatus = this.filterStack.receiveTrailers(status); - this.trace( - 'ended with status: code=' + - filteredStatus.code + - ' details="' + - filteredStatus.details + - '"' - ); + if (this.traceEnabled) { + this.trace( + 'ended with status: code=' + + filteredStatus.code + + ' details="' + + filteredStatus.details + + '"' + ); + } this.statusWatchers.forEach(watcher => watcher(filteredStatus)); process.nextTick(() => { this.listener?.onReceiveStatus(filteredStatus); @@ -251,9 +265,12 @@ export class ResolvingCall implements Call { this.method, this.host, this.credentials, - this.deadline + this.deadline, + this.callNumber ); - this.trace('Created child [' + this.child.getCallNumber() + ']'); + if (this.traceEnabled) { + this.trace('Created child [' + this.child.getCallNumber() + ']'); + } this.childStartTime = new Date(); this.child.start(filteredMetadata, { onReceiveMetadata: metadata => { @@ -314,9 +331,11 @@ export class ResolvingCall implements Call { } } cancelWithStatus(status: Status, details: string): void { - this.trace( - 'cancelWithStatus code: ' + status + ' details: "' + details + '"' - ); + if (this.traceEnabled) { + this.trace( + 'cancelWithStatus code: ' + status + ' details: "' + details + '"' + ); + } this.child?.cancelWithStatus(status, details); this.outputStatus({ code: status, @@ -334,7 +353,9 @@ export class ResolvingCall implements Call { this.getConfig(); } sendMessageWithContext(context: MessageContext, message: Buffer): void { - this.trace('write() called with message of length ' + message.length); + if (this.traceEnabled) { + this.trace('write() called with message of length ' + message.length); + } if (this.child) { this.sendMessageOnChild(context, message); } else { diff --git a/packages/grpc-js/src/retrying-call.ts b/packages/grpc-js/src/retrying-call.ts index 61ff58fa1..9023b5241 100644 --- a/packages/grpc-js/src/retrying-call.ts +++ b/packages/grpc-js/src/retrying-call.ts @@ -21,6 +21,7 @@ import { Deadline, formatDateDifference } from './deadline'; import { Metadata } from './metadata'; import { CallConfig } from './resolver'; import * as logging from './logging'; +import { getNextCallNumber } from './call-number'; import { Call, DeadlineInfoProvider, @@ -276,23 +277,31 @@ export class RetryingCall implements Call, DeadlineInfoProvider { return this.callNumber; } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '[' + this.callNumber + '] ' + text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '[' + this.callNumber + '] ' + text + ); + } } private reportStatus(statusObject: StatusObject) { - this.trace( - 'ended with status: code=' + - statusObject.code + - ' details="' + - statusObject.details + - '" start time=' + - this.startTime.toISOString() - ); + if (this.traceEnabled) { + this.trace( + 'ended with status: code=' + + statusObject.code + + ' details="' + + statusObject.details + + '" start time=' + + this.startTime.toISOString() + ); + } this.bufferTracker.freeAll(this.callNumber); this.writeBufferOffset = this.writeBufferOffset + this.writeBuffer.length; this.writeBuffer = []; @@ -307,9 +316,11 @@ export class RetryingCall implements Call, DeadlineInfoProvider { } cancelWithStatus(status: Status, details: string): void { - this.trace( - 'cancelWithStatus code: ' + status + ' details: "' + details + '"' - ); + if (this.traceEnabled) { + this.trace( + 'cancelWithStatus code: ' + status + ' details: "' + details + '"' + ); + } this.reportStatus({ code: status, details, metadata: new Metadata() }); for (const { call } of this.underlyingCalls) { call.cancelWithStatus(status, details); @@ -372,12 +383,14 @@ export class RetryingCall implements Call, DeadlineInfoProvider { if (this.state === 'COMMITTED') { return; } - this.trace( - 'Committing call [' + - this.underlyingCalls[index].call.getCallNumber() + - '] at index ' + - index - ); + if (this.traceEnabled) { + this.trace( + 'Committing call [' + + this.underlyingCalls[index].call.getCallNumber() + + '] at index ' + + index + ); + } this.state = 'COMMITTED'; this.callConfig.onCommitted?.(); this.committedCallIndex = index; @@ -586,16 +599,18 @@ export class RetryingCall implements Call, DeadlineInfoProvider { if (this.underlyingCalls[callIndex].state === 'COMPLETED') { return; } - this.trace( - 'state=' + - this.state + - ' handling status with progress ' + - status.progress + - ' from child [' + - this.underlyingCalls[callIndex].call.getCallNumber() + - '] in state ' + - this.underlyingCalls[callIndex].state - ); + if (this.traceEnabled) { + this.trace( + 'state=' + + this.state + + ' handling status with progress ' + + status.progress + + ' from child [' + + this.underlyingCalls[callIndex].call.getCallNumber() + + '] in state ' + + this.underlyingCalls[callIndex].state + ); + } this.underlyingCalls[callIndex].state = 'COMPLETED'; if (status.code === Status.OK) { this.retryThrottler?.addCallSucceeded(); @@ -677,19 +692,24 @@ export class RetryingCall implements Call, DeadlineInfoProvider { } private startNewAttempt() { + const childCallNumber = + this.underlyingCalls.length > 0 ? getNextCallNumber() : this.callNumber; const child = this.channel.createLoadBalancingCall( this.callConfig, this.methodName, this.host, this.credentials, - this.deadline - ); - this.trace( - 'Created child call [' + - child.getCallNumber() + - '] for attempt ' + - this.attempts + this.deadline, + childCallNumber ); + if (this.traceEnabled) { + this.trace( + 'Created child call [' + + child.getCallNumber() + + '] for attempt ' + + this.attempts + ); + } const index = this.underlyingCalls.length; this.underlyingCalls.push({ state: 'ACTIVE', @@ -708,9 +728,11 @@ export class RetryingCall implements Call, DeadlineInfoProvider { let receivedMetadata = false; child.start(initialMetadata, { onReceiveMetadata: metadata => { - this.trace( - 'Received metadata from child [' + child.getCallNumber() + ']' - ); + if (this.traceEnabled) { + this.trace( + 'Received metadata from child [' + child.getCallNumber() + ']' + ); + } this.commitCall(index); receivedMetadata = true; if (previousAttempts > 0) { @@ -724,18 +746,22 @@ export class RetryingCall implements Call, DeadlineInfoProvider { } }, onReceiveMessage: message => { - this.trace( - 'Received message from child [' + child.getCallNumber() + ']' - ); + if (this.traceEnabled) { + this.trace( + 'Received message from child [' + child.getCallNumber() + ']' + ); + } this.commitCall(index); if (this.underlyingCalls[index].state === 'ACTIVE') { this.listener!.onReceiveMessage(message); } }, onReceiveStatus: status => { - this.trace( - 'Received status from child [' + child.getCallNumber() + ']' - ); + if (this.traceEnabled) { + this.trace( + 'Received status from child [' + child.getCallNumber() + ']' + ); + } if (!receivedMetadata && previousAttempts > 0) { status.metadata.set( PREVIONS_RPC_ATTEMPTS_METADATA_KEY, @@ -792,11 +818,13 @@ export class RetryingCall implements Call, DeadlineInfoProvider { // has already been passed to the underlying transport. const nextEntry = this.getBufferEntry(messageIndex + 1); if (nextEntry.entryType === 'HALF_CLOSE') { - this.trace( - 'Sending halfClose immediately after message to child [' + - childCall.call.getCallNumber() + - '] - optimizing for unary/final message' - ); + if (this.traceEnabled) { + this.trace( + 'Sending halfClose immediately after message to child [' + + childCall.call.getCallNumber() + + '] - optimizing for unary/final message' + ); + } childCall.nextMessageToSend += 1; childCall.call.halfClose(); } @@ -813,7 +841,9 @@ export class RetryingCall implements Call, DeadlineInfoProvider { } sendMessageWithContext(context: MessageContext, message: Buffer): void { - this.trace('write() called with message of length ' + message.length); + if (this.traceEnabled) { + this.trace('write() called with message of length ' + message.length); + } const writeObj: WriteObject = { message, flags: context.flags, @@ -891,11 +921,13 @@ export class RetryingCall implements Call, DeadlineInfoProvider { // - nextMessageToSend === halfCloseIndex: all messages sent and acknowledged if (call.nextMessageToSend === halfCloseIndex || call.nextMessageToSend === halfCloseIndex - 1) { - this.trace( - 'Sending halfClose immediately to child [' + - call.call.getCallNumber() + - '] - all messages already sent' - ); + if (this.traceEnabled) { + this.trace( + 'Sending halfClose immediately to child [' + + call.call.getCallNumber() + + '] - all messages already sent' + ); + } call.nextMessageToSend += 1; call.call.halfClose(); } diff --git a/packages/grpc-js/src/single-subchannel-channel.ts b/packages/grpc-js/src/single-subchannel-channel.ts index c1a1fd1b0..4e286795c 100644 --- a/packages/grpc-js/src/single-subchannel-channel.ts +++ b/packages/grpc-js/src/single-subchannel-channel.ts @@ -142,7 +142,13 @@ class SubchannelCallWrapper implements Call { } } } - this.childCall = this.subchannel.createCall(credsMetadata, this.options.host, this.method, childListener); + this.childCall = this.subchannel.createCall( + credsMetadata, + this.options.host, + this.method, + childListener, + this.callNumber + ); if (this.readPending) { this.childCall.startRead(); } diff --git a/packages/grpc-js/src/subchannel-call.ts b/packages/grpc-js/src/subchannel-call.ts index 207b781c7..51878e76f 100644 --- a/packages/grpc-js/src/subchannel-call.ts +++ b/packages/grpc-js/src/subchannel-call.ts @@ -154,11 +154,13 @@ export class Http2SubchannelCall implements SubchannelCall { const maxReceiveMessageLength = transport.getOptions()['grpc.max_receive_message_length'] ?? DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH; this.decoder = new StreamDecoder(maxReceiveMessageLength); http2Stream.on('response', (headers, flags) => { - let headersString = ''; - for (const header of Object.keys(headers)) { - headersString += '\t\t' + header + ': ' + headers[header] + '\n'; + if (this.traceEnabled) { + let headersString = ''; + for (const header of Object.keys(headers)) { + headersString += '\t\t' + header + ': ' + headers[header] + '\n'; + } + this.trace('Received server headers:\n' + headersString); } - this.trace('Received server headers:\n' + headersString); this.httpStatusCode = headers[':status']; if (flags & http2.constants.NGHTTP2_FLAG_END_STREAM) { @@ -187,7 +189,9 @@ export class Http2SubchannelCall implements SubchannelCall { if (this.statusOutput) { return; } - this.trace('receive HTTP/2 data frame of length ' + data.length); + if (this.traceEnabled) { + this.trace('receive HTTP/2 data frame of length ' + data.length); + } let messages: Buffer[]; try { messages = this.decoder.write(data); @@ -212,7 +216,9 @@ export class Http2SubchannelCall implements SubchannelCall { } for (const message of messages) { - this.trace('parsed message of length ' + message.length); + if (this.traceEnabled) { + this.trace('parsed message of length ' + message.length); + } this.callEventTracker!.addMessageReceived(); this.tryPush(message); } @@ -227,7 +233,9 @@ export class Http2SubchannelCall implements SubchannelCall { * "error" event that may be emitted at about the same time, so that * we can bubble up the error message from that event. */ process.nextTick(() => { - this.trace('HTTP/2 stream closed with code ' + http2Stream.rstCode); + if (this.traceEnabled) { + this.trace('HTTP/2 stream closed with code ' + http2Stream.rstCode); + } /* If we have a final status with an OK status code, that means that * we have received all of the messages and we have processed the * trailers and the call completed successfully, so it doesn't matter @@ -328,16 +336,18 @@ export class Http2SubchannelCall implements SubchannelCall { * https://github.com/nodejs/node/blob/8b8620d580314050175983402dfddf2674e8e22a/lib/internal/http2/core.js#L2267 */ if (err.code !== 'ERR_HTTP2_STREAM_ERROR') { - this.trace( - 'Node error event: message=' + - err.message + - ' code=' + - err.code + - ' errno=' + - getSystemErrorName(err.errno) + - ' syscall=' + - err.syscall - ); + if (this.traceEnabled) { + this.trace( + 'Node error event: message=' + + err.message + + ' code=' + + err.code + + ' errno=' + + getSystemErrorName(err.errno) + + ' syscall=' + + err.syscall + ); + } this.internalError = err; } this.callEventTracker.onStreamEnd(false); @@ -364,13 +374,15 @@ export class Http2SubchannelCall implements SubchannelCall { /* Precondition: this.finalStatus !== null */ if (!this.statusOutput) { this.statusOutput = true; - this.trace( - 'ended with status: code=' + - this.finalStatus!.code + - ' details="' + - this.finalStatus!.details + - '"' - ); + if (this.traceEnabled) { + this.trace( + 'ended with status: code=' + + this.finalStatus!.code + + ' details="' + + this.finalStatus!.details + + '"' + ); + } this.callEventTracker.onCallEnd(this.finalStatus!); /* We delay the actual action of bubbling up the status to insulate the * cleanup code in this class from any errors that may be thrown in the @@ -389,12 +401,18 @@ export class Http2SubchannelCall implements SubchannelCall { } } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '[' + this.callId + '] ' + text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '[' + this.callId + '] ' + text + ); + } } /** @@ -430,10 +448,12 @@ export class Http2SubchannelCall implements SubchannelCall { } private push(message: Buffer): void { - this.trace( - 'pushing to reader message of length ' + - (message instanceof Buffer ? message.length : null) - ); + if (this.traceEnabled) { + this.trace( + 'pushing to reader message of length ' + + (message instanceof Buffer ? message.length : null) + ); + } this.canPush = false; this.isPushPending = true; process.nextTick(() => { @@ -455,9 +475,11 @@ export class Http2SubchannelCall implements SubchannelCall { this.http2Stream!.pause(); this.push(messageBytes); } else { - this.trace( - 'unpushedReadMessages.push message of length ' + messageBytes.length - ); + if (this.traceEnabled) { + this.trace( + 'unpushedReadMessages.push message of length ' + messageBytes.length + ); + } this.unpushedReadMessages.push(messageBytes); } } @@ -465,11 +487,13 @@ export class Http2SubchannelCall implements SubchannelCall { private handleTrailers(headers: http2.IncomingHttpHeaders) { this.serverEndedCall = true; this.callEventTracker.onStreamEnd(true); - let headersString = ''; - for (const header of Object.keys(headers)) { - headersString += '\t\t' + header + ': ' + headers[header] + '\n'; + if (this.traceEnabled) { + let headersString = ''; + for (const header of Object.keys(headers)) { + headersString += '\t\t' + header + ': ' + headers[header] + '\n'; + } + this.trace('Received server trailers:\n' + headersString); } - this.trace('Received server trailers:\n' + headersString); let metadata: Metadata; try { metadata = Metadata.fromHttp2Headers(headers); @@ -480,7 +504,9 @@ export class Http2SubchannelCall implements SubchannelCall { let status: StatusObject; if (typeof metadataMap['grpc-status'] === 'string') { const receivedStatus: Status = Number(metadataMap['grpc-status']); - this.trace('received status code ' + receivedStatus + ' from server'); + if (this.traceEnabled) { + this.trace('received status code ' + receivedStatus + ' from server'); + } metadata.remove('grpc-status'); let details = ''; if (typeof metadataMap['grpc-message'] === 'string') { @@ -490,9 +516,11 @@ export class Http2SubchannelCall implements SubchannelCall { details = metadataMap['grpc-message']; } metadata.remove('grpc-message'); - this.trace( - 'received status details string "' + details + '" from server' - ); + if (this.traceEnabled) { + this.trace( + 'received status details string "' + details + '" from server' + ); + } } status = { code: receivedStatus, @@ -534,15 +562,19 @@ export class Http2SubchannelCall implements SubchannelCall { } else { code = http2.constants.NGHTTP2_CANCEL; } - this.trace('close http2 stream with code ' + code); + if (this.traceEnabled) { + this.trace('close http2 stream with code ' + code); + } this.http2Stream.close(code); } } cancelWithStatus(status: Status, details: string): void { - this.trace( - 'cancelWithStatus code: ' + status + ' details: "' + details + '"' - ); + if (this.traceEnabled) { + this.trace( + 'cancelWithStatus code: ' + status + ' details: "' + details + '"' + ); + } this.endCall({ code: status, details, metadata: new Metadata() }); } @@ -582,7 +614,9 @@ export class Http2SubchannelCall implements SubchannelCall { } sendMessageWithContext(context: MessageContext, message: Buffer) { - this.trace('write() called with message of length ' + message.length); + if (this.traceEnabled) { + this.trace('write() called with message of length ' + message.length); + } const cb: WriteCallback = (error?: Error | null) => { /* nextTick here ensures that no stream action can be taken in the call * stack of the write callback, in order to hopefully work around @@ -601,7 +635,9 @@ export class Http2SubchannelCall implements SubchannelCall { context.callback?.(); }); }; - this.trace('sending data chunk of length ' + message.length); + if (this.traceEnabled) { + this.trace('sending data chunk of length ' + message.length); + } this.callEventTracker.addMessageSent(); try { this.http2Stream!.write(message, cb); diff --git a/packages/grpc-js/src/subchannel.ts b/packages/grpc-js/src/subchannel.ts index 1156a0c79..8242c1154 100644 --- a/packages/grpc-js/src/subchannel.ts +++ b/packages/grpc-js/src/subchannel.ts @@ -166,10 +166,12 @@ export class Subchannel implements SubchannelInterface { ); this.channelzTrace.addTrace('CT_INFO', 'Subchannel created'); - this.trace( - 'Subchannel constructed with options ' + - JSON.stringify(options, undefined, 2) - ); + if (this.traceEnabled) { + this.trace( + 'Subchannel constructed with options ' + + JSON.stringify(options, undefined, 2) + ); + } this.secureConnector = credentials._createSecureConnector(channelTarget, options); } @@ -183,30 +185,42 @@ export class Subchannel implements SubchannelInterface { }; } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + + private get refTraceEnabled(): boolean { + return logging.isTracerEnabled('subchannel_refcount'); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } private refTrace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - 'subchannel_refcount', - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.refTraceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + 'subchannel_refcount', + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } private handleBackoffTimer() { @@ -306,20 +320,23 @@ export class Subchannel implements SubchannelInterface { if (oldStates.indexOf(this.connectivityState) === -1) { return false; } - if (errorMessage) { - this.trace( - ConnectivityState[this.connectivityState] + - ' -> ' + - ConnectivityState[newState] + - ' with error "' + errorMessage + '"' - ); - - } else { - this.trace( - ConnectivityState[this.connectivityState] + - ' -> ' + - ConnectivityState[newState] - ); + if (this.traceEnabled) { + if (errorMessage) { + this.trace( + ConnectivityState[this.connectivityState] + + ' -> ' + + ConnectivityState[newState] + + ' with error "' + + errorMessage + + '"' + ); + } else { + this.trace( + ConnectivityState[this.connectivityState] + + ' -> ' + + ConnectivityState[newState] + ); + } } if (this.channelzEnabled) { this.channelzTrace.addTrace( @@ -370,12 +387,16 @@ export class Subchannel implements SubchannelInterface { } ref() { - this.refTrace('refcount ' + this.refcount + ' -> ' + (this.refcount + 1)); + if (this.refTraceEnabled) { + this.refTrace('refcount ' + this.refcount + ' -> ' + (this.refcount + 1)); + } this.refcount += 1; } unref() { - this.refTrace('refcount ' + this.refcount + ' -> ' + (this.refcount - 1)); + if (this.refTraceEnabled) { + this.refTrace('refcount ' + this.refcount + ' -> ' + (this.refcount - 1)); + } this.refcount -= 1; if (this.refcount === 0) { this.channelzTrace.addTrace('CT_INFO', 'Shutting down'); @@ -402,7 +423,8 @@ export class Subchannel implements SubchannelInterface { metadata: Metadata, host: string, method: string, - listener: SubchannelCallInterceptingListener + listener: SubchannelCallInterceptingListener, + callId?: number ): SubchannelCall { if (!this.transport) { throw new Error('Cannot create call, subchannel not READY'); @@ -428,7 +450,8 @@ export class Subchannel implements SubchannelInterface { host, method, listener, - statsTracker + statsTracker, + callId ); } diff --git a/packages/grpc-js/src/transport.ts b/packages/grpc-js/src/transport.ts index 8d0554fd3..bc19ca2a6 100644 --- a/packages/grpc-js/src/transport.ts +++ b/packages/grpc-js/src/transport.ts @@ -90,7 +90,8 @@ export interface Transport { host: string, method: string, listener: SubchannelCallInterceptingListener, - subchannelCallStatsTracker: Partial + subchannelCallStatsTracker: Partial, + callId?: number ): SubchannelCall; addDisconnectListener(listener: TransportDisconnectListener): void; shutdown(): void; @@ -214,27 +215,33 @@ class Http2Transport implements Transport { ) { tooManyPings = true; } - this.trace( - 'connection closed by GOAWAY with code ' + - errorCode + - ' and data ' + - opaqueData?.toString() - ); + if (this.traceEnabled) { + this.trace( + 'connection closed by GOAWAY with code ' + + errorCode + + ' and data ' + + opaqueData?.toString() + ); + } this.reportDisconnectToOwner(tooManyPings); } ); session.once('error', error => { - this.trace('connection closed with error ' + (error as Error).message); + if (this.traceEnabled) { + this.trace('connection closed with error ' + (error as Error).message); + } this.handleDisconnect(); }); session.socket.once('close', (hadError) => { - this.trace('connection closed. hadError=' + hadError); + if (this.traceEnabled) { + this.trace('connection closed. hadError=' + hadError); + } this.handleDisconnect(); }); - if (logging.isTracerEnabled(TRACER_NAME)) { + if (this.traceEnabled) { session.on('remoteSettings', (settings: http2.Settings) => { this.trace( 'new settings received' + @@ -325,56 +332,80 @@ class Http2Transport implements Transport { return socketInfo; } + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + + private get keepaliveTraceEnabled(): boolean { + return logging.isTracerEnabled('keepalive'); + } + + private get flowControlTraceEnabled(): boolean { + return logging.isTracerEnabled(FLOW_CONTROL_TRACER_NAME); + } + + private get internalsTraceEnabled(): boolean { + return logging.isTracerEnabled('transport_internals'); + } + private trace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } private keepaliveTrace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - 'keepalive', - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.keepaliveTraceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + 'keepalive', + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } private flowControlTrace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - FLOW_CONTROL_TRACER_NAME, - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.flowControlTraceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + FLOW_CONTROL_TRACER_NAME, + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } private internalsTrace(text: string): void { - logging.trace( - LogVerbosity.DEBUG, - 'transport_internals', - '(' + - this.channelzRef.id + - ') ' + - this.subchannelAddressString + - ' ' + - text - ); + if (this.internalsTraceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + 'transport_internals', + '(' + + this.channelzRef.id + + ') ' + + this.subchannelAddressString + + ' ' + + text + ); + } } /** @@ -433,9 +464,11 @@ class Http2Transport implements Transport { if (this.channelzEnabled) { this.keepalivesSent += 1; } - this.keepaliveTrace( - 'Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms' - ); + if (this.keepaliveTraceEnabled) { + this.keepaliveTrace( + 'Sending ping with timeout ' + this.keepaliveTimeoutMs + 'ms' + ); + } this.keepaliveTimer = setTimeout(() => { this.keepaliveTimer = null; this.keepaliveTrace('Ping timeout passed without response'); @@ -448,7 +481,9 @@ class Http2Transport implements Transport { (err: Error | null, duration: number, payload: Buffer) => { this.clearKeepaliveTimeout(); if (err) { - this.keepaliveTrace('Ping failed with error ' + err.message); + if (this.keepaliveTraceEnabled) { + this.keepaliveTrace('Ping failed with error ' + err.message); + } this.handleDisconnect(); } else { this.keepaliveTrace('Received ping response'); @@ -464,7 +499,9 @@ class Http2Transport implements Transport { pingSendError = (e instanceof Error ? e.message : '') || 'Unknown error'; } if (pingSendError) { - this.keepaliveTrace('Ping send failed: ' + pingSendError); + if (this.keepaliveTraceEnabled) { + this.keepaliveTrace('Ping send failed: ' + pingSendError); + } this.handleDisconnect(); } } @@ -528,7 +565,8 @@ class Http2Transport implements Transport { host: string, method: string, listener: SubchannelCallInterceptingListener, - subchannelCallStatsTracker: Partial + subchannelCallStatsTracker: Partial, + callId?: number ): Http2SubchannelCall { const headers = metadata.toHttp2Headers(); headers[HTTP2_HEADER_AUTHORITY] = host; @@ -552,20 +590,24 @@ class Http2Transport implements Transport { this.handleDisconnect(); throw e; } - this.flowControlTrace( - 'local window size: ' + - this.session.state.localWindowSize + - ' remote window size: ' + - this.session.state.remoteWindowSize - ); - this.internalsTrace( - 'session.closed=' + - this.session.closed + - ' session.destroyed=' + - this.session.destroyed + - ' session.socket.destroyed=' + - this.session.socket.destroyed - ); + if (this.flowControlTraceEnabled) { + this.flowControlTrace( + 'local window size: ' + + this.session.state.localWindowSize + + ' remote window size: ' + + this.session.state.remoteWindowSize + ); + } + if (this.internalsTraceEnabled) { + this.internalsTrace( + 'session.closed=' + + this.session.closed + + ' session.destroyed=' + + this.session.destroyed + + ' session.socket.destroyed=' + + this.session.socket.destroyed + ); + } let eventTracker: CallEventTracker; // eslint-disable-next-line prefer-const let call: Http2SubchannelCall; @@ -617,7 +659,7 @@ class Http2Transport implements Transport { eventTracker, listener, this, - getNextCallNumber() + callId ?? getNextCallNumber() ); this.addActiveCall(call); return call; @@ -659,12 +701,18 @@ export class Http2SubchannelConnector implements SubchannelConnector { private isShutdown = false; constructor(private channelTarget: GrpcUri) {} + private get traceEnabled(): boolean { + return logging.isTracerEnabled(TRACER_NAME); + } + private trace(text: string) { - logging.trace( - LogVerbosity.DEBUG, - TRACER_NAME, - uriToString(this.channelTarget) + ' ' + text - ); + if (this.traceEnabled) { + logging.trace( + LogVerbosity.DEBUG, + TRACER_NAME, + uriToString(this.channelTarget) + ' ' + text + ); + } } private createSession( @@ -706,7 +754,9 @@ export class Http2SubchannelConnector implements SubchannelConnector { const errorHandler = (error: Error) => { this.session?.destroy(); errorMessage = (error as Error).message; - this.trace('connection failed with error ' + errorMessage); + if (this.traceEnabled) { + this.trace('connection failed with error ' + errorMessage); + } if (!reportedError) { reportedError = true; reject(`${errorMessage} (${new Date().toISOString()})`); @@ -801,14 +851,22 @@ export class Http2SubchannelConnector implements SubchannelConnector { let secureConnectResult: SecureConnectResult | null = null; const addressString = subchannelAddressToString(address); try { - this.trace(addressString + ' Waiting for secureConnector to be ready'); + if (this.traceEnabled) { + this.trace(addressString + ' Waiting for secureConnector to be ready'); + } await secureConnector.waitForReady(); - this.trace(addressString + ' secureConnector is ready'); + if (this.traceEnabled) { + this.trace(addressString + ' secureConnector is ready'); + } tcpConnection = await this.tcpConnect(address, options); tcpConnection.setNoDelay(); - this.trace(addressString + ' Established TCP connection'); + if (this.traceEnabled) { + this.trace(addressString + ' Established TCP connection'); + } secureConnectResult = await secureConnector.connect(tcpConnection); - this.trace(addressString + ' Established secure connection'); + if (this.traceEnabled) { + this.trace(addressString + ' Established secure connection'); + } return this.createSession(secureConnectResult, address, options); } catch (e) { tcpConnection?.destroy(); diff --git a/packages/grpc-js/test/test-call-number.ts b/packages/grpc-js/test/test-call-number.ts new file mode 100644 index 000000000..583f2e097 --- /dev/null +++ b/packages/grpc-js/test/test-call-number.ts @@ -0,0 +1,574 @@ +/* + * Copyright 2026 gRPC authors. + * + * 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 + * + * http://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 { execFile } from 'child_process'; +import * as path from 'path'; + +import * as grpc from '../src'; +import { Server, ServerCredentials } from '../src'; +import { getNextCallNumber } from '../src/call-number'; +import { ServiceClientConstructor } from '../src/make-client'; + +import { loadProtoFile } from './common'; + +const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto'); +const EchoService = loadProtoFile(protoFile) + .EchoService as ServiceClientConstructor; + +describe('Call number unification', () => { + let server: Server; + let serverPort: number; + + before(done => { + server = new Server(); + server.addService(EchoService.service, { + echo( + call: grpc.ServerUnaryCall, + callback: grpc.sendUnaryData + ) { + const succeedOnRetryAttempt = call.metadata.get( + 'succeed-on-retry-attempt' + ); + const previousAttempts = call.metadata.get( + 'grpc-previous-rpc-attempts' + ); + if ( + succeedOnRetryAttempt.length === 0 || + (previousAttempts.length > 0 && + previousAttempts[0] === succeedOnRetryAttempt[0]) + ) { + callback(null, call.request); + } else { + callback({ + code: grpc.status.UNAVAILABLE, + details: `Failed on attempt ${previousAttempts[0] ?? 0}`, + }); + } + }, + echoClientStream( + call: grpc.ServerReadableStream, + callback: grpc.sendUnaryData + ) { + let lastMessage: any; + call.on('data', message => { + lastMessage = message; + }); + call.on('end', () => { + callback(null, lastMessage ?? { value: '', value2: 0 }); + }); + }, + echoServerStream(call: grpc.ServerWritableStream) { + call.write(call.request); + call.end(); + }, + echoBidiStream(call: grpc.ServerDuplexStream) { + call.on('data', message => { + call.write(message); + }); + call.on('end', () => { + call.end(); + }); + }, + }); + + server.bindAsync( + 'localhost:0', + ServerCredentials.createInsecure(), + (error, port) => { + assert.ifError(error); + serverPort = port; + done(); + } + ); + }); + + after(done => { + server.tryShutdown(done); + }); + + it('getNextCallNumber increases monotonically', () => { + const initialCallNumber = getNextCallNumber(); + assert.strictEqual(getNextCallNumber(), initialCallNumber + 1); + assert.strictEqual(getNextCallNumber(), initialCallNumber + 2); + }); + + it('allocates exactly one call number per logical RPC', done => { + const client = new EchoService( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + + const callNumberBefore = getNextCallNumber(); + client.echo( + { value: 'call-number-test', value2: 42 }, + (error: grpc.ServiceError | null, response: any) => { + assert.ifError(error); + assert.strictEqual(response.value, 'call-number-test'); + + const callNumberAfter = getNextCallNumber(); + // callNumberBefore was N. + // The RPC allocated exactly 1 call number: N + 1. + // callNumberAfter is therefore N + 2. + assert.strictEqual( + callNumberAfter, + callNumberBefore + 2, + 'Expected single RPC to consume exactly 1 call number' + ); + + // A second sequential RPC should allocate N + 3. + client.echo( + { value: 'second-call', value2: 43 }, + (secondError: grpc.ServiceError | null, secondResponse: any) => { + assert.ifError(secondError); + assert.strictEqual(secondResponse.value, 'second-call'); + + const callNumberFinal = getNextCallNumber(); + // Second RPC consumed N + 3, so callNumberFinal is N + 4. + assert.strictEqual( + callNumberFinal, + callNumberAfter + 2, + 'Expected sequential RPC to consume exactly 1 call number' + ); + client.close(); + done(); + } + ); + } + ); + }); + + it('allocates a new call number for retry attempts', done => { + const serviceConfig = { + methodConfig: [ + { + name: [ + { + service: 'EchoService', + }, + ], + retryPolicy: { + maxAttempts: 3, + initialBackoff: '0.01s', + maxBackoff: '0.1s', + backoffMultiplier: 1.2, + retryableStatusCodes: [grpc.status.UNAVAILABLE], + }, + }, + ], + }; + + const client = new EchoService( + `localhost:${serverPort}`, + grpc.credentials.createInsecure(), + { + 'grpc.service_config': JSON.stringify(serviceConfig), + } + ); + + const callNumberBefore = getNextCallNumber(); + const metadata = new grpc.Metadata(); + metadata.set('succeed-on-retry-attempt', '1'); + + client.echo( + { value: 'retry-test', value2: 100 }, + metadata, + (error: grpc.ServiceError | null, response: any) => { + assert.ifError(error); + assert.strictEqual(response.value, 'retry-test'); + + const callNumberAfter = getNextCallNumber(); + // Initial call allocated 1 call number (attempt 0), + // and retry attempt 1 allocated 1 additional call number. + // So exactly 2 call numbers were allocated for the retried RPC. + assert.strictEqual( + callNumberAfter, + callNumberBefore + 3, + 'Expected retried RPC with 1 retry to consume exactly 2 call numbers' + ); + client.close(); + done(); + } + ); + }); + + it('threads call number across InternalChannel call creation helpers', () => { + const client = new grpc.Client( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + const internalChannel = (client.getChannel() as any).internalChannel; + + const customCallNumber = getNextCallNumber() + 100; + const callConfig = { + methodConfig: { name: [] }, + pickInformation: {}, + status: grpc.status.OK, + dynamicFilterFactories: [], + }; + + // createRetryingCall accepts an optional callNumber + const retryingCallWithId = internalChannel.createRetryingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity, + customCallNumber + ); + assert.strictEqual(retryingCallWithId.getCallNumber(), customCallNumber); + + // createRetryingCall generates a callNumber if omitted + const retryingCallWithoutId = internalChannel.createRetryingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity + ); + assert(typeof retryingCallWithoutId.getCallNumber() === 'number'); + + // createLoadBalancingCall accepts an optional callNumber + const loadBalancingCallWithId = internalChannel.createLoadBalancingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity, + customCallNumber + ); + assert.strictEqual( + loadBalancingCallWithId.getCallNumber(), + customCallNumber + ); + + // createLoadBalancingCall generates a callNumber if omitted + const loadBalancingCallWithoutId = internalChannel.createLoadBalancingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity + ); + assert(typeof loadBalancingCallWithoutId.getCallNumber() === 'number'); + client.close(); + }); + + it('unifies call numbers across trace logs across all layers when tracing is enabled', function (done) { + this.timeout(5000); + const packageDirectory = __dirname.includes('build') + ? path.resolve(__dirname, '../..') + : path.resolve(__dirname, '..'); + const script = ` +const grpc = require('./build/src'); +const path = require('path'); +const protoFile = path.join( + __dirname, 'test', 'fixtures', 'echo_service.proto' +); +const { loadProtoFile } = require('./build/test/common'); +const EchoService = loadProtoFile(protoFile).EchoService; + +const server = new grpc.Server(); +server.addService(EchoService.service, { + echo(call, callback) { callback(null, call.request); } +}); +server.bindAsync( + 'localhost:0', + grpc.ServerCredentials.createInsecure(), + (err, port) => { + const logs = []; + grpc.setLogger({ + error(...args) { logs.push(args.join(' ')); } + }); + grpc.setLogVerbosity(grpc.logVerbosity.DEBUG); + + const client = new EchoService( + 'localhost:' + port, + grpc.credentials.createInsecure() + ); + client.echo({ value: 'hi', value2: 1 }, (err, resp) => { + client.close(); + server.forceShutdown(); + process.stdout.write(JSON.stringify(logs)); + }); + } +); +`; + execFile( + process.execPath, + ['-e', script], + { + cwd: packageDirectory, + env: { + ...process.env, + GRPC_TRACE: 'resolving_call,subchannel_call,load_balancing_call', + }, + }, + (error, stdout) => { + assert.ifError(error); + const logs: string[] = JSON.parse(stdout); + assert(logs.length > 0, 'Expected trace logs to be emitted'); + for (const line of logs) { + assert( + line.includes('[0]'), + `Expected log line to contain [0]: ${line}` + ); + } + done(); + } + ); + }); + + it('allocates exactly one call number per client streaming RPC', done => { + const client = new EchoService( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + + const callNumberBefore = getNextCallNumber(); + const stream = client.echoClientStream( + (error: grpc.ServiceError | null, response: any) => { + assert.ifError(error); + assert.strictEqual(response.value, 'streaming-test'); + const callNumberAfter = getNextCallNumber(); + assert.strictEqual( + callNumberAfter, + callNumberBefore + 2, + 'Expected client streaming RPC to consume exactly 1 call number' + ); + client.close(); + done(); + } + ); + stream.write({ value: 'streaming-test', value2: 1 }); + stream.end(); + }); + + it('allocates exactly one call number per server streaming RPC', done => { + const client = new EchoService( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + + const callNumberBefore = getNextCallNumber(); + const stream = client.echoServerStream({ + value: 'server-streaming-test', + value2: 2, + }); + stream.on('data', (response: any) => { + assert.strictEqual(response.value, 'server-streaming-test'); + }); + stream.on('end', () => { + const callNumberAfter = getNextCallNumber(); + assert.strictEqual( + callNumberAfter, + callNumberBefore + 2, + 'Expected server streaming RPC to consume exactly 1 call number' + ); + client.close(); + done(); + }); + }); + + it('allocates exactly one call number per bidi streaming RPC', done => { + const client = new EchoService( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + + const callNumberBefore = getNextCallNumber(); + const stream = client.echoBidiStream(); + stream.on('data', (response: any) => { + assert.strictEqual(response.value, 'bidi-test'); + stream.end(); + }); + stream.on('end', () => { + const callNumberAfter = getNextCallNumber(); + assert.strictEqual( + callNumberAfter, + callNumberBefore + 2, + 'Expected bidi streaming RPC to consume exactly 1 call number' + ); + client.close(); + done(); + }); + stream.write({ value: 'bidi-test', value2: 3 }); + }); + + it('allocates new call numbers for transparent retries', () => { + const client = new grpc.Client( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + const internalChannel = (client.getChannel() as any).internalChannel; + const initialCallNumber = getNextCallNumber(); + const callConfig = { + methodConfig: { name: [] }, + pickInformation: {}, + status: grpc.status.OK, + dynamicFilterFactories: [], + }; + + const retryingCall = internalChannel.createRetryingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity, + initialCallNumber + ) as any; + + retryingCall.start(new grpc.Metadata(), { + onReceiveMetadata: () => {}, + onReceiveMessage: () => {}, + onReceiveStatus: () => {}, + }); + + // First attempt uses initial call number + assert.strictEqual(retryingCall.underlyingCalls.length, 1); + assert.strictEqual( + retryingCall.underlyingCalls[0].call.getCallNumber(), + initialCallNumber + ); + + // Simulate transparent retry (e.g. REFUSED) + retryingCall.handleChildStatus( + { + code: grpc.status.UNAVAILABLE, + details: 'Stream refused', + metadata: new grpc.Metadata(), + progress: 'REFUSED', + }, + 0 + ); + + // Transparent retry attempt gets a fresh call number + assert.strictEqual(retryingCall.underlyingCalls.length, 2); + assert.strictEqual( + retryingCall.underlyingCalls[1].call.getCallNumber(), + initialCallNumber + 1 + ); + + client.close(); + }); + + it('allocates new call numbers for hedged attempts', () => { + const client = new grpc.Client( + `localhost:${serverPort}`, + grpc.credentials.createInsecure() + ); + const internalChannel = (client.getChannel() as any).internalChannel; + const initialCallNumber = getNextCallNumber(); + const callConfig = { + methodConfig: { + name: [], + hedgingPolicy: { + maxAttempts: 3, + hedgingDelay: '100s', + nonFatalStatusCodes: [grpc.status.UNAVAILABLE], + }, + }, + pickInformation: {}, + status: grpc.status.OK, + dynamicFilterFactories: [], + }; + + const retryingCall = internalChannel.createRetryingCall( + callConfig, + '/EchoService/echo', + 'localhost', + grpc.credentials.createInsecure(), + Infinity, + initialCallNumber + ) as any; + + retryingCall.start(new grpc.Metadata(), { + onReceiveMetadata: () => {}, + onReceiveMessage: () => {}, + onReceiveStatus: () => {}, + }); + + assert.strictEqual(retryingCall.underlyingCalls.length, 1); + assert.strictEqual( + retryingCall.underlyingCalls[0].call.getCallNumber(), + initialCallNumber + ); + + // Trigger hedging attempt + retryingCall.maybeStartHedgingAttempt(); + + assert.strictEqual(retryingCall.underlyingCalls.length, 2); + assert.strictEqual( + retryingCall.underlyingCalls[1].call.getCallNumber(), + initialCallNumber + 1 + ); + + retryingCall.cancelWithStatus(grpc.status.CANCELLED, 'test done'); + client.close(); + }); + + it('propagates call number in SingleSubchannelChannel', async () => { + let capturedCallId: number | undefined; + const mockSubchannel: any = { + getAddress: () => 'localhost:12345', + getConnectivityState: () => 2, // READY + getCallCredentials: () => ({ + generateMetadata: async () => new grpc.Metadata(), + }), + getChannelzRef: () => ({ id: 1, kind: 'subchannel', name: 'subchannel' }), + createCall: ( + metadata: any, + host: string, + method: string, + listener: any, + callId?: number + ) => { + capturedCallId = callId; + return { + getPeer: () => 'localhost:12345', + startRead: () => {}, + sendMessageWithContext: () => {}, + halfClose: () => {}, + cancelWithStatus: () => {}, + getCallNumber: () => callId, + }; + }, + }; + + const SingleSubchannelChannel = + require('../src/single-subchannel-channel').SingleSubchannelChannel; + const singleChannel = new SingleSubchannelChannel( + mockSubchannel, + { scheme: 'dns', path: 'localhost:12345' }, + { 'grpc.enable_channelz': 0 } + ); + + const callNumber = getNextCallNumber(); + // Next call from singleChannel.createCall should receive callNumber + 1 + const call = singleChannel.createCall('/service/method', Infinity); + assert.strictEqual(call.getCallNumber(), callNumber + 1); + + await call.start(new grpc.Metadata(), { + onReceiveMetadata: () => {}, + onReceiveMessage: () => {}, + onReceiveStatus: () => {}, + }); + + // Verify mockSubchannel.createCall received the exact same callNumber + assert.strictEqual(capturedCallId, callNumber + 1); + }); +}); diff --git a/packages/grpc-js/test/test-logging.ts b/packages/grpc-js/test/test-logging.ts index d275158cf..9fe9936ef 100644 --- a/packages/grpc-js/test/test-logging.ts +++ b/packages/grpc-js/test/test-logging.ts @@ -70,4 +70,169 @@ describe('Logging', () => { ['j', 'k'], ]); }); + + describe('isTracerEnabled and trace', () => { + function reloadLoggingWithEnv(env: { + grpcTrace?: string; + grpcNodeTrace?: string; + }): typeof logging { + const originalGrpcTrace = process.env.GRPC_TRACE; + const originalGrpcNodeTrace = process.env.GRPC_NODE_TRACE; + const resolvedPath = require.resolve('../src/logging'); + const originalModule = require.cache[resolvedPath]; + + try { + if (env.grpcTrace !== undefined) { + process.env.GRPC_TRACE = env.grpcTrace; + } else { + delete process.env.GRPC_TRACE; + } + if (env.grpcNodeTrace !== undefined) { + process.env.GRPC_NODE_TRACE = env.grpcNodeTrace; + } else { + delete process.env.GRPC_NODE_TRACE; + } + delete require.cache[resolvedPath]; + return require('../src/logging'); + } finally { + if (originalGrpcTrace !== undefined) { + process.env.GRPC_TRACE = originalGrpcTrace; + } else { + delete process.env.GRPC_TRACE; + } + if (originalGrpcNodeTrace !== undefined) { + process.env.GRPC_NODE_TRACE = originalGrpcNodeTrace; + } else { + delete process.env.GRPC_NODE_TRACE; + } + if (originalModule) { + require.cache[resolvedPath] = originalModule; + } else { + delete require.cache[resolvedPath]; + } + } + } + + it('returns false when no tracers are configured', () => { + const reloaded = reloadLoggingWithEnv({}); + assert.strictEqual(reloaded.isTracerEnabled('channel'), false); + assert.strictEqual(reloaded.isTracerEnabled('subchannel'), false); + assert.strictEqual(reloaded.isTracerEnabled('all'), false); + }); + + it('enables specified comma-separated tracers and ignores empty entries', () => { + const reloaded = reloadLoggingWithEnv({ + grpcTrace: ',channel,,subchannel,', + }); + assert.strictEqual(reloaded.isTracerEnabled('channel'), true); + assert.strictEqual(reloaded.isTracerEnabled('subchannel'), true); + assert.strictEqual(reloaded.isTracerEnabled('transport'), false); + }); + + it('supports all and disabled tracers', () => { + const reloaded = reloadLoggingWithEnv({ + grpcTrace: 'all,-channel', + }); + assert.strictEqual(reloaded.isTracerEnabled('channel'), false); + assert.strictEqual(reloaded.isTracerEnabled('subchannel'), true); + assert.strictEqual(reloaded.isTracerEnabled('other_tracer'), true); + }); + + it('prefers GRPC_NODE_TRACE over GRPC_TRACE', () => { + const reloaded = reloadLoggingWithEnv({ + grpcTrace: 'channel', + grpcNodeTrace: 'subchannel', + }); + assert.strictEqual(reloaded.isTracerEnabled('subchannel'), true); + assert.strictEqual(reloaded.isTracerEnabled('channel'), false); + }); + + it('does not log when tracer is disabled', () => { + const output: Array = []; + const logger: Partial = { + error(...args: string[]): void { + output.push(args); + }, + }; + const reloaded = reloadLoggingWithEnv({}); + reloaded.setLogger(logger); + reloaded.setLoggerVerbosity(grpc.logVerbosity.DEBUG); + + reloaded.trace(grpc.logVerbosity.DEBUG, 'channel', 'test message'); + assert.strictEqual(output.length, 0); + }); + + it('logs formatted message when tracer is enabled', () => { + const output: Array = []; + const logger: Partial = { + error(...args: string[]): void { + output.push(args); + }, + }; + const reloaded = reloadLoggingWithEnv({ + grpcTrace: 'test_tracer', + }); + reloaded.setLogger(logger); + reloaded.setLoggerVerbosity(grpc.logVerbosity.DEBUG); + reloaded.trace( + grpc.logVerbosity.DEBUG, + 'test_tracer', + 'hello trace message' + ); + assert.strictEqual(output.length, 1); + const loggedLine = String(output[0]); + assert(loggedLine.includes('test_tracer')); + assert(loggedLine.includes('hello trace message')); + }); + + it('correctly enables specific built-in tracers', () => { + const reloaded = reloadLoggingWithEnv({ + grpcTrace: + 'channel_stacktrace,subchannel_refcount,transport_internals,keepalive,transport_flowctrl,backoff', + }); + assert.strictEqual(reloaded.isTracerEnabled('channel_stacktrace'), true); + assert.strictEqual(reloaded.isTracerEnabled('subchannel_refcount'), true); + assert.strictEqual(reloaded.isTracerEnabled('transport_internals'), true); + assert.strictEqual(reloaded.isTracerEnabled('keepalive'), true); + assert.strictEqual(reloaded.isTracerEnabled('transport_flowctrl'), true); + assert.strictEqual(reloaded.isTracerEnabled('backoff'), true); + assert.strictEqual(reloaded.isTracerEnabled('unrelated_tracer'), false); + }); + + it('does not evaluate expensive trace arguments when tracing is disabled (negative proof)', () => { + const circularOptions: Record = {}; + circularOptions.self = circularOptions; + + // When channel tracer is disabled, creating the client succeeds without triggering JSON.stringify + assert.doesNotThrow(() => { + const client = new grpc.Client( + 'localhost:1234', + grpc.credentials.createInsecure(), + circularOptions + ); + client.close(); + }); + + // A deadline with a throwing toISOString method will throw if deadlineToString is evaluated + const throwingDeadline = new Date(Date.now() + 10000); + throwingDeadline.toISOString = () => { + throw new Error('deadlineToString should not be evaluated'); + }; + const client = new grpc.Client( + 'localhost:1234', + grpc.credentials.createInsecure() + ); + const internalChannel = (client.getChannel() as any).internalChannel; + assert.doesNotThrow(() => { + internalChannel.createResolvingCall( + '/Service/Method', + throwingDeadline as any, + 'localhost', + null, + null + ); + }); + client.close(); + }); + }); }); From 26a8d3d7d919098ded374b1c246d113161d69eb1 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 2/3] 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 51878e76f..a9cbe08d8 100644 --- a/packages/grpc-js/src/subchannel-call.ts +++ b/packages/grpc-js/src/subchannel-call.ts @@ -549,9 +549,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 @@ -652,6 +663,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 526702d8ed335a99960878aea19cf6748f6d2314 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 3/3] 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,