Merge pull request #2877 from grpc/@grpc/[email protected]

Merge grpc-js 1.12.x into master

packages/grpc-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@grpc/grpc-js",
-  "version": "1.12.1",
+  "version": "1.12.5",
   "description": "gRPC Library for Node - pure JS implementation",
   "homepage": "https://grpc.io/",
   "repository": "https://github.com/grpc/grpc-node/tree/master/packages/grpc-js",

packages/grpc-js/src/certificate-provider.ts

@@ -15,9 +15,10 @@
  *
  */
 
-import * as fs from 'fs/promises';
+import * as fs from 'fs';
 import * as logging from './logging';
 import { LogVerbosity } from './constants';
+import { promisify } from 'util';
 
 const TRACER_NAME = 'certificate_provider';
 
@@ -56,6 +57,8 @@ export interface FileWatcherCertificateProviderConfig {
   refreshIntervalMs: number;
 }
 
+const readFilePromise = promisify(fs.readFile);
+
 export class FileWatcherCertificateProvider implements CertificateProvider {
   private refreshTimer: NodeJS.Timeout | null = null;
   private fileResultPromise: Promise<[PromiseSettledResult<Buffer>, PromiseSettledResult<Buffer>, PromiseSettledResult<Buffer>]> | null = null;
@@ -82,9 +85,9 @@ export class FileWatcherCertificateProvider implements CertificateProvider {
       return;
     }
     this.fileResultPromise = Promise.allSettled([
-      this.config.certificateFile ? fs.readFile(this.config.certificateFile) : Promise.reject<Buffer>(),
-      this.config.privateKeyFile ? fs.readFile(this.config.privateKeyFile) : Promise.reject<Buffer>(),
-      this.config.caCertificateFile ? fs.readFile(this.config.caCertificateFile) : Promise.reject<Buffer>()
+      this.config.certificateFile ? readFilePromise(this.config.certificateFile) : Promise.reject<Buffer>(),
+      this.config.privateKeyFile ? readFilePromise(this.config.privateKeyFile) : Promise.reject<Buffer>(),
+      this.config.caCertificateFile ? readFilePromise(this.config.caCertificateFile) : Promise.reject<Buffer>()
     ]);
     this.fileResultPromise.then(([certificateResult, privateKeyResult, caCertificateResult]) => {
       if (!this.refreshTimer) {

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

@@ -140,6 +140,8 @@ export class Http2SubchannelCall implements SubchannelCall {
 
   private serverEndedCall = false;
 
+  private connectionDropped = false;
+
   constructor(
     private readonly http2Stream: http2.ClientHttp2Stream,
     private readonly callEventTracker: CallEventTracker,
@@ -188,7 +190,22 @@ export class Http2SubchannelCall implements SubchannelCall {
       try {
         messages = this.decoder.write(data);
       } catch (e) {
-        this.cancelWithStatus(Status.RESOURCE_EXHAUSTED, (e as Error).message);
+        /* Some servers send HTML error pages along with HTTP status codes.
+         * When the client attempts to parse this as a length-delimited
+         * message, the parsed message size is greater than the default limit,
+         * resulting in a message decoding error. In that situation, the HTTP
+         * error code information is more useful to the user than the
+         * RESOURCE_EXHAUSTED error is, so we report that instead. Normally,
+         * we delay processing the HTTP status until after the stream ends, to
+         * prioritize reporting the gRPC status from trailers if it is present,
+         * but when there is a message parsing error we end the stream early
+         * before processing trailers. */
+        if (this.httpStatusCode !== undefined && this.httpStatusCode !== 200) {
+          const mappedStatus = mapHttpStatusCode(this.httpStatusCode);
+          this.cancelWithStatus(mappedStatus.code, mappedStatus.details);
+        } else {
+          this.cancelWithStatus(Status.RESOURCE_EXHAUSTED, (e as Error).message);
+        }
         return;
       }
 
@@ -240,8 +257,16 @@ export class Http2SubchannelCall implements SubchannelCall {
             details = 'Stream refused by server';
             break;
           case http2.constants.NGHTTP2_CANCEL:
-            code = Status.CANCELLED;
-            details = 'Call cancelled';
+            /* Bug reports indicate that Node synthesizes a NGHTTP2_CANCEL
+             * code from connection drops. We want to prioritize reporting
+             * an unavailable status when that happens. */
+            if (this.connectionDropped) {
+              code = Status.UNAVAILABLE;
+              details = 'Connection dropped';
+            } else {
+              code = Status.CANCELLED;
+              details = 'Call cancelled';
+            }
             break;
           case http2.constants.NGHTTP2_ENHANCE_YOUR_CALM:
             code = Status.RESOURCE_EXHAUSTED;
@@ -321,10 +346,15 @@ export class Http2SubchannelCall implements SubchannelCall {
   }
 
   public onDisconnect() {
-    this.endCall({
-      code: Status.UNAVAILABLE,
-      details: 'Connection dropped',
-      metadata: new Metadata(),
+    this.connectionDropped = true;
+    /* Give the call an event loop cycle to finish naturally before reporting
+     * the disconnection as an error. */
+    setImmediate(() => {
+      this.endCall({
+        code: Status.UNAVAILABLE,
+        details: 'Connection dropped',
+        metadata: new Metadata(),
+      });
     });
   }
 

packages/grpc-js/src/transport.ts

@@ -225,6 +225,11 @@ class Http2Transport implements Transport {
       this.handleDisconnect();
     });
 
+    session.socket.once('close', () => {
+      this.trace('connection closed');
+      this.handleDisconnect();
+    });
+
     if (logging.isTracerEnabled(TRACER_NAME)) {
       session.on('remoteSettings', (settings: http2.Settings) => {
         this.trace(
@@ -380,17 +385,13 @@ class Http2Transport implements Transport {
    * Handle connection drops, but not GOAWAYs.
    */
   private handleDisconnect() {
-    if (this.disconnectHandled) {
-      return;
-    }
     this.clearKeepaliveTimeout();
     this.reportDisconnectToOwner(false);
-    /* Give calls an event loop cycle to finish naturally before reporting the
-     * disconnnection to them. */
+    for (const call of this.activeCalls) {
+      call.onDisconnect();
+    }
+    // Wait an event loop cycle before destroying the connection
     setImmediate(() => {
-      for (const call of this.activeCalls) {
-        call.onDisconnect();
-      }
       this.session.destroy();
     });
   }