From 421db8e76d09c533c2bc955612268f88cc7a0b82 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 13 Nov 2026 08:37:47 +0000 Subject: [PATCH 1/2] fix: recover Thrift CloudFetch downloads from rejected URLs, 4xx and stalls A one-shot 401, 403 or 404 from cloud storage failed the query: CloudFetchResultHandler threw on any non-OK response and HttpRetryPolicy retries only 429 and 5xx. Other 4xx responses failed the same way, and a stalled attempt was never timed out, so it waited until retriesTimeout instead of being retried. On 401/403/404 the presigned URL itself was rejected, so fetch a fresh link for the same rows with FetchResults(startRowOffset) and download it, counting each refresh against retryMaxAttempts. RowSetProvider serializes its FetchResults calls and pages by explicit offset after a refresh, since the refresh may move the server cursor. Retry other 4xx with the same URL, and apply socketTimeout to each download attempt so a stalled one is retried. Signed-off-by: Vu Anh Phung Co-authored-by: Isaac --- CHANGELOG.md | 1 + lib/DBSQLClient.ts | 4 + lib/connection/connections/HttpConnection.ts | 6 +- lib/connection/connections/HttpRetryPolicy.ts | 15 +- .../contracts/IConnectionProvider.ts | 10 +- lib/contracts/IDBSQLClient.ts | 4 + lib/result/CloudFetchResultHandler.ts | 72 +++++++- lib/result/RowSetProvider.ts | 60 ++++++- tests/unit/.stubs/ConnectionProviderStub.ts | 8 +- .../connections/HttpRetryPolicy.test.ts | 23 +++ .../result/CloudFetchResultHandler.test.ts | 156 ++++++++++++++++++ tests/unit/result/RowSetProvider.test.ts | 81 ++++++++- 12 files changed, 418 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 34532b62..d55a01d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Thrift CloudFetch downloads now recover from more failures. When cloud storage rejects a presigned URL with 401, 403 or 404 (an expired or invalid signature), the driver fetches a fresh link for the same rows from the server and retries with it; each refresh counts as an attempt against `retryMaxAttempts`. Other 4xx responses are retried with the same URL. `socketTimeout` now also bounds each download attempt, so a stalled attempt is retried instead of waiting until `retriesTimeout`. Previously these failures failed the query, and a stalled attempt waited out the whole deadline. - 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/DBSQLClient.ts b/lib/DBSQLClient.ts index 4256873a..e1bda5e0 100644 --- a/lib/DBSQLClient.ts +++ b/lib/DBSQLClient.ts @@ -742,6 +742,10 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I if (options.retryDelayMax !== undefined) { this.config.retryDelayMax = options.retryDelayMax; } + // Also bounds each CloudFetch download attempt on the Thrift backend + if (options.socketTimeout !== undefined) { + this.config.socketTimeout = options.socketTimeout; + } // Override telemetry config if provided in options. Per-key narrowed copy // preserves the structural type system: `ConnectionOptions` and diff --git a/lib/connection/connections/HttpConnection.ts b/lib/connection/connections/HttpConnection.ts index 6472265a..5dc1a3ce 100644 --- a/lib/connection/connections/HttpConnection.ts +++ b/lib/connection/connections/HttpConnection.ts @@ -4,7 +4,7 @@ import http from 'http'; import { HeadersInit } from 'node-fetch'; import { ProxyAgent } from 'proxy-agent'; -import IConnectionProvider, { HttpTransactionDetails } from '../contracts/IConnectionProvider'; +import IConnectionProvider, { HttpRetryPolicyOptions, HttpTransactionDetails } from '../contracts/IConnectionProvider'; import IConnectionOptions, { ProxyOptions } from '../contracts/IConnectionOptions'; import IClientContext from '../../contracts/IClientContext'; import { LogLevel } from '../../contracts/IDBSQLLogger'; @@ -227,7 +227,7 @@ export default class HttpConnection implements IConnectionProvider { return this.connection; } - public async getRetryPolicy(): Promise> { - return new HttpRetryPolicy(this.context); + public async getRetryPolicy(options?: HttpRetryPolicyOptions): Promise> { + return new HttpRetryPolicy(this.context, options); } } diff --git a/lib/connection/connections/HttpRetryPolicy.ts b/lib/connection/connections/HttpRetryPolicy.ts index 3e6a2e53..41cb4633 100644 --- a/lib/connection/connections/HttpRetryPolicy.ts +++ b/lib/connection/connections/HttpRetryPolicy.ts @@ -1,5 +1,5 @@ import IRetryPolicy, { ShouldRetryResult, RetryableOperation } from '../contracts/IRetryPolicy'; -import { HttpTransactionDetails } from '../contracts/IConnectionProvider'; +import { HttpRetryPolicyOptions, HttpTransactionDetails } from '../contracts/IConnectionProvider'; import IClientContext from '../../contracts/IClientContext'; import RetryError, { RetryErrorCode } from '../../errors/RetryError'; @@ -53,10 +53,13 @@ export default class HttpRetryPolicy implements IRetryPolicy { @@ -193,6 +196,7 @@ export default class HttpRetryPolicy implements IRetryPolicy= 500 && statusCode !== 501); + (statusCode >= 500 && statusCode !== 501) || + // ...and, when enabled, on the other `4xx` codes + (retryClientErrorsExcept !== undefined && + statusCode >= 400 && + statusCode < 500 && + !retryClientErrorsExcept.has(statusCode)); return result; } diff --git a/lib/connection/contracts/IConnectionProvider.ts b/lib/connection/contracts/IConnectionProvider.ts index a81bbe82..935ead37 100644 --- a/lib/connection/contracts/IConnectionProvider.ts +++ b/lib/connection/contracts/IConnectionProvider.ts @@ -7,6 +7,14 @@ export interface HttpTransactionDetails { response: Response; } +export interface HttpRetryPolicyOptions { + /** + * Also retry 4xx responses, except these statuses. For idempotent GETs of presigned + * URLs (CloudFetch), where a 4xx from storage does not mean the request was wrong. + */ + retryClientErrorsExcept?: ReadonlySet; +} + export default interface IConnectionProvider { getThriftConnection(): Promise; @@ -18,5 +26,5 @@ export default interface IConnectionProvider { setHeaders(headers: HeadersInit): void; - getRetryPolicy(): Promise>; + getRetryPolicy(options?: HttpRetryPolicyOptions): Promise>; } diff --git a/lib/contracts/IDBSQLClient.ts b/lib/contracts/IDBSQLClient.ts index 5c302bb0..2d6c3929 100644 --- a/lib/contracts/IDBSQLClient.ts +++ b/lib/contracts/IDBSQLClient.ts @@ -70,6 +70,10 @@ export type ConnectionOptions = { port?: number; path: string; userAgentEntry?: string; + /** + * Request timeout in milliseconds. On the Thrift backend it also bounds each CloudFetch + * download attempt; an attempt that runs out is retried. Default 900000 (15 minutes). + */ socketTimeout?: number; proxy?: ProxyOptions; enableMetricViewMetadata?: boolean; diff --git a/lib/result/CloudFetchResultHandler.ts b/lib/result/CloudFetchResultHandler.ts index fbfe5c7c..36baf436 100644 --- a/lib/result/CloudFetchResultHandler.ts +++ b/lib/result/CloudFetchResultHandler.ts @@ -1,4 +1,5 @@ import fetch, { RequestInfo, RequestInit, Request, Response } from 'node-fetch'; +import Int64 from 'node-int64'; import { TGetResultSetMetadataResp, TRowSet, TSparkArrowResultLink } from '../../thrift/TCLIService_types'; import HiveDriverError from '../errors/HiveDriverError'; import RetryError, { RetryErrorCode } from '../errors/RetryError'; @@ -20,6 +21,19 @@ type SettledDownload = { ok: true; batch: ArrowBatch } | { ok: false; error: unk // Largest delay `setTimeout` accepts; longer values fire immediately, so they are clamped. const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; +// Statuses that mean the presigned URL itself was rejected (expired or invalid signature; GCS +// answers 404 for an expired one). Retrying the same URL can't succeed, so the link is refreshed. +// Other 4xx responses are retried with the same URL. +const LINK_REFRESH_STATUSES: ReadonlySet = new Set([401, 403, 404]); + +interface LinkRefresher { + refreshLinks(startRowOffset: Int64, maxRows: Int64): Promise>; +} + +function canRefreshLinks(source: unknown): source is LinkRefresher { + return typeof (source as Partial | undefined)?.refreshLinks === 'function'; +} + export default class CloudFetchResultHandler implements IResultsProvider { private readonly context: IClientContext; @@ -128,7 +142,20 @@ export default class CloudFetchResultHandler implements IResultsProvider { + if (!canRefreshLinks(this.source)) { + return undefined; + } + const offset = link.startRowOffset.toNumber(); + this.context + .getLogger() + .log(LogLevel.debug, `CloudFetch download got HTTP ${status}; refreshing the link at row offset ${offset}`); + const links = await this.source.refreshLinks(link.startRowOffset, link.rowCount); + return links.find((candidate) => candidate.startRowOffset.toNumber() === offset); + } + private async fetch(url: RequestInfo, init?: RequestInit): Promise<{ response: Response; body?: ArrayBuffer }> { const connectionProvider = await this.context.getConnectionProvider(); let requestUrl = url; @@ -156,8 +196,11 @@ export default class CloudFetchResultHandler implements IResultsProvider 0 ? Math.min(socketTimeout, MAX_TIMER_DELAY_MS) : undefined; const requestConfig: RequestInit = { agent, ...init }; // The retry policy checks `retriesTimeout` only between attempts, so a stalled attempt @@ -187,9 +230,21 @@ export default class CloudFetchResultHandler implements IResultsProvider attempt.abort(); + controller?.signal.addEventListener('abort', abortAttempt, { once: true }); + let attemptTimedOut = false; + const attemptTimer = + attemptTimeoutMs === undefined + ? undefined + : setTimeout(() => { + attemptTimedOut = true; + attempt.abort(); + }, attemptTimeoutMs); const request = new Request(url, { ...requestConfig, - signal: controller?.signal as unknown as RequestInit['signal'], + signal: attempt.signal as unknown as RequestInit['signal'], }); try { const response = await fetch(request); @@ -203,8 +258,13 @@ export default class CloudFetchResultHandler implements IResultsProvider = Promise.resolve(); + + private serializeFetch(fetchResults: () => Promise): Promise { + const result = this.lastFetch.then(fetchResults, fetchResults); + this.lastFetch = result.catch(() => undefined); + return result; + } + private get hasMoreRows(): boolean { // `hasMoreRowsFlag` is populated only after fetching the first row set. // Prior to that, we use a `operationHandle.hasResultSet` flag which @@ -129,9 +145,38 @@ export default class RowSetProvider implements IResultsProvider { + this.nextLinkRowOffset = Math.max( + this.nextLinkRowOffset, + link.startRowOffset.toNumber() + link.rowCount.toNumber(), + ); + }); return response.results; } + /** + * Fetches fresh result links starting at `startRowOffset`, e.g. to replace a presigned + * URL that cloud storage rejected. Returns no links when the operation is already closed. + */ + public async refreshLinks(startRowOffset: Int64, maxRows: Int64): Promise> { + if (this.returnOnlyPrefetchedResults) { + return []; + } + const driver = await this.context.getDriver(); + const response = await this.serializeFetch(async () => { + this.fetchByOffset = true; + return driver.fetchResults({ + operationHandle: this.operationHandle, + orientation: TFetchOrientation.FETCH_NEXT, + maxRows, + fetchType: FetchType.Data, + startRowOffset, + }); + }); + Status.assert(response.status); + return response.results?.resultLinks ?? []; + } + public async fetchNext({ limit }: ResultsProviderFetchNextOptions) { const prefetchedResponse = this.prefetchedResults.shift(); if (prefetchedResponse) { @@ -150,12 +195,15 @@ export default class RowSetProvider implements IResultsProvider + driver.fetchResults({ + operationHandle: this.operationHandle, + orientation: this.fetchOrientation, + maxRows: new Int64(limit), + fetchType: FetchType.Data, + ...(this.fetchByOffset ? { startRowOffset: new Int64(this.nextLinkRowOffset) } : {}), + }), + ); const latencyMs = Date.now() - startTime; this.emitChunkEvent(latencyMs, response); diff --git a/tests/unit/.stubs/ConnectionProviderStub.ts b/tests/unit/.stubs/ConnectionProviderStub.ts index 20b54dfd..579d441b 100644 --- a/tests/unit/.stubs/ConnectionProviderStub.ts +++ b/tests/unit/.stubs/ConnectionProviderStub.ts @@ -1,6 +1,9 @@ import http from 'http'; import { HeadersInit } from 'node-fetch'; -import IConnectionProvider, { HttpTransactionDetails } from '../../../lib/connection/contracts/IConnectionProvider'; +import IConnectionProvider, { + HttpRetryPolicyOptions, + HttpTransactionDetails, +} from '../../../lib/connection/contracts/IConnectionProvider'; import IRetryPolicy from '../../../lib/connection/contracts/IRetryPolicy'; import NullRetryPolicy from '../../../lib/connection/connections/NullRetryPolicy'; @@ -19,7 +22,8 @@ export default class ConnectionProviderStub implements IConnectionProvider { this.headers = headers; } - public async getRetryPolicy(): Promise> { + // eslint-disable-next-line @typescript-eslint/no-unused-vars + public async getRetryPolicy(options?: HttpRetryPolicyOptions): Promise> { return new NullRetryPolicy(); } } diff --git a/tests/unit/connection/connections/HttpRetryPolicy.test.ts b/tests/unit/connection/connections/HttpRetryPolicy.test.ts index 35e0c0fc..96ccece0 100644 --- a/tests/unit/connection/connections/HttpRetryPolicy.test.ts +++ b/tests/unit/connection/connections/HttpRetryPolicy.test.ts @@ -103,6 +103,29 @@ describe('HttpRetryPolicy', () => { expect(result.shouldRetry).to.be.false; }); + it('should retry 4xx responses only when enabled, except the listed statuses', async () => { + function createStub(status: number) { + return { + request: new Request('http://localhost'), + response: new Response(undefined, { status }), + }; + } + + const context = new ClientContextStub({ retryMaxAttempts: 100 }); + const defaultPolicy = new HttpRetryPolicy(context); + const cloudFetchPolicy = new HttpRetryPolicy(context, { retryClientErrorsExcept: new Set([401, 403, 404]) }); + + for (const status of [400, 405, 408, 412]) { + expect((await defaultPolicy.shouldRetry(createStub(status))).shouldRetry, `${status}`).to.be.false; + expect((await cloudFetchPolicy.shouldRetry(createStub(status))).shouldRetry, `${status}`).to.be.true; + } + for (const status of [401, 403, 404]) { + expect((await cloudFetchPolicy.shouldRetry(createStub(status))).shouldRetry, `${status}`).to.be.false; + } + expect((await cloudFetchPolicy.shouldRetry(createStub(501))).shouldRetry).to.be.false; + expect((await cloudFetchPolicy.shouldRetry(createStub(503))).shouldRetry).to.be.true; + }); + it('should use `Retry-After` header as a base for backoff', async () => { const context = new ClientContextStub({ retryDelayMin: 3, retryDelayMax: 100, retryMaxAttempts: 10 }); const policy = new HttpRetryPolicy(context); diff --git a/tests/unit/result/CloudFetchResultHandler.test.ts b/tests/unit/result/CloudFetchResultHandler.test.ts index 4498d177..7e50ad36 100644 --- a/tests/unit/result/CloudFetchResultHandler.test.ts +++ b/tests/unit/result/CloudFetchResultHandler.test.ts @@ -902,6 +902,162 @@ describe('CloudFetchResultHandler', () => { } }); + describe('recovery', () => { + const body = Buffer.concat([sampleArrowSchema, sampleArrowBatch]); + + function makeLink(fileLink: string): TSparkArrowResultLink { + return { + fileLink, + expiryTime: new Int64(defaultLinkExpiryTime), + rowCount: new Int64(1), + startRowOffset: new Int64(7), + bytesNum: new Int64(0), + }; + } + + // Like `createHandler`, with a source that can refresh links + function createRefreshingHandler( + link: TSparkArrowResultLink, + refreshedLinks: () => Array, + configOverrides: Partial = {}, + ) { + const context = new BaseClientContextStub({ cloudFetchConcurrentDownloads: 1, ...configOverrides }); + sinon + .stub(context.connectionProvider, 'getRetryPolicy') + .callsFake(async (options) => new HttpRetryPolicy(context, options)); + const source = new ResultsProviderStub( + [{ startRowOffset: new Int64(0), rows: [], resultLinks: [link] }], + undefined, + ); + const refreshLinks = sinon.stub().callsFake(async () => refreshedLinks()); + const handler = new CloudFetchResultHandler(context, Object.assign(source, { refreshLinks }), { + lz4Compressed: false, + status: { statusCode: TStatusCode.SUCCESS_STATUS }, + }); + return { handler, refreshLinks }; + } + + [401, 403, 404].forEach((status) => { + it(`should refresh the link and download the fresh URL after HTTP ${status}`, async () => { + const hits: Array = []; + const url = await startServer((req, res) => { + hits.push(req.url ?? ''); + if (req.url?.includes('sig=old')) { + res.writeHead(status); + res.end(); + } else { + res.end(body); + } + }); + const { handler, refreshLinks } = createRefreshingHandler(makeLink(`${url}?sig=old`), () => [ + makeLink(`${url}?sig=new`), + ]); + + const { value, error } = await fetchWithin(handler, 1500); + + expect(error).to.be.undefined; + expect(value?.batches).to.deep.eq([body]); + expect(hits).to.deep.eq(['/chunk?sig=old', '/chunk?sig=new']); + expect(refreshLinks.callCount).to.be.equal(1); + expect(refreshLinks.firstCall.args[0].toNumber()).to.be.equal(7); + expect(refreshLinks.firstCall.args[1].toNumber()).to.be.equal(1); + }); + }); + + it('should fail with the HTTP error when the source cannot refresh links', async () => { + const url = await startServer((req, res) => { + res.writeHead(403, 'Forbidden'); + res.end(); + }); + const handler = createHandler(url, {}); + + const { error } = await fetchWithin(handler, 1500); + + expect(error.message).to.be.equal('CloudFetch HTTP error 403 Forbidden'); + }); + + it('should count link refreshes against retryMaxAttempts', async () => { + const url = await startServer((req, res) => { + res.writeHead(403, 'Forbidden'); + res.end(); + }); + const { handler, refreshLinks } = createRefreshingHandler( + makeLink(`${url}?sig=0`), + () => [makeLink(`${url}?sig=${Math.random()}`)], + { retryMaxAttempts: 3 }, + ); + + const { error } = await fetchWithin(handler, 1500); + + expect(error.message).to.be.equal('CloudFetch HTTP error 403 Forbidden'); + expect(refreshLinks.callCount).to.be.equal(2); + }); + + it('should retry other 4xx responses with the same URL', async () => { + const hits: Array = []; + const url = await startServer((req, res) => { + hits.push(req.url ?? ''); + if (hits.length === 1) { + res.writeHead(412, 'Precondition Failed'); + res.end(); + } else { + res.end(body); + } + }); + const { handler, refreshLinks } = createRefreshingHandler(makeLink(`${url}?sig=a`), () => [], { + retryDelayMin: 10, + retryDelayMax: 20, + }); + + const { value, error } = await fetchWithin(handler, 1500); + + expect(error).to.be.undefined; + expect(value?.batches).to.deep.eq([body]); + expect(hits).to.deep.eq(['/chunk?sig=a', '/chunk?sig=a']); + expect(refreshLinks.called).to.be.false; + }); + + it('should time out a stalled attempt after socketTimeout and retry it', async () => { + let hits = 0; + const url = await startServer((req, res) => { + hits += 1; + if (hits > 1) { + res.end(body); + } + // ...the first request never gets a response + }); + const handler = createHandler(url, { socketTimeout: 200, retryDelayMin: 10, retryDelayMax: 20 }); + + const { value, error, elapsedMs } = await fetchWithin(handler, 1500); + + expect(error).to.be.undefined; + expect(value?.batches).to.deep.eq([body]); + expect(hits).to.be.equal(2); + expect(elapsedMs).to.be.gte(150); + }); + + it('should report a timed-out last attempt without the presigned URL', async () => { + const url = await startServer(() => { + // never respond + }); + const handler = createHandler(`${url}?sig=SECRET_SIGNATURE`, { + socketTimeout: 100, + retryMaxAttempts: 2, + retryDelayMin: 10, + retryDelayMax: 20, + }); + + const { error } = await fetchWithin(handler, 1500); + + expect(error).to.be.instanceOf(RetryError); + expect(error.errorCode).to.be.equal(RetryErrorCode.AttemptsExceeded); + expect(error.message).to.be.equal( + 'CloudFetch result download failed: Max retry count exceeded (last error: network error ETIMEDOUT)', + ); + expect(error.message).to.not.contain('SECRET_SIGNATURE'); + }); + }); + it('should not enforce a deadline when retriesTimeout is not positive', async () => { const body = Buffer.concat([sampleArrowSchema, sampleArrowBatch]); const url = await startServer((req, res) => { diff --git a/tests/unit/result/RowSetProvider.test.ts b/tests/unit/result/RowSetProvider.test.ts index f2aebe44..ab29bff1 100644 --- a/tests/unit/result/RowSetProvider.test.ts +++ b/tests/unit/result/RowSetProvider.test.ts @@ -11,7 +11,13 @@ import { expect } from 'chai'; import sinon from 'sinon'; import Int64 from 'node-int64'; -import { TOperationHandle, TOperationType, THandleIdentifier } from '../../../thrift/TCLIService_types'; +import { + TFetchResultsResp, + TOperationHandle, + TOperationType, + THandleIdentifier, + TStatusCode, +} from '../../../thrift/TCLIService_types'; import RowSetProvider from '../../../lib/result/RowSetProvider'; import ClientContextStub from '../.stubs/ClientContextStub'; @@ -159,4 +165,77 @@ describe('RowSetProvider', () => { expect(emitCloudFetchChunk.firstCall.args[0].bytes).to.equal(64); }); }); + + describe('link refresh', () => { + function linksResponse(fileLink: string, startRowOffset: number, rowCount: number): TFetchResultsResp { + return { + status: { statusCode: TStatusCode.SUCCESS_STATUS }, + hasMoreRows: true, + results: { + startRowOffset: new Int64(startRowOffset), + rows: [], + resultLinks: [ + { + fileLink, + expiryTime: new Int64(Date.now() + 60_000), + startRowOffset: new Int64(startRowOffset), + rowCount: new Int64(rowCount), + bytesNum: new Int64(0), + }, + ], + }, + }; + } + + it('fetches fresh links from the offset, then pages by offset', async () => { + const context = new ClientContextStub(); + const fetchResults = sinon.spy(context.driver, 'fetchResults'); + const provider = new RowSetProvider(context, makeOperationHandle(), [], false); + + context.driver.fetchResultsResp = linksResponse('https://storage/0?sig=old', 0, 100); + await provider.fetchNext({ limit: 1000 }); + expect(fetchResults.firstCall.args[0].startRowOffset).to.be.undefined; + + context.driver.fetchResultsResp = linksResponse('https://storage/0?sig=new', 0, 100); + const links = await provider.refreshLinks(new Int64(0), new Int64(100)); + expect(links.map((link) => link.fileLink)).to.deep.equal(['https://storage/0?sig=new']); + expect(fetchResults.secondCall.args[0].startRowOffset?.toNumber()).to.equal(0); + expect(fetchResults.secondCall.args[0].maxRows.toNumber()).to.equal(100); + + // The refresh may have moved the server cursor, so the next page is requested by offset + context.driver.fetchResultsResp = linksResponse('https://storage/100?sig=a', 100, 50); + await provider.fetchNext({ limit: 1000 }); + expect(fetchResults.thirdCall.args[0].startRowOffset?.toNumber()).to.equal(100); + }); + + it('serializes a refresh with a page fetch', async () => { + const context = new ClientContextStub(); + const provider = new RowSetProvider(context, makeOperationHandle(), [], false); + context.driver.fetchResultsResp = linksResponse('https://storage/0?sig=old', 0, 100); + await provider.fetchNext({ limit: 1000 }); + + let inFlight = 0; + let maxInFlight = 0; + sinon.stub(context.driver, 'fetchResults').callsFake(async () => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => { + setTimeout(resolve, 10); + }); + inFlight -= 1; + return linksResponse('https://storage/0?sig=new', 0, 100); + }); + await Promise.all([provider.refreshLinks(new Int64(0), new Int64(100)), provider.fetchNext({ limit: 1000 })]); + expect(maxInFlight).to.equal(1); + }); + + it('returns no links for an operation closed by direct results', async () => { + const context = new ClientContextStub(); + const fetchResults = sinon.spy(context.driver, 'fetchResults'); + const provider = new RowSetProvider(context, makeOperationHandle(), [], true); + + expect(await provider.refreshLinks(new Int64(0), new Int64(100))).to.deep.equal([]); + expect(fetchResults.called).to.be.false; + }); + }); }); From 0b03672eec897021e891b08266507b389e347485 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 13 Nov 2026 09:04:58 +0000 Subject: [PATCH 2/2] fix: make socketTimeout an idle timeout and share the deadline across refreshes Review follow-up. The per-attempt download timer now restarts on every received chunk, so socketTimeout aborts only an attempt that stops receiving data, not a large download still in progress. Link refreshes reuse the download's retriesTimeout deadline instead of starting a new one each. Signed-off-by: Vu Anh Phung Co-authored-by: Isaac --- CHANGELOG.md | 2 +- lib/contracts/IDBSQLClient.ts | 5 +- lib/result/CloudFetchResultHandler.ts | 59 +++++++++++++------ .../result/CloudFetchResultHandler.test.ts | 53 +++++++++++++++++ 4 files changed, 97 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d55a01d3..3fb46bde 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Unreleased -- Thrift CloudFetch downloads now recover from more failures. When cloud storage rejects a presigned URL with 401, 403 or 404 (an expired or invalid signature), the driver fetches a fresh link for the same rows from the server and retries with it; each refresh counts as an attempt against `retryMaxAttempts`. Other 4xx responses are retried with the same URL. `socketTimeout` now also bounds each download attempt, so a stalled attempt is retried instead of waiting until `retriesTimeout`. Previously these failures failed the query, and a stalled attempt waited out the whole deadline. +- Thrift CloudFetch downloads now recover from more failures. When cloud storage rejects a presigned URL with 401, 403 or 404 (an expired or invalid signature), the driver fetches a fresh link for the same rows from the server and retries with it; each refresh counts as an attempt against `retryMaxAttempts`. Other 4xx responses are retried with the same URL. `socketTimeout` is now also the inactivity timeout of each download attempt, so an attempt that stops receiving data is retried instead of waiting until `retriesTimeout`; one `retriesTimeout` deadline still covers the whole download, including link refreshes. Previously these failures failed the query, and a stalled attempt waited out the whole deadline. - 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/contracts/IDBSQLClient.ts b/lib/contracts/IDBSQLClient.ts index 2d6c3929..fb036cb1 100644 --- a/lib/contracts/IDBSQLClient.ts +++ b/lib/contracts/IDBSQLClient.ts @@ -71,8 +71,9 @@ export type ConnectionOptions = { path: string; userAgentEntry?: string; /** - * Request timeout in milliseconds. On the Thrift backend it also bounds each CloudFetch - * download attempt; an attempt that runs out is retried. Default 900000 (15 minutes). + * Request timeout in milliseconds. On the Thrift backend it is also the inactivity timeout + * of each CloudFetch download attempt: an attempt that receives nothing for this long is + * aborted and retried. Default 900000 (15 minutes). */ socketTimeout?: number; proxy?: ProxyOptions; diff --git a/lib/result/CloudFetchResultHandler.ts b/lib/result/CloudFetchResultHandler.ts index 36baf436..2ec2d3f7 100644 --- a/lib/result/CloudFetchResultHandler.ts +++ b/lib/result/CloudFetchResultHandler.ts @@ -142,10 +142,12 @@ export default class CloudFetchResultHandler implements IResultsProvider candidate.startRowOffset.toNumber() === offset); } - private async fetch(url: RequestInfo, init?: RequestInit): Promise<{ response: Response; body?: ArrayBuffer }> { + // End (epoch ms) of a download's `retriesTimeout` deadline; `undefined` when it is off + private downloadDeadline(): number | undefined { + const { retriesTimeout } = this.context.getConfig(); + return retriesTimeout > 0 ? Date.now() + retriesTimeout : undefined; + } + + // `deadlineAt` is shared by a download and its link refreshes + private async fetch( + url: RequestInfo, + init?: RequestInit, + deadlineAt: number | undefined = this.downloadDeadline(), + ): Promise<{ response: Response; body?: ArrayBuffer }> { const connectionProvider = await this.context.getConnectionProvider(); let requestUrl = url; if (typeof requestUrl !== 'string') { @@ -197,10 +210,10 @@ export default class CloudFetchResultHandler implements IResultsProvider 0 ? Math.min(socketTimeout, MAX_TIMER_DELAY_MS) : undefined; + const { socketTimeout } = this.context.getConfig(); + // `socketTimeout` is an inactivity timeout for each attempt: an attempt that receives + // nothing for that long is aborted and retried like a network timeout. + const idleTimeoutMs = socketTimeout > 0 ? Math.min(socketTimeout, MAX_TIMER_DELAY_MS) : undefined; const requestConfig: RequestInit = { agent, ...init }; // The retry policy checks `retriesTimeout` only between attempts, so a stalled attempt @@ -208,7 +221,8 @@ export default class CloudFetchResultHandler implements IResultsProvider 0 ? Math.min(retriesTimeout, MAX_TIMER_DELAY_MS) : undefined; + const deadlineMs = + deadlineAt === undefined ? undefined : Math.min(Math.max(deadlineAt - Date.now(), 0), MAX_TIMER_DELAY_MS); // Aborted only once, by the deadline, so a single controller serves every attempt and backoff. const controller = deadlineMs === undefined ? undefined : new AbortController(); let timeoutError: RetryError | undefined; @@ -230,26 +244,33 @@ export default class CloudFetchResultHandler implements IResultsProvider attempt.abort(); controller?.signal.addEventListener('abort', abortAttempt, { once: true }); let attemptTimedOut = false; - const attemptTimer = - attemptTimeoutMs === undefined - ? undefined - : setTimeout(() => { - attemptTimedOut = true; - attempt.abort(); - }, attemptTimeoutMs); + let idleTimer: NodeJS.Timeout | undefined; + const restartIdleTimer = () => { + if (idleTimeoutMs !== undefined) { + clearTimeout(idleTimer); + idleTimer = setTimeout(() => { + attemptTimedOut = true; + attempt.abort(); + }, idleTimeoutMs); + } + }; + restartIdleTimer(); const request = new Request(url, { ...requestConfig, signal: attempt.signal as unknown as RequestInit['signal'], }); try { const response = await fetch(request); + restartIdleTimer(); if (response.ok) { - downloaded = await response.arrayBuffer(); + const bodyRead = response.arrayBuffer(); + response.body.on('data', restartIdleTimer); + downloaded = await bodyRead; } else { lastFailure = { request, response }; } @@ -263,7 +284,7 @@ export default class CloudFetchResultHandler implements IResultsProvider { ); expect(error.message).to.not.contain('SECRET_SIGNATURE'); }); + + it('should not time out an attempt that keeps receiving data', async () => { + let hits = 0; + const url = await startServer((req, res) => { + hits += 1; + res.writeHead(200, { 'Content-Length': `${body.length}` }); + // Slower than socketTimeout in total, but never idle for that long + let sent = 0; + const interval = setInterval(() => { + const next = Math.min(sent + 40, body.length); + res.write(body.subarray(sent, next)); + sent = next; + if (sent === body.length) { + clearInterval(interval); + res.end(); + } + }, 50); + }); + const handler = createHandler(url, { socketTimeout: 150 }); + + const { value, error, elapsedMs } = await fetchWithin(handler, 3000); + + expect(error).to.be.undefined; + expect(value?.batches).to.deep.eq([body]); + expect(hits).to.be.equal(1); + expect(elapsedMs).to.be.gte(150); + }); + + it('should share one retriesTimeout deadline across link refreshes', async () => { + const url = await startServer((req, res) => { + if (req.url?.includes('sig=0')) { + res.writeHead(403, 'Forbidden'); + res.end(); + } + // ...every refreshed URL stalls + }); + let refreshes = 0; + const { handler } = createRefreshingHandler( + makeLink(`${url}?sig=0`), + () => { + refreshes += 1; + return [makeLink(`${url}?sig=${refreshes}`)]; + }, + { retriesTimeout: 300, retryMaxAttempts: 10 }, + ); + + const { error, elapsedMs } = await fetchWithin(handler, 1500); + + expect(error).to.be.instanceOf(RetryError); + expect(error.errorCode).to.be.equal(RetryErrorCode.TimeoutExceeded); + expect(elapsedMs).to.be.lt(1000); + expect(refreshes).to.be.equal(1); + }); }); it('should not enforce a deadline when retriesTimeout is not positive', async () => {