[coverage] Conformance findings: CLOUDFETCH-017 (#471)

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- CloudFetch retry-exhaustion errors now identify the download and the last failure: the `RetryError` message reads `CloudFetch result download failed: <original message> (last error: <cause>)`. `<cause>` is `HTTP <status>: <statusText>` (or `HTTP <status>` when the status text is empty) for HTTP failures, and `network error <code>`, `network error (<type>)` (e.g. `aborted`, when there is no `code`) or a bare `network error` for network failures (the raw network-error message, which can contain the presigned URL, is not copied). If no cause can be determined, the ` (last error: …)` suffix is omitted. The `RetryError` instance, `errorCode` and `payload` (including the node-fetch `Request` / `Response`) are unchanged. (#471)
 - Report untyped NULL columns as STRING in kernel result metadata, matching Thrift. (#470)
 
 ## 3.0.0

lib/result/CloudFetchResultHandler.ts

@@ -1,6 +1,7 @@
 import fetch, { RequestInfo, RequestInit, Request, Response } from 'node-fetch';
 import { TGetResultSetMetadataResp, TRowSet, TSparkArrowResultLink } from '../../thrift/TCLIService_types';
 import HiveDriverError from '../errors/HiveDriverError';
+import RetryError from '../errors/RetryError';
 import IClientContext from '../contracts/IClientContext';
 import IResultsProvider, { ResultsProviderFetchNextOptions } from './IResultsProvider';
 import { ArrowBatch } from './utils';
@@ -157,15 +158,20 @@ export default class CloudFetchResultHandler implements IResultsProvider<ArrowBa
     // Pulling `arrayBuffer()` in here lets the retry policy treat those
     // body-stream failures the same as connect-time failures.
     let downloaded: ArrayBuffer | undefined;
-    const result = await retryPolicy.invokeWithRetry(async () => {
-      downloaded = undefined;
-      const request = new Request(url, requestConfig);
-      const response = await fetch(request);
-      if (response.ok) {
-        downloaded = await response.arrayBuffer();
-      }
-      return { request, response };
-    });
+    let result: { request: Request; response: Response };
+    try {
+      result = await retryPolicy.invokeWithRetry(async () => {
+        downloaded = undefined;
+        const request = new Request(url, requestConfig);
+        const response = await fetch(request);
+        if (response.ok) {
+          downloaded = await response.arrayBuffer();
+        }
+        return { request, response };
+      });
+    } catch (error) {
+      throw CloudFetchResultHandler.describeDownloadError(error);
+    }
     // Fall back to reading the body here if the retry policy returned a
     // response without consuming it via our operation (e.g. unit-test stubs
     // that hand back a pre-baked Response without invoking the operation
@@ -177,6 +183,58 @@ export default class CloudFetchResultHandler implements IResultsProvider<ArrowBa
     return { response: result.response, body: downloaded };
   }
 
+  /**
+   * Make a CloudFetch retry-exhaustion failure easier to diagnose (#471): the generic
+   * "Max retry count exceeded" / "Retry timeout exceeded" message of a `RetryError` is enriched with
+   * the CloudFetch context and the last cause — the HTTP status / status text, or for network errors
+   * only the error code / type. Raw network-error messages are deliberately NOT copied, because
+   * node-fetch embeds the presigned (credential-bearing) request URL in them.
+   * The same error instance, class, error code and payload (including the node-fetch
+   * `Request` / `Response` objects) are kept. Errors other than `RetryError` are returned unchanged.
+   */
+  private static describeDownloadError(error: unknown): unknown {
+    if (!(error instanceof RetryError)) {
+      return error;
+    }
+
+    const prefix = 'CloudFetch result download failed: ';
+    // Idempotent: the same instance may come through again (e.g. a custom retry policy rethrowing a
+    // cached error) — don't stack the prefix / cause a second time.
+    if (error.message.startsWith(prefix)) {
+      return error;
+    }
+
+    let cause: string | undefined;
+    const { payload } = error;
+    if (payload instanceof Error) {
+      const { code, type } = payload as Error & { code?: unknown; type?: unknown };
+      if (typeof code === 'string' && code) {
+        cause = `network error ${code}`;
+      } else if (typeof type === 'string' && type) {
+        cause = `network error (${type})`;
+      } else {
+        cause = 'network error';
+      }
+    } else if (payload && typeof payload === 'object' && 'response' in payload) {
+      const { response } = payload as { response?: { status?: unknown; statusText?: unknown } };
+      if (response && typeof response.status === 'number') {
+        const { statusText } = response;
+        cause = `HTTP ${response.status}${typeof statusText === 'string' && statusText ? `: ${statusText}` : ''}`;
+      }
+    }
+
+    // Mutate the original error rather than creating a new one, so the stack trace still points at
+    // where the retry policy gave up. V8 bakes the message into the `stack` header at construction
+    // time, so rewrite that header too — otherwise loggers that print `err.stack` miss the context.
+    const originalHeader = `${error.name}: ${error.message}`;
+    const message = `${prefix}${error.message}${cause ? ` (last error: ${cause})` : ''}`;
+    error.message = message;
+    if (typeof error.stack === 'string' && error.stack.startsWith(originalHeader)) {
+      error.stack = `${error.name}: ${message}${error.stack.slice(originalHeader.length)}`;
+    }
+    return error;
+  }
+
   /**
    * Emit cloudfetch.chunk telemetry event.
    * CRITICAL: All exceptions swallowed and logged at LogLevel.debug ONLY.

tests/e2e/cloudfetch.test.ts

@@ -1,10 +1,13 @@
 import { expect } from 'chai';
 import sinon from 'sinon';
+import http from 'http';
+import { AddressInfo } from 'net';
 import { DBSQLClient } from '../../lib';
 import { ClientConfig } from '../../lib/contracts/IClientContext';
 import CloudFetchResultHandler from '../../lib/result/CloudFetchResultHandler';
 import ArrowResultConverter from '../../lib/result/ArrowResultConverter';
 import ResultSlicer from '../../lib/result/ResultSlicer';
+import RetryError, { RetryErrorCode } from '../../lib/errors/RetryError';
 
 import config from './utils/config';
 
@@ -132,4 +135,103 @@ describe('CloudFetch', () => {
     const chunk = await operation.fetchChunk({ maxRows: 100000, disableBuffering: true });
     expect(chunk.length).to.be.gt(0);
   });
+
+  it('should surface HTTP status and download context when CloudFetch retries are exhausted', async () => {
+    const retryMaxAttempts = 2;
+    const session = await openSession({
+      cloudFetchConcurrentDownloads: 1,
+      useLZ4Compression: false,
+      retryMaxAttempts,
+      retryDelayMin: 1,
+      retryDelayMax: 10,
+    });
+
+    let serverHits = 0;
+    const server = http.createServer((req, res) => {
+      serverHits += 1;
+      res.writeHead(500, 'Internal Server Error', { 'Content-Type': 'text/plain' });
+      res.end('Internal Server Error');
+    });
+    await new Promise<void>((resolve) => {
+      server.listen(0, '127.0.0.1', () => resolve());
+    });
+    const { port } = server.address() as AddressInfo;
+
+    try {
+      const queriedRowsCount = 10000000; // result has to be quite big to enable CloudFetch
+      const operation = await session.executeStatement(
+        `
+          SELECT *
+          FROM range(0, ${queriedRowsCount}) AS t1
+          LEFT JOIN (SELECT 1) AS t2
+        `,
+        {
+          maxRows: null, // disable DirectResults
+          useCloudFetch: true,
+        },
+      );
+
+      await operation.finished();
+
+      // @ts-expect-error TS2339: Property getResultHandler does not exist on type IOperation
+      const resultHandler = await operation.getResultHandler();
+      expect(resultHandler.source.source).to.be.instanceOf(CloudFetchResultHandler);
+      const cfResultHandler = resultHandler.source.source;
+
+      // Plain-HTTP local server: let node-fetch pick its default agent for the download.
+      // The stub is scoped to the `fetchChunk` call below and restored right after it, so
+      // later requests on this connection provider (e.g. `session.close()`) keep the real agent
+      // (proxy / custom CA / keep-alive config)
+      const connectionProvider = await cfResultHandler.context.getConnectionProvider();
+      const getAgentStub = sinon.stub(connectionProvider, 'getAgent').resolves(undefined);
+
+      // Redirect every presigned link to the local server that always responds with HTTP 500,
+      // keeping the real expiry time and row count returned by the warehouse
+      const rowSetProvider = cfResultHandler.source;
+      const originalFetchNext = rowSetProvider.fetchNext.bind(rowSetProvider);
+      const originalFileLinks: Array<string> = [];
+      sinon.stub(rowSetProvider, 'fetchNext').callsFake(async (...args: Array<any>) => {
+        const rowSet = await originalFetchNext(...args);
+        rowSet?.resultLinks?.forEach((link: any) => {
+          originalFileLinks.push(link.fileLink);
+          link.fileLink = `http://127.0.0.1:${port}/result`;
+          link.httpHeaders = {};
+        });
+        return rowSet;
+      });
+
+      let thrown: unknown;
+      try {
+        await operation.fetchChunk({ maxRows: 100000, disableBuffering: true });
+      } catch (error) {
+        thrown = error;
+      } finally {
+        getAgentStub.restore();
+      }
+
+      // Pin the retry-exhaustion path (not the plain `CloudFetch HTTP error ...` thrown by `downloadLink`)
+      expect(thrown).to.be.instanceOf(RetryError);
+      expect((thrown as RetryError).errorCode).to.be.equal(RetryErrorCode.AttemptsExceeded);
+      expect(serverHits).to.be.equal(retryMaxAttempts);
+      const { message } = thrown as RetryError;
+      expect(message).to.contain('CloudFetch result download failed');
+      expect(message).to.contain('Max retry count exceeded');
+      expect(message).to.contain('HTTP 500: Internal Server Error');
+
+      // The real presigned link (host and credentialed query string) must not leak into the message
+      expect(originalFileLinks.length).to.be.gt(0);
+      originalFileLinks.forEach((fileLink) => {
+        const { host, search } = new URL(fileLink);
+        expect(message).to.not.contain(host);
+        if (search) {
+          expect(message).to.not.contain(search);
+        }
+      });
+    } finally {
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+      await session.close();
+    }
+  });
 });