diff --git a/CHANGELOG.md b/CHANGELOG.md index 34532b62..e8119d4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Thrift `OpenSession` is now retried on transient failures (HTTP 408, 429, 5xx except 501, and network errors), like other idempotent requests. Thrift `ExecuteStatement` is now also retried on HTTP 429 and on HTTP 503 with a `Retry-After` header, which the server returns without running the statement; other HTTP errors on `ExecuteStatement` are still not retried, so a statement can't run twice. HTTP 408 is now retried on every idempotent request. These retries use the existing `retryMaxAttempts` / `retriesTimeout` budget, matching the kernel backend. **Behavior change:** when that budget runs out, these calls now fail like other retried requests, with a `HiveDriverError` (`Hive driver: when connecting to resource. Max retry count exceeded.` or `... Retry timeout exceeded.`) instead of a `THTTPException` with `statusCode`. - Thrift CloudFetch downloads are now bounded end to end by `retriesTimeout` (default 15 minutes), including every attempt and retry backoff. Previously a stalled download, where the response headers or body never arrived, blocked the fetch forever. A download still running when the deadline expires is aborted and fails with `RetryError` (`TIMEOUT_EXCEEDED`, message `CloudFetch result download failed: Retry timeout exceeded`). (#520) - Thrift connections made through `proxy` now apply `checkServerCertificate`, `customCaCert` and `clientCert`/`clientKey`. Previously the proxied path ignored them and verified against Node's process-wide trust store, so a certificate outside the configured trust set could be accepted. **Compatibility:** proxied connections now verify unless `checkServerCertificate: false` is set; as on direct connections, `NODE_TLS_REJECT_UNAUTHORIZED=0` no longer turns verification off. For an `https` proxy these options also apply to the proxy's own certificate. (#485) - Honor the client `useCloudFetch: false` default on the kernel backend so complete results can be fetched without cloud-storage downloads. (#574) diff --git a/lib/connection/connections/HttpConnection.ts b/lib/connection/connections/HttpConnection.ts index 6472265a..a47cc22b 100644 --- a/lib/connection/connections/HttpConnection.ts +++ b/lib/connection/connections/HttpConnection.ts @@ -13,7 +13,7 @@ import HiveDriverError from '../../errors/HiveDriverError'; import ThriftHttpConnection from './ThriftHttpConnection'; import IRetryPolicy from '../contracts/IRetryPolicy'; import HttpRetryPolicy from './HttpRetryPolicy'; -import UnsentRequestRetryPolicy from './UnsentRequestRetryPolicy'; +import NonIdempotentRetryPolicy from './NonIdempotentRetryPolicy'; // Set by the driver, auth provider or Thrift transport. node-fetch merges // same-named headers case-insensitively into one value, so a caller copy @@ -214,7 +214,7 @@ export default class HttpConnection implements IConnectionProvider { transport: thrift.TBufferedTransport, protocol: thrift.TBinaryProtocol, getRetryPolicy: () => this.getRetryPolicy(), - getUnsentRequestRetryPolicy: async () => new UnsentRequestRetryPolicy(this.context), + getNonIdempotentRetryPolicy: async () => new NonIdempotentRetryPolicy(this.context), }, { agent, diff --git a/lib/connection/connections/HttpRetryPolicy.ts b/lib/connection/connections/HttpRetryPolicy.ts index 3e6a2e53..43512924 100644 --- a/lib/connection/connections/HttpRetryPolicy.ts +++ b/lib/connection/connections/HttpRetryPolicy.ts @@ -197,7 +197,8 @@ export default class HttpRetryPolicy implements IRetryPolicy= 500 && statusCode !== 501); diff --git a/lib/connection/connections/UnsentRequestRetryPolicy.ts b/lib/connection/connections/NonIdempotentRetryPolicy.ts similarity index 73% rename from lib/connection/connections/UnsentRequestRetryPolicy.ts rename to lib/connection/connections/NonIdempotentRetryPolicy.ts index 14357c52..175dd2ea 100644 --- a/lib/connection/connections/UnsentRequestRetryPolicy.ts +++ b/lib/connection/connections/NonIdempotentRetryPolicy.ts @@ -7,8 +7,10 @@ import RetryError from '../../errors/RetryError'; // socket (ECONNRESET, EPIPE, timeouts, ...) are excluded: the server may already have the request. const UNSENT_REQUEST_ERROR_CODES = new Set(['ECONNREFUSED', 'ENOTFOUND', 'EAI_AGAIN']); -// For operations that are not safe to repeat: retries only requests that never left the client. -export default class UnsentRequestRetryPolicy extends HttpRetryPolicy { +// For operations that are not safe to repeat: retries only requests the server did not run, i.e. +// requests that never left the client, a 429, or a 503 with `Retry-After` (the server asks for a +// retry). Other failures are ambiguous and are not retried. Same rules as the kernel backend. +export default class NonIdempotentRetryPolicy extends HttpRetryPolicy { public async invokeWithRetry(operation: RetryableOperation): Promise { try { return await super.invokeWithRetry(operation); @@ -21,8 +23,8 @@ export default class UnsentRequestRetryPolicy extends HttpRetryPolicy { } } - protected isRetryable(): boolean { - return false; + protected isRetryable({ response }: HttpTransactionDetails): boolean { + return response.status === 429 || (response.status === 503 && response.headers.has('Retry-After')); } protected isRetryableNetworkError(error: unknown): boolean { diff --git a/lib/connection/connections/ThriftHttpConnection.ts b/lib/connection/connections/ThriftHttpConnection.ts index fa686596..bc14d808 100644 --- a/lib/connection/connections/ThriftHttpConnection.ts +++ b/lib/connection/connections/ThriftHttpConnection.ts @@ -60,7 +60,7 @@ interface ThriftHttpConnectionOptions { transport?: TTransportType; protocol?: TProtocolConstructor; getRetryPolicy(): Promise>; - getUnsentRequestRetryPolicy?(): Promise>; + getNonIdempotentRetryPolicy?(): Promise>; } // This type describes a shape of internals of Thrift client object. @@ -76,6 +76,8 @@ type ThriftClient = { }; const retryableThriftMethods = new Set([ + // A repeated OpenSession leaves at most an unused session, which the server expires + 'OpenSession', 'GetOperationStatus', 'CancelOperation', 'CloseOperation', @@ -93,8 +95,8 @@ const retryableThriftMethods = new Set([ 'GetCrossReference', ]); -// Not safe to repeat once the server may have the request, so retried only if it was never sent -const unsentRequestRetryableThriftMethods = new Set(['ExecuteStatement']); +// Not safe to repeat once the server may have run the request, so retried only when it did not +const nonIdempotentThriftMethods = new Set(['ExecuteStatement']); export default class ThriftHttpConnection extends EventEmitter { private readonly url: string; @@ -128,10 +130,10 @@ export default class ThriftHttpConnection extends EventEmitter { } if ( thriftMethodName && - unsentRequestRetryableThriftMethods.has(thriftMethodName) && - this.options.getUnsentRequestRetryPolicy + nonIdempotentThriftMethods.has(thriftMethodName) && + this.options.getNonIdempotentRetryPolicy ) { - return this.options.getUnsentRequestRetryPolicy(); + return this.options.getNonIdempotentRetryPolicy(); } // Don't retry everything that is not explicitly allowed to retry return new NullRetryPolicy(); diff --git a/tests/unit/connection/connections/HttpRetryPolicy.test.ts b/tests/unit/connection/connections/HttpRetryPolicy.test.ts index 35e0c0fc..d2ef8815 100644 --- a/tests/unit/connection/connections/HttpRetryPolicy.test.ts +++ b/tests/unit/connection/connections/HttpRetryPolicy.test.ts @@ -64,9 +64,9 @@ describe('HttpRetryPolicy', () => { } // Status codes between 100 (including) and 500 (excluding) should not be retried - // The only exception is 429 (Too many requests) + // The only exceptions are 408 (Request Timeout) and 429 (Too many requests) for (let status = 100; status < 500; status += 1) { - const expectedResult = status === 429 ? true : false; + const expectedResult = status === 408 || status === 429; expect(policy['isRetryable'](createStub(status))).to.equal(expectedResult); } diff --git a/tests/unit/connection/connections/UnsentRequestRetryPolicy.test.ts b/tests/unit/connection/connections/NonIdempotentRetryPolicy.test.ts similarity index 64% rename from tests/unit/connection/connections/UnsentRequestRetryPolicy.test.ts rename to tests/unit/connection/connections/NonIdempotentRetryPolicy.test.ts index ca01a9a7..3c207f81 100644 --- a/tests/unit/connection/connections/UnsentRequestRetryPolicy.test.ts +++ b/tests/unit/connection/connections/NonIdempotentRetryPolicy.test.ts @@ -1,7 +1,7 @@ import { expect, AssertionError } from 'chai'; import sinon from 'sinon'; import { Request, Response, FetchError } from 'node-fetch'; -import UnsentRequestRetryPolicy from '../../../../lib/connection/connections/UnsentRequestRetryPolicy'; +import NonIdempotentRetryPolicy from '../../../../lib/connection/connections/NonIdempotentRetryPolicy'; import RetryError, { RetryErrorCode } from '../../../../lib/errors/RetryError'; import ClientContextStub from '../../.stubs/ClientContextStub'; @@ -17,7 +17,7 @@ function okResponse() { } function createPolicy(retryMaxAttempts = 5) { - return new UnsentRequestRetryPolicy(new ClientContextStub({ retryDelayMin: 1, retryDelayMax: 2, retryMaxAttempts })); + return new NonIdempotentRetryPolicy(new ClientContextStub({ retryDelayMin: 1, retryDelayMax: 2, retryMaxAttempts })); } async function expectRejection(promise: Promise): Promise { @@ -32,7 +32,11 @@ async function expectRejection(promise: Promise): Promise { return expect.fail('It should throw an error'); } -describe('UnsentRequestRetryPolicy', () => { +function statusResponse(status: number, headers: Record = {}) { + return { request: new Request('http://localhost'), response: new Response(undefined, { status, headers }) }; +} + +describe('NonIdempotentRetryPolicy', () => { for (const code of ['ECONNREFUSED', 'ENOTFOUND', 'EAI_AGAIN']) { it(`retries ${code}, which is raised before the request is sent`, async () => { const operation = sinon.stub(); @@ -64,9 +68,46 @@ describe('UnsentRequestRetryPolicy', () => { }); } - for (const status of [429, 500, 503]) { - it(`does not retry an HTTP ${status} response`, async () => { - const details = { request: new Request('http://localhost'), response: new Response(undefined, { status }) }; + const serverRejections: Array<[string, ReturnType]> = [ + ['HTTP 429', statusResponse(429)], + ['HTTP 429 with Retry-After', statusResponse(429, { 'Retry-After': '1' })], + ['HTTP 503 with Retry-After', statusResponse(503, { 'Retry-After': '1' })], + ]; + + for (const [label, rejection] of serverRejections) { + it(`retries an ${label} response, which the server did not run`, async () => { + const operation = sinon.stub(); + operation.onCall(0).resolves(rejection); + operation.onCall(1).resolves(okResponse()); + + const result = await createPolicy().invokeWithRetry(operation); + + expect(result.response.status).to.equal(200); + expect(operation.callCount).to.equal(2); + }); + } + + it('throws a RetryError once a persistent 429 runs out of attempts', async () => { + const operation = sinon.stub().resolves(statusResponse(429)); + + const error = await expectRejection(createPolicy(3).invokeWithRetry(operation)); + + expect(error).to.be.instanceOf(RetryError); + expect((error as RetryError).errorCode).to.equal(RetryErrorCode.AttemptsExceeded); + expect(operation.callCount).to.equal(3); + }); + + const ambiguousStatuses: Array<[string, ReturnType]> = [ + ['HTTP 503 without Retry-After', statusResponse(503)], + ['HTTP 500', statusResponse(500)], + ['HTTP 500 with Retry-After', statusResponse(500, { 'Retry-After': '1' })], + ['HTTP 502', statusResponse(502)], + ['HTTP 504', statusResponse(504)], + ['HTTP 408', statusResponse(408)], + ]; + + for (const [label, details] of ambiguousStatuses) { + it(`does not retry an ${label} response`, async () => { const policy = createPolicy(); const operation = sinon.stub().resolves(details); @@ -86,7 +127,7 @@ describe('UnsentRequestRetryPolicy', () => { it('rethrows the original error once the retry timeout is exceeded', async () => { const error = systemError('ECONNREFUSED'); - const policy = new UnsentRequestRetryPolicy( + const policy = new NonIdempotentRetryPolicy( new ClientContextStub({ retryDelayMin: 1, retryDelayMax: 2, retryMaxAttempts: 100, retriesTimeout: 0 }), ); const operation = sinon.stub().rejects(error); diff --git a/tests/unit/connection/connections/ThriftHttpConnectionRetry.test.ts b/tests/unit/connection/connections/ThriftHttpConnectionRetry.test.ts index 4492f69b..9c3bdb28 100644 --- a/tests/unit/connection/connections/ThriftHttpConnectionRetry.test.ts +++ b/tests/unit/connection/connections/ThriftHttpConnectionRetry.test.ts @@ -20,7 +20,13 @@ function thriftMessage(name: string, type: number, seqId: number): Buffer { const CALL = 1; const REPLY = 2; -type ServerAction = 'reply' | 'close-after-request' | 'status-503'; +type ServerAction = + | 'reply' + | 'close-after-request' + | 'status-408' + | 'status-429' + | 'status-503' + | 'status-503-retry-after'; interface TestServer { port: number; @@ -39,8 +45,11 @@ async function startServer(actions: ServerAction[]): Promise { if (action === 'close-after-request') { // The full request has been received; drop the connection without a response req.socket.destroy(); - } else if (action === 'status-503') { - res.writeHead(503); + } else if (action === 'status-503-retry-after') { + res.writeHead(503, { 'Retry-After': '1' }); + res.end(); + } else if (action.startsWith('status-')) { + res.writeHead(Number(action.slice('status-'.length))); res.end(); } else { const body = Buffer.concat(chunks); @@ -179,20 +188,44 @@ describe('ThriftHttpConnection retries', () => { expect(server.requests).to.equal(2); }); - it('does not retry ExecuteStatement on an HTTP error status', async () => { - const { server, connection } = await setup(['status-503'], 0); + for (const status of ['status-503', 'status-408'] as const) { + it(`does not retry ExecuteStatement on an ambiguous HTTP error (${status})`, async () => { + const { server, connection } = await setup([status], 0); - const error = await expectRejection(send(connection, 'ExecuteStatement', 1)); + const error = await expectRejection(send(connection, 'ExecuteStatement', 1)); - expect(error).to.be.instanceOf(THTTPException); - expect(error.statusCode).to.equal(503); - expect(server.requests).to.equal(1); - }); + expect(error).to.be.instanceOf(THTTPException); + expect(error.statusCode).to.equal(Number(status.slice('status-'.length))); + expect(server.requests).to.equal(1); + }); + } + + for (const status of ['status-429', 'status-503-retry-after'] as const) { + it(`retries ExecuteStatement when the server rejects it without running it (${status})`, async () => { + const { server, connection } = await setup([status, 'reply'], 0); + + const result = await send(connection, 'ExecuteStatement', 1); + + expect(result).to.equal('ExecuteStatement result'); + expect(server.requests).to.equal(2); + }); + } + + for (const status of ['status-503', 'status-429', 'status-408'] as const) { + it(`retries OpenSession on a transient HTTP error (${status})`, async () => { + const { server, connection } = await setup([status, 'reply'], 0); + + const result = await send(connection, 'OpenSession', 1); + + expect(result).to.equal('OpenSession result'); + expect(server.requests).to.equal(2); + }); + } it('does not retry other non-idempotent operations when the connection is refused', async () => { const { server, agent, connection } = await setup(['reply'], 1); - const error = await expectRejection(send(connection, 'OpenSession', 1)); + const error = await expectRejection(send(connection, 'FetchResults', 1)); expect(error).to.be.instanceOf(FetchError); expect(error.code).to.equal('ECONNREFUSED');