diff --git a/core/packages/gax/src/fallbackServiceStub.ts b/core/packages/gax/src/fallbackServiceStub.ts index 119e3b3aab4e..b1dc131d734f 100644 --- a/core/packages/gax/src/fallbackServiceStub.ts +++ b/core/packages/gax/src/fallbackServiceStub.ts @@ -22,6 +22,8 @@ import * as serializer from 'proto3-json-serializer'; 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'; @@ -82,6 +84,81 @@ function _formatEmptyResponse(rpc: protobuf.Method) { 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 @@ export function generateServiceStub( 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, }; @@ -245,15 +326,16 @@ export function generateServiceStub( } }) .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; } }); diff --git a/core/packages/gax/test/unit/grpc-fallback.ts b/core/packages/gax/test/unit/grpc-fallback.ts index 732c52e298d4..42f745c04fbe 100644 --- a/core/packages/gax/test/unit/grpc-fallback.ts +++ b/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 { + return gaxGrpc.createStub(echoService, stubOptions).then( + echoStub => + new Promise(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(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); + }); + }); }); diff --git a/core/packages/gax/test/unit/utils.ts b/core/packages/gax/test/unit/utils.ts index 81440541a1fb..017d94b440bb 100644 --- a/core/packages/gax/test/unit/utils.ts +++ b/core/packages/gax/test/unit/utils.ts @@ -143,3 +143,80 @@ export function setMockFallbackResponse( const authClient = new MockedResponseAuthClient(); gaxGrpc.auth = new GoogleAuth({authClient}); } + +/** + * Makes a Fallback request fail the way the real transport does: by rejecting + * from the auth client rather than resolving with a failed response. This is + * what gaxios does for network-level failures, and for HTTP statuses that do + * not pass `validateStatus`. + * + * @param gaxGrpc The gRPC Client to use + * @param error The error the transport should reject with + * @returns The request options the transport was called with + */ +export function setMockFallbackError(gaxGrpc: GrpcClient, error: Error) { + const requestOptions: gaxios.GaxiosOptions[] = []; + + class MockedErrorAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + requestOptions.push(opts); + throw error; + } + } + + const authClient = new MockedErrorAuthClient(); + gaxGrpc.auth = new GoogleAuth({authClient}); + + return requestOptions; +} + +/** + * Sets a response for a Fallback request, reproducing how gaxios decides + * between resolving and rejecting. + * + * Unlike {@link setMockFallbackResponse}, which always resolves, this applies + * the request's `validateStatus` predicate (defaulting to gaxios' own 2xx-only + * rule when the caller does not supply one) and rejects with a GaxiosError-shaped + * error when it fails. Transports that do not opt in to receiving error + * responses therefore see a rejection here, exactly as they would in production. + * + * @param gaxGrpc The gRPC Client to use + * @param response The Response object to use + */ +export function setMockFallbackHttpResponse( + gaxGrpc: GrpcClient, + response: Response, +) { + class MockedHttpAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + const validateStatus = + opts.validateStatus ?? + ((status: number) => status >= 200 && status < 300); + + if (!validateStatus(response.status)) { + throw Object.assign( + new Error(`Request failed with status code ${response.status}`), + { + status: response.status, + response: {status: response.status}, + }, + ); + } + + return Object.assign(response, { + config: { + headers: response.headers, + url: new URL(opts.url || 'https://example.com'), + }, + data: response.body as T, + }); + } + } + + const authClient = new MockedHttpAuthClient(); + gaxGrpc.auth = new GoogleAuth({authClient}); +}