grpc-js: Allow per-channel request compression from the client and decompression from the server

packages/grpc-js/package.json

@@ -52,13 +52,15 @@
     "test": "gulp test",
     "check": "gts check src/**/*.ts",
     "fix": "gts fix src/*.ts",
-    "pretest": "npm run generate-types && npm run compile",
+    "pretest": "npm run generate-types && npm run generate-test-types && npm run compile",
     "posttest": "npm run check && madge -c ./build/src",
-    "generate-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --includeDirs proto/ -O src/generated/ --grpcLib ../index channelz.proto"
+    "generate-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --includeDirs proto/ --include-dirs test/fixtures/ -O src/generated/ --grpcLib ../index channelz.proto",
+    "generate-test-types": "proto-loader-gen-types --keepCase --longs String --enums String --defaults --oneofs --includeComments --include-dirs test/fixtures/ -O test/generated/ --grpcLib ../../src/index test_service.proto"
   },
   "dependencies": {
     "@grpc/proto-loader": "^0.6.4",
-    "@types/node": ">=12.12.47"
+    "@types/node": "16.10.0",
+    "@types/semver": "^7.3.9"
   },
   "files": [
     "src/**/*.ts",

packages/grpc-js/src/channel-options.ts

@@ -36,6 +36,7 @@ export interface ChannelOptions {
   'grpc.enable_http_proxy'?: number;
   'grpc.http_connect_target'?: string;
   'grpc.http_connect_creds'?: string;
+  'grpc.default_compression_algorithm'?: number;
   'grpc-node.max_session_memory'?: number;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   [key: string]: any;

packages/grpc-js/src/channel.ts

@@ -45,7 +45,6 @@ import { MaxMessageSizeFilterFactory } from './max-message-size-filter';
 import { mapProxyName } from './http_proxy';
 import { GrpcUri, parseUri, uriToString } from './uri-parser';
 import { ServerSurfaceCall } from './server-call';
-import { SurfaceCall } from './call';
 import { Filter } from './filter';
 
 import { ConnectivityState } from './connectivity-state';
@@ -309,7 +308,7 @@ export class ChannelImplementation implements Channel {
       new CallCredentialsFilterFactory(this),
       new DeadlineFilterFactory(this),
       new MaxMessageSizeFilterFactory(this.options),
-      new CompressionFilterFactory(this),
+      new CompressionFilterFactory(this, this.options),
     ]);
     this.trace('Channel constructed with options ' + JSON.stringify(options, undefined, 2));
   }

packages/grpc-js/src/compression-filter.ts

@@ -17,10 +17,25 @@
 
 import * as zlib from 'zlib';
 
-import { Call, WriteFlags, WriteObject } from './call-stream';
+import { Call, WriteObject, WriteFlags } from './call-stream';
 import { Channel } from './channel';
 import { BaseFilter, Filter, FilterFactory } from './filter';
 import { Metadata, MetadataValue } from './metadata';
+import { ChannelOptions } from './channel-options';
+
+export const CompressionAlgorithms = {
+  0: 'identity',
+  1: 'deflate',
+  2: 'gzip'
+} as const;
+
+const CompressionAlgorithKeys = new Set(Object.keys(CompressionAlgorithms));
+
+const isCompressionAlgorithmKey = (key: number | undefined): key is keyof typeof CompressionAlgorithms => {
+  return typeof key === 'number' && CompressionAlgorithKeys.has(key.toString());
+}
+
+type CompressionAlgorithm = (typeof CompressionAlgorithms)[keyof typeof CompressionAlgorithms];
 
 abstract class CompressionHandler {
   protected abstract compressMessage(message: Buffer): Promise<Buffer>;
@@ -167,10 +182,29 @@ function getCompressionHandler(compressionName: string): CompressionHandler {
 export class CompressionFilter extends BaseFilter implements Filter {
   private sendCompression: CompressionHandler = new IdentityHandler();
   private receiveCompression: CompressionHandler = new IdentityHandler();
+  private defaultCompressionAlgorithm: CompressionAlgorithm | undefined;
+
+  constructor(channelOptions: ChannelOptions) {
+    super();
+
+    const compressionAlgorithmKey = channelOptions['grpc.default_compression_algorithm'];
+    if (isCompressionAlgorithmKey(compressionAlgorithmKey)) {
+      this.defaultCompressionAlgorithm = CompressionAlgorithms[compressionAlgorithmKey];
+      this.sendCompression = getCompressionHandler(this.defaultCompressionAlgorithm);
+    }
+  }
+
   async sendMetadata(metadata: Promise<Metadata>): Promise<Metadata> {
     const headers: Metadata = await metadata;
     headers.set('grpc-accept-encoding', 'identity,deflate,gzip');
     headers.set('accept-encoding', 'identity');
+
+    if (this.defaultCompressionAlgorithm && ['deflate', 'gzip'].includes(this.defaultCompressionAlgorithm)) {
+      headers.set('grpc-encoding', this.defaultCompressionAlgorithm);
+    } else {
+      headers.remove('grpc-encoding');
+    }
+
     return headers;
   }
 
@@ -183,6 +217,18 @@ export class CompressionFilter extends BaseFilter implements Filter {
       }
     }
     metadata.remove('grpc-encoding');
+
+    /* Check to see if the compression we're using to send messages is supported by the server
+     * If not, reset the sendCompression filter and have it use the default IdentityHandler */
+    const serverSupportedEncodingsHeader = metadata.get('grpc-accept-encoding')[0] as string | undefined;
+    if (serverSupportedEncodingsHeader) {
+      const serverSupportedEncodings = serverSupportedEncodingsHeader.split(',');
+
+      if ((this.sendCompression instanceof DeflateHandler && !serverSupportedEncodings.includes('deflate'))
+          || (this.sendCompression instanceof GzipHandler && !serverSupportedEncodings.includes('gzip'))) {
+          this.sendCompression = new IdentityHandler();
+        }
+    }
     metadata.remove('grpc-accept-encoding');
     return metadata;
   }
@@ -194,7 +240,7 @@ export class CompressionFilter extends BaseFilter implements Filter {
     const resolvedMessage: WriteObject = await message;
     const compress =
       resolvedMessage.flags === undefined
-        ? false
+        ? !(this.sendCompression instanceof IdentityHandler)
         : (resolvedMessage.flags & WriteFlags.NoCompress) === 0;
     return {
       message: await this.sendCompression.writeMessage(
@@ -216,8 +262,8 @@ export class CompressionFilter extends BaseFilter implements Filter {
 
 export class CompressionFilterFactory
   implements FilterFactory<CompressionFilter> {
-  constructor(private readonly channel: Channel) {}
+  constructor(private readonly channel: Channel, private readonly options: ChannelOptions) {}
   createFilter(callStream: Call): CompressionFilter {
-    return new CompressionFilter();
+    return new CompressionFilter(this.options);
   }
 }

packages/grpc-js/src/index.ts

@@ -25,6 +25,7 @@ import {
 import { CallCredentials, OAuth2Client } from './call-credentials';
 import { Deadline, StatusObject } from './call-stream';
 import { Channel, ChannelImplementation } from './channel';
+import { CompressionAlgorithms } from './compression-filter';
 import { ConnectivityState } from './connectivity-state';
 import { ChannelCredentials } from './channel-credentials';
 import {
@@ -124,6 +125,7 @@ export {
   Status as status,
   ConnectivityState as connectivityState,
   Propagate as propagate,
+  CompressionAlgorithms as compressionAlgorithms
   // TODO: Other constants as well
 };
 

packages/grpc-js/src/server-call.ts

@@ -18,6 +18,7 @@
 import { EventEmitter } from 'events';
 import * as http2 from 'http2';
 import { Duplex, Readable, Writable } from 'stream';
+import * as zlib from 'zlib';
 
 import { Deadline, StatusObject } from './call-stream';
 import {
@@ -32,6 +33,7 @@ import { StreamDecoder } from './stream-decoder';
 import { ObjectReadable, ObjectWritable } from './object-stream';
 import { ChannelOptions } from './channel-options';
 import * as logging from './logging';
+import { MetadataValue } from '.';
 
 const TRACER_NAME = 'server_call';
 
@@ -60,7 +62,7 @@ const deadlineUnitsToMs: DeadlineUnitIndexSignature = {
 const defaultResponseHeaders = {
   // TODO(cjihrig): Remove these encoding headers from the default response
   // once compression is integrated.
-  [GRPC_ACCEPT_ENCODING_HEADER]: 'identity',
+  [GRPC_ACCEPT_ENCODING_HEADER]: 'identity,deflate,gzip',
   [GRPC_ENCODING_HEADER]: 'identity',
   [http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
   [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
@@ -136,12 +138,13 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
   constructor(
     private call: Http2ServerCallStream<RequestType, ResponseType>,
     public metadata: Metadata,
-    public deserialize: Deserialize<RequestType>
+    public deserialize: Deserialize<RequestType>,
+    encoding?: MetadataValue
   ) {
     super({ objectMode: true });
     this.cancelled = false;
     this.call.setupSurfaceCall(this);
-    this.call.setupReadable(this);
+    this.call.setupReadable(this, encoding);
   }
 
   _read(size: number) {
@@ -250,13 +253,14 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
     private call: Http2ServerCallStream<RequestType, ResponseType>,
     public metadata: Metadata,
     public serialize: Serialize<ResponseType>,
-    public deserialize: Deserialize<RequestType>
+    public deserialize: Deserialize<RequestType>,
+    encoding?: MetadataValue
   ) {
     super({ objectMode: true });
     this.cancelled = false;
     this.trailingMetadata = new Metadata();
     this.call.setupSurfaceCall(this);
-    this.call.setupReadable(this);
+    this.call.setupReadable(this, encoding);
 
     this.on('error', (err) => {
       this.call.sendError(err);
@@ -439,6 +443,47 @@ export class Http2ServerCallStream<
     return this.cancelled;
   }
 
+  private getDecompressedMessage(message: Buffer, encoding?: MetadataValue) {
+    switch (encoding) {
+      case 'deflate': {
+        return new Promise<Buffer | undefined>((resolve, reject) => {
+          zlib.inflate(message.slice(5), (err, output) => {
+            if (err) {
+              this.sendError({
+                code: Status.INTERNAL,
+                details: `Received "grpc-encoding" header "${encoding}" but ${encoding} decompression failed`,
+              });
+              resolve();
+            } else {
+              const joined = Buffer.concat([message.slice(0, 5), output]);
+              resolve(joined);
+            }
+          });
+        });
+      }
+
+      case 'gzip': {
+        return new Promise<Buffer | undefined>((resolve, reject) => {
+          zlib.unzip(message.slice(5), (err, output) => {
+            if (err) {
+              this.sendError({
+                code: Status.INTERNAL,
+                details: `Received "grpc-encoding" header "${encoding}" but ${encoding} decompression failed`,
+              });
+              resolve();
+            } else {
+              const joined = Buffer.concat([message.slice(0, 5), output]);
+              resolve(joined);
+            }
+          });
+        });
+      }
+
+      default:
+        return Promise.resolve(message);
+    }
+  }
+
   sendMetadata(customMetadata?: Metadata) {
     if (this.checkCancelled()) {
       return;
@@ -469,7 +514,7 @@ export class Http2ServerCallStream<
         const err = new Error('Invalid deadline') as ServerErrorResponse;
         err.code = Status.OUT_OF_RANGE;
         this.sendError(err);
-        return;
+        return metadata;
       }
 
       const timeout = (+match[1] * deadlineUnitsToMs[match[2]]) | 0;
@@ -484,13 +529,12 @@ export class Http2ServerCallStream<
     metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
     metadata.remove(http2.constants.HTTP2_HEADER_TE);
     metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
-    metadata.remove('grpc-encoding');
     metadata.remove('grpc-accept-encoding');
 
     return metadata;
   }
 
-  receiveUnaryMessage(): Promise<RequestType> {
+  receiveUnaryMessage(encoding?: MetadataValue): Promise<RequestType> {
     return new Promise((resolve, reject) => {
       const stream = this.stream;
       const chunks: Buffer[] = [];
@@ -516,7 +560,19 @@ export class Http2ServerCallStream<
           }
 
           this.emit('receiveMessage');
-          resolve(this.deserializeMessage(requestBytes));
+
+          const compressed = requestBytes.readUInt8(0) === 1;
+          const compressedMessageEncoding = compressed ? encoding : undefined;
+          const decompressedMessage = await this.getDecompressedMessage(requestBytes, compressedMessageEncoding);
+
+          // Encountered an error with decompression; it'll already have been propogated back
+          // Just return early
+          if (!decompressedMessage) {
+            resolve();
+          }
+          else {
+            resolve(this.deserializeMessage(decompressedMessage));
+          }
         } catch (err) {
           err.code = Status.INTERNAL;
           this.sendError(err);
@@ -673,7 +729,8 @@ export class Http2ServerCallStream<
   setupReadable(
     readable:
       | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>
+      | ServerDuplexStream<RequestType, ResponseType>,
+    encoding?: MetadataValue
   ) {
     const decoder = new StreamDecoder();
 
@@ -692,7 +749,16 @@ export class Http2ServerCallStream<
           return;
         }
         this.emit('receiveMessage');
-        this.pushOrBufferMessage(readable, message);
+
+        const compressed = message.readUInt8(0) === 1;
+        const compressedMessageEncoding = compressed ? encoding : undefined;
+        const decompressedMessage = await this.getDecompressedMessage(message, compressedMessageEncoding);
+
+        // Encountered an error with decompression; it'll already have been propogated back
+        // Just return early
+        if (!decompressedMessage) return;
+
+        this.pushOrBufferMessage(readable, decompressedMessage);
       }
     });