fix: retry Thrift OpenSession and server-rejected ExecuteStatement

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: <status> 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)

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,

lib/connection/connections/HttpRetryPolicy.ts

@@ -197,7 +197,8 @@ export default class HttpRetryPolicy implements IRetryPolicy<HttpTransactionDeta
     const result =
       // Retry on all codes below 100
       statusCode < 100 ||
-      // ...and on `429 Too Many Requests`
+      // ...and on `408 Request Timeout` and `429 Too Many Requests`
+      statusCode === 408 ||
       statusCode === 429 ||
       // ...and on all `5xx` codes except for `501 Not Implemented`
       (statusCode >= 500 && statusCode !== 501);

lib/connection/connections/UnsentRequestRetryPolicy.ts → 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<HttpTransactionDetails>): Promise<HttpTransactionDetails> {
     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 {

lib/connection/connections/ThriftHttpConnection.ts

@@ -60,7 +60,7 @@ interface ThriftHttpConnectionOptions {
   transport?: TTransportType;
   protocol?: TProtocolConstructor;
   getRetryPolicy(): Promise<IRetryPolicy<HttpTransactionDetails>>;
-  getUnsentRequestRetryPolicy?(): Promise<IRetryPolicy<HttpTransactionDetails>>;
+  getNonIdempotentRetryPolicy?(): Promise<IRetryPolicy<HttpTransactionDetails>>;
 }
 
 // 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();

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);
     }
 

tests/unit/connection/connections/UnsentRequestRetryPolicy.test.ts → 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<unknown>): Promise<unknown> {
@@ -32,7 +32,11 @@ async function expectRejection(promise: Promise<unknown>): Promise<unknown> {
   return expect.fail('It should throw an error');
 }
 
-describe('UnsentRequestRetryPolicy', () => {
+function statusResponse(status: number, headers: Record<string, string> = {}) {
+  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<typeof statusResponse>]> = [
+    ['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<typeof statusResponse>]> = [
+    ['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);

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<TestServer> {
       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');