From ea90811604294964f7f4665a5dfa21874d39acb3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Mon, 28 Sep 2026 08:24:51 +0200 Subject: [PATCH 1/2] perf(grpc-js): add synchronous fast path for FilterStack and CompressionFilter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, `FilterStack` and `CompressionFilter` always executed `sendMetadata`, `sendMessage`, and `receiveMessage` through `async` functions and `Promise` chains—even when no compression was used and no asynchronous filters were present in the stack. As a result, every uncompressed RPC had to allocate multiple intermediate `Promise`s and closures and hop through the microtask queue for metadata filtering, request framing, and response deframing. This change adds an optional synchronous fast path (`sendMetadataMaybeSync`, `sendMessageMaybeSync`, and `receiveMessageMaybeSync`) to `Filter`, `FilterStack`, `CompressionFilter`, and `ResolvingCall`. - **Synchronous metadata and uncompressed message framing/deframing**: `CompressionHandler.writeMessage` and `readMessage` now return a `Buffer` synchronously when no compression or decompression is required (such as `identity` encoding or when `WriteFlags.NoCompress` is set), only returning a `Promise` when `zlib` compression or decompression is actually needed. - **Zero-allocation pass-through for `BaseFilter` methods**: `FilterStack` skips methods inherited unchanged from `BaseFilter` (such as pass-through methods on `RouterFilter`) without allocating a `Promise`. - **Shared `IDENTITY_HANDLER` singleton**: `IdentityHandler` has no instance state, so `CompressionFilter` now reuses a module-level singleton instead of allocating two new `IdentityHandler` instances per call. - **Synchronous child call start on warm channels**: When channel configuration is already resolved and metadata filters are synchronous, `ResolvingCall` now starts the child `RetryingCall` and forwards the initial request message and `halfClose` in the same turn, avoiding the intermediate `pendingMessage` queuing in `ResolvingCall` and allowing `LoadBalancingCall` to flush the initial message and half-close together when call credentials resolve. - **Full backward compatibility for existing filters and callers**: The existing `sendMetadata`, `sendMessage`, and `receiveMessage` methods on `Filter`, `BaseFilter`, and `FilterStack` are preserved. When a legacy filter is present in the stack, `FilterStack` seamlessly transitions to `Promise` chaining for that filter and any subsequent filters in the pipeline. - **Unchanged ordering and async guarantees**: When any filter in the stack returns a `Promise` (for example, `gzip` or `deflate` compression, or a custom async filter), `ResolvingCall` continues to set `writeFilterPending` and `readFilterPending` to defer `halfClose` and `onReceiveStatus` until the pending filter finishes. Status delivery in `outputStatus` continues to use `process.nextTick`. - **Minor housekeeping**: - Clear `this.pendingMessage = null` in `ResolvingCall` and `LoadBalancingCall` once the queued message has been forwarded to the child call so the request buffer is not retained for the rest of the call. - Check `if (this.ended) return;` when asynchronous metadata or message filters resolve so cancelled calls do not start a child call or forward in-flight messages after cancellation. --- packages/grpc-js/src/compression-filter.ts | 135 +- packages/grpc-js/src/filter-stack.ts | 80 +- packages/grpc-js/src/filter.ts | 57 +- packages/grpc-js/src/load-balancing-call.ts | 6 +- packages/grpc-js/src/resolving-call.ts | 241 ++- packages/grpc-js/test/test-filter-stack.ts | 1701 +++++++++++++++++++ 6 files changed, 2098 insertions(+), 122 deletions(-) create mode 100644 packages/grpc-js/test/test-filter-stack.ts diff --git a/packages/grpc-js/src/compression-filter.ts b/packages/grpc-js/src/compression-filter.ts index e4428a1fb..2bb2a24ca 100644 --- a/packages/grpc-js/src/compression-filter.ts +++ b/packages/grpc-js/src/compression-filter.ts @@ -21,7 +21,12 @@ 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 { + DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, + DEFAULT_MAX_SEND_MESSAGE_LENGTH, + LogVerbosity, + Status, +} from './constants'; import { BaseFilter, Filter, FilterFactory } 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; + sendMetadataMaybeSync(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 { + sendMessageMaybeSync( + 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 (writeResult instanceof Promise) { + 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) { + receiveMessageMaybeSync(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..baa93b85a 100644 --- a/packages/grpc-js/src/filter-stack.ts +++ b/packages/grpc-js/src/filter-stack.ts @@ -16,7 +16,7 @@ */ import { StatusObject, WriteObject } from './call-interface'; -import { Filter, FilterFactory } from './filter'; +import { BaseFilter, Filter, FilterFactory } from './filter'; import { Metadata } from './metadata'; export class FilterStack implements Filter { @@ -32,6 +32,34 @@ export class FilterStack implements Filter { return result; } + /* Note: sendMetadataMaybeSync, sendMessageMaybeSync, and + * receiveMessageMaybeSync are intentionally kept as separate methods rather + * than sharing a dynamic helper so that property accesses on each filter + * remain monomorphic in V8 inline caches. */ + sendMetadataMaybeSync(metadata: Metadata): Metadata | Promise { + let result: Metadata | Promise = metadata; + + for (let i = 0; i < this.filters.length; i++) { + const filter = this.filters[i]; + if (filter.sendMetadataMaybeSync) { + result = + result instanceof Promise + ? result.then(resolvedMetadata => + filter.sendMetadataMaybeSync!(resolvedMetadata) + ) + : filter.sendMetadataMaybeSync(result); + } else if (filter.sendMetadata === BaseFilter.prototype.sendMetadata) { + // Default pass-through: no-op for both sync and Promise values + } else { + /* Normalize custom thenables or cross-realm promises into a native + * Promise so downstream `instanceof Promise` checks remain valid. */ + result = Promise.resolve(filter.sendMetadata(Promise.resolve(result))); + } + } + + return result; + } + receiveMetadata(metadata: Metadata) { let result: Metadata = metadata; @@ -52,6 +80,30 @@ export class FilterStack implements Filter { return result; } + sendMessageMaybeSync( + message: WriteObject + ): WriteObject | Promise { + let result: WriteObject | Promise = message; + + for (let i = 0; i < this.filters.length; i++) { + const filter = this.filters[i]; + if (filter.sendMessageMaybeSync) { + result = + result instanceof Promise + ? result.then(resolvedMessage => + filter.sendMessageMaybeSync!(resolvedMessage) + ) + : filter.sendMessageMaybeSync(result); + } else if (filter.sendMessage === BaseFilter.prototype.sendMessage) { + // Default pass-through: no-op for both sync and Promise values + } else { + result = Promise.resolve(filter.sendMessage(Promise.resolve(result))); + } + } + + return result; + } + receiveMessage(message: Promise): Promise { let result: Promise = message; @@ -62,6 +114,32 @@ export class FilterStack implements Filter { return result; } + receiveMessageMaybeSync(message: Buffer): Buffer | Promise { + let result: Buffer | Promise = message; + + for (let i = this.filters.length - 1; i >= 0; i--) { + const filter = this.filters[i]; + if (filter.receiveMessageMaybeSync) { + result = + result instanceof Promise + ? result.then(resolvedMessage => + filter.receiveMessageMaybeSync!(resolvedMessage) + ) + : filter.receiveMessageMaybeSync(result); + } else if ( + filter.receiveMessage === BaseFilter.prototype.receiveMessage + ) { + // Default pass-through: no-op for both sync and Promise values + } else { + result = Promise.resolve( + filter.receiveMessage(Promise.resolve(result)) + ); + } + } + + return result; + } + receiveTrailers(status: StatusObject): StatusObject { let result: StatusObject = status; diff --git a/packages/grpc-js/src/filter.ts b/packages/grpc-js/src/filter.ts index 5313f91a8..93473f7a7 100644 --- a/packages/grpc-js/src/filter.ts +++ b/packages/grpc-js/src/filter.ts @@ -34,11 +34,58 @@ export interface Filter { receiveMessage(message: Promise): Promise; receiveTrailers(status: StatusObject): StatusObject; + + /** + * Optional synchronous fast path for sendMetadata. When present, FilterStack + * calls this method instead of sendMetadata, so both must behave identically. + * Return the Metadata directly when no asynchronous work is needed, or a + * Promise (checked via `instanceof Promise`) otherwise. Report failures by + * returning a rejected Promise whose reason is a StatusObject rather than + * throwing synchronously. + */ + sendMetadataMaybeSync?(metadata: Metadata): Metadata | Promise; + + /** + * Optional synchronous fast path for sendMessage. When present, FilterStack + * calls this method instead of sendMessage, so both must behave identically. + * Return the WriteObject directly when no asynchronous work is needed, or a + * Promise (checked via `instanceof Promise`) otherwise. Report failures by + * returning a rejected Promise whose reason is a StatusObject rather than + * throwing synchronously. + */ + sendMessageMaybeSync?( + message: WriteObject + ): WriteObject | Promise; + + /** + * Optional synchronous fast path for receiveMessage. When present, + * FilterStack calls this method instead of receiveMessage, so both must + * behave identically. Return the Buffer directly when no asynchronous work is + * needed, or a Promise (checked via `instanceof Promise`) otherwise. Report + * failures by returning a rejected Promise whose reason is a StatusObject + * rather than throwing synchronously. + */ + receiveMessageMaybeSync?(message: Buffer): Buffer | Promise; } +/** + * Base class for filters with no-op default implementations. FilterStack's + * synchronous fast-path methods detect and skip methods inherited unchanged + * from BaseFilter without allocating a Promise. + */ export abstract class BaseFilter implements Filter { + sendMetadataMaybeSync?(metadata: Metadata): Metadata | Promise; + + sendMessageMaybeSync?( + message: WriteObject + ): WriteObject | Promise; + + receiveMessageMaybeSync?(message: Buffer): Buffer | Promise; + async sendMetadata(metadata: Promise): Promise { - return metadata; + return this.sendMetadataMaybeSync + ? this.sendMetadataMaybeSync(await metadata) + : metadata; } receiveMetadata(metadata: Metadata): Metadata { @@ -46,11 +93,15 @@ export abstract class BaseFilter implements Filter { } async sendMessage(message: Promise): Promise { - return message; + return this.sendMessageMaybeSync + ? this.sendMessageMaybeSync(await message) + : message; } async receiveMessage(message: Promise): Promise { - return message; + return this.receiveMessageMaybeSync + ? this.receiveMessageMaybeSync(await message) + : message; } receiveTrailers(status: StatusObject): StatusObject { diff --git a/packages/grpc-js/src/load-balancing-call.ts b/packages/grpc-js/src/load-balancing-call.ts index 74e7f71d7..94670f7ab 100644 --- a/packages/grpc-js/src/load-balancing-call.ts +++ b/packages/grpc-js/src/load-balancing-call.ts @@ -293,9 +293,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..48856fe20 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,14 @@ import { getRelativeTimeout, minDeadline, } from './deadline'; +import { getErrorMessage } from './error'; 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 +216,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 | Promise; + try { + // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. + filterResult = this.filterStack!.sendMessageMaybeSync({ + message: message, + flags: context.flags, + }); + } catch (error) { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to filter outgoing message: ${getErrorMessage(error)}` + ); + return; + } + if (filterResult instanceof Promise) { + // 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 | Promise; + try { + // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. + filterResult = this.filterStack!.receiveMessageMaybeSync(message); + } catch (error) { + this.cancelWithStatus( + Status.INTERNAL, + `Failed to filter incoming message: ${getErrorMessage(error)}` + ); + return; + } + if (filterResult instanceof Promise) { + // 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 +398,42 @@ 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.sendMetadataMaybeSync( + this.metadata + ); + if (filterResult instanceof Promise) { + 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 +457,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/test/test-filter-stack.ts b/packages/grpc-js/test/test-filter-stack.ts new file mode 100644 index 000000000..c0f10ccc0 --- /dev/null +++ b/packages/grpc-js/test/test-filter-stack.ts @@ -0,0 +1,1701 @@ +/* + * 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 } 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: Promise): Promise { + const resolvedMetadata = await metadata; + await Promise.resolve(); + resolvedMetadata.set(this.headerKey, this.headerValue); + return resolvedMetadata; + } + + async sendMessage(message: Promise): Promise { + const resolvedMessage = await message; + await Promise.resolve(); + return { + message: Buffer.concat([Buffer.from('prefix:'), resolvedMessage.message]), + flags: resolvedMessage.flags, + }; + } + + async receiveMessage(message: Promise): Promise { + const resolvedMessage = await message; + await Promise.resolve(); + return Buffer.concat([resolvedMessage, 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); + } +} + +function createCustomThenable(promise: Promise): Promise { + return { + then( + onFulfilled?: ((value: T) => TResult1 | PromiseLike) | null, + onRejected?: + | ((reason: unknown) => TResult2 | PromiseLike) + | null + ) { + return promise.then(onFulfilled, onRejected); + }, + } as unknown as Promise; +} + +class ThenableLegacyFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Promise): Promise { + return createCustomThenable( + metadata.then(resolvedMetadata => { + resolvedMetadata.set('x-thenable', 'true'); + return resolvedMetadata; + }) + ); + } + + sendMessage(message: Promise): Promise { + return createCustomThenable( + message.then(resolvedMessage => ({ + message: Buffer.concat([ + Buffer.from('thenable:'), + resolvedMessage.message, + ]), + flags: resolvedMessage.flags, + })) + ); + } + + receiveMessage(message: Promise): Promise { + return createCustomThenable( + message.then(resolvedMessage => + Buffer.concat([resolvedMessage, Buffer.from(':thenable')]) + ) + ); + } +} + +class ThenableLegacyFilterFactory + implements FilterFactory +{ + createFilter(): ThenableLegacyFilter { + return new ThenableLegacyFilter(); + } +} + +class MaybeSyncOnlyFilter extends BaseFilter implements Filter { + sendMetadataMaybeSync(metadata: Metadata): Metadata { + metadata.set('x-maybe-sync-only', 'true'); + return metadata; + } + + sendMessageMaybeSync(message: WriteObject): WriteObject { + return { + message: Buffer.concat([Buffer.from('sync:'), message.message]), + flags: message.flags, + }; + } + + receiveMessageMaybeSync(message: Buffer): Buffer { + return Buffer.concat([message, Buffer.from(':sync')]); + } +} + +class MaybeSyncOnlyFilterFactory implements FilterFactory { + createFilter(): MaybeSyncOnlyFilter { + return new MaybeSyncOnlyFilter(); + } +} + +class AsyncMetadataOnlyFilter extends BaseFilter implements Filter { + async sendMetadata(metadata: Promise): Promise { + const resolvedMetadata = await metadata; + await Promise.resolve(); + resolvedMetadata.set('x-metadata-only-async', 'true'); + return resolvedMetadata; + } +} + +class AsyncMetadataOnlyFilterFactory + implements FilterFactory +{ + createFilter(): AsyncMetadataOnlyFilter { + return new AsyncMetadataOnlyFilter(); + } +} + +class RejectingMetadataFilter extends BaseFilter implements Filter { + sendMetadata(_metadata: Promise): 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 sendMetadataMaybeSync synchronously', () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter: CompressionFilter = factory.createFilter(); + const metadata = new Metadata(); + metadata.set('grpc-encoding', 'gzip'); + + const result = filter.sendMetadataMaybeSync(metadata); + assert(!(result instanceof Promise)); + 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.sendMetadataMaybeSync(metadata); + assert(!(result instanceof Promise)); + assert.deepStrictEqual(result.get('grpc-encoding'), ['gzip']); + }); + + it('executes sendMessageMaybeSync synchronously for identity compression', () => { + const factory = new CompressionFilterFactory(dummyChannel, {}); + const filter = factory.createFilter(); + const payload = Buffer.from('hello world'); + + const result = filter.sendMessageMaybeSync({ message: payload }); + assert(!(result instanceof Promise)); + 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 sendMessageMaybeSync 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.sendMessageMaybeSync({ + message: payload, + flags: WriteFlags.NoCompress, + }); + assert(!(result instanceof Promise)); + 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 sendMessageMaybeSync 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.sendMessageMaybeSync({ + message: payload, + flags: 0, + }); + assert(result instanceof Promise); + 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 sendMessageMaybeSync 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.sendMessageMaybeSync({ message: payload }); + assert(result instanceof Promise); + await assert.rejects(result, (error: StatusObject) => { + assert.strictEqual(error.code, Status.RESOURCE_EXHAUSTED); + return true; + }); + }); + + it('executes receiveMessageMaybeSync 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.receiveMessageMaybeSync(framed); + assert(!(result instanceof Promise)); + assert.deepStrictEqual(result, payload); + }); + + it('returns a Promise from receiveMessageMaybeSync 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.receiveMessageMaybeSync(framed); + assert(result instanceof Promise); + 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, sendMetadataMaybeSync sets grpc-encoding: gzip + const initialMetadata = filter.sendMetadataMaybeSync(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.sendMessageMaybeSync({ message: payload }); + assert(!(sendResult instanceof Promise)); + 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.sendMetadataMaybeSync(new Metadata()); + assert.deepStrictEqual(nextMetadata.get('grpc-encoding'), []); + const nextSendResult = nextFilter.sendMessageMaybeSync({ + message: payload, + }); + assert(!(nextSendResult instanceof Promise)); + assert.strictEqual(nextSendResult.message.readUInt8(0), 0); + }); + }); + + describe('FilterStack', () => { + it('executes synchronously with CompressionFilter, MaybeSyncOnlyFilter, and BaseFilter pass-throughs', () => { + const stackFactory = new FilterStackFactory([ + new NoopPassThroughFilterFactory(), + new MaybeSyncOnlyFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadata = new Metadata(); + const metadataResult = stack.sendMetadataMaybeSync(metadata); + assert(!(metadataResult instanceof Promise)); + assert.deepStrictEqual(metadataResult.get('x-maybe-sync-only'), ['true']); + assert.deepStrictEqual(metadataResult.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const payload = Buffer.from('sync stack message'); + const sendResult = stack.sendMessageMaybeSync({ message: payload }); + assert(!(sendResult instanceof Promise)); + assert.strictEqual(sendResult.message.readUInt8(0), 0); + assert.deepStrictEqual( + sendResult.message.subarray(5), + Buffer.from('sync:sync stack message') + ); + + const receiveResult = stack.receiveMessageMaybeSync(sendResult.message); + assert(!(receiveResult instanceof Promise)); + 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 MaybeSyncOnlyFilterFactory(), + new NoopPassThroughFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadata = new Metadata(); + const metadataResult = stack.sendMetadataMaybeSync(metadata); + assert(metadataResult instanceof Promise); + const resolvedMetadata = await metadataResult; + assert.deepStrictEqual(resolvedMetadata.get('x-custom-async'), [ + 'enabled', + ]); + assert.deepStrictEqual(resolvedMetadata.get('x-maybe-sync-only'), [ + 'true', + ]); + assert.deepStrictEqual(resolvedMetadata.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const payload = Buffer.from('body'); + const sendResult = stack.sendMessageMaybeSync({ message: payload }); + assert(sendResult instanceof Promise); + const resolvedSend = await sendResult; + const expectedSentPayload = Buffer.from('sync:prefix:body'); + assert.deepStrictEqual( + resolvedSend.message.subarray(5), + expectedSentPayload + ); + + // Test receiveMessageMaybeSync when inner CompressionFilter and MaybeSyncOnlyFilter are synchronous + // and outer AsyncHeaderFilter transitions to a Promise + const receiveResult = stack.receiveMessageMaybeSync(resolvedSend.message); + assert(receiveResult instanceof Promise); + const resolvedReceive = await receiveResult; + assert.deepStrictEqual( + resolvedReceive, + Buffer.from('sync:prefix:body:sync:suffix') + ); + + // Test receiveMessageMaybeSync when inner CompressionFilter returns a Promise (gzip), + // MaybeSyncOnlyFilter 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.receiveMessageMaybeSync(framedCompressed); + assert(compressedReceiveResult instanceof Promise); + const resolvedCompressedReceive = await compressedReceiveResult; + assert.deepStrictEqual( + resolvedCompressedReceive, + Buffer.from('compressed-body:sync:suffix') + ); + }); + + it('normalizes non-native thenables returned by legacy filters into Promises', async () => { + const stackFactory = new FilterStackFactory([ + new ThenableLegacyFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadataResult = stack.sendMetadataMaybeSync(new Metadata()); + assert(metadataResult instanceof Promise); + const resolvedMetadata = await metadataResult; + assert.deepStrictEqual(resolvedMetadata.get('x-thenable'), ['true']); + assert.deepStrictEqual(resolvedMetadata.get('grpc-accept-encoding'), [ + 'identity,deflate,gzip', + ]); + + const sendResult = stack.sendMessageMaybeSync({ + message: Buffer.from('msg'), + flags: WriteFlags.NoCompress, + }); + assert(sendResult instanceof Promise); + const resolvedSend = await sendResult; + assert.strictEqual(resolvedSend.flags, WriteFlags.NoCompress); + assert.deepStrictEqual( + resolvedSend.message.subarray(5), + Buffer.from('thenable:msg') + ); + + const receiveResult = stack.receiveMessageMaybeSync(resolvedSend.message); + assert(receiveResult instanceof Promise); + const resolvedReceive = await receiveResult; + assert.deepStrictEqual( + resolvedReceive, + Buffer.from('thenable:msg:thenable') + ); + }); + + it('preserves legacy async FilterStack and BaseFilter methods', async () => { + const noopFilter = new NoopPassThroughFilter(); + // Verify BaseFilter methods always return a Promise even if called with a raw value + const rawBufferPromise = noopFilter.receiveMessage( + Buffer.from('raw') as unknown as Promise + ); + assert(rawBufferPromise instanceof Promise); + assert.deepStrictEqual(await rawBufferPromise, Buffer.from('raw')); + + const stackFactory = new FilterStackFactory([ + new NoopPassThroughFilterFactory(), + new MaybeSyncOnlyFilterFactory(), + new CompressionFilterFactory(dummyChannel, {}), + ]); + const stack: FilterStack = stackFactory.createFilter(); + + const metadata = await stack.sendMetadata( + Promise.resolve(new Metadata()) + ); + assert.deepStrictEqual(metadata.get('accept-encoding'), ['identity']); + assert.deepStrictEqual(metadata.get('x-maybe-sync-only'), ['true']); + + const payload = Buffer.from('legacy test'); + const framed = await stack.sendMessage( + Promise.resolve({ message: payload }) + ); + const deframed = await stack.receiveMessage( + Promise.resolve(framed.message) + ); + assert.deepStrictEqual(deframed, Buffer.from('sync:legacy test:sync')); + }); + }); + + 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 sendMetadataMaybeSync with async gzip sendMessageMaybeSync and receiveMessageMaybeSync', 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); + }, + }); + + // sendMetadataMaybeSync 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('does not start child if call is cancelled while sendMetadataMaybeSync is pending', async () => { + let releaseMetadataFilter!: () => void; + const metadataFilterGate = new Promise(resolve => { + releaseMetadataFilter = resolve; + }); + + class GatedMetadataFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Promise): Promise { + return metadata.then(async resolvedMetadata => { + await metadataFilterGate; + return resolvedMetadata; + }); + } + } + + 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 { + sendMessage(message: Promise): Promise { + return message.then(async resolvedMessage => { + await sendFilterGate; + return resolvedMessage; + }); + } + receiveMessage(message: Promise): Promise { + return message.then(async resolvedMessage => { + await receiveFilterGate; + return resolvedMessage; + }); + } + } + + 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 sendMetadataMaybeSync 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 sendMessageMaybeSync 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 receiveMessageMaybeSync 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 MaybeSync filters into INTERNAL call status without throwing', async () => { + class ThrowingSyncFilter extends BaseFilter implements Filter { + constructor( + private readonly throwOn: + | 'metadata' + | 'sendMessage' + | 'receiveMessage' + ) { + super(); + } + sendMetadataMaybeSync(metadata: Metadata): Metadata { + if (this.throwOn === 'metadata') { + throw Object.assign(new Error('sync metadata boom'), { + code: 'ERR_SYNC_FILTER', + }); + } + return metadata; + } + sendMessageMaybeSync(message: WriteObject): WriteObject { + if (this.throwOn === 'sendMessage') { + throw Object.assign(new Error('sync send boom'), { + code: 'ERR_SYNC_FILTER', + }); + } + return message; + } + receiveMessageMaybeSync(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 sendMetadataMaybeSync + { + 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 sendMessageMaybeSync + { + 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 receiveMessageMaybeSync + { + 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 sendMessageMaybeSync 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 MaybeSync 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 = { + 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); + }); + }); +}); From f66fcd066db3f37a5cc687983c07bd6978a84588 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Knut=20Olav=20L=C3=B8ite?= Date: Wed, 30 Sep 2026 17:41:19 +0200 Subject: [PATCH 2/2] refactor(grpc-js): replace MaybeSync filter methods with a single Filter API Address review feedback on the synchronous filter fast path. Instead of adding separate *MaybeSync methods next to the existing Promise-based ones, the Filter methods themselves now take a plain value and may return either the result directly or a thenable when they need to do asynchronous work. BaseFilter's methods are now simple synchronous pass-throughs. Filter results are now detected with an isThenable check instead of `instanceof Promise`, so the result of every filter is handled the same way, including the last one in the stack. Promises from other realms and custom thenables are handled correctly as well. The special case that skipped BaseFilter pass-through methods in FilterStack has been removed, as it is no longer needed now that those methods are synchronous. This is a breaking change to the experimental Filter API: filters now receive the value itself instead of a Promise of it. The fault injection filter in grpc-js-xds has been updated to match. Filters that simply `await` their input keep working. --- .../src/http-filter/fault-injection-filter.ts | 3 +- packages/grpc-js/src/compression-filter.ts | 10 +- packages/grpc-js/src/filter-stack.ts | 106 +---- packages/grpc-js/src/filter.ts | 89 ++-- packages/grpc-js/src/resolving-call.ts | 19 +- .../grpc-js/src/single-subchannel-channel.ts | 4 +- packages/grpc-js/test/test-filter-stack.ts | 413 ++++++++++-------- 7 files changed, 301 insertions(+), 343 deletions(-) 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 2bb2a24ca..0fe28a1de 100644 --- a/packages/grpc-js/src/compression-filter.ts +++ b/packages/grpc-js/src/compression-filter.ts @@ -27,7 +27,7 @@ import { LogVerbosity, Status, } from './constants'; -import { BaseFilter, Filter, FilterFactory } from './filter'; +import { BaseFilter, Filter, FilterFactory, isThenable } from './filter'; import * as logging from './logging'; import { Metadata, MetadataValue } from './metadata'; @@ -296,7 +296,7 @@ export class CompressionFilter extends BaseFilter implements Filter { } } - sendMetadataMaybeSync(headers: Metadata): Metadata { + sendMetadata(headers: Metadata): Metadata { headers.set('grpc-accept-encoding', 'identity,deflate,gzip'); headers.set('accept-encoding', 'identity'); @@ -345,7 +345,7 @@ export class CompressionFilter extends BaseFilter implements Filter { return metadata; } - sendMessageMaybeSync( + sendMessage( resolvedMessage: WriteObject ): WriteObject | Promise { /* This filter is special. The input message is the bare message bytes, @@ -371,7 +371,7 @@ export class CompressionFilter extends BaseFilter implements Filter { resolvedMessage.message, compress ); - if (writeResult instanceof Promise) { + if (isThenable(writeResult)) { return writeResult.then(message => ({ message, flags: resolvedMessage.flags, @@ -383,7 +383,7 @@ export class CompressionFilter extends BaseFilter implements Filter { }; } - receiveMessageMaybeSync(message: Buffer): Buffer | 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 diff --git a/packages/grpc-js/src/filter-stack.ts b/packages/grpc-js/src/filter-stack.ts index baa93b85a..c9177cd4e 100644 --- a/packages/grpc-js/src/filter-stack.ts +++ b/packages/grpc-js/src/filter-stack.ts @@ -16,45 +16,25 @@ */ import { StatusObject, WriteObject } from './call-interface'; -import { BaseFilter, 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; - - for (let i = 0; i < this.filters.length; i++) { - result = this.filters[i].sendMetadata(result); - } - - return result; - } - - /* Note: sendMetadataMaybeSync, sendMessageMaybeSync, and - * receiveMessageMaybeSync are intentionally kept as separate methods rather - * than sharing a dynamic helper so that property accesses on each filter - * remain monomorphic in V8 inline caches. */ - sendMetadataMaybeSync(metadata: Metadata): Metadata | Promise { - let result: Metadata | Promise = metadata; + sendMetadata(metadata: Metadata): Metadata | PromiseLike { + let result: Metadata | PromiseLike = metadata; for (let i = 0; i < this.filters.length; i++) { const filter = this.filters[i]; - if (filter.sendMetadataMaybeSync) { - result = - result instanceof Promise - ? result.then(resolvedMetadata => - filter.sendMetadataMaybeSync!(resolvedMetadata) - ) - : filter.sendMetadataMaybeSync(result); - } else if (filter.sendMetadata === BaseFilter.prototype.sendMetadata) { - // Default pass-through: no-op for both sync and Promise values - } else { - /* Normalize custom thenables or cross-realm promises into a native - * Promise so downstream `instanceof Promise` checks remain valid. */ - result = Promise.resolve(filter.sendMetadata(Promise.resolve(result))); - } + result = isThenable(result) + ? result.then(resolvedMetadata => filter.sendMetadata(resolvedMetadata)) + : filter.sendMetadata(result); } return result; @@ -70,71 +50,27 @@ export class FilterStack implements Filter { return result; } - sendMessage(message: Promise): Promise { - let result: Promise = message; - - for (let i = 0; i < this.filters.length; i++) { - result = this.filters[i].sendMessage(result); - } - - return result; - } - - sendMessageMaybeSync( - message: WriteObject - ): WriteObject | Promise { - let result: WriteObject | Promise = message; + sendMessage(message: WriteObject): WriteObject | PromiseLike { + let result: WriteObject | PromiseLike = message; for (let i = 0; i < this.filters.length; i++) { const filter = this.filters[i]; - if (filter.sendMessageMaybeSync) { - result = - result instanceof Promise - ? result.then(resolvedMessage => - filter.sendMessageMaybeSync!(resolvedMessage) - ) - : filter.sendMessageMaybeSync(result); - } else if (filter.sendMessage === BaseFilter.prototype.sendMessage) { - // Default pass-through: no-op for both sync and Promise values - } else { - result = Promise.resolve(filter.sendMessage(Promise.resolve(result))); - } - } - - return result; - } - - receiveMessage(message: Promise): Promise { - let result: Promise = message; - - for (let i = this.filters.length - 1; i >= 0; i--) { - result = this.filters[i].receiveMessage(result); + result = isThenable(result) + ? result.then(resolvedMessage => filter.sendMessage(resolvedMessage)) + : filter.sendMessage(result); } return result; } - receiveMessageMaybeSync(message: Buffer): Buffer | Promise { - let result: Buffer | Promise = message; + receiveMessage(message: Buffer): Buffer | PromiseLike { + let result: Buffer | PromiseLike = message; for (let i = this.filters.length - 1; i >= 0; i--) { const filter = this.filters[i]; - if (filter.receiveMessageMaybeSync) { - result = - result instanceof Promise - ? result.then(resolvedMessage => - filter.receiveMessageMaybeSync!(resolvedMessage) - ) - : filter.receiveMessageMaybeSync(result); - } else if ( - filter.receiveMessage === BaseFilter.prototype.receiveMessage - ) { - // Default pass-through: no-op for both sync and Promise values - } else { - result = Promise.resolve( - filter.receiveMessage(Promise.resolve(result)) - ); - } + 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 93473f7a7..62f16aaf9 100644 --- a/packages/grpc-js/src/filter.ts +++ b/packages/grpc-js/src/filter.ts @@ -20,88 +20,39 @@ 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; - - /** - * Optional synchronous fast path for sendMetadata. When present, FilterStack - * calls this method instead of sendMetadata, so both must behave identically. - * Return the Metadata directly when no asynchronous work is needed, or a - * Promise (checked via `instanceof Promise`) otherwise. Report failures by - * returning a rejected Promise whose reason is a StatusObject rather than - * throwing synchronously. - */ - sendMetadataMaybeSync?(metadata: Metadata): Metadata | Promise; - - /** - * Optional synchronous fast path for sendMessage. When present, FilterStack - * calls this method instead of sendMessage, so both must behave identically. - * Return the WriteObject directly when no asynchronous work is needed, or a - * Promise (checked via `instanceof Promise`) otherwise. Report failures by - * returning a rejected Promise whose reason is a StatusObject rather than - * throwing synchronously. - */ - sendMessageMaybeSync?( - message: WriteObject - ): WriteObject | Promise; - - /** - * Optional synchronous fast path for receiveMessage. When present, - * FilterStack calls this method instead of receiveMessage, so both must - * behave identically. Return the Buffer directly when no asynchronous work is - * needed, or a Promise (checked via `instanceof Promise`) otherwise. Report - * failures by returning a rejected Promise whose reason is a StatusObject - * rather than throwing synchronously. - */ - receiveMessageMaybeSync?(message: Buffer): Buffer | Promise; } -/** - * Base class for filters with no-op default implementations. FilterStack's - * synchronous fast-path methods detect and skip methods inherited unchanged - * from BaseFilter without allocating a Promise. - */ export abstract class BaseFilter implements Filter { - sendMetadataMaybeSync?(metadata: Metadata): Metadata | Promise; - - sendMessageMaybeSync?( - message: WriteObject - ): WriteObject | Promise; - - receiveMessageMaybeSync?(message: Buffer): Buffer | Promise; - - async sendMetadata(metadata: Promise): Promise { - return this.sendMetadataMaybeSync - ? this.sendMetadataMaybeSync(await metadata) - : metadata; + sendMetadata(metadata: Metadata): Metadata | PromiseLike { + return metadata; } receiveMetadata(metadata: Metadata): Metadata { return metadata; } - async sendMessage(message: Promise): Promise { - return this.sendMessageMaybeSync - ? this.sendMessageMaybeSync(await message) - : message; + sendMessage(message: WriteObject): WriteObject | PromiseLike { + return message; } - async receiveMessage(message: Promise): Promise { - return this.receiveMessageMaybeSync - ? this.receiveMessageMaybeSync(await message) - : message; + receiveMessage(message: Buffer): Buffer | PromiseLike { + return message; } receiveTrailers(status: StatusObject): StatusObject { @@ -112,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/resolving-call.ts b/packages/grpc-js/src/resolving-call.ts index 48856fe20..e252eb3f2 100644 --- a/packages/grpc-js/src/resolving-call.ts +++ b/packages/grpc-js/src/resolving-call.ts @@ -34,6 +34,7 @@ import { 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'; @@ -216,10 +217,10 @@ export class ResolvingCall implements Call { throw new Error('sendMessageonChild called with child not populated'); } const child = this.child; - let filterResult: WriteObject | Promise; + let filterResult: WriteObject | PromiseLike; try { // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. - filterResult = this.filterStack!.sendMessageMaybeSync({ + filterResult = this.filterStack!.sendMessage({ message: message, flags: context.flags, }); @@ -230,7 +231,7 @@ export class ResolvingCall implements Call { ); return; } - if (filterResult instanceof Promise) { + if (isThenable(filterResult)) { // Slow path: asynchronous filter. Pause new writes and forward when resolved. this.writeFilterPending = true; filterResult.then( @@ -264,10 +265,10 @@ export class ResolvingCall implements Call { private handleChildReceiveMessage(message: Buffer): void { this.trace('Received message'); - let filterResult: Buffer | Promise; + let filterResult: Buffer | PromiseLike; try { // Attempt synchronous filtering to bypass microtask deferral and promise allocations on the hot path. - filterResult = this.filterStack!.receiveMessageMaybeSync(message); + filterResult = this.filterStack!.receiveMessage(message); } catch (error) { this.cancelWithStatus( Status.INTERNAL, @@ -275,7 +276,7 @@ export class ResolvingCall implements Call { ); return; } - if (filterResult instanceof Promise) { + if (isThenable(filterResult)) { // Slow path: asynchronous filter. Pause reading and forward when resolved. this.readFilterPending = true; filterResult.then( @@ -406,10 +407,8 @@ export class ResolvingCall implements Call { this.filterStackFactory.push(config.dynamicFilterFactories); this.filterStack = this.filterStackFactory.createFilter(); try { - const filterResult = this.filterStack.sendMetadataMaybeSync( - this.metadata - ); - if (filterResult instanceof Promise) { + const filterResult = this.filterStack.sendMetadata(this.metadata); + if (isThenable(filterResult)) { filterResult.then( filteredMetadata => { this.startChild(config, filteredMetadata); 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 index c0f10ccc0..3dd40b801 100644 --- a/packages/grpc-js/test/test-filter-stack.ts +++ b/packages/grpc-js/test/test-filter-stack.ts @@ -33,7 +33,7 @@ import { } from '../src/compression-filter'; import { ConnectivityState } from '../src/connectivity-state'; import { Status } from '../src/constants'; -import { BaseFilter, Filter, FilterFactory } from '../src/filter'; +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'; @@ -59,26 +59,23 @@ class AsyncHeaderFilter extends BaseFilter implements Filter { super(); } - async sendMetadata(metadata: Promise): Promise { - const resolvedMetadata = await metadata; + async sendMetadata(metadata: Metadata): Promise { await Promise.resolve(); - resolvedMetadata.set(this.headerKey, this.headerValue); - return resolvedMetadata; + metadata.set(this.headerKey, this.headerValue); + return metadata; } - async sendMessage(message: Promise): Promise { - const resolvedMessage = await message; + async sendMessage(message: WriteObject): Promise { await Promise.resolve(); return { - message: Buffer.concat([Buffer.from('prefix:'), resolvedMessage.message]), - flags: resolvedMessage.flags, + message: Buffer.concat([Buffer.from('prefix:'), message.message]), + flags: message.flags, }; } - async receiveMessage(message: Promise): Promise { - const resolvedMessage = await message; + async receiveMessage(message: Buffer): Promise { await Promise.resolve(); - return Buffer.concat([resolvedMessage, Buffer.from(':suffix')]); + return Buffer.concat([message, Buffer.from(':suffix')]); } } @@ -93,88 +90,80 @@ class AsyncHeaderFilterFactory implements FilterFactory { } } -function createCustomThenable(promise: Promise): Promise { +/** + * 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); }, - } as unknown as Promise; + }; } -class ThenableLegacyFilter extends BaseFilter implements Filter { - sendMetadata(metadata: Promise): Promise { - return createCustomThenable( - metadata.then(resolvedMetadata => { - resolvedMetadata.set('x-thenable', 'true'); - return resolvedMetadata; - }) - ); +class ThenableFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Metadata): PromiseLike { + metadata.set('x-thenable', 'true'); + return createCustomThenable(metadata); } - sendMessage(message: Promise): Promise { - return createCustomThenable( - message.then(resolvedMessage => ({ - message: Buffer.concat([ - Buffer.from('thenable:'), - resolvedMessage.message, - ]), - flags: resolvedMessage.flags, - })) - ); + sendMessage(message: WriteObject): PromiseLike { + return createCustomThenable({ + message: Buffer.concat([Buffer.from('thenable:'), message.message]), + flags: message.flags, + }); } - receiveMessage(message: Promise): Promise { + receiveMessage(message: Buffer): PromiseLike { return createCustomThenable( - message.then(resolvedMessage => - Buffer.concat([resolvedMessage, Buffer.from(':thenable')]) - ) + Buffer.concat([message, Buffer.from(':thenable')]) ); } } -class ThenableLegacyFilterFactory - implements FilterFactory -{ - createFilter(): ThenableLegacyFilter { - return new ThenableLegacyFilter(); +class ThenableFilterFactory implements FilterFactory { + createFilter(): ThenableFilter { + return new ThenableFilter(); } } -class MaybeSyncOnlyFilter extends BaseFilter implements Filter { - sendMetadataMaybeSync(metadata: Metadata): Metadata { - metadata.set('x-maybe-sync-only', 'true'); +class SyncFilter extends BaseFilter implements Filter { + sendMetadata(metadata: Metadata): Metadata { + metadata.set('x-sync', 'true'); return metadata; } - sendMessageMaybeSync(message: WriteObject): WriteObject { + sendMessage(message: WriteObject): WriteObject { return { message: Buffer.concat([Buffer.from('sync:'), message.message]), flags: message.flags, }; } - receiveMessageMaybeSync(message: Buffer): Buffer { + receiveMessage(message: Buffer): Buffer { return Buffer.concat([message, Buffer.from(':sync')]); } } -class MaybeSyncOnlyFilterFactory implements FilterFactory { - createFilter(): MaybeSyncOnlyFilter { - return new MaybeSyncOnlyFilter(); +class SyncFilterFactory implements FilterFactory { + createFilter(): SyncFilter { + return new SyncFilter(); } } class AsyncMetadataOnlyFilter extends BaseFilter implements Filter { - async sendMetadata(metadata: Promise): Promise { - const resolvedMetadata = await metadata; + async sendMetadata(metadata: Metadata): Promise { await Promise.resolve(); - resolvedMetadata.set('x-metadata-only-async', 'true'); - return resolvedMetadata; + metadata.set('x-metadata-only-async', 'true'); + return metadata; } } @@ -187,7 +176,7 @@ class AsyncMetadataOnlyFilterFactory } class RejectingMetadataFilter extends BaseFilter implements Filter { - sendMetadata(_metadata: Promise): Promise { + sendMetadata(_metadata: Metadata): Promise { return Promise.reject({ code: Status.PERMISSION_DENIED, details: 'Rejected by metadata filter', @@ -216,14 +205,14 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { const dummyChannel = {} as Channel; describe('CompressionFilter', () => { - it('executes sendMetadataMaybeSync synchronously', () => { + 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.sendMetadataMaybeSync(metadata); - assert(!(result instanceof Promise)); + const result = filter.sendMetadata(metadata); + assert(!isThenable(result)); assert.strictEqual(result, metadata); assert.deepStrictEqual(result.get('grpc-accept-encoding'), [ 'identity,deflate,gzip', @@ -239,54 +228,54 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { const filter = factory.createFilter(); const metadata = new Metadata(); - const result = filter.sendMetadataMaybeSync(metadata); - assert(!(result instanceof Promise)); + const result = filter.sendMetadata(metadata); + assert(!isThenable(result)); assert.deepStrictEqual(result.get('grpc-encoding'), ['gzip']); }); - it('executes sendMessageMaybeSync synchronously for identity compression', () => { + 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.sendMessageMaybeSync({ message: payload }); - assert(!(result instanceof Promise)); + 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 sendMessageMaybeSync synchronously when WriteFlags.NoCompress is set with gzip', () => { + 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.sendMessageMaybeSync({ + const result = filter.sendMessage({ message: payload, flags: WriteFlags.NoCompress, }); - assert(!(result instanceof Promise)); + 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 sendMessageMaybeSync when compressing with gzip', async () => { + 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.sendMessageMaybeSync({ + const result = filter.sendMessage({ message: payload, flags: 0, }); - assert(result instanceof Promise); + assert(isThenable(result)); const resolved = await result; assert.strictEqual(resolved.flags, 0); assert.strictEqual(resolved.message.readUInt8(0), 1); @@ -298,33 +287,33 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.deepStrictEqual(zlib.gunzipSync(compressedBytes), payload); }); - it('returns a rejected Promise from sendMessageMaybeSync when maxSendMessageLength is exceeded', async () => { + 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.sendMessageMaybeSync({ message: payload }); - assert(result instanceof Promise); + 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 receiveMessageMaybeSync synchronously for uncompressed messages', () => { + 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.receiveMessageMaybeSync(framed); - assert(!(result instanceof Promise)); + const result = filter.receiveMessage(framed); + assert(!isThenable(result)); assert.deepStrictEqual(result, payload); }); - it('returns a Promise from receiveMessageMaybeSync for compressed messages', async () => { + it('returns a Promise from receiveMessage for compressed messages', async () => { const factory = new CompressionFilterFactory(dummyChannel, {}); const filter = factory.createFilter(); const responseMetadata = new Metadata(); @@ -334,8 +323,8 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { const payload = Buffer.from('compressed response payload'); const framed = frameMessage(zlib.gzipSync(payload), true); - const result = filter.receiveMessageMaybeSync(framed); - assert(result instanceof Promise); + const result = filter.receiveMessage(framed); + assert(isThenable(result)); const decompressed = await result; assert.deepStrictEqual(decompressed, payload); }); @@ -346,8 +335,8 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { }); const filter = factory.createFilter(); - // Before receiveMetadata, sendMetadataMaybeSync sets grpc-encoding: gzip - const initialMetadata = filter.sendMetadataMaybeSync(new Metadata()); + // 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 @@ -361,52 +350,52 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { // Current filter now frames messages synchronously with identity (uncompressed) const payload = Buffer.from('fallback to identity'); - const sendResult = filter.sendMessageMaybeSync({ message: payload }); - assert(!(sendResult instanceof Promise)); + 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.sendMetadataMaybeSync(new Metadata()); + const nextMetadata = nextFilter.sendMetadata(new Metadata()); assert.deepStrictEqual(nextMetadata.get('grpc-encoding'), []); - const nextSendResult = nextFilter.sendMessageMaybeSync({ + const nextSendResult = nextFilter.sendMessage({ message: payload, }); - assert(!(nextSendResult instanceof Promise)); + assert(!isThenable(nextSendResult)); assert.strictEqual(nextSendResult.message.readUInt8(0), 0); }); }); describe('FilterStack', () => { - it('executes synchronously with CompressionFilter, MaybeSyncOnlyFilter, and BaseFilter pass-throughs', () => { + it('executes synchronously with CompressionFilter, SyncFilter, and BaseFilter pass-throughs', () => { const stackFactory = new FilterStackFactory([ new NoopPassThroughFilterFactory(), - new MaybeSyncOnlyFilterFactory(), + new SyncFilterFactory(), new CompressionFilterFactory(dummyChannel, {}), ]); const stack: FilterStack = stackFactory.createFilter(); const metadata = new Metadata(); - const metadataResult = stack.sendMetadataMaybeSync(metadata); - assert(!(metadataResult instanceof Promise)); - assert.deepStrictEqual(metadataResult.get('x-maybe-sync-only'), ['true']); + 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.sendMessageMaybeSync({ message: payload }); - assert(!(sendResult instanceof Promise)); + 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.receiveMessageMaybeSync(sendResult.message); - assert(!(receiveResult instanceof Promise)); + const receiveResult = stack.receiveMessage(sendResult.message); + assert(!isThenable(receiveResult)); assert.deepStrictEqual( receiveResult, Buffer.from('sync:sync stack message:sync') @@ -416,29 +405,27 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { 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 MaybeSyncOnlyFilterFactory(), + new SyncFilterFactory(), new NoopPassThroughFilterFactory(), new CompressionFilterFactory(dummyChannel, {}), ]); const stack: FilterStack = stackFactory.createFilter(); const metadata = new Metadata(); - const metadataResult = stack.sendMetadataMaybeSync(metadata); - assert(metadataResult instanceof Promise); + 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-maybe-sync-only'), [ - 'true', - ]); + 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.sendMessageMaybeSync({ message: payload }); - assert(sendResult instanceof Promise); + const sendResult = stack.sendMessage({ message: payload }); + assert(isThenable(sendResult)); const resolvedSend = await sendResult; const expectedSentPayload = Buffer.from('sync:prefix:body'); assert.deepStrictEqual( @@ -446,18 +433,18 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { expectedSentPayload ); - // Test receiveMessageMaybeSync when inner CompressionFilter and MaybeSyncOnlyFilter are synchronous + // Test receiveMessage when inner CompressionFilter and SyncFilter are synchronous // and outer AsyncHeaderFilter transitions to a Promise - const receiveResult = stack.receiveMessageMaybeSync(resolvedSend.message); - assert(receiveResult instanceof 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 receiveMessageMaybeSync when inner CompressionFilter returns a Promise (gzip), - // MaybeSyncOnlyFilter chains via result.then(...), and outer AsyncHeaderFilter chains onto that Promise + // 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); @@ -467,9 +454,8 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { true ); - const compressedReceiveResult = - stack.receiveMessageMaybeSync(framedCompressed); - assert(compressedReceiveResult instanceof Promise); + const compressedReceiveResult = stack.receiveMessage(framedCompressed); + assert(isThenable(compressedReceiveResult)); const resolvedCompressedReceive = await compressedReceiveResult; assert.deepStrictEqual( resolvedCompressedReceive, @@ -477,72 +463,82 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { ); }); - it('normalizes non-native thenables returned by legacy filters into Promises', async () => { + it('chains custom thenables returned by filters in the middle of the stack', async () => { const stackFactory = new FilterStackFactory([ - new ThenableLegacyFilterFactory(), + new SyncFilterFactory(), + new ThenableFilterFactory(), new CompressionFilterFactory(dummyChannel, {}), ]); const stack: FilterStack = stackFactory.createFilter(); - const metadataResult = stack.sendMetadataMaybeSync(new Metadata()); - assert(metadataResult instanceof Promise); + // 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.sendMessageMaybeSync({ + const sendResult = stack.sendMessage({ message: Buffer.from('msg'), flags: WriteFlags.NoCompress, }); - assert(sendResult instanceof Promise); + 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:msg') + Buffer.from('thenable:sync:msg') ); - const receiveResult = stack.receiveMessageMaybeSync(resolvedSend.message); - assert(receiveResult instanceof Promise); + // CompressionFilter -> ThenableFilter -> SyncFilter + const receiveResult = stack.receiveMessage(resolvedSend.message); + assert(isThenable(receiveResult)); const resolvedReceive = await receiveResult; assert.deepStrictEqual( resolvedReceive, - Buffer.from('thenable:msg:thenable') + Buffer.from('thenable:sync:msg:thenable:sync') ); }); - it('preserves legacy async FilterStack and BaseFilter methods', async () => { - const noopFilter = new NoopPassThroughFilter(); - // Verify BaseFilter methods always return a Promise even if called with a raw value - const rawBufferPromise = noopFilter.receiveMessage( - Buffer.from('raw') as unknown as Promise - ); - assert(rawBufferPromise instanceof Promise); - assert.deepStrictEqual(await rawBufferPromise, Buffer.from('raw')); + it('returns a custom thenable from the last filter without wrapping it', async () => { + const stack: FilterStack = new FilterStackFactory([ + new ThenableFilterFactory(), + ]).createFilter(); - const stackFactory = new FilterStackFactory([ - new NoopPassThroughFilterFactory(), - new MaybeSyncOnlyFilterFactory(), - new CompressionFilterFactory(dummyChannel, {}), + const metadataResult = stack.sendMetadata(new Metadata()); + assert(isThenable(metadataResult)); + assert(!(metadataResult instanceof Promise)); + assert.deepStrictEqual((await metadataResult).get('x-thenable'), [ + 'true', ]); - const stack: FilterStack = stackFactory.createFilter(); + }); - const metadata = await stack.sendMetadata( - Promise.resolve(new Metadata()) - ); - assert.deepStrictEqual(metadata.get('accept-encoding'), ['identity']); - assert.deepStrictEqual(metadata.get('x-maybe-sync-only'), ['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(), + }; - const payload = Buffer.from('legacy test'); - const framed = await stack.sendMessage( - Promise.resolve({ message: payload }) - ); - const deframed = await stack.receiveMessage( - Promise.resolve(framed.message) - ); - assert.deepStrictEqual(deframed, Buffer.from('sync:legacy test:sync')); + 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); }); }); @@ -721,7 +717,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { }); }); - it('handles sync sendMetadataMaybeSync with async gzip sendMessageMaybeSync and receiveMessageMaybeSync', async () => { + it('handles sync sendMetadata with async gzip sendMessage and receiveMessage', async () => { const { mockChannel, getChildStartedMetadata, @@ -778,7 +774,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { }, }); - // sendMetadataMaybeSync is still synchronous even when gzip compression is configured + // sendMetadata is still synchronous even when gzip compression is configured assert(getChildStartedMetadata() !== null); assert.deepStrictEqual(getChildStartedMetadata()!.get('grpc-encoding'), [ 'gzip', @@ -1024,18 +1020,86 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(isHalfClosed(), true); }); - it('does not start child if call is cancelled while sendMetadataMaybeSync is pending', async () => { + 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 { - sendMetadata(metadata: Promise): Promise { - return metadata.then(async resolvedMetadata => { - await metadataFilterGate; - return resolvedMetadata; - }); + async sendMetadata(metadata: Metadata): Promise { + await metadataFilterGate; + return metadata; } } @@ -1094,17 +1158,13 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { }); class GatedMessageFilter extends BaseFilter implements Filter { - sendMessage(message: Promise): Promise { - return message.then(async resolvedMessage => { - await sendFilterGate; - return resolvedMessage; - }); + async sendMessage(message: WriteObject): Promise { + await sendFilterGate; + return message; } - receiveMessage(message: Promise): Promise { - return message.then(async resolvedMessage => { - await receiveFilterGate; - return resolvedMessage; - }); + async receiveMessage(message: Buffer): Promise { + await receiveFilterGate; + return message; } } @@ -1165,7 +1225,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(receivedMessages.length, 0); }); - it('outputs status when sendMetadataMaybeSync rejects', async () => { + it('outputs status when sendMetadata rejects', async () => { const { mockChannel, getChildStartedMetadata } = createMockChannel([ new RejectingMetadataFilterFactory(), ]); @@ -1205,7 +1265,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getChildStartedMetadata(), null); }); - it('cancels call with status when sendMessageMaybeSync rejects', async () => { + it('cancels call with status when sendMessage rejects', async () => { const { mockChannel, getCancelledStatus } = createMockChannel(); const stackFactory = new FilterStackFactory([ new CompressionFilterFactory(dummyChannel, { @@ -1245,7 +1305,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getCancelledStatus()?.code, Status.RESOURCE_EXHAUSTED); }); - it('cancels call with status when receiveMessageMaybeSync rejects', async () => { + it('cancels call with status when receiveMessage rejects', async () => { const { mockChannel, getChildListener, getCancelledStatus } = createMockChannel(); const stackFactory = new FilterStackFactory([ @@ -1297,7 +1357,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getCancelledStatus()?.code, Status.RESOURCE_EXHAUSTED); }); - it('converts synchronous exceptions in MaybeSync filters into INTERNAL call status without throwing', async () => { + it('converts synchronous exceptions in filters into INTERNAL call status without throwing', async () => { class ThrowingSyncFilter extends BaseFilter implements Filter { constructor( private readonly throwOn: @@ -1307,7 +1367,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { ) { super(); } - sendMetadataMaybeSync(metadata: Metadata): Metadata { + sendMetadata(metadata: Metadata): Metadata { if (this.throwOn === 'metadata') { throw Object.assign(new Error('sync metadata boom'), { code: 'ERR_SYNC_FILTER', @@ -1315,7 +1375,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { } return metadata; } - sendMessageMaybeSync(message: WriteObject): WriteObject { + sendMessage(message: WriteObject): WriteObject { if (this.throwOn === 'sendMessage') { throw Object.assign(new Error('sync send boom'), { code: 'ERR_SYNC_FILTER', @@ -1323,7 +1383,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { } return message; } - receiveMessageMaybeSync(message: Buffer): Buffer { + receiveMessage(message: Buffer): Buffer { if (this.throwOn === 'receiveMessage') { throw Object.assign(new Error('sync receive boom'), { code: 'ERR_SYNC_FILTER', @@ -1333,7 +1393,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { } } - // 1. Synchronous throw in sendMetadataMaybeSync + // 1. Synchronous throw in sendMetadata { const { mockChannel, getChildStartedMetadata } = createMockChannel([ { createFilter: () => new ThrowingSyncFilter('metadata') }, @@ -1364,7 +1424,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getChildStartedMetadata(), null); } - // 2. Synchronous throw in sendMessageMaybeSync + // 2. Synchronous throw in sendMessage { const { mockChannel, getCancelledStatus } = createMockChannel([ { createFilter: () => new ThrowingSyncFilter('sendMessage') }, @@ -1396,7 +1456,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getCancelledStatus()?.code, Status.INTERNAL); } - // 3. Synchronous throw in receiveMessageMaybeSync + // 3. Synchronous throw in receiveMessage { const { mockChannel, getChildListener, getCancelledStatus } = createMockChannel([ @@ -1431,7 +1491,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { assert.strictEqual(getCancelledStatus()?.code, Status.INTERNAL); } - // 4. Synchronous throw in sendMessageMaybeSync when flushing a queued message inside getConfig() + // 4. Synchronous throw in sendMessage when flushing a queued message inside getConfig() { let configResolved = false; const queuedCalls: ResolvingCall[] = []; @@ -1490,7 +1550,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { ); } - // 5. Synchronous throw in MaybeSync filter preceded by an async filter (Error rejection normalization) + // 5. Synchronous throw in filter preceded by an async filter (Error rejection normalization) for (const stage of [ 'metadata', 'sendMessage', @@ -1656,6 +1716,7 @@ describe('CompressionFilter and FilterStack synchronous fast paths', () => { }; const mockInternalChannel = { + getServiceUrl: () => 'https://localhost/test.Service', doPick: () => ({ pickResultType: PickResultType.COMPLETE, subchannel: mockSubchannel,