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

@@ -15,6 +15,8 @@
  *
  */
 
+import { CompressionAlgorithms } from './compression-algorithms';
+
 /**
  * An interface that contains options used when initializing a Channel instance.
  */
@@ -36,6 +38,7 @@ export interface ChannelOptions {
   'grpc.enable_http_proxy'?: number;
   'grpc.http_connect_target'?: string;
   'grpc.http_connect_creds'?: string;
+  'grpc.default_compression_algorithm'?: keyof typeof CompressionAlgorithms;
   '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-algorithms.ts

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021 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.
+ *
+ */
+
+export const CompressionAlgorithms = {
+  0: 'identity',
+  1: 'deflate',
+  2: 'gzip'
+} as const;

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

@@ -17,11 +17,23 @@
 
 import * as zlib from 'zlib';
 
-import { Call, WriteFlags, WriteObject } from './call-stream';
+import { Call, WriteObject, WriteFlags } from './call-stream';
 import { Channel } from './channel';
+import { ChannelOptions } from './channel-options';
+import { CompressionAlgorithms } from './compression-algorithms';
+import { LogVerbosity } from './constants';
 import { BaseFilter, Filter, FilterFactory } from './filter';
+import * as logging from './logging';
 import { Metadata, MetadataValue } from './metadata';
 
+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>;
   protected abstract decompressMessage(data: Buffer): Promise<Buffer>;
@@ -167,10 +179,31 @@ 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);
+    } else {
+      logging.log(LogVerbosity.ERROR, 'Invalid value provided for grpc.default_compression_algorithm option');
+    }
+  }
+
   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 +216,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 +239,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 +261,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-algorithms';
 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 {
@@ -60,7 +61,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 +137,13 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
   constructor(
     private call: Http2ServerCallStream<RequestType, ResponseType>,
     public metadata: Metadata,
-    public deserialize: Deserialize<RequestType>
+    public deserialize: Deserialize<RequestType>,
+    encoding?: string
   ) {
     super({ objectMode: true });
     this.cancelled = false;
     this.call.setupSurfaceCall(this);
-    this.call.setupReadable(this);
+    this.call.setupReadable(this, encoding);
   }
 
   _read(size: number) {
@@ -250,13 +252,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?: string
   ) {
     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 +442,47 @@ export class Http2ServerCallStream<
     return this.cancelled;
   }
 
+  private getDecompressedMessage(message: Buffer, encoding?: string) {
+    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 +513,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 +528,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?: string): Promise<RequestType> {
     return new Promise((resolve, reject) => {
       const stream = this.stream;
       const chunks: Buffer[] = [];
@@ -516,7 +559,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 +728,8 @@ export class Http2ServerCallStream<
   setupReadable(
     readable:
       | ServerReadableStream<RequestType, ResponseType>
-      | ServerDuplexStream<RequestType, ResponseType>
+      | ServerDuplexStream<RequestType, ResponseType>,
+    encoding?: string
   ) {
     const decoder = new StreamDecoder();
 
@@ -692,7 +748,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);
       }
     });