From a943257a0402b26fd80196057a9724fd28fc5c1b Mon Sep 17 00:00:00 2001 From: surbhigarg92 Date: Wed, 26 Jun 2024 11:50:21 +0200 Subject: [PATCH 1/2] fix: Retry with timeout (#2071) Use `gaxOptions.timeout` during retry in streaming calls. Earlier the timeout value was only used for a single RPC not for the whole operation including retries. Now if RPC returns `Unavailable` error and the timeout value has been reached, library will throw an Deadline exceeded error. ``` const query = { sql: 'Select 1', gaxOptions: {timeout: 500} } const [rows] = await database.run(query); ``` --- src/partial-result-stream.ts | 18 ++++++++++++++++-- src/transaction.ts | 2 ++ test/partial-result-stream.ts | 32 ++++++++++++++++++++++++++++++++ test/transaction.ts | 8 ++++++++ 4 files changed, 58 insertions(+), 2 deletions(-) diff --git a/src/partial-result-stream.ts b/src/partial-result-stream.ts index 616cdc88b..69439f534 100644 --- a/src/partial-result-stream.ts +++ b/src/partial-result-stream.ts @@ -22,8 +22,8 @@ import mergeStream = require('merge-stream'); import {common as p} from 'protobufjs'; import {Readable, Transform} from 'stream'; import * as streamEvents from 'stream-events'; -import {grpc} from 'google-gax'; -import {isRetryableInternalError} from './transaction-runner'; +import {grpc, CallOptions} from 'google-gax'; +import {DeadlineError, isRetryableInternalError} from './transaction-runner'; import {codec, JSONOptions, Json, Field, Value} from './codec'; import {google} from '../protos/protos'; @@ -96,6 +96,7 @@ export interface RowOptions { * }; */ columnsMetadata?: object; + gaxOptions?: CallOptions; } /** @@ -491,6 +492,8 @@ export function partialResultStream( const maxQueued = 10; let lastResumeToken: ResumeToken; let lastRequestStream: Readable; + const startTime = Date.now(); + const timeout = options?.gaxOptions?.timeout ?? Infinity; // mergeStream allows multiple streams to be connected into one. This is good; // if we need to retry a request and pipe more data to the user's stream. @@ -541,6 +544,17 @@ export function partialResultStream( }; const retry = (err: grpc.ServiceError): void => { + const elapsed = Date.now() - startTime; + if (elapsed >= timeout) { + // The timeout has reached so this will flush any rows the + // checkpoint stream has queued. After that, we will destroy the + // user's stream with the Deadline exceeded error. + setImmediate(() => + batchAndSplitOnTokenStream.destroy(new DeadlineError(err)) + ); + return; + } + if ( !( err.code && diff --git a/src/transaction.ts b/src/transaction.ts index d5ec34a30..479262a27 100644 --- a/src/transaction.ts +++ b/src/transaction.ts @@ -704,6 +704,7 @@ export class Snapshot extends EventEmitter { jsonOptions, maxResumeRetries, columnsMetadata, + gaxOptions, }) ?.on('response', response => { if (response.metadata && response.metadata!.transaction && !this.id) { @@ -1210,6 +1211,7 @@ export class Snapshot extends EventEmitter { jsonOptions, maxResumeRetries, columnsMetadata, + gaxOptions, }) .on('response', response => { if (response.metadata && response.metadata!.transaction && !this.id) { diff --git a/test/partial-result-stream.ts b/test/partial-result-stream.ts index 8256221f1..799d29b00 100644 --- a/test/partial-result-stream.ts +++ b/test/partial-result-stream.ts @@ -311,6 +311,38 @@ describe('PartialResultStream', () => { ); }); + it('should get Deadline exceeded error if timeout has reached', done => { + const fakeCheckpointStream = through.obj(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + (fakeCheckpointStream as any).reset = () => {}; + + sandbox.stub(checkpointStream, 'obj').returns(fakeCheckpointStream); + + const firstFakeRequestStream = through.obj(); + + const requestFnStub = sandbox.stub(); + + requestFnStub.onCall(0).callsFake(() => { + setTimeout(() => { + // This causes a new request stream to be created. + firstFakeRequestStream.emit('error', { + code: grpc.status.UNAVAILABLE, + message: 'Error.', + } as grpc.ServiceError); + }, 50); + + return firstFakeRequestStream; + }); + + partialResultStream(requestFnStub, {gaxOptions: {timeout: 0}}) + .on('data', row => {}) + .on('error', err => { + assert.strictEqual(err.code, grpc.status.DEADLINE_EXCEEDED); + assert.strictEqual(requestFnStub.callCount, 1); + done(); + }); + }); + it('should resume if there was a retryable error', done => { // This test will emit four rows total: // - Two rows diff --git a/test/transaction.ts b/test/transaction.ts index 392370ab7..28be1543f 100644 --- a/test/transaction.ts +++ b/test/transaction.ts @@ -391,11 +391,15 @@ describe('Transaction', () => { }); it('should pass along row options', () => { + const gaxOptions = { + timeout: 60, + }; const fakeOptions = { json: true, jsonOptions: {a: 'b'}, maxResumeRetries: 10, columnsMetadata: {column1: {test: 'ss'}, column2: Function}, + gaxOptions: gaxOptions, }; snapshot.createReadStream(TABLE, fakeOptions); @@ -766,11 +770,15 @@ describe('Transaction', () => { }); it('should pass along row options', () => { + const gaxOptions = { + timeout: 60, + }; const expectedOptions = { json: true, jsonOptions: {a: 'b'}, maxResumeRetries: 10, columnsMetadata: {column1: {test: 'ss'}, column2: Function}, + gaxOptions: gaxOptions, }; const fakeQuery = Object.assign({}, QUERY, expectedOptions); From 30151cad2723d6a9f2c70342c81f1a578ee02f6a Mon Sep 17 00:00:00 2001 From: "release-please[bot]" <55107282+release-please[bot]@users.noreply.github.com> Date: Wed, 26 Jun 2024 10:58:21 +0000 Subject: [PATCH 2/2] chore(main): release 7.9.1 (#2072) :robot: I have created a release *beep* *boop* --- ## [7.9.1](https://togithub.com/googleapis/nodejs-spanner/compare/v7.9.0...v7.9.1) (2024-06-26) ### Bug Fixes * Retry with timeout ([#2071](https://togithub.com/googleapis/nodejs-spanner/issues/2071)) ([a943257](https://togithub.com/googleapis/nodejs-spanner/commit/a943257a0402b26fd80196057a9724fd28fc5c1b)) --- This PR was generated with [Release Please](https://togithub.com/googleapis/release-please). See [documentation](https://togithub.com/googleapis/release-please#release-please). --- CHANGELOG.md | 7 +++++++ package.json | 2 +- samples/package.json | 2 +- 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f5b06a1c1..8639238df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ [1]: https://www.npmjs.com/package/nodejs-spanner?activeTab=versions +## [7.9.1](https://github.com/googleapis/nodejs-spanner/compare/v7.9.0...v7.9.1) (2024-06-26) + + +### Bug Fixes + +* Retry with timeout ([#2071](https://github.com/googleapis/nodejs-spanner/issues/2071)) ([a943257](https://github.com/googleapis/nodejs-spanner/commit/a943257a0402b26fd80196057a9724fd28fc5c1b)) + ## [7.9.0](https://github.com/googleapis/nodejs-spanner/compare/v7.8.0...v7.9.0) (2024-06-21) diff --git a/package.json b/package.json index d840d1343..8345d73c2 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "@google-cloud/spanner", "description": "Cloud Spanner Client Library for Node.js", - "version": "7.9.0", + "version": "7.9.1", "license": "Apache-2.0", "author": "Google Inc.", "engines": { diff --git a/samples/package.json b/samples/package.json index 1b9815460..c00361667 100644 --- a/samples/package.json +++ b/samples/package.json @@ -17,7 +17,7 @@ "dependencies": { "@google-cloud/kms": "^4.0.0", "@google-cloud/precise-date": "^4.0.0", - "@google-cloud/spanner": "^7.9.0", + "@google-cloud/spanner": "^7.9.1", "yargs": "^17.0.0", "protobufjs": "^7.0.0" },