fix(gax): map REST fallback transport errors to gRPC status codes

core/packages/gax/src/fallbackServiceStub.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
 import type {Response as NodeFetchResponse} from 'node-fetch' with {'resolution-mode': 'import'};
 
 import {AuthClient, GoogleAuth, gaxios} from 'google-auth-library';
 import * as serializer from 'proto3-json-serializer';
@@ -22,6 +22,8 @@
 import {isNodeJS} from './featureDetection';
 import {StreamArrayParser} from './streamArrayParser';
 import {defaultToObjectOptions} from './fallback';
+import {GoogleError} from './googleError';
+import {rpcCodeFromHttpStatusCode, Status} from './status';
 import {pipeline, PipelineSource} from 'stream';
 import type {Agent as HttpAgent} from 'http';
 import type {Agent as HttpsAgent} from 'https';
@@ -33,7 +35,7 @@
 // - https://github.com/node-fetch/node-fetch#custom-agent
 // - https://github.com/googleapis/gax-nodejs/pull/1534
 let agentOption:
   | ((parsedUrl: {protocol: string}) => HttpAgent | HttpsAgent)
   | null = null;
 if (isNodeJS()) {
   const http = require('http');
@@ -82,6 +84,81 @@
   return resp;
 }
 
+/**
+ * Translates an error thrown by the underlying fetch implementation into a
+ * {@link GoogleError} carrying a numeric gRPC status code.
+ *
+ * Retry logic (see `normalCalls/retries.ts`) and user code both match on
+ * numeric gRPC status codes. An untranslated error from `auth.fetch()` carries
+ * either a system error string (e.g. `'ECONNRESET'` for a socket hang up) or an
+ * HTTP status number, so it never matches a retry code and is silently treated
+ * as a permanent failure.
+ *
+ * An error carrying an HTTP status maps through `rpcCodeFromHttpStatusCode`.
+ * Otherwise the call failed before producing a response, which is reported as
+ * UNAVAILABLE, except for an explicit cancellation (CANCELLED) or an elapsed
+ * deadline (DEADLINE_EXCEEDED).
+ *
+ * @param err The error thrown by `auth.fetch()`.
+ * @returns A GoogleError with a numeric `code`, or the original value if it is
+ *   not an Error.
+ */
+function _toGoogleError(err: unknown): unknown {
+  if (err instanceof GoogleError) {
+    return err;
+  }
+  if (!(err instanceof Error)) {
+    return err;
+  }
+
+  const error = new GoogleError(err.message);
+  error.cause = err;
+
+  // `GaxiosError` shape, described structurally to avoid depending on the
+  // error instance originating from any particular copy of gaxios.
+  const fetchError = err as Partial<{
+    status: number;
+    response: {status?: number};
+    code: string | number;
+  }>;
+
+  // Errors that carry an HTTP status map through the standard HTTP-to-gRPC
+  // table.
+  const httpStatus =
+    typeof fetchError.status === 'number'
+      ? fetchError.status
+      : fetchError.response?.status;
+  if (typeof httpStatus === 'number') {
+    error.code = rpcCodeFromHttpStatusCode(httpStatus);
+    return error;
+  }
+
+  // An explicit cancellation and an elapsed deadline are distinct conditions in
+  // gRPC, so they are separated out before the general case below. Native fetch
+  // reports both as a DOMException, where `code` is a numeric DOMException
+  // value (20 and 23) rather than a string, so match on `name` as the rest of
+  // this file does when it detects cancellation. A string `code` is also
+  // accepted, because gaxios normalizes a DOMException's name onto `code`.
+  if (err.name === 'AbortError' || fetchError.code === 'AbortError') {
+    error.code = Status.CANCELLED;
+    return error;
+  }
+  if (err.name === 'TimeoutError' || fetchError.code === 'TimeoutError') {
+    error.code = Status.DEADLINE_EXCEEDED;
+    return error;
+  }
+
+  // Anything else that rejects here failed before producing a response, which
+  // gRPC reports as UNAVAILABLE irrespective of the underlying system error:
+  // @grpc/grpc-js defaults transport failures to UNAVAILABLE and only inspects
+  // errno to refine an HTTP/2 INTERNAL_ERROR. Enumerating errnos here would
+  // classify anything left off the list as non-retryable, so follow gRPC and
+  // treat the whole category uniformly. Errors raised while decoding a response
+  // are handled nearer the decoder and do not reach this point.
+  error.code = Status.UNAVAILABLE;
+  return error;
+}
+
 export function generateServiceStub(
   rpcs: {[name: string]: protobuf.Method},
   protocol: string,
@@ -164,6 +241,10 @@
         method: fetchParameters.method,
         signal: cancelSignal,
         responseType: 'stream', // ensure gaxios returns the data directly so that it handle data/streams itself
+        // Error responses must resolve so that they are decoded below into a
+        // GoogleError carrying a gRPC status code. 401 and 403 keep rejecting
+        // so that the auth client can refresh credentials and retry.
+        validateStatus: (status: number) => status !== 401 && status !== 403,
         agent: agentOption || undefined,
       };
 
@@ -198,7 +279,7 @@
                     (err instanceof Error && err.name !== 'AbortError'))
                 ) {
                   if (callback) {
                     callback(err);
                   }
                   streamArrayParser.emit('error', err);
                 }
@@ -210,15 +291,15 @@
               Promise.resolve(response.ok),
               response.arrayBuffer(),
             ])
               .then(([ok, buffer]: [boolean, Buffer | ArrayBuffer]) => {
                 const response = responseDecoder(rpc, ok, buffer);
                 callback!(null, response);
               })
               .catch((err: Error) => {
                 if (!cancelRequested || err.name !== 'AbortError') {
                   if (rpc.responseStream) {
                     if (callback) {
                       callback(err);
                     }
                     streamArrayParser.emit('error', err);
                   } else {
@@ -245,15 +326,16 @@
           }
         })
         .catch((err: unknown) => {
+          const translated = _toGoogleError(err);
           if (rpc.responseStream) {
             if (callback) {
-              callback(err);
+              callback(translated);
             }
-            streamArrayParser.emit('error', err);
+            streamArrayParser.emit('error', translated);
           } else if (callback) {
-            callback(err);
+            callback(translated);
           } else {
-            throw err;
+            throw translated;
           }
         });
 

core/packages/gax/test/unit/grpc-fallback.ts

@@ -23,9 +23,14 @@ import * as protobuf from 'protobufjs';
 import * as sinon from 'sinon';
 import echoProtoJson = require('../fixtures/echo.json');
 import {GrpcClient} from '../../src/fallback';
-import {ClientStubOptions, GoogleAuth, GoogleError} from '../../src';
+import {ClientStubOptions, GoogleAuth, GoogleError, Status} from '../../src';
+import {StreamArrayParser} from '../../src/streamArrayParser';
 import {PassThroughClient} from 'google-auth-library';
-import {setMockFallbackResponse} from './utils';
+import {
+  setMockFallbackError,
+  setMockFallbackHttpResponse,
+  setMockFallbackResponse,
+} from './utils';
 
 let authClient = new PassThroughClient();
 let opts = {
@@ -539,4 +544,237 @@ describe('grpc-fallback', () => {
     const stub = await gaxGrpc.createStub(echoService, stubOptions);
     stub.close({}, {}, {}, () => {});
   });
+
+  describe('transport error translation', () => {
+    // Errors surfaced by the transport must carry a numeric gRPC status code:
+    // retry logic in normalCalls/retries.ts matches `err.code` against the
+    // numeric retry codes from the service config, and client libraries branch
+    // on the same codes.
+    function callEcho(): Promise<GoogleError> {
+      return gaxGrpc.createStub(echoService, stubOptions).then(
+        echoStub =>
+          new Promise<GoogleError>(resolve => {
+            echoStub.echo({content: 'test-content'}, {}, {}, (err?: Error) =>
+              resolve(err as GoogleError),
+            );
+          }),
+      );
+    }
+
+    it('should translate a connection failure into UNAVAILABLE', async () => {
+      // e.g. a "socket hang up" when the server closes a keep-alive socket.
+      const fetchError = Object.assign(
+        new Error(
+          'request to https://foo.example.com failed, reason: socket hang up',
+        ),
+        {code: 'ECONNRESET'},
+      );
+      setMockFallbackError(gaxGrpc, fetchError);
+
+      const err = await callEcho();
+
+      assert(err instanceof GoogleError);
+      assert.strictEqual(err.code, Status.UNAVAILABLE);
+      assert.strictEqual(err.cause, fetchError);
+    });
+
+    it('should translate a timeout into DEADLINE_EXCEEDED', async () => {
+      // Native fetch rejects with a DOMException whose `code` is the numeric
+      // DOMException value (23), not a string, so the name is what identifies it.
+      setMockFallbackError(
+        gaxGrpc,
+        new DOMException(
+          'The operation was aborted due to timeout',
+          'TimeoutError',
+        ),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED);
+    });
+
+    it('should translate a timeout reported only by code into DEADLINE_EXCEEDED', async () => {
+      setMockFallbackError(
+        gaxGrpc,
+        Object.assign(new Error('The operation was aborted due to timeout'), {
+          code: 'TimeoutError',
+        }),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED);
+    });
+
+    it('should translate an aborted request into CANCELLED', async () => {
+      // As above: native fetch reports abort as DOMException code 20.
+      setMockFallbackError(
+        gaxGrpc,
+        new DOMException('This operation was aborted', 'AbortError'),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.CANCELLED);
+    });
+
+    it('should translate an abort reported only by code into CANCELLED', async () => {
+      setMockFallbackError(
+        gaxGrpc,
+        Object.assign(new Error('This operation was aborted'), {
+          code: 'AbortError',
+        }),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.CANCELLED);
+    });
+
+    it('should use UNAVAILABLE for an unrecognized transport error', async () => {
+      // An error that rejected before a response was produced is a transport
+      // failure, which gRPC reports as UNAVAILABLE whatever the cause. Codes
+      // left off a list would otherwise be classified as non-retryable.
+      setMockFallbackError(gaxGrpc, new Error('something unexpected'));
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.UNAVAILABLE);
+      assert.strictEqual(err.message, 'something unexpected');
+    });
+
+    it('should translate a socket timeout into UNAVAILABLE', async () => {
+      // @grpc/grpc-js maps ETIMEDOUT to UNAVAILABLE, not DEADLINE_EXCEEDED,
+      // which is reserved for an elapsed call deadline.
+      setMockFallbackError(
+        gaxGrpc,
+        Object.assign(new Error('socket timeout'), {code: 'ETIMEDOUT'}),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.UNAVAILABLE);
+    });
+
+    it('should map a rejection carrying an HTTP status onto a gRPC status', async () => {
+      // 401 and 403 responses are rejected by the transport on purpose, so that
+      // the auth client can refresh credentials and retry.
+      setMockFallbackError(
+        gaxGrpc,
+        Object.assign(
+          new Error('Request had invalid authentication credentials'),
+          {
+            status: 401,
+          },
+        ),
+      );
+
+      const err = await callEcho();
+
+      assert.strictEqual(err.code, Status.UNAUTHENTICATED);
+    });
+
+    it('should not rewrap an error that is already a GoogleError', async () => {
+      const googleError = new GoogleError('already translated');
+      googleError.code = Status.FAILED_PRECONDITION;
+      setMockFallbackError(gaxGrpc, googleError);
+
+      const err = await callEcho();
+
+      assert.strictEqual(err, googleError);
+      assert.strictEqual(err.code, Status.FAILED_PRECONDITION);
+    });
+
+    it('should translate transport errors on server streaming calls', async () => {
+      setMockFallbackError(
+        gaxGrpc,
+        Object.assign(new Error('socket hang up'), {code: 'ECONNRESET'}),
+      );
+
+      const echoStub = await gaxGrpc.createStub(echoService, stubOptions);
+      const stream = echoStub.expand(
+        {content: 'test content'},
+        {},
+        {},
+        () => {},
+      );
+
+      const err = await new Promise<GoogleError>(resolve => {
+        (stream as StreamArrayParser).on('error', resolve);
+      });
+
+      assert.strictEqual(err.code, Status.UNAVAILABLE);
+    });
+
+    it('should let the auth client handle 401 and 403, and decode every other status', async () => {
+      const requestOptions = setMockFallbackError(gaxGrpc, new Error('unused'));
+
+      await callEcho();
+
+      const validateStatus = requestOptions[0].validateStatus;
+      assert(validateStatus, 'the transport must set validateStatus');
+      // Rejected, so that AuthClient can refresh credentials and retry.
+      assert.strictEqual(validateStatus(401), false);
+      assert.strictEqual(validateStatus(403), false);
+      // Resolved, so that the response is decoded into a GoogleError with a
+      // gRPC status code rather than surfacing as a raw transport error.
+      assert.strictEqual(validateStatus(404), true);
+      assert.strictEqual(validateStatus(429), true);
+      assert.strictEqual(validateStatus(503), true);
+      assert.strictEqual(validateStatus(200), true);
+    });
+
+    it('should decode a non-2xx response into its canonical gRPC status', async () => {
+      // The primary half of the fix: validateStatus lets this response resolve,
+      // so it reaches decodeResponse and is parsed rather than surfacing as a
+      // raw transport rejection. parseHttpError prefers the canonical status
+      // name in the body.
+      setMockFallbackHttpResponse(
+        gaxGrpc,
+        new Response(
+          Buffer.from(
+            JSON.stringify({
+              error: {
+                code: 409,
+                message: 'Too much contention on these documents.',
+                status: 'ABORTED',
+              },
+            }),
+          ),
+          {status: 409},
+        ),
+      );
+
+      const err = await callEcho();
+
+      assert(err instanceof GoogleError);
+      assert.strictEqual(err.code, Status.ABORTED);
+      // The code must denote the status the server named, which is the same
+      // value the gRPC transport reports for this condition.
+      assert.strictEqual(Status[err.code!], 'ABORTED');
+      // The HTTP status must not leak through as the error code.
+      assert.notStrictEqual(err.code as number, 409);
+    });
+
+    it('should derive a code from the HTTP status when the body has none', async () => {
+      setMockFallbackHttpResponse(
+        gaxGrpc,
+        new Response(
+          Buffer.from(
+            JSON.stringify({
+              error: {code: 503, message: 'The service is currently down.'},
+            }),
+          ),
+          {status: 503},
+        ),
+      );
+
+      const err = await callEcho();
+
+      assert(err instanceof GoogleError);
+      assert.strictEqual(err.code, Status.UNAVAILABLE);
+      assert.notStrictEqual(err.code as number, 503);
+    });
+  });
 });