diff --git a/packages/grpc-js-xds/src/http-filter/fault-injection-filter.ts b/packages/grpc-js-xds/src/http-filter/fault-injection-filter.ts index 2d251dc40..139465749 100644 --- a/packages/grpc-js-xds/src/http-filter/fault-injection-filter.ts +++ b/packages/grpc-js-xds/src/http-filter/fault-injection-filter.ts @@ -235,8 +235,7 @@ class FaultInjectionFilter extends BaseFilter implements Filter { super(); } - async sendMetadata(metadataPromise: Promise): Promise { - const metadata = await metadataPromise; + async sendMetadata(metadata: Metadata): Promise { // Handle delay if (totalActiveFaults < this.config.maxActiveFaults && this.config.delay) { let duration = 0; diff --git a/packages/grpc-js/src/compression-filter.ts b/packages/grpc-js/src/compression-filter.ts index e4428a1fb..0fe28a1de 100644 --- a/packages/grpc-js/src/compression-filter.ts +++ b/packages/grpc-js/src/compression-filter.ts @@ -21,8 +21,13 @@ import { WriteObject, WriteFlags } from './call-interface'; import { Channel } from './channel'; import { ChannelOptions } from './channel-options'; import { CompressionAlgorithms } from './compression-algorithms'; -import { DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, DEFAULT_MAX_SEND_MESSAGE_LENGTH, LogVerbosity, Status } from './constants'; -import { BaseFilter, Filter, FilterFactory } from './filter'; +import { + DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, + DEFAULT_MAX_SEND_MESSAGE_LENGTH, + LogVerbosity, + Status, +} from './constants'; +import { BaseFilter, Filter, FilterFactory, isThenable } from './filter'; import * as logging from './logging'; import { Metadata, MetadataValue } from './metadata'; @@ -43,34 +48,46 @@ type SharedCompressionFilterConfig = { abstract class CompressionHandler { protected abstract compressMessage(message: Buffer): Promise; protected abstract decompressMessage(data: Buffer): Promise; + /** * @param message Raw uncompressed message bytes * @param compress Indicates whether the message should be compressed * @return Framed message, compressed if applicable */ - async writeMessage(message: Buffer, compress: boolean): Promise { - let messageBuffer = message; - if (compress) { - messageBuffer = await this.compressMessage(messageBuffer); + writeMessage(message: Buffer, compress: boolean): Buffer | Promise { + if (!compress) { + return this.frameMessage(message, false); } - const output = Buffer.allocUnsafe(messageBuffer.length + 5); - output.writeUInt8(compress ? 1 : 0, 0); - output.writeUInt32BE(messageBuffer.length, 1); - messageBuffer.copy(output, 5); - return output; + return this.compressMessage(message).then(messageBuffer => + this.frameMessage(messageBuffer, true) + ); } /** * @param data Framed message, possibly compressed * @return Uncompressed message */ - async readMessage(data: Buffer): Promise { + readMessage(data: Buffer): Buffer | Promise { const compressed = data.readUInt8(0) === 1; - let messageBuffer: Buffer = data.slice(5); + const messageBuffer: Buffer = data.slice(5); if (compressed) { - messageBuffer = await this.decompressMessage(messageBuffer); + return this.decompressMessage(messageBuffer); } return messageBuffer; } + + /** + * Frames a message by prepending the 1-byte compression flag and 4-byte big-endian length prefix. + * @param message Message payload bytes + * @param compressed Indicates whether the message payload is compressed + * @return 5-byte framed message buffer + */ + protected frameMessage(message: Buffer, compressed: boolean): Buffer { + const output = Buffer.allocUnsafe(message.length + 5); + output.writeUInt8(compressed ? 1 : 0, 0); + output.writeUInt32BE(message.length, 1); + message.copy(output, 5); + return output; + } } class IdentityHandler extends CompressionHandler { @@ -78,14 +95,10 @@ class IdentityHandler extends CompressionHandler { return message; } - async writeMessage(message: Buffer, compress: boolean): Promise { - const output = Buffer.allocUnsafe(message.length + 5); + writeMessage(message: Buffer, compress: boolean): Buffer { /* With "identity" compression, messages should always be marked as * uncompressed */ - output.writeUInt8(0, 0); - output.writeUInt32BE(message.length, 1); - message.copy(output, 5); - return output; + return this.frameMessage(message, false); } decompressMessage(message: Buffer): Promise { @@ -122,11 +135,14 @@ class DeflateHandler extends CompressionHandler { decompresser.on('data', (chunk: Buffer) => { messageParts.push(chunk); totalLength += chunk.byteLength; - if (this.maxRecvMessageLength !== -1 && totalLength > this.maxRecvMessageLength) { + if ( + this.maxRecvMessageLength !== -1 && + totalLength > this.maxRecvMessageLength + ) { decompresser.destroy(); reject({ code: Status.RESOURCE_EXHAUSTED, - details: `Received message that decompresses to a size larger than ${this.maxRecvMessageLength}` + details: `Received message that decompresses to a size larger than ${this.maxRecvMessageLength}`, }); } }); @@ -164,11 +180,14 @@ class GzipHandler extends CompressionHandler { decompresser.on('data', (chunk: Buffer) => { messageParts.push(chunk); totalLength += chunk.byteLength; - if (this.maxRecvMessageLength !== -1 && totalLength > this.maxRecvMessageLength) { + if ( + this.maxRecvMessageLength !== -1 && + totalLength > this.maxRecvMessageLength + ) { decompresser.destroy(); reject({ code: Status.RESOURCE_EXHAUSTED, - details: `Received message that decompresses to a size larger than ${this.maxRecvMessageLength}` + details: `Received message that decompresses to a size larger than ${this.maxRecvMessageLength}`, }); } }); @@ -201,10 +220,19 @@ class UnknownHandler extends CompressionHandler { } } -function getCompressionHandler(compressionName: string, maxReceiveMessageSize: number): CompressionHandler { +/** + * IdentityHandler is completely stateless and immutable, so a single instance + * can safely be shared across all channels, filters, and concurrent calls. + */ +const IDENTITY_HANDLER = new IdentityHandler(); + +function getCompressionHandler( + compressionName: string, + maxReceiveMessageSize: number +): CompressionHandler { switch (compressionName) { case 'identity': - return new IdentityHandler(); + return IDENTITY_HANDLER; case 'deflate': return new DeflateHandler(maxReceiveMessageSize); case 'gzip': @@ -215,8 +243,8 @@ function getCompressionHandler(compressionName: string, maxReceiveMessageSize: n } export class CompressionFilter extends BaseFilter implements Filter { - private sendCompression: CompressionHandler = new IdentityHandler(); - private receiveCompression: CompressionHandler = new IdentityHandler(); + private sendCompression: CompressionHandler = IDENTITY_HANDLER; + private receiveCompression: CompressionHandler = IDENTITY_HANDLER; private currentCompressionAlgorithm: CompressionAlgorithm = 'identity'; private maxReceiveMessageLength: number; private maxSendMessageLength: number; @@ -229,8 +257,12 @@ export class CompressionFilter extends BaseFilter implements Filter { const compressionAlgorithmKey = channelOptions['grpc.default_compression_algorithm']; - this.maxReceiveMessageLength = channelOptions['grpc.max_receive_message_length'] ?? DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH; - this.maxSendMessageLength = channelOptions['grpc.max_send_message_length'] ?? DEFAULT_MAX_SEND_MESSAGE_LENGTH; + this.maxReceiveMessageLength = + channelOptions['grpc.max_receive_message_length'] ?? + DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH; + this.maxSendMessageLength = + channelOptions['grpc.max_send_message_length'] ?? + DEFAULT_MAX_SEND_MESSAGE_LENGTH; if (compressionAlgorithmKey !== undefined) { if (isCompressionAlgorithmKey(compressionAlgorithmKey)) { const clientSelectedEncoding = CompressionAlgorithms[ @@ -264,8 +296,7 @@ export class CompressionFilter extends BaseFilter implements Filter { } } - async sendMetadata(metadata: Promise): Promise { - const headers: Metadata = await metadata; + sendMetadata(headers: Metadata): Metadata { headers.set('grpc-accept-encoding', 'identity,deflate,gzip'); headers.set('accept-encoding', 'identity'); @@ -284,7 +315,10 @@ export class CompressionFilter extends BaseFilter implements Filter { if (receiveEncoding.length > 0) { const encoding: MetadataValue = receiveEncoding[0]; if (typeof encoding === 'string') { - this.receiveCompression = getCompressionHandler(encoding, this.maxReceiveMessageLength); + this.receiveCompression = getCompressionHandler( + encoding, + this.maxReceiveMessageLength + ); } } metadata.remove('grpc-encoding'); @@ -303,7 +337,7 @@ export class CompressionFilter extends BaseFilter implements Filter { if ( !serverSupportedEncodings.includes(this.currentCompressionAlgorithm) ) { - this.sendCompression = new IdentityHandler(); + this.sendCompression = IDENTITY_HANDLER; this.currentCompressionAlgorithm = 'identity'; } } @@ -311,16 +345,20 @@ export class CompressionFilter extends BaseFilter implements Filter { return metadata; } - async sendMessage(message: Promise): Promise { + sendMessage( + resolvedMessage: WriteObject + ): WriteObject | Promise { /* This filter is special. The input message is the bare message bytes, * and the output is a framed and possibly compressed message. For this * reason, this filter should be at the bottom of the filter stack */ - const resolvedMessage: WriteObject = await message; - if (this.maxSendMessageLength !== -1 && resolvedMessage.message.length > this.maxSendMessageLength) { - throw { + if ( + this.maxSendMessageLength !== -1 && + resolvedMessage.message.length > this.maxSendMessageLength + ) { + return Promise.reject({ code: Status.RESOURCE_EXHAUSTED, - details: `Attempted to send message with a size larger than ${this.maxSendMessageLength}` - }; + details: `Attempted to send message with a size larger than ${this.maxSendMessageLength}`, + }); } let compress: boolean; if (this.sendCompression instanceof IdentityHandler) { @@ -329,21 +367,28 @@ export class CompressionFilter extends BaseFilter implements Filter { compress = ((resolvedMessage.flags ?? 0) & WriteFlags.NoCompress) === 0; } + const writeResult = this.sendCompression.writeMessage( + resolvedMessage.message, + compress + ); + if (isThenable(writeResult)) { + return writeResult.then(message => ({ + message, + flags: resolvedMessage.flags, + })); + } return { - message: await this.sendCompression.writeMessage( - resolvedMessage.message, - compress - ), + message: writeResult, flags: resolvedMessage.flags, }; } - async receiveMessage(message: Promise) { + receiveMessage(message: Buffer): Buffer | Promise { /* This filter is also special. The input message is framed and possibly * compressed, and the output message is deframed and uncompressed. So * this is another reason that this filter should be at the bottom of the * filter stack. */ - return this.receiveCompression.readMessage(await message); + return this.receiveCompression.readMessage(message); } } diff --git a/packages/grpc-js/src/filter-stack.ts b/packages/grpc-js/src/filter-stack.ts index 910f5aa36..c9177cd4e 100644 --- a/packages/grpc-js/src/filter-stack.ts +++ b/packages/grpc-js/src/filter-stack.ts @@ -16,17 +16,25 @@ */ import { StatusObject, WriteObject } from './call-interface'; -import { Filter, FilterFactory } from './filter'; +import { Filter, FilterFactory, isThenable } from './filter'; import { Metadata } from './metadata'; +/** + * Runs a list of filters in order. Each filter is called synchronously as long + * as every previous filter returned its result synchronously. Once a filter + * returns a thenable, the remaining filters are chained onto it. + */ export class FilterStack implements Filter { constructor(private readonly filters: Filter[]) {} - sendMetadata(metadata: Promise): Promise { - let result: Promise = metadata; + sendMetadata(metadata: Metadata): Metadata | PromiseLike { + let result: Metadata | PromiseLike = metadata; for (let i = 0; i < this.filters.length; i++) { - result = this.filters[i].sendMetadata(result); + const filter = this.filters[i]; + result = isThenable(result) + ? result.then(resolvedMetadata => filter.sendMetadata(resolvedMetadata)) + : filter.sendMetadata(result); } return result; @@ -42,21 +50,27 @@ export class FilterStack implements Filter { return result; } - sendMessage(message: Promise): Promise { - let result: Promise = message; + sendMessage(message: WriteObject): WriteObject | PromiseLike { + let result: WriteObject | PromiseLike = message; for (let i = 0; i < this.filters.length; i++) { - result = this.filters[i].sendMessage(result); + const filter = this.filters[i]; + result = isThenable(result) + ? result.then(resolvedMessage => filter.sendMessage(resolvedMessage)) + : filter.sendMessage(result); } return result; } - receiveMessage(message: Promise): Promise { - let result: Promise = message; + receiveMessage(message: Buffer): Buffer | PromiseLike { + let result: Buffer | PromiseLike = message; for (let i = this.filters.length - 1; i >= 0; i--) { - result = this.filters[i].receiveMessage(result); + const filter = this.filters[i]; + result = isThenable(result) + ? result.then(resolvedMessage => filter.receiveMessage(resolvedMessage)) + : filter.receiveMessage(result); } return result; diff --git a/packages/grpc-js/src/filter.ts b/packages/grpc-js/src/filter.ts index 5313f91a8..62f16aaf9 100644 --- a/packages/grpc-js/src/filter.ts +++ b/packages/grpc-js/src/filter.ts @@ -20,24 +20,26 @@ import { Metadata } from './metadata'; /** * Filter classes represent related per-call logic and state that is primarily - * used to modify incoming and outgoing data. All async filters can be - * rejected. The rejection error must be a StatusObject, and a rejection will - * cause the call to end with that status. + * used to modify incoming and outgoing data. The sendMetadata, sendMessage, + * and receiveMessage methods may return their result directly when no + * asynchronous work is needed, or a thenable otherwise. All async filters can + * be rejected. The rejection error must be a StatusObject, and a rejection + * will cause the call to end with that status. */ export interface Filter { - sendMetadata(metadata: Promise): Promise; + sendMetadata(metadata: Metadata): Metadata | PromiseLike; receiveMetadata(metadata: Metadata): Metadata; - sendMessage(message: Promise): Promise; + sendMessage(message: WriteObject): WriteObject | PromiseLike; - receiveMessage(message: Promise): Promise; + receiveMessage(message: Buffer): Buffer | PromiseLike; receiveTrailers(status: StatusObject): StatusObject; } export abstract class BaseFilter implements Filter { - async sendMetadata(metadata: Promise): Promise { + sendMetadata(metadata: Metadata): Metadata | PromiseLike { return metadata; } @@ -45,11 +47,11 @@ export abstract class BaseFilter implements Filter { return metadata; } - async sendMessage(message: Promise): Promise { + sendMessage(message: WriteObject): WriteObject | PromiseLike { return message; } - async receiveMessage(message: Promise): Promise { + receiveMessage(message: Buffer): Buffer | PromiseLike { return message; } @@ -61,3 +63,15 @@ export abstract class BaseFilter implements Filter { export interface FilterFactory { createFilter(): T; } + +/** + * Checks whether a filter result is a thenable that must be awaited, rather + * than a value that is already available. This is used instead of + * `instanceof Promise` so that promises from other realms and custom + * thenables are also recognized. + */ +export function isThenable( + value: T | PromiseLike +): value is PromiseLike { + return typeof (value as PromiseLike)?.then === 'function'; +} diff --git a/packages/grpc-js/src/load-balancing-call.ts b/packages/grpc-js/src/load-balancing-call.ts index 75b1d3e45..ca8f13db3 100644 --- a/packages/grpc-js/src/load-balancing-call.ts +++ b/packages/grpc-js/src/load-balancing-call.ts @@ -281,9 +281,11 @@ export class LoadBalancingCall implements Call, DeadlineInfoProvider { this.child.startRead(); } if (this.pendingMessage) { + const pendingMessage = this.pendingMessage; + this.pendingMessage = null; this.child.sendMessageWithContext( - this.pendingMessage.context, - this.pendingMessage.message + pendingMessage.context, + pendingMessage.message ); } if (this.pendingHalfClose) { diff --git a/packages/grpc-js/src/resolving-call.ts b/packages/grpc-js/src/resolving-call.ts index 2c507280d..e252eb3f2 100644 --- a/packages/grpc-js/src/resolving-call.ts +++ b/packages/grpc-js/src/resolving-call.ts @@ -23,6 +23,7 @@ import { InterceptingListener, MessageContext, StatusObject, + WriteObject, } from './call-interface'; import { LogVerbosity, Propagate, Status } from './constants'; import { @@ -32,12 +33,15 @@ import { getRelativeTimeout, minDeadline, } from './deadline'; +import { getErrorMessage } from './error'; +import { isThenable } from './filter'; import { FilterStack, FilterStackFactory } from './filter-stack'; import { InternalChannel } from './internal-channel'; import { Metadata } from './metadata'; import * as logging from './logging'; import { restrictControlPlaneStatusCode } from './control-plane-status'; import { AuthContext } from './auth-context'; +import { CallConfig } from './resolver'; const TRACER_NAME = 'resolving_call'; @@ -213,21 +217,141 @@ export class ResolvingCall implements Call { throw new Error('sendMessageonChild called with child not populated'); } const child = this.child; - this.writeFilterPending = true; - this.filterStack!.sendMessage( - Promise.resolve({ message: message, flags: context.flags }) - ).then( - filteredMessage => { - this.writeFilterPending = false; - child.sendMessageWithContext(context, filteredMessage.message); - if (this.pendingHalfClose) { - child.halfClose(); + let filterResult: WriteObject | PromiseLike; + try { + // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. + filterResult = this.filterStack!.sendMessage({ + message: message, + flags: context.flags, + }); + } catch (error) { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to filter outgoing message: ${getErrorMessage(error)}` + ); + return; + } + if (isThenable(filterResult)) { + // Slow path: asynchronous filter. Pause new writes and forward when resolved. + this.writeFilterPending = true; + filterResult.then( + filteredMessage => { + this.writeFilterPending = false; + if (this.ended) { + return; + } + child.sendMessageWithContext(context, filteredMessage.message); + if (this.pendingHalfClose) { + child.halfClose(); + } + }, + (status: StatusObject) => { + this.cancelWithStatus( + typeof status.code === 'number' ? status.code : Status.INTERNAL, + typeof status.details === 'string' + ? status.details + : `Failed to filter outgoing message: ${getErrorMessage(status)}` + ); } - }, - (status: StatusObject) => { - this.cancelWithStatus(status.code, status.details); + ); + } else { + // Fast path: synchronous filter completion. Forward immediately. + child.sendMessageWithContext(context, filterResult.message); + if (this.pendingHalfClose) { + child.halfClose(); } + } + } + + private handleChildReceiveMessage(message: Buffer): void { + this.trace('Received message'); + let filterResult: Buffer | PromiseLike; + try { + // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. + filterResult = this.filterStack!.receiveMessage(message); + } catch (error) { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to filter incoming message: ${getErrorMessage(error)}` + ); + return; + } + if (isThenable(filterResult)) { + // Slow path: asynchronous filter. Pause reading and forward when resolved. + this.readFilterPending = true; + filterResult.then( + filteredMessage => { + this.trace('Finished filtering received message'); + this.readFilterPending = false; + if (this.ended) { + return; + } + this.listener!.onReceiveMessage(filteredMessage); + if (this.pendingChildStatus) { + this.outputStatus(this.pendingChildStatus); + } + }, + (status: StatusObject) => { + this.cancelWithStatus( + typeof status.code === 'number' ? status.code : Status.INTERNAL, + typeof status.details === 'string' + ? status.details + : `Failed to filter incoming message: ${getErrorMessage(status)}` + ); + } + ); + } else { + // Fast path: synchronous filter completion. Forward immediately. + this.trace('Finished filtering received message'); + this.listener!.onReceiveMessage(filterResult); + } + } + + private startChild(config: CallConfig, filteredMetadata: Metadata): void { + if (this.ended) { + return; + } + this.child = this.channel.createRetryingCall( + config, + this.method, + this.host, + this.credentials, + this.deadline, + this.callNumber ); + if (this.traceEnabled) { + this.trace('Created child [' + this.child.getCallNumber() + ']'); + } + this.childStartTime = Date.now(); + this.child.start(filteredMetadata, { + onReceiveMetadata: metadata => { + this.trace('Received metadata'); + this.listener!.onReceiveMetadata( + this.filterStack!.receiveMetadata(metadata) + ); + }, + onReceiveMessage: message => { + this.handleChildReceiveMessage(message); + }, + onReceiveStatus: status => { + this.trace('Received status'); + if (this.readFilterPending) { + this.pendingChildStatus = status; + } else { + this.outputStatus(status); + } + }, + }); + if (this.readPending) { + this.child.startRead(); + } + if (this.pendingMessage) { + const pendingMessage = this.pendingMessage; + this.pendingMessage = null; + this.sendMessageOnChild(pendingMessage.context, pendingMessage.message); + } else if (this.pendingHalfClose) { + this.child.halfClose(); + } } getConfig(): void { @@ -275,71 +399,40 @@ export class ResolvingCall implements Call { this.runDeadlineTimer(); } + /* Create the filterStack before entering the try block so that: + * 1. A synchronous throw in createFilter() does not enter catch -> + * cancelWithStatus() -> outputStatus(), which would otherwise call + * createFilter() a second time while this.filterStack is still null. + * 2. V8 does not allocate an extra context object per call in getConfig(). */ this.filterStackFactory.push(config.dynamicFilterFactories); this.filterStack = this.filterStackFactory.createFilter(); - this.filterStack.sendMetadata(Promise.resolve(this.metadata)).then( - filteredMetadata => { - this.child = this.channel.createRetryingCall( - config, - this.method, - this.host, - this.credentials, - this.deadline, - this.callNumber - ); - if (this.traceEnabled) { - this.trace('Created child [' + this.child.getCallNumber() + ']'); - } - this.childStartTime = Date.now(); - this.child.start(filteredMetadata, { - onReceiveMetadata: metadata => { - this.trace('Received metadata'); - this.listener!.onReceiveMetadata( - this.filterStack!.receiveMetadata(metadata) - ); - }, - onReceiveMessage: message => { - this.trace('Received message'); - this.readFilterPending = true; - this.filterStack!.receiveMessage(message).then( - filteredMesssage => { - this.trace('Finished filtering received message'); - this.readFilterPending = false; - this.listener!.onReceiveMessage(filteredMesssage); - if (this.pendingChildStatus) { - this.outputStatus(this.pendingChildStatus); - } - }, - (status: StatusObject) => { - this.cancelWithStatus(status.code, status.details); - } - ); + try { + const filterResult = this.filterStack.sendMetadata(this.metadata); + if (isThenable(filterResult)) { + filterResult.then( + filteredMetadata => { + this.startChild(config, filteredMetadata); }, - onReceiveStatus: status => { - this.trace('Received status'); - if (this.readFilterPending) { - this.pendingChildStatus = status; - } else { + (status: StatusObject) => { + if (typeof status.code === 'number') { this.outputStatus(status); + } else { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to start call: ${getErrorMessage(status)}` + ); } - }, - }); - if (this.readPending) { - this.child.startRead(); - } - if (this.pendingMessage) { - this.sendMessageOnChild( - this.pendingMessage.context, - this.pendingMessage.message - ); - } else if (this.pendingHalfClose) { - this.child.halfClose(); - } - }, - (status: StatusObject) => { - this.outputStatus(status); + } + ); + } else { + this.startChild(config, filterResult); } - ); + } catch (error) { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to start call: ${getErrorMessage(error)}` + ); + } } reportResolverError(status: StatusObject) { @@ -363,6 +456,11 @@ export class ResolvingCall implements Call { }); } getPeer(): string { + /* While waiting for channel config resolution or an async metadata filter, + * this.child is null and getPeer() falls back to the channel target. Once + * this.child (RetryingCall) is created, it delegates to child.getPeer(), + * which returns 'unknown' until a call attempt commits and then returns the + * connected subchannel peer address. */ return this.child?.getPeer() ?? this.channel.getTarget(); } start(metadata: Metadata, listener: InterceptingListener): void { diff --git a/packages/grpc-js/src/single-subchannel-channel.ts b/packages/grpc-js/src/single-subchannel-channel.ts index df6016b0c..5b33d5a60 100644 --- a/packages/grpc-js/src/single-subchannel-channel.ts +++ b/packages/grpc-js/src/single-subchannel-channel.ts @@ -88,7 +88,7 @@ class SubchannelCallWrapper implements Call { }); return; } - const filteredMetadata = await this.filterStack.sendMetadata(Promise.resolve(metadata)); + const filteredMetadata = await this.filterStack.sendMetadata(metadata); let credsMetadata: Metadata; try { credsMetadata = await this.subchannel.getCallCredentials() @@ -150,7 +150,7 @@ class SubchannelCallWrapper implements Call { } async sendMessageWithContext(context: MessageContext, message: Buffer): Promise { this.writeFilterPending = true; - const filteredMessage = await this.filterStack.sendMessage(Promise.resolve({message: message, flags: context.flags})); + const filteredMessage = await this.filterStack.sendMessage({message: message, flags: context.flags}); this.writeFilterPending = false; if (this.childCall) { this.childCall.sendMessageWithContext(context, filteredMessage.message); diff --git a/packages/grpc-js/test/test-filter-stack.ts b/packages/grpc-js/test/test-filter-stack.ts new file mode 100644 index 000000000..3dd40b801 --- /dev/null +++ b/packages/grpc-js/test/test-filter-stack.ts @@ -0,0 +1,1762 @@ +/* + * 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 * as zlib from 'zlib'; + +import { CallCredentials } from '../src/call-credentials'; +import { + InterceptingListener, + StatusObject, + WriteFlags, + WriteObject, +} from '../src/call-interface'; +import { Channel } from '../src/channel'; +import { CompressionAlgorithms } from '../src/compression-algorithms'; +import { + CompressionFilter, + CompressionFilterFactory, +} from '../src/compression-filter'; +import { ConnectivityState } from '../src/connectivity-state'; +import { Status } from '../src/constants'; +import { BaseFilter, Filter, FilterFactory, isThenable } from '../src/filter'; +import { FilterStack, FilterStackFactory } from '../src/filter-stack'; +import { InternalChannel } from '../src/internal-channel'; +import { LoadBalancingCall } from '../src/load-balancing-call'; +import { Metadata } from '../src/metadata'; +import { PickResultType } from '../src/picker'; +import { ResolvingCall } from '../src/resolving-call'; + +class NoopPassThroughFilter extends BaseFilter implements Filter {} + +class NoopPassThroughFilterFactory + implements FilterFactory +{ + createFilter(): NoopPassThroughFilter { + return new NoopPassThroughFilter(); + } +} + +class AsyncHeaderFilter extends BaseFilter implements Filter { + constructor( + private readonly headerKey: string, + private readonly headerValue: string + ) { + super(); + } + + async sendMetadata(metadata: Metadata): Promise { + await Promise.resolve(); + metadata.set(this.headerKey, this.headerValue); + return metadata; + } + + async sendMessage(message: WriteObject): Promise { + await Promise.resolve(); + return { + message: Buffer.concat([Buffer.from('prefix:'), message.message]), + flags: message.flags, + }; + } + + async receiveMessage(message: Buffer): Promise { + await Promise.resolve(); + return Buffer.concat([message, Buffer.from(':suffix')]); + } +} + +class AsyncHeaderFilterFactory implements FilterFactory { + constructor( + private readonly headerKey: string, + private readonly headerValue: string + ) {} + + createFilter(): AsyncHeaderFilter { + return new AsyncHeaderFilter(this.headerKey, this.headerValue); + } +} + +/** + * Returns a minimal thenable that resolves to the given value and is not an + * instance of Promise, to simulate promises from other realms or custom + * promise implementations. + */ +function createCustomThenable(value: T): PromiseLike { + const promise = Promise.resolve(value); + return { + then( + onFulfilled?: ((value: T) => TResult1 | PromiseLike) | null, + onRejected?: + | ((reason: unknown) => TResult2 | PromiseLike) + | null + ): PromiseLike { + return promise.then(onFulfilled, onRejected); + }, + }; +} + +class ThenableFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Metadata): PromiseLike { + metadata.set('x-thenable', 'true'); + return createCustomThenable(metadata); + } + + sendMessage(message: WriteObject): PromiseLike { + return createCustomThenable({ + message: Buffer.concat([Buffer.from('thenable:'), message.message]), + flags: message.flags, + }); + } + + receiveMessage(message: Buffer): PromiseLike { + return createCustomThenable( + Buffer.concat([message, Buffer.from(':thenable')]) + ); + } +} + +class ThenableFilterFactory implements FilterFactory { + createFilter(): ThenableFilter { + return new ThenableFilter(); + } +} + +class SyncFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Metadata): Metadata { + metadata.set('x-sync', 'true'); + return metadata; + } + + sendMessage(message: WriteObject): WriteObject { + return { + message: Buffer.concat([Buffer.from('sync:'), message.message]), + flags: message.flags, + }; + } + + receiveMessage(message: Buffer): Buffer { + return Buffer.concat([message, Buffer.from(':sync')]); + } +} + +class SyncFilterFactory implements FilterFactory { + createFilter(): SyncFilter { + return new SyncFilter(); + } +} + +class AsyncMetadataOnlyFilter extends BaseFilter implements Filter { + async sendMetadata(metadata: Metadata): Promise { + await Promise.resolve(); + metadata.set('x-metadata-only-async', 'true'); + return metadata; + } +} + +class AsyncMetadataOnlyFilterFactory + implements FilterFactory +{ + createFilter(): AsyncMetadataOnlyFilter { + return new AsyncMetadataOnlyFilter(); + } +} + +class RejectingMetadataFilter extends BaseFilter implements Filter { + sendMetadata(_metadata: Metadata): Promise { + return Promise.reject({ + code: Status.PERMISSION_DENIED, + details: 'Rejected by metadata filter', + metadata: new Metadata(), + }); + } +} + +class RejectingMetadataFilterFactory + implements FilterFactory +{ + createFilter(): RejectingMetadataFilter { + return new RejectingMetadataFilter(); + } +} + +function frameMessage(payload: Buffer, compressed = false): Buffer { + const framed = Buffer.allocUnsafe(payload.length + 5); + framed.writeUInt8(compressed ? 1 : 0, 0); + framed.writeUInt32BE(payload.length, 1); + payload.copy(framed, 5); + return framed; +} + +describe('CompressionFilter and FilterStack synchronous fast paths', () => { + const dummyChannel = {} as Channel; + + describe('CompressionFilter', () => { + it('executes sendMetadata synchronously', () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter: CompressionFilter = factory.createFilter(); + const metadata = new Metadata(); + metadata.set('grpc-encoding', 'gzip'); + + const result = filter.sendMetadata(metadata); + assert(!isThenable(result)); + assert.strictEqual(result, metadata); + assert.deepStrictEqual(result.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + assert.deepStrictEqual(result.get('accept-encoding'), ['identity']); + assert.deepStrictEqual(result.get('grpc-encoding'), []); + }); + + it('sets grpc-encoding synchronously when default compression is configured', () => { + const factory = new CompressionFilterFactory(dummyChannel, { + 'grpc.default_compression_algorithm': CompressionAlgorithms.gzip, + }); + const filter = factory.createFilter(); + const metadata = new Metadata(); + + const result = filter.sendMetadata(metadata); + assert(!isThenable(result)); + assert.deepStrictEqual(result.get('grpc-encoding'), ['gzip']); + }); + + it('executes sendMessage synchronously for identity compression', () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter = factory.createFilter(); + const payload = Buffer.from('hello world'); + + const result = filter.sendMessage({ message: payload }); + assert(!isThenable(result)); + assert.strictEqual(result.flags, undefined); + assert.strictEqual(result.message.readUInt8(0), 0); + assert.strictEqual(result.message.readUInt32BE(1), payload.length); + assert.deepStrictEqual(result.message.subarray(5), payload); + }); + + it('executes sendMessage synchronously when WriteFlags.NoCompress is set with gzip', () => { + const factory = new CompressionFilterFactory(dummyChannel, { + 'grpc.default_compression_algorithm': CompressionAlgorithms.gzip, + }); + const filter = factory.createFilter(); + const payload = Buffer.from('hello uncompressed'); + + const result = filter.sendMessage({ + message: payload, + flags: WriteFlags.NoCompress, + }); + assert(!isThenable(result)); + assert.strictEqual(result.flags, WriteFlags.NoCompress); + assert.strictEqual(result.message.readUInt8(0), 0); + assert.strictEqual(result.message.readUInt32BE(1), payload.length); + assert.deepStrictEqual(result.message.subarray(5), payload); + }); + + it('returns a Promise from sendMessage when compressing with gzip', async () => { + const factory = new CompressionFilterFactory(dummyChannel, { + 'grpc.default_compression_algorithm': CompressionAlgorithms.gzip, + }); + const filter = factory.createFilter(); + const payload = Buffer.from('hello compressed world'); + + const result = filter.sendMessage({ + message: payload, + flags: 0, + }); + assert(isThenable(result)); + const resolved = await result; + assert.strictEqual(resolved.flags, 0); + assert.strictEqual(resolved.message.readUInt8(0), 1); + const compressedBytes = resolved.message.subarray(5); + assert.strictEqual( + resolved.message.readUInt32BE(1), + compressedBytes.length + ); + assert.deepStrictEqual(zlib.gunzipSync(compressedBytes), payload); + }); + + it('returns a rejected Promise from sendMessage when maxSendMessageLength is exceeded', async () => { + const factory = new CompressionFilterFactory(dummyChannel, { + 'grpc.max_send_message_length': 5, + }); + const filter = factory.createFilter(); + const payload = Buffer.from('message longer than 5 bytes'); + + const result = filter.sendMessage({ message: payload }); + assert(isThenable(result)); + await assert.rejects(result, (error: StatusObject) => { + assert.strictEqual(error.code, Status.RESOURCE_EXHAUSTED); + return true; + }); + }); + + it('executes receiveMessage synchronously for uncompressed messages', () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter = factory.createFilter(); + const payload = Buffer.from('response payload'); + const framed = frameMessage(payload, false); + + const result = filter.receiveMessage(framed); + assert(!isThenable(result)); + assert.deepStrictEqual(result, payload); + }); + + it('returns a Promise from receiveMessage for compressed messages', async () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter = factory.createFilter(); + const responseMetadata = new Metadata(); + responseMetadata.set('grpc-encoding', 'gzip'); + filter.receiveMetadata(responseMetadata); + + const payload = Buffer.from('compressed response payload'); + const framed = frameMessage(zlib.gzipSync(payload), true); + + const result = filter.receiveMessage(framed); + assert(isThenable(result)); + const decompressed = await result; + assert.deepStrictEqual(decompressed, payload); + }); + + it('resets sendCompression to identity when server grpc-accept-encoding excludes configured algorithm', () => { + const factory = new CompressionFilterFactory(dummyChannel, { + 'grpc.default_compression_algorithm': CompressionAlgorithms.gzip, + }); + const filter = factory.createFilter(); + + // Before receiveMetadata, sendMetadata sets grpc-encoding: gzip + const initialMetadata = filter.sendMetadata(new Metadata()); + assert.deepStrictEqual(initialMetadata.get('grpc-encoding'), ['gzip']); + + // Server responds with grpc-accept-encoding that only supports identity + const responseMetadata = new Metadata(); + responseMetadata.set('grpc-accept-encoding', 'identity'); + const filteredResponseMetadata = filter.receiveMetadata(responseMetadata); + assert.deepStrictEqual( + filteredResponseMetadata.get('grpc-accept-encoding'), + [] + ); + + // Current filter now frames messages synchronously with identity (uncompressed) + const payload = Buffer.from('fallback to identity'); + const sendResult = filter.sendMessage({ message: payload }); + assert(!isThenable(sendResult)); + assert.strictEqual(sendResult.message.readUInt8(0), 0); + assert.strictEqual(sendResult.message.readUInt32BE(1), payload.length); + assert.deepStrictEqual(sendResult.message.subarray(5), payload); + + // Subsequent filters created by the same factory also default to identity + const nextFilter = factory.createFilter(); + const nextMetadata = nextFilter.sendMetadata(new Metadata()); + assert.deepStrictEqual(nextMetadata.get('grpc-encoding'), []); + const nextSendResult = nextFilter.sendMessage({ + message: payload, + }); + assert(!isThenable(nextSendResult)); + assert.strictEqual(nextSendResult.message.readUInt8(0), 0); + }); + }); + + describe('FilterStack', () => { + it('executes synchronously with CompressionFilter, SyncFilter, and BaseFilter pass-throughs', () => { + const stackFactory = new FilterStackFactory([ + new NoopPassThroughFilterFactory(), + new SyncFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadata = new Metadata(); + const metadataResult = stack.sendMetadata(metadata); + assert(!isThenable(metadataResult)); + assert.deepStrictEqual(metadataResult.get('x-sync'), ['true']); + assert.deepStrictEqual(metadataResult.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const payload = Buffer.from('sync stack message'); + const sendResult = stack.sendMessage({ message: payload }); + assert(!isThenable(sendResult)); + assert.strictEqual(sendResult.message.readUInt8(0), 0); + assert.deepStrictEqual( + sendResult.message.subarray(5), + Buffer.from('sync:sync stack message') + ); + + const receiveResult = stack.receiveMessage(sendResult.message); + assert(!isThenable(receiveResult)); + assert.deepStrictEqual( + receiveResult, + Buffer.from('sync:sync stack message:sync') + ); + }); + + it('transitions to Promise chaining when an async filter is in the stack', async () => { + const stackFactory = new FilterStackFactory([ + new AsyncHeaderFilterFactory('x-custom-async', 'enabled'), + new SyncFilterFactory(), + new NoopPassThroughFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadata = new Metadata(); + const metadataResult = stack.sendMetadata(metadata); + assert(isThenable(metadataResult)); + const resolvedMetadata = await metadataResult; + assert.deepStrictEqual(resolvedMetadata.get('x-custom-async'), [ + 'enabled', + ]); + assert.deepStrictEqual(resolvedMetadata.get('x-sync'), ['true']); + assert.deepStrictEqual(resolvedMetadata.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const payload = Buffer.from('body'); + const sendResult = stack.sendMessage({ message: payload }); + assert(isThenable(sendResult)); + const resolvedSend = await sendResult; + const expectedSentPayload = Buffer.from('sync:prefix:body'); + assert.deepStrictEqual( + resolvedSend.message.subarray(5), + expectedSentPayload + ); + + // Test receiveMessage when inner CompressionFilter and SyncFilter are synchronous + // and outer AsyncHeaderFilter transitions to a Promise + const receiveResult = stack.receiveMessage(resolvedSend.message); + assert(isThenable(receiveResult)); + const resolvedReceive = await receiveResult; + assert.deepStrictEqual( + resolvedReceive, + Buffer.from('sync:prefix:body:sync:suffix') + ); + + // Test receiveMessage when inner CompressionFilter returns a Promise (gzip), + // SyncFilter chains via result.then(...), and outer AsyncHeaderFilter chains onto that Promise + const gzipResponseMetadata = new Metadata(); + gzipResponseMetadata.set('grpc-encoding', 'gzip'); + stack.receiveMetadata(gzipResponseMetadata); + + const framedCompressed = frameMessage( + zlib.gzipSync(Buffer.from('compressed-body')), + true + ); + + const compressedReceiveResult = stack.receiveMessage(framedCompressed); + assert(isThenable(compressedReceiveResult)); + const resolvedCompressedReceive = await compressedReceiveResult; + assert.deepStrictEqual( + resolvedCompressedReceive, + Buffer.from('compressed-body:sync:suffix') + ); + }); + + it('chains custom thenables returned by filters in the middle of the stack', async () => { + const stackFactory = new FilterStackFactory([ + new SyncFilterFactory(), + new ThenableFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + // SyncFilter -> ThenableFilter -> CompressionFilter + const metadataResult = stack.sendMetadata(new Metadata()); + assert(isThenable(metadataResult)); + const resolvedMetadata = await metadataResult; + assert.deepStrictEqual(resolvedMetadata.get('x-sync'), ['true']); + assert.deepStrictEqual(resolvedMetadata.get('x-thenable'), ['true']); + assert.deepStrictEqual(resolvedMetadata.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const sendResult = stack.sendMessage({ + message: Buffer.from('msg'), + flags: WriteFlags.NoCompress, + }); + assert(isThenable(sendResult)); + const resolvedSend = await sendResult; + assert.strictEqual(resolvedSend.flags, WriteFlags.NoCompress); + assert.strictEqual(resolvedSend.message.readUInt8(0), 0); + assert.deepStrictEqual( + resolvedSend.message.subarray(5), + Buffer.from('thenable:sync:msg') + ); + + // CompressionFilter -> ThenableFilter -> SyncFilter + const receiveResult = stack.receiveMessage(resolvedSend.message); + assert(isThenable(receiveResult)); + const resolvedReceive = await receiveResult; + assert.deepStrictEqual( + resolvedReceive, + Buffer.from('thenable:sync:msg:thenable:sync') + ); + }); + + it('returns a custom thenable from the last filter without wrapping it', async () => { + const stack: FilterStack = new FilterStackFactory([ + new ThenableFilterFactory(), + ]).createFilter(); + + const metadataResult = stack.sendMetadata(new Metadata()); + assert(isThenable(metadataResult)); + assert(!(metadataResult instanceof Promise)); + assert.deepStrictEqual((await metadataResult).get('x-thenable'), [ + 'true', + ]); + }); + + it('BaseFilter pass-through methods return their input synchronously', () => { + const filter = new NoopPassThroughFilter(); + const metadata = new Metadata(); + const writeObject: WriteObject = { message: Buffer.from('out') }; + const buffer = Buffer.from('in'); + const status: StatusObject = { + code: Status.OK, + details: '', + metadata: new Metadata(), + }; + + assert.strictEqual(filter.sendMetadata(metadata), metadata); + assert.strictEqual(filter.receiveMetadata(metadata), metadata); + assert.strictEqual(filter.sendMessage(writeObject), writeObject); + assert.strictEqual(filter.receiveMessage(buffer), buffer); + assert.strictEqual(filter.receiveTrailers(status), status); + }); + + it('isThenable returns false for null, undefined and non-function then properties', () => { + assert.strictEqual(isThenable(null), false); + assert.strictEqual(isThenable(undefined), false); + assert.strictEqual(isThenable({ then: 'not a function' }), false); + }); + }); + + describe('ResolvingCall filter execution', () => { + function createMockChannel( + dynamicFilterFactories: FilterFactory[] = [] + ) { + let childStartedMetadata: Metadata | null = null; + let childListener: InterceptingListener | null = null; + const sentMessages: Buffer[] = []; + let halfClosed = false; + let readStarted = false; + let cancelledStatus: { code: Status; details: string } | null = null; + let resolveHalfClose: () => void; + const halfClosePromise = new Promise(resolve => { + resolveHalfClose = resolve; + }); + let messageWaiter: (() => void) | null = null; + + const mockChild = { + start(metadata: Metadata, listener: InterceptingListener) { + childStartedMetadata = metadata; + childListener = listener; + }, + sendMessageWithContext(_context: unknown, message: Buffer) { + sentMessages.push(message); + if (messageWaiter) { + const waiter = messageWaiter; + messageWaiter = null; + waiter(); + } + }, + startRead() { + readStarted = true; + }, + halfClose() { + halfClosed = true; + resolveHalfClose(); + }, + cancelWithStatus(code: Status, details: string) { + cancelledStatus = { code, details }; + }, + getCallNumber() { + return 42; + }, + getDeadlineInfo() { + return []; + }, + getPeer() { + return 'localhost:50051'; + }, + getAuthContext() { + return null; + }, + }; + + const mockChannel = { + getConfig() { + return { + type: 'SUCCESS', + config: { + methodConfig: { name: [] }, + pickInformation: {}, + status: Status.OK, + dynamicFilterFactories, + }, + }; + }, + createRetryingCall() { + return mockChild; + }, + getTarget() { + return 'dns:///localhost:50051'; + }, + }; + + return { + mockChannel: mockChannel as unknown as InternalChannel, + getChildStartedMetadata: () => childStartedMetadata, + getChildListener: () => childListener, + sentMessages, + isHalfClosed: () => halfClosed, + isReadStarted: () => readStarted, + getCancelledStatus: () => cancelledStatus, + halfClosePromise, + waitForNextSentMessage: () => + new Promise(resolve => { + messageWaiter = resolve; + }), + }; + } + + it('starts child, sends message, and receives message synchronously when using default CompressionFilter', done => { + const { + mockChannel, + getChildStartedMetadata, + getChildListener, + sentMessages, + isHalfClosed, + isReadStarted, + } = createMockChannel(); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 1 + ); + + // Before start(), child is null so getPeer() returns the channel target + assert.strictEqual(call.getPeer(), 'dns:///localhost:50051'); + + let receivedMetadata: Metadata | null = null; + let receivedMessage: Buffer | null = null; + + call.start(new Metadata(), { + onReceiveMetadata(metadata) { + receivedMetadata = metadata; + }, + onReceiveMessage(message) { + receivedMessage = message; + }, + onReceiveStatus(status) { + assert.strictEqual(status.code, Status.OK); + done(); + }, + }); + + // Child call must already be started synchronously after call.start(), + // and getPeer() now delegates to child.getPeer() + assert.strictEqual(call.getPeer(), 'localhost:50051'); + assert(getChildStartedMetadata() !== null); + assert.deepStrictEqual( + getChildStartedMetadata()!.get('grpc-accept-encoding'), + ['identity,deflate,gzip'] + ); + + call.startRead(); + assert.strictEqual(isReadStarted(), true); + + const requestPayload = Buffer.from('sync request'); + call.sendMessageWithContext({}, requestPayload); + // Message must be framed and sent to child synchronously without waiting for a microtask + assert.strictEqual(sentMessages.length, 1); + assert.deepStrictEqual(sentMessages[0].subarray(5), requestPayload); + + call.halfClose(); + assert.strictEqual(isHalfClosed(), true); + + const childListener = getChildListener()!; + const serverHeaders = new Metadata(); + serverHeaders.set('grpc-encoding', 'identity'); + childListener.onReceiveMetadata(serverHeaders); + assert(receivedMetadata !== null); + + const responsePayload = Buffer.from('sync response'); + const framedResponse = frameMessage(responsePayload, false); + + childListener.onReceiveMessage(framedResponse); + // Message must be deframed and delivered to listener synchronously + assert.deepStrictEqual(receivedMessage, responsePayload); + + childListener.onReceiveStatus({ + code: Status.OK, + details: 'OK', + metadata: new Metadata(), + }); + }); + + it('handles sync sendMetadata with async gzip sendMessage and receiveMessage', async () => { + const { + mockChannel, + getChildStartedMetadata, + getChildListener, + sentMessages, + isHalfClosed, + halfClosePromise, + waitForNextSentMessage, + } = createMockChannel(); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, { + 'grpc.default_compression_algorithm': CompressionAlgorithms.gzip, + }), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/GzipBidi', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 2 + ); + + const events: string[] = []; + const receivedMessages: Buffer[] = []; + let resolveFirstReceivedMessage: () => void; + const firstReceivedMessagePromise = new Promise(resolve => { + resolveFirstReceivedMessage = resolve; + }); + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() { + events.push('metadata'); + }, + onReceiveMessage(message) { + events.push('message'); + receivedMessages.push(message); + if (receivedMessages.length === 1) { + resolveFirstReceivedMessage(); + } + }, + onReceiveStatus(status: StatusObject) { + events.push('status'); + resolveStatus(status); + }, + }); + + // sendMetadata is still synchronous even when gzip compression is configured + assert(getChildStartedMetadata() !== null); + assert.deepStrictEqual(getChildStartedMetadata()!.get('grpc-encoding'), [ + 'gzip', + ]); + + // Send 1st message without calling halfClose() yet (covers pendingHalfClose === false branch) + const firstRequestPayload = Buffer.from('first gzip request'); + const firstSentPromise = waitForNextSentMessage(); + call.sendMessageWithContext({}, firstRequestPayload); + assert.strictEqual(sentMessages.length, 0); + await firstSentPromise; + assert.strictEqual(sentMessages.length, 1); + assert.strictEqual(isHalfClosed(), false); + + // Send 2nd message and immediately call halfClose() while writeFilterPending is true + const secondRequestPayload = Buffer.from('second gzip request'); + call.sendMessageWithContext({}, secondRequestPayload); + call.halfClose(); + assert.strictEqual(isHalfClosed(), false); + + await halfClosePromise; + assert.strictEqual(sentMessages.length, 2); + assert.strictEqual(sentMessages[1].readUInt8(0), 1); + assert.deepStrictEqual( + zlib.gunzipSync(sentMessages[1].subarray(5)), + secondRequestPayload + ); + assert.strictEqual(isHalfClosed(), true); + + const childListener = getChildListener()!; + const responseMetadata = new Metadata(); + responseMetadata.set('grpc-encoding', 'gzip'); + childListener.onReceiveMetadata(responseMetadata); + + // Receive 1st compressed message before status arrives (covers pendingChildStatus === null branch) + const firstResponsePayload = Buffer.from('first gzip response'); + const firstFramedResponse = frameMessage( + zlib.gzipSync(firstResponsePayload), + true + ); + + childListener.onReceiveMessage(firstFramedResponse); + await firstReceivedMessagePromise; + assert.strictEqual(receivedMessages.length, 1); + assert.deepStrictEqual(receivedMessages[0], firstResponsePayload); + + // Receive 2nd compressed message followed immediately by onReceiveStatus (covers pendingChildStatus !== null branch) + const secondResponsePayload = Buffer.from('second gzip response'); + const secondFramedResponse = frameMessage( + zlib.gzipSync(secondResponsePayload), + true + ); + + childListener.onReceiveMessage(secondFramedResponse); + childListener.onReceiveStatus({ + code: Status.OK, + details: 'OK', + metadata: new Metadata(), + }); + + const finalStatus = await statusPromise; + assert.strictEqual(finalStatus.code, Status.OK); + assert.deepStrictEqual(events, [ + 'metadata', + 'message', + 'message', + 'status', + ]); + assert.deepStrictEqual(receivedMessages[1], secondResponsePayload); + }); + + it('handles async dynamic filters and defers status while readFilterPending is true', async () => { + const { + mockChannel, + getChildStartedMetadata, + getChildListener, + sentMessages, + isHalfClosed, + isReadStarted, + halfClosePromise, + } = createMockChannel([ + new AsyncHeaderFilterFactory('x-dynamic-async', 'true'), + ]); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 3 + ); + + const events: string[] = []; + let receivedMessage: Buffer | null = null; + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() { + events.push('metadata'); + }, + onReceiveMessage(message) { + events.push('message'); + receivedMessage = message; + }, + onReceiveStatus(status: StatusObject) { + events.push('status'); + resolveStatus(status); + }, + }); + + // Child is not started synchronously because AsyncHeaderFilter.sendMetadata is async + assert.strictEqual(getChildStartedMetadata(), null); + assert.strictEqual(call.getPeer(), 'dns:///localhost:50051'); + + call.startRead(); + call.sendMessageWithContext({}, Buffer.from('req')); + call.halfClose(); + + await halfClosePromise; + assert.strictEqual(isReadStarted(), true); + assert(getChildStartedMetadata() !== null); + assert.deepStrictEqual( + getChildStartedMetadata()!.get('x-dynamic-async'), + ['true'] + ); + assert.strictEqual(sentMessages.length, 1); + assert.deepStrictEqual( + sentMessages[0].subarray(5), + Buffer.from('prefix:req') + ); + assert.strictEqual(isHalfClosed(), true); + // Verify pendingMessage reference was cleared after flushing in startChild + assert.strictEqual((call as any).pendingMessage, null); + + const childListener = getChildListener()!; + childListener.onReceiveMetadata(new Metadata()); + + const framedResponse = frameMessage(Buffer.from('resp'), false); + + // Trigger async receiveMessage filter followed immediately by onReceiveStatus + childListener.onReceiveMessage(framedResponse); + childListener.onReceiveStatus({ + code: Status.OK, + details: 'OK', + metadata: new Metadata(), + }); + + const finalStatus = await statusPromise; + assert.strictEqual(finalStatus.code, Status.OK); + assert.deepStrictEqual(events, ['metadata', 'message', 'status']); + assert.deepStrictEqual(receivedMessage, Buffer.from('resp:suffix')); + }); + + it('flushes queued message and halfClose synchronously in startChild when only sendMetadata is async', async () => { + const { + mockChannel, + getChildStartedMetadata, + sentMessages, + isHalfClosed, + halfClosePromise, + } = createMockChannel([new AsyncMetadataOnlyFilterFactory()]); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 4 + ); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus() {}, + }); + + assert.strictEqual(getChildStartedMetadata(), null); + call.sendMessageWithContext({}, Buffer.from('sync-send-after-async-md')); + call.halfClose(); + + await halfClosePromise; + assert.deepStrictEqual( + getChildStartedMetadata()!.get('x-metadata-only-async'), + ['true'] + ); + assert.strictEqual(sentMessages.length, 1); + assert.deepStrictEqual( + sentMessages[0].subarray(5), + Buffer.from('sync-send-after-async-md') + ); + assert.strictEqual(isHalfClosed(), true); + assert.strictEqual((call as any).pendingMessage, null); + }); + + it('flushes pendingHalfClose in startChild when no message was queued', async () => { + const { mockChannel, isHalfClosed, halfClosePromise } = createMockChannel( + [new AsyncMetadataOnlyFilterFactory()] + ); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/ClientStream', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 5 + ); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus() {}, + }); + + call.halfClose(); + await halfClosePromise; + assert.strictEqual(isHalfClosed(), true); + }); + + it('handles custom thenables returned by the filter stack for metadata, sent messages and received messages', async () => { + const { + mockChannel, + getChildStartedMetadata, + getChildListener, + sentMessages, + isHalfClosed, + halfClosePromise, + } = createMockChannel([new ThenableFilterFactory()]); + // No CompressionFilter, so that the custom thenable returned by + // ThenableFilter is the final result of the filter stack and reaches + // ResolvingCall without being chained onto a native Promise. + const stackFactory = new FilterStackFactory([]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + stackFactory, + 17 + ); + + const events: string[] = []; + let receivedMessage: Buffer | null = null; + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() { + events.push('metadata'); + }, + onReceiveMessage(message) { + events.push('message'); + receivedMessage = message; + }, + onReceiveStatus(status) { + events.push('status'); + resolveStatus(status); + }, + }); + assert.strictEqual(getChildStartedMetadata(), null); + + call.sendMessageWithContext({}, Buffer.from('req')); + call.halfClose(); + await halfClosePromise; + + assert.deepStrictEqual(getChildStartedMetadata()!.get('x-thenable'), [ + 'true', + ]); + assert.strictEqual(sentMessages.length, 1); + assert.deepStrictEqual(sentMessages[0], Buffer.from('thenable:req')); + assert.strictEqual(isHalfClosed(), true); + + const childListener = getChildListener()!; + childListener.onReceiveMetadata(new Metadata()); + childListener.onReceiveMessage(Buffer.from('resp')); + childListener.onReceiveStatus({ + code: Status.OK, + details: 'OK', + metadata: new Metadata(), + }); + + const status = await statusPromise; + assert.strictEqual(status.code, Status.OK); + assert.deepStrictEqual(events, ['metadata', 'message', 'status']); + assert.deepStrictEqual(receivedMessage, Buffer.from('resp:thenable')); + }); + + it('does not start child if call is cancelled while sendMetadata is pending', async () => { + let releaseMetadataFilter!: () => void; + const metadataFilterGate = new Promise(resolve => { + releaseMetadataFilter = resolve; + }); + + class GatedMetadataFilter extends BaseFilter implements Filter { + async sendMetadata(metadata: Metadata): Promise { + await metadataFilterGate; + return metadata; + } + } + + const { mockChannel, getChildStartedMetadata } = createMockChannel([ + { createFilter: () => new GatedMetadataFilter() }, + ]); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 6 + ); + + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + + call.cancelWithStatus(Status.CANCELLED, 'Cancelled before metadata done'); + const status = await statusPromise; + assert.strictEqual(status.code, Status.CANCELLED); + + // Release the gated filter and wait for the check phase so all chained + // microtasks in FilterStack and ResolvingCall have drained. + releaseMetadataFilter(); + await new Promise(resolve => setImmediate(resolve)); + assert.strictEqual(getChildStartedMetadata(), null); + }); + + it('does not forward sent or received messages if call is cancelled while async message filters are pending', async () => { + let releaseSendFilter!: () => void; + const sendFilterGate = new Promise(resolve => { + releaseSendFilter = resolve; + }); + let releaseReceiveFilter!: () => void; + const receiveFilterGate = new Promise(resolve => { + releaseReceiveFilter = resolve; + }); + + class GatedMessageFilter extends BaseFilter implements Filter { + async sendMessage(message: WriteObject): Promise { + await sendFilterGate; + return message; + } + async receiveMessage(message: Buffer): Promise { + await receiveFilterGate; + return message; + } + } + + const { mockChannel, getChildListener, sentMessages, isHalfClosed } = + createMockChannel([{ createFilter: () => new GatedMessageFilter() }]); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/BidiStream', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 60 + ); + + const receivedMessages: Buffer[] = []; + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage(message) { + receivedMessages.push(message); + }, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + + call.sendMessageWithContext({}, Buffer.from('in-flight-request')); + call.halfClose(); + getChildListener()!.onReceiveMessage( + frameMessage(Buffer.from('in-flight-response'), false) + ); + + call.cancelWithStatus( + Status.CANCELLED, + 'Cancelled while message filters pending' + ); + const status = await statusPromise; + assert.strictEqual(status.code, Status.CANCELLED); + + releaseSendFilter(); + releaseReceiveFilter(); + await new Promise(resolve => setImmediate(resolve)); + + assert.strictEqual(sentMessages.length, 0); + assert.strictEqual(isHalfClosed(), false); + assert.strictEqual(receivedMessages.length, 0); + }); + + it('outputs status when sendMetadata rejects', async () => { + const { mockChannel, getChildStartedMetadata } = createMockChannel([ + new RejectingMetadataFilterFactory(), + ]); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 7 + ); + + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + + const status = await statusPromise; + assert.strictEqual(status.code, Status.PERMISSION_DENIED); + assert.strictEqual(status.details, 'Rejected by metadata filter'); + assert.strictEqual(getChildStartedMetadata(), null); + }); + + it('cancels call with status when sendMessage rejects', async () => { + const { mockChannel, getCancelledStatus } = createMockChannel(); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, { + 'grpc.max_send_message_length': 4, + }), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 8 + ); + + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + + call.sendMessageWithContext({}, Buffer.from('message exceeding 4 bytes')); + const status = await statusPromise; + assert.strictEqual(status.code, Status.RESOURCE_EXHAUSTED); + assert.strictEqual(getCancelledStatus()?.code, Status.RESOURCE_EXHAUSTED); + }); + + it('cancels call with status when receiveMessage rejects', async () => { + const { mockChannel, getChildListener, getCancelledStatus } = + createMockChannel(); + const stackFactory = new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, { + 'grpc.max_receive_message_length': 4, + }), + ]); + + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { + deadline: Infinity, + flags: 0, + host: 'localhost', + parentCall: null, + }, + stackFactory, + 9 + ); + + let resolveStatus: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + + const childListener = getChildListener()!; + const responseMetadata = new Metadata(); + responseMetadata.set('grpc-encoding', 'gzip'); + childListener.onReceiveMetadata(responseMetadata); + + const framedResponse = frameMessage( + zlib.gzipSync(Buffer.from('decompressed payload exceeding 4 bytes')), + true + ); + + childListener.onReceiveMessage(framedResponse); + + const status = await statusPromise; + assert.strictEqual(status.code, Status.RESOURCE_EXHAUSTED); + assert.strictEqual(getCancelledStatus()?.code, Status.RESOURCE_EXHAUSTED); + }); + + it('converts synchronous exceptions in filters into INTERNAL call status without throwing', async () => { + class ThrowingSyncFilter extends BaseFilter implements Filter { + constructor( + private readonly throwOn: + | 'metadata' + | 'sendMessage' + | 'receiveMessage' + ) { + super(); + } + sendMetadata(metadata: Metadata): Metadata { + if (this.throwOn === 'metadata') { + throw Object.assign(new Error('sync metadata boom'), { + code: 'ERR_SYNC_FILTER', + }); + } + return metadata; + } + sendMessage(message: WriteObject): WriteObject { + if (this.throwOn === 'sendMessage') { + throw Object.assign(new Error('sync send boom'), { + code: 'ERR_SYNC_FILTER', + }); + } + return message; + } + receiveMessage(message: Buffer): Buffer { + if (this.throwOn === 'receiveMessage') { + throw Object.assign(new Error('sync receive boom'), { + code: 'ERR_SYNC_FILTER', + }); + } + return message; + } + } + + // 1. Synchronous throw in sendMetadata + { + const { mockChannel, getChildStartedMetadata } = createMockChannel([ + { createFilter: () => new ThrowingSyncFilter('metadata') }, + ]); + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 11 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('sync metadata boom')); + assert.strictEqual(getChildStartedMetadata(), null); + } + + // 2. Synchronous throw in sendMessage + { + const { mockChannel, getCancelledStatus } = createMockChannel([ + { createFilter: () => new ThrowingSyncFilter('sendMessage') }, + ]); + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 12 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + call.sendMessageWithContext({}, Buffer.from('msg')); + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('sync send boom')); + assert.strictEqual(getCancelledStatus()?.code, Status.INTERNAL); + } + + // 3. Synchronous throw in receiveMessage + { + const { mockChannel, getChildListener, getCancelledStatus } = + createMockChannel([ + { createFilter: () => new ThrowingSyncFilter('receiveMessage') }, + ]); + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 13 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + const framed = frameMessage(Buffer.from('resp'), false); + getChildListener()!.onReceiveMessage(framed); + + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('sync receive boom')); + assert.strictEqual(getCancelledStatus()?.code, Status.INTERNAL); + } + + // 4. Synchronous throw in sendMessage when flushing a queued message inside getConfig() + { + let configResolved = false; + const queuedCalls: ResolvingCall[] = []; + const baseMock = createMockChannel([ + { createFilter: () => new ThrowingSyncFilter('sendMessage') }, + ]); + const delayedMockChannel = { + getConfig(method: string, metadata: Metadata) { + if (!configResolved) { + return { type: 'NONE' as const }; + } + return baseMock.mockChannel.getConfig(method, metadata); + }, + queueCallForConfig(queuedCall: ResolvingCall) { + queuedCalls.push(queuedCall); + }, + createRetryingCall: baseMock.mockChannel.createRetryingCall, + getTarget: baseMock.mockChannel.getTarget, + } as unknown as InternalChannel; + + const call = new ResolvingCall( + delayedMockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 14 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + call.sendMessageWithContext({}, Buffer.from('queued-msg')); + assert.strictEqual(queuedCalls.length, 1); + + // Resolving config and draining queuedCalls must not throw out of getConfig() + configResolved = true; + for (const queuedCall of queuedCalls) { + queuedCall.getConfig(); + } + + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('sync send boom')); + assert.strictEqual( + baseMock.getCancelledStatus()?.code, + Status.INTERNAL + ); + } + + // 5. Synchronous throw in filter preceded by an async filter (Error rejection normalization) + for (const stage of [ + 'metadata', + 'sendMessage', + 'receiveMessage', + ] as const) { + const dynamicFactories: FilterFactory[] = + stage === 'receiveMessage' + ? [ + { createFilter: () => new ThrowingSyncFilter(stage) }, + new AsyncHeaderFilterFactory('x-async', '1'), + ] + : [ + new AsyncHeaderFilterFactory('x-async', '1'), + { createFilter: () => new ThrowingSyncFilter(stage) }, + ]; + const { mockChannel, getChildListener } = + createMockChannel(dynamicFactories); + const call = new ResolvingCall( + mockChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 140 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + if (stage === 'sendMessage') { + call.sendMessageWithContext({}, Buffer.from('msg')); + } else if (stage === 'receiveMessage') { + await new Promise(resolve => setImmediate(resolve)); + getChildListener()!.onReceiveMessage( + frameMessage(Buffer.from('resp'), false) + ); + } + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('boom')); + } + }); + + it('cancels child and outputs INTERNAL status if child.start throws synchronously in startChild', async () => { + let childCancelledWith = null as { + code: Status; + details: string; + } | null; + const throwingChildChannel = { + getConfig() { + return { + type: 'SUCCESS' as const, + config: { + methodConfig: { name: [] }, + status: Status.OK, + dynamicFilterFactories: [], + pickInformation: {}, + }, + }; + }, + createRetryingCall() { + return { + start() { + throw new Error('sync child.start boom'); + }, + cancelWithStatus(code: Status, details: string) { + childCancelledWith = { code, details }; + }, + getCallNumber: () => 99, + }; + }, + getTarget: () => 'dns:///localhost:50051', + } as unknown as InternalChannel; + + const call = new ResolvingCall( + throwingChildChannel, + '/test.Service/Unary', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 15 + ); + let resolveStatus!: (status: StatusObject) => void; + const statusPromise = new Promise(resolve => { + resolveStatus = resolve; + }); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus(status) { + resolveStatus(status); + }, + }); + const status = await statusPromise; + assert.strictEqual(status.code, Status.INTERNAL); + assert(status.details.includes('sync child.start boom')); + assert.strictEqual(childCancelledWith?.code, Status.INTERNAL); + }); + + it('propagates an exception thrown by listener.onReceiveMessage synchronously for uncompressed messages', () => { + const { mockChannel, getChildListener } = createMockChannel(); + const call = new ResolvingCall( + mockChannel, + '/test.Service/ServerStream', + { deadline: Infinity, flags: 0, host: 'localhost', parentCall: null }, + new FilterStackFactory([ + new CompressionFilterFactory(dummyChannel, {}), + ]), + 16 + ); + call.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() { + throw new Error('listener boom'); + }, + onReceiveStatus() {}, + }); + const framed = frameMessage(Buffer.from('resp'), false); + // Like onReceiveMetadata and onReceiveStatus, a listener exception is + // not caught or converted to a call status: it propagates to the caller. + assert.throws( + () => getChildListener()!.onReceiveMessage(framed), + /listener boom/ + ); + }); + }); + + describe('LoadBalancingCall', () => { + it('flushes pendingMessage and clears it to null once generateMetadata resolves', async () => { + const sentMessages: Buffer[] = []; + let resolveChildSent: () => void; + const childSentPromise = new Promise(resolve => { + resolveChildSent = resolve; + }); + + const mockSubchannelCall = { + getCallNumber: () => 42, + startRead: () => {}, + sendMessageWithContext: (_context: unknown, message: Buffer) => { + sentMessages.push(message); + resolveChildSent(); + }, + halfClose: () => {}, + }; + + const mockSubchannel = { + getCallCredentials: () => CallCredentials.createEmpty(), + getConnectivityState: () => ConnectivityState.READY, + getRealSubchannel: () => ({ + createCall: () => mockSubchannelCall, + }), + getChannelzRef: () => ({ id: 1 }), + getAddress: () => '127.0.0.1:50051', + }; + + const mockInternalChannel = { + getServiceUrl: () => 'https://localhost/test.Service', + doPick: () => ({ + pickResultType: PickResultType.COMPLETE, + subchannel: mockSubchannel, + status: null, + onCallStarted: null, + onCallEnded: null, + }), + }; + + const loadBalancingCall = new LoadBalancingCall( + mockInternalChannel as unknown as InternalChannel, + { + methodConfig: { name: [] }, + status: Status.OK, + dynamicFilterFactories: [], + pickInformation: {}, + }, + '/test.Service/ServerStream', + 'localhost', + CallCredentials.createEmpty(), + Infinity, + 10 + ); + + loadBalancingCall.start(new Metadata(), { + onReceiveMetadata() {}, + onReceiveMessage() {}, + onReceiveStatus() {}, + }); + + const requestPayload = Buffer.from('framed-request-payload'); + // CallCredentials.generateMetadata resolves asynchronously on a microtask, + // so sendMessageWithContext queues onto loadBalancingCall.pendingMessage + loadBalancingCall.sendMessageWithContext({}, requestPayload); + assert.notStrictEqual((loadBalancingCall as any).pendingMessage, null); + + await childSentPromise; + assert.strictEqual(sentMessages.length, 1); + assert.strictEqual(sentMessages[0], requestPayload); + assert.strictEqual((loadBalancingCall as any).pendingMessage, null); + }); + }); +});