diff --git a/core/packages/gax/src/fallbackServiceStub.ts b/core/packages/gax/src/fallbackServiceStub.ts index 119e3b3aab4..836d6830f5c 100644 --- a/core/packages/gax/src/fallbackServiceStub.ts +++ b/core/packages/gax/src/fallbackServiceStub.ts @@ -14,7 +14,9 @@ * limitations under the License. */ -import type {Response as NodeFetchResponse} from 'node-fetch' with {'resolution-mode': 'import'}; +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 +24,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 {Status} from './status'; import {pipeline, PipelineSource} from 'stream'; import type {Agent as HttpAgent} from 'http'; import type {Agent as HttpsAgent} from 'https'; @@ -33,8 +37,7 @@ import type {Agent as HttpsAgent} from 'https'; // - 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; + ((parsedUrl: {protocol: string}) => HttpAgent | HttpsAgent) | null = null; if (isNodeJS()) { const http = require('http'); const https = require('https'); @@ -82,6 +85,41 @@ function _formatEmptyResponse(rpc: protobuf.Method) { return resp; } +/** + * Reports an expired deadline the way the rest of gax expects. + * + * `retryCodes` matching, caller `err.code` comparisons and telemetry's + * `error.type` all key off the numeric gRPC status, and gRPC reports this + * condition as DEADLINE_EXCEEDED, so a REST deadline is surfaced identically + * rather than leaking a transport-specific error for a failure both transports + * share. The original error is kept as `cause`. + * + * Whether the deadline expired is passed in rather than inferred from `err`, + * because the error carries no usable evidence of it. Measured against a server + * that accepts a connection and never replies: node-fetch discards + * `signal.reason` and throws its own AbortError, gaxios wraps that in a + * GaxiosError which never sets `name` (so it stays the inherited 'Error') and + * only copies `code` from a DOMException cause, which this is not. The result + * is `name: 'Error'`, `code: undefined` — byte-identical to what `cancel()` + * produces. Only the caller, which armed the timer, knows which happened. + */ +function toDeadlineExceeded( + err: unknown, + rpcName: string, + timeoutMs: number | undefined, + timedOut: boolean, +): unknown { + if (!timedOut || timeoutMs === undefined) { + return err; + } + const error = new GoogleError( + `Deadline exceeded: ${rpcName} did not respond within ${timeoutMs} milliseconds.`, + {cause: err}, + ); + error.code = Status.DEADLINE_EXCEEDED; + return error; +} + export function generateServiceStub( rpcs: {[name: string]: protobuf.Method}, protocol: string, @@ -112,13 +150,38 @@ export function generateServiceStub( }, }; for (const [rpcName, rpc] of Object.entries(rpcs)) { + // Named for what gax actually passes, which is its `UnaryCall` order: + // (request, metadata, options, callback). `FallbackServiceStub` declares + // the middle two the other way round, so the third argument — the one gRPC + // calls `options`, carrying the deadline — reads as metadata there and was + // long ignored as such. serviceStub[rpcName] = ( request: {}, - options?: {[name: string]: string}, - _metadata?: {} | Function, + metadata?: {[name: string]: string | string[]}, + callOptions?: {deadline?: Date}, callback?: Function, ) => { - options ??= {}; + metadata ??= {}; + + // `addTimeoutArg` sets a deadline on every call and `CallSettings.timeout` + // defaults to 30s, so one is essentially always present. gRPC enforces its + // own deadline, but nothing here ever read this one, so an endpoint that + // accepted the connection and then went quiet left the request — and the + // promise or callback waiting on it — outstanding forever. Convert it to + // the remaining duration and arm an abort signal with it below. + // + // Server-streaming RPCs are deliberately excluded. Their response is + // long-lived by design and the signal stays armed once the body starts + // flowing, so forwarding the deadline would abort a healthy stream + // mid-read. Bounding those is a separate, user-visible change. + let timeoutMs: number | undefined; + if (callOptions?.deadline && !rpc.responseStream) { + // `AbortSignal.timeout` rejects a negative delay with a RangeError, so + // an already-expired deadline is clamped. Zero is a fine value here: it + // aborts on the next tick, which is the right answer for a deadline + // that has already passed. + timeoutMs = Math.max(0, callOptions.deadline.getTime() - Date.now()); + } // We cannot use async-await in this function because we need to return the canceller object as soon as possible. // Using plain old promises instead. @@ -148,10 +211,40 @@ export function generateServiceStub( const cancelController = new AbortController(); const cancelSignal = cancelController.signal as AbortSignal; let cancelRequested = false; + + // Arm the deadline here rather than handing `timeout` to gaxios, which + // would build the identical `AbortSignal.timeout` internally. The + // difference is bookkeeping: both a deadline expiry and a `cancel()` + // abort the same request and surface the same error, so unless we record + // which one fired, the handlers below cannot tell them apart. + let timedOut = false; + let requestSignal = cancelSignal; + if (timeoutMs !== undefined) { + const timeoutSignal = AbortSignal.timeout(timeoutMs); + timeoutSignal.addEventListener('abort', () => (timedOut = true), { + once: true, + }); + requestSignal = AbortSignal.any([cancelSignal, timeoutSignal]); + } + const url = fetchParameters.url; const headers = new Headers(fetchParameters.headers); - for (const key of Object.keys(options)) { - headers.set(key, options[key][0]); + // gRPC metadata is multi-valued, and `buildMetadata` normalizes every + // value to an array for exactly that reason. This used to read + // `metadata[key][0]`, which dropped every value after the first and, for + // a value that was a plain string rather than an array, sent only its + // first character. Replace whatever the request encoder set, as the + // single-value `set` did, then keep all of the values. + for (const key of Object.keys(metadata)) { + const value = metadata[key]; + if (Array.isArray(value)) { + headers.delete(key); + for (const item of value) { + headers.append(key, String(item)); + } + } else { + headers.set(key, String(value)); + } } const streamArrayParser = new StreamArrayParser(rpc); let response204Ok = false; @@ -162,7 +255,7 @@ export function generateServiceStub( ? fetchParameters.body : Buffer.from(fetchParameters.body), method: fetchParameters.method, - signal: cancelSignal, + signal: requestSignal, responseType: 'stream', // ensure gaxios returns the data directly so that it handle data/streams itself agent: agentOption || undefined, }; @@ -213,14 +306,30 @@ export function generateServiceStub( .then(([ok, buffer]: [boolean, Buffer | ArrayBuffer]) => { const response = responseDecoder(rpc, ok, buffer); callback!(null, response); + return; }) .catch((err: Error) => { - if (!cancelRequested || err.name !== 'AbortError') { + // The deadline can expire after the response headers arrive but + // before the body is fully read, which rejects here rather than + // in the outer handler. + const callErr = toDeadlineExceeded( + err, + rpcName, + timeoutMs, + timedOut, + ); + // A caller that cancelled does not need the resulting abort + // reported back to it, but a deadline always does. This used to + // test `err.name !== 'AbortError'`; gaxios wraps node-fetch's + // AbortError and never sets its own `name`, leaving the + // inherited 'Error', so the check never matched and cancelled + // calls still reported an error. Use the state we recorded. + if (timedOut || !cancelRequested) { if (rpc.responseStream) { if (callback) { - callback(err); + callback(callErr); } - streamArrayParser.emit('error', err); + streamArrayParser.emit('error', callErr); } else { // This supports a legacy Apiary behavior that allows // empty 204 responses. If we do not intercept this potential error @@ -232,7 +341,7 @@ export function generateServiceStub( if (!response204Ok) { // by this point, we're guaranteed to have added a callback // it is added in the library before calling this.innerApiCalls - callback!(err); + callback!(callErr); } else { const resp = _formatEmptyResponse(rpc); // by this point, we're guaranteed to have added a callback @@ -244,7 +353,10 @@ export function generateServiceStub( }); } }) - .catch((err: unknown) => { + .catch((rawErr: unknown) => { + // The usual timeout path: the deadline expired before any response + // was received, so the fetch itself rejects. + const err = toDeadlineExceeded(rawErr, rpcName, timeoutMs, timedOut); if (rpc.responseStream) { if (callback) { callback(err); diff --git a/core/packages/gax/test/unit/grpc-fallback.ts b/core/packages/gax/test/unit/grpc-fallback.ts index 732c52e298d..ca026839d90 100644 --- a/core/packages/gax/test/unit/grpc-fallback.ts +++ b/core/packages/gax/test/unit/grpc-fallback.ts @@ -21,10 +21,20 @@ import assert from 'assert'; import {describe, it, beforeEach, afterEach, after} from 'mocha'; import * as protobuf from 'protobufjs'; import * as sinon from 'sinon'; +import * as stream from 'stream'; import echoProtoJson = require('../fixtures/echo.json'); import {GrpcClient} from '../../src/fallback'; -import {ClientStubOptions, GoogleAuth, GoogleError} from '../../src'; -import {PassThroughClient} from 'google-auth-library'; +import { + CallSettings, + ClientStubOptions, + GoogleAuth, + GoogleError, + Status, + createApiCall, +} from '../../src'; +import {GRPCCall} from '../../src/apitypes'; +import {StreamArrayParser} from '../../src/streamArrayParser'; +import {gaxios, PassThroughClient} from 'google-auth-library'; import {setMockFallbackResponse} from './utils'; let authClient = new PassThroughClient(); @@ -539,4 +549,407 @@ describe('grpc-fallback', () => { const stub = await gaxGrpc.createStub(echoService, stubOptions); stub.close({}, {}, {}, () => {}); }); + + // `setMockFallbackResponse` discards the options it is handed, but the + // deadline and metadata handling under test are only observable there, so + // record them. + function recordRequests( + client: GrpcClient, + response: Response, + ): gaxios.GaxiosOptions[] { + const requests: gaxios.GaxiosOptions[] = []; + class RecordingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + requests.push(opts); + return Object.assign(response, { + config: { + headers: response.headers, + url: new URL(opts.url || 'https://example.com'), + }, + data: response.body as T, + }); + } + } + client.auth = new GoogleAuth({authClient: new RecordingAuthClient()}); + return requests; + } + + describe('call metadata', () => { + async function headersSentFor(metadata: { + [name: string]: string | string[]; + }): Promise { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo({content: 'test'}, metadata, {}, () => resolve()); + }); + + return requests[0].headers as Headers; + } + + it('should send every value of a multi-valued header', async () => { + const headers = await headersSentFor({'x-multi': ['a', 'b', 'c']}); + + // gRPC metadata is multi-valued and `buildMetadata` normalizes every + // value to an array precisely because of that. Reading index 0 silently + // dropped the rest. The Headers API joins repeated values with ', '. + assert.strictEqual(headers.get('x-multi'), 'a, b, c'); + }); + + it('should send a single-valued header as its only value', async () => { + const headers = await headersSentFor({'x-single': ['one']}); + + assert.strictEqual(headers.get('x-single'), 'one'); + }); + + it('should send a plain string value whole', async () => { + const headers = await headersSentFor({'x-plain': 'hello'}); + + // Indexing a string yields its first character, so this used to arrive + // as 'h'. The declared parameter type said the values were strings while + // the code indexed them as arrays; both could not be right. + assert.strictEqual(headers.get('x-plain'), 'hello'); + }); + + it('should let metadata replace a header set by the request encoder', async () => { + const headers = await headersSentFor({ + 'content-type': ['application/x-custom'], + }); + + // The previous `headers.set` replaced rather than appended, so keeping + // every value must not turn an override into an accumulation. + assert.strictEqual(headers.get('content-type'), 'application/x-custom'); + }); + }); + + describe('call deadline', () => { + function signalOf(request: gaxios.GaxiosOptions): AbortSignal | undefined { + return request.signal as AbortSignal | undefined; + } + + // Resolves true if the signal aborts within the budget, false if it does + // not. A budget rather than a bare `aborted` read, because the abort is + // asynchronous and a test that only sampled it would pass for the wrong + // reason. + function abortedWithin( + signal: AbortSignal | undefined, + ms: number, + ): Promise { + if (!signal) { + return Promise.resolve(false); + } + if (signal.aborted) { + return Promise.resolve(true); + } + return new Promise(resolve => { + const timer = setTimeout(() => resolve(false), ms); + signal.addEventListener( + 'abort', + () => { + clearTimeout(timer); + resolve(true); + }, + {once: true}, + ); + }); + } + + // The error an aborted request actually produces, measured end to end + // against a server that accepts the connection and never replies: + // node-fetch discards `signal.reason` and throws its own AbortError, + // gaxios wraps that without setting `name` (so it stays the inherited + // 'Error') and only copies `code` from a DOMException cause, which this is + // not. A `cancel()` produces a byte-identical error. Earlier versions of + // these tests fabricated a `TimeoutError` that never occurs in production + // and so passed against a translation that was dead code. + function abortError(): Error { + const cause = new Error('The operation was aborted.'); + cause.name = 'AbortError'; + return new Error('The operation was aborted.', {cause}); + } + + // Rejects as soon as the request is aborted. `cancel()` can run before the + // asynchronous auth chain ever reaches the transport, and a listener added + // to an already-aborted signal never fires, so check the state first. + function rejectWhenAborted( + signal: AbortSignal | undefined, + ): Promise { + return new Promise((_resolve, reject) => { + if (!signal) { + return; + } + if (signal.aborted) { + reject(abortError()); + } else { + signal.addEventListener('abort', () => reject(abortError()), { + once: true, + }); + } + }); + } + + // A transport that is actually bound by the signal: it stays pending until + // the request is aborted, then rejects the way the real one does. + function rejectOnAbort(client: GrpcClient): gaxios.GaxiosOptions[] { + const requests: gaxios.GaxiosOptions[] = []; + class AbortingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + requests.push(opts); + return rejectWhenAborted(signalOf(opts)); + } + } + client.auth = new GoogleAuth({authClient: new AbortingAuthClient()}); + return requests; + } + + // Aborts after the response headers arrive but before the body is read, + // which rejects in the stub's inner handler rather than the outer one. + function rejectDuringBodyRead(client: GrpcClient) { + class BodyAbortingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + const signal = signalOf(opts); + return { + ok: true, + status: 200, + headers: new Headers(), + arrayBuffer: () => rejectWhenAborted(signal), + } as unknown as gaxios.GaxiosResponse; + } + } + client.auth = new GoogleAuth({authClient: new BodyAbortingAuthClient()}); + } + + it('should abort the in-flight request when the deadline expires', async () => { + const requests = rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + () => resolve(), + ); + }); + + // The transport never answered. Before this, nothing read the deadline, + // so the request and the callback waiting on it stayed outstanding. + assert.strictEqual(signalOf(requests[0])?.aborted, true); + }); + + it('should carry CallSettings.timeout through to the transport', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + // Every other test here hands the stub a deadline directly, which only + // exercises the stub itself. This one goes through `createApiCall`, the + // path a generated client takes, so `addTimeoutArg` is what produces the + // deadline. That hand-off is the seam where the deadline used to be + // dropped, and no direct call to the stub can see it. + const apiCall = createApiCall( + Promise.resolve(echoStub.echo as unknown as GRPCCall), + new CallSettings({timeout: 50}), + ); + + await assert.rejects( + apiCall({content: 'test'}, {}) as unknown as Promise, + (err: unknown) => { + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + return true; + }, + ); + }); + + it('should not abort a call that has no deadline', async () => { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo({content: 'test'}, {}, {}, () => resolve()); + }); + + assert.strictEqual( + await abortedWithin(signalOf(requests[0]), 100), + false, + ); + }); + + it('should abort promptly, and not throw, for an already-expired deadline', async () => { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + // `AbortSignal.timeout` rejects a negative delay with a RangeError, so + // an expired deadline that was not clamped would throw out of the stub + // before the request was ever made. Zero is the right clamp: the call + // has no time left, so it should abort on the next tick. + await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() - 60000)}, + () => resolve(), + ); + }); + + assert.strictEqual(await abortedWithin(signalOf(requests[0]), 100), true); + }); + + it('should not bound server-streaming calls by the deadline', async () => { + const responseStream = new stream.Readable(); + responseStream.push(JSON.stringify([{content: 'test'}])); + responseStream.push(null); + const requests = recordRequests( + gaxGrpc, + new Response(responseStream as unknown as BodyInit), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const responses = echoStub.expand( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + () => {}, + ) as StreamArrayParser; + await new Promise((resolve, reject) => { + responses.on('data', () => {}); + responses.on('error', reject); + responses.on('end', resolve); + }); + + // A server stream is long-lived by design; the signal would stay armed + // once the body starts flowing and abort a healthy read. + assert.strictEqual( + await abortedWithin(signalOf(requests[0]), 100), + false, + ); + }); + + it('should report an expired deadline as DEADLINE_EXCEEDED', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + (err?: Error) => resolve(err), + ); + }); + + // gRPC reports this condition with a numeric status, and retryCodes, + // caller `err.code` checks and telemetry all key off that, so the REST + // path must not leak the transport's own error shape. + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + assert.match(err.message, /Deadline exceeded/); + }); + + it('should report a deadline that expires while the body is being read', async () => { + rejectDuringBodyRead(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + (err?: Error) => resolve(err), + ); + }); + + // Headers arriving in time does not mean the call met its deadline. + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + }); + + it('should not report a cancelled call as DEADLINE_EXCEEDED', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + const call = echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 5000)}, + (err?: Error) => resolve(err), + ); + (call as {cancel: () => void}).cancel(); + }); + + // A deadline was armed here but never expired; the caller gave up first. + // This is the case no amount of error inspection can get right, because + // the abort a cancel produces is byte-identical to the one a timeout + // produces. Only the stub, which armed the timer, knows which fired. + assert(!(err instanceof GoogleError)); + }); + + it('should leave an abort error alone when no deadline was forwarded', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + const call = echoStub.echo({content: 'test'}, {}, {}, (err?: Error) => + resolve(err), + ); + (call as {cancel: () => void}).cancel(); + }); + + // Nothing armed a deadline, so this abort came from the caller, and + // reporting a deadline that was never set would be a fabrication. The + // error itself is byte-identical to a timeout's, which is why the + // translation is gated on the flag the stub records rather than on + // anything read back off the error. + assert(!(err instanceof GoogleError)); + assert.strictEqual((err?.cause as Error | undefined)?.name, 'AbortError'); + }); + + it('should not report an error when the caller cancelled', async () => { + rejectDuringBodyRead(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + let callbackErr: unknown; + let callbackCalled = false; + const call = echoStub.echo( + {content: 'test'}, + {}, + {}, + (err?: Error, resp?: {}) => { + callbackCalled = true; + callbackErr = err ?? resp; + }, + ); + (call as {cancel: () => void}).cancel(); + + await new Promise(resolve => setTimeout(resolve, 100)); + + // A caller that cancelled does not need the resulting abort reported + // back to it. The guard here used to test `err.name !== 'AbortError'`, + // but gaxios wraps node-fetch's AbortError and never sets its own + // `name`, leaving the inherited 'Error', so the check never matched and + // cancelled calls reported an error anyway. + assert.strictEqual( + callbackCalled, + false, + `callback was invoked with ${callbackErr}`, + ); + }); + }); });