From 0811c9914656c95c5a5b12ae615348b0dd7b6497 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 13 Nov 2026 08:35:13 +0000 Subject: [PATCH 1/2] fix: report a close or cancel during a wait as OperationStateError Closing or cancelling an operation while fetchAll(), fetchChunk() or finished() waits on it could reject the wait with the server's answer for the released statement. On Thrift, closing the session while a query runs makes the in-flight GetOperationStatus return HTTP 409 ("A concurrent GetOperationStatus request updated the command. Please retry."), which surfaced as a THTTPException. DBSQLOperation now records a close or cancel from the moment it is requested and reports such failures as OperationStateError CLOSED or CANCELED. Signed-off-by: Vu Anh Phung Co-authored-by: Isaac --- CHANGELOG.md | 1 + lib/DBSQLOperation.ts | 50 +++++++++-- tests/unit/DBSQLOperation.test.ts | 133 ++++++++++++++++++++++++++++++ 3 files changed, 177 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 34532b62..2b13c5f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,7 @@ - Fix `DECIMAL` parameters losing their fractional digits on the Thrift backend: `"123.45"` was bound as `decimal(3,0)` and came back as `123`. Both backends now send `DECIMAL(p,s)` derived from the value (also for the `DEC` / `NUMERIC` synonyms), stringify `Int64` values beyond 2^53 exactly instead of as `Infinity`, accept exponent forms such as `1e-7` (rewritten as a plain numeral), and throw a `ParameterError` for non-numeric values or values that need more than 38 digits. (#474) - `customHeaders` are now sent on every Thrift request (session, statement and metadata calls), as the kernel backend already did, instead of only on telemetry and feature-flag requests. On these requests, caller values for driver-managed headers (`Authorization`, `User-Agent`, `Content-Type`, `Content-Length`, `Transfer-Encoding`, `Connection`) are ignored on both backends, and the Thrift backend rejects header names or values that cannot be sent at connect time. (#491) - Error telemetry now reports only the error class name and the server SQLSTATE. It no longer includes the error message or stack trace, which could contain table and column names from the query. On the Thrift backend, `statement.complete` telemetry now also reports the result delivery format (inline Arrow, inline columnar, or CloudFetch) and whether the server LZ4-compressed the result. (#508) +- Closing or cancelling an operation (directly, or through `session.close()` / `client.close()`) while `fetchAll()`, `fetchChunk()` or `finished()` is waiting on it now rejects that wait with `OperationStateError` (`CLOSED` or `CANCELED`). Previously the request in flight could fail with the server's answer for the released statement, such as a Thrift `THTTPException` for HTTP 409 (`A concurrent GetOperationStatus request updated the command. Please retry.`). ## 3.0.0 diff --git a/lib/DBSQLOperation.ts b/lib/DBSQLOperation.ts index 2da1506a..be56c02c 100644 --- a/lib/DBSQLOperation.ts +++ b/lib/DBSQLOperation.ts @@ -57,6 +57,11 @@ export default class DBSQLOperation implements IOperation { private cancelled: boolean = false; + // Set from the moment a close/cancel request is sent (see `clientStopError`). + private closeRequested: boolean = false; + + private cancelRequested: boolean = false; + public readonly _data?: unknown; private metadata?: ResultMetadata; @@ -159,11 +164,16 @@ export default class DBSQLOperation implements IOperation { const defaultMaxRows = this.context.getConfig().fetchChunkDefaultMaxRows; const limit = options?.maxRows ?? defaultMaxRows; - const result = await this.backend.fetchChunk({ - limit, - disableBuffering: options?.disableBuffering, - isClosed: () => this.closed || this.cancelled, - }); + let result: Array; + try { + result = await this.backend.fetchChunk({ + limit, + disableBuffering: options?.disableBuffering, + isClosed: () => this.closed || this.cancelled, + }); + } catch (err) { + throw (err instanceof OperationStateError ? undefined : this.clientStopError()) ?? err; + } await this.failIfClosed(); this.context.getLogger().log(LogLevel.debug, `Fetched chunk of size: ${limit} from operation with id: ${this.id}`); @@ -207,7 +217,14 @@ export default class DBSQLOperation implements IOperation { if (this.closed || this.cancelled) { return Status.success(); } - const result = await this.backend.cancel(); + this.cancelRequested = true; + let result: Status; + try { + result = await this.backend.cancel(); + } catch (error) { + this.cancelRequested = false; + throw error; + } this.cancelled = true; this.onClose?.(); return result; @@ -226,6 +243,7 @@ export default class DBSQLOperation implements IOperation { if (this.closed || this.cancelled) { return Status.success(); } + this.closeRequested = true; let result: Status; try { result = await this.backend.close(); @@ -235,6 +253,7 @@ export default class DBSQLOperation implements IOperation { // The backend answers from completion state it already holds (including // direct results), without another status RPC. if (!this.backend.isFinished()) { + this.closeRequested = false; throw error; } // Afterwards the close only releases server-side results, which closing @@ -383,6 +402,22 @@ export default class DBSQLOperation implements IOperation { } } + /** + * The error for a backend call that failed because this operation was closed + * or cancelled while the call was in flight. Such a call fails with whatever + * the server answers for the released statement (Thrift answers an in-flight + * GetOperationStatus with HTTP 409 ABORTED), so report the client's action. + */ + private clientStopError(): OperationStateError | undefined { + if (this.closed || this.closeRequested) { + return new OperationStateError(OperationStateErrorCode.Closed); + } + if (this.cancelled || this.cancelRequested) { + return new OperationStateError(OperationStateErrorCode.Canceled); + } + return undefined; + } + private async waitUntilReadyThroughBackend(options?: WaitUntilReadyOptions) { // The backend-facing `waitUntilReady` takes a neutral // `IOperationBackendWaitOptions` whose `callback` receives an @@ -412,8 +447,9 @@ export default class DBSQLOperation implements IOperation { } else if (err.errorCode === OperationStateErrorCode.Closed) { this.closed = true; } + throw err; } - throw err; + throw this.clientStopError() ?? err; } } diff --git a/tests/unit/DBSQLOperation.test.ts b/tests/unit/DBSQLOperation.test.ts index e37b5059..cc8b0a05 100644 --- a/tests/unit/DBSQLOperation.test.ts +++ b/tests/unit/DBSQLOperation.test.ts @@ -608,6 +608,139 @@ describe('DBSQLOperation', () => { }); }); + describe('close or cancel while a request is in flight', () => { + // How the HTTP layer reports the 409 the server answers an in-flight + // GetOperationStatus with once the operation is closed. + const abortedRequest = () => + new Error( + 'Received a response with a bad HTTP status code: 409: ABORTED: A concurrent GetOperationStatus request updated the command. Please retry.', + ); + + const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); + + function pending(stub: sinon.SinonStub) { + let failRequest: (error: Error) => void = () => {}; + stub.callsFake( + () => + new Promise((_resolve, reject) => { + failRequest = reject; + }), + ); + return (error: Error) => failRequest(error); + } + + function runningOperation() { + const context = new ClientContextStub(); + context.driver.getOperationStatusResp.operationState = TOperationState.RUNNING_STATE; + const getOperationStatus = sinon.stub(context.driver, 'getOperationStatus'); + const failPoll = pending(getOperationStatus); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + return { context, operation, getOperationStatus, failPoll }; + } + + async function expectStateError(promise: Promise, errorCode: OperationStateErrorCode) { + try { + await promise; + expect.fail('It should throw an OperationStateError'); + } catch (e) { + if (e instanceof AssertionError) { + throw e; + } + expect(e).to.be.instanceOf(OperationStateError); + expect((e as OperationStateError).errorCode).to.equal(errorCode); + } + } + + it('should report a status poll failed by close() as closed', async () => { + const { operation, getOperationStatus, failPoll } = runningOperation(); + + const fetching = operation.fetchAll(); + await flush(); + expect(getOperationStatus.calledOnce).to.be.true; + + const closing = operation.close(); + failPoll(abortedRequest()); + + await closing; + await expectStateError(fetching, OperationStateErrorCode.Closed); + }); + + it('should report a status poll failed by cancel() as canceled', async () => { + const { operation, getOperationStatus, failPoll } = runningOperation(); + + const waiting = operation.finished(); + await flush(); + expect(getOperationStatus.calledOnce).to.be.true; + + const cancelling = operation.cancel(); + failPoll(abortedRequest()); + + await cancelling; + await expectStateError(waiting, OperationStateErrorCode.Canceled); + }); + + it('should keep the error of a status poll that failed on its own', async () => { + const { operation, failPoll } = runningOperation(); + + const fetching = operation.fetchAll(); + await flush(); + const error = abortedRequest(); + failPoll(error); + + try { + await fetching; + expect.fail('It should throw the poll error'); + } catch (e) { + if (e instanceof AssertionError) { + throw e; + } + expect(e).to.equal(error); + } + }); + + it('should keep the error of a status poll when the close request failed', async () => { + const { context, operation, failPoll } = runningOperation(); + context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + + const fetching = operation.fetchAll(); + await flush(); + await expectFailure(() => operation.close()); + const error = abortedRequest(); + failPoll(error); + + try { + await fetching; + expect.fail('It should throw the poll error'); + } catch (e) { + if (e instanceof AssertionError) { + throw e; + } + expect(e).to.equal(error); + } + }); + + it('should report a result fetch failed by close() as closed', async () => { + const context = new ClientContextStub(); + context.driver.getOperationStatusResp.operationState = TOperationState.FINISHED_STATE; + context.driver.getOperationStatusResp.hasResultSet = true; + const fetchResults = sinon.stub(context.driver, 'fetchResults'); + const failFetch = pending(fetchResults); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const fetching = operation.fetchChunk({ disableBuffering: true }); + for (let i = 0; i < 20 && !fetchResults.called; i += 1) { + await flush(); // eslint-disable-line no-await-in-loop + } + expect(fetchResults.calledOnce).to.be.true; + + const closing = operation.close(); + failFetch(abortedRequest()); + + await closing; + await expectStateError(fetching, OperationStateErrorCode.Closed); + }); + }); + describe('Symbol.asyncDispose', () => { it('should close the operation', async () => { const context = new ClientContextStub(); From f12a671fc148699f97c09b976497e98227b208e0 Mon Sep 17 00:00:00 2001 From: Vu Anh Phung Date: Fri, 13 Nov 2026 09:54:08 +0000 Subject: [PATCH 2/2] fix: decide the reported error after a pending close or cancel settles A status poll or result fetch could fail while the close or cancel that caused it was still pending. The waiter was then told CLOSED or CANCELED even if that request went on to fail, leaving the operation open and dropping the original error. Keep the latest close/cancel request and await it before choosing the error: report the client's action only if it succeeded, otherwise rethrow the original error. Signed-off-by: Vu Anh Phung Co-authored-by: Isaac --- lib/DBSQLOperation.ts | 37 +++++++++++++------------------ tests/unit/DBSQLOperation.test.ts | 25 +++++++++++++++++++++ 2 files changed, 41 insertions(+), 21 deletions(-) diff --git a/lib/DBSQLOperation.ts b/lib/DBSQLOperation.ts index be56c02c..9543e0b1 100644 --- a/lib/DBSQLOperation.ts +++ b/lib/DBSQLOperation.ts @@ -57,10 +57,8 @@ export default class DBSQLOperation implements IOperation { private cancelled: boolean = false; - // Set from the moment a close/cancel request is sent (see `clientStopError`). - private closeRequested: boolean = false; - - private cancelRequested: boolean = false; + // The latest close or cancel request (see `clientStopError`). + private stopRequest?: Promise; public readonly _data?: unknown; @@ -172,7 +170,7 @@ export default class DBSQLOperation implements IOperation { isClosed: () => this.closed || this.cancelled, }); } catch (err) { - throw (err instanceof OperationStateError ? undefined : this.clientStopError()) ?? err; + throw (err instanceof OperationStateError ? undefined : await this.clientStopError()) ?? err; } await this.failIfClosed(); @@ -217,14 +215,8 @@ export default class DBSQLOperation implements IOperation { if (this.closed || this.cancelled) { return Status.success(); } - this.cancelRequested = true; - let result: Status; - try { - result = await this.backend.cancel(); - } catch (error) { - this.cancelRequested = false; - throw error; - } + this.stopRequest = this.backend.cancel(); + const result = await this.stopRequest; this.cancelled = true; this.onClose?.(); return result; @@ -243,17 +235,16 @@ export default class DBSQLOperation implements IOperation { if (this.closed || this.cancelled) { return Status.success(); } - this.closeRequested = true; + this.stopRequest = this.backend.close(); let result: Status; try { - result = await this.backend.close(); + result = await this.stopRequest; } catch (error: any) { // Until the statement finished, closing it is what stops it server-side, // so the failure is the caller's to handle (and the close can be retried). // The backend answers from completion state it already holds (including // direct results), without another status RPC. if (!this.backend.isFinished()) { - this.closeRequested = false; throw error; } // Afterwards the close only releases server-side results, which closing @@ -406,13 +397,17 @@ export default class DBSQLOperation implements IOperation { * The error for a backend call that failed because this operation was closed * or cancelled while the call was in flight. Such a call fails with whatever * the server answers for the released statement (Thrift answers an in-flight - * GetOperationStatus with HTTP 409 ABORTED), so report the client's action. + * GetOperationStatus with HTTP 409 ABORTED), so report the client's action, + * but only once that close or cancel has succeeded. */ - private clientStopError(): OperationStateError | undefined { - if (this.closed || this.closeRequested) { + private async clientStopError(): Promise { + // `closeInternal` / `cancelInternal` awaited the same request first, so + // their flags are already updated when this resumes. + await this.stopRequest?.catch(() => undefined); + if (this.closed) { return new OperationStateError(OperationStateErrorCode.Closed); } - if (this.cancelled || this.cancelRequested) { + if (this.cancelled) { return new OperationStateError(OperationStateErrorCode.Canceled); } return undefined; @@ -449,7 +444,7 @@ export default class DBSQLOperation implements IOperation { } throw err; } - throw this.clientStopError() ?? err; + throw (await this.clientStopError()) ?? err; } } diff --git a/tests/unit/DBSQLOperation.test.ts b/tests/unit/DBSQLOperation.test.ts index cc8b0a05..08b382a4 100644 --- a/tests/unit/DBSQLOperation.test.ts +++ b/tests/unit/DBSQLOperation.test.ts @@ -719,6 +719,31 @@ describe('DBSQLOperation', () => { } }); + it('should keep the error of a status poll that failed while a close request that then failed was pending', async () => { + const { context, operation, failPoll } = runningOperation(); + const failClose = pending(sinon.stub(context.driver, 'closeOperation')); + + const fetching = operation.fetchAll(); + await flush(); + const closing = operation.close(); + const error = abortedRequest(); + failPoll(error); + await flush(); + failClose(new Error('close failed')); + + await expectFailure(() => closing); + expect(operation['closed']).to.be.false; + try { + await fetching; + expect.fail('It should throw the poll error'); + } catch (e) { + if (e instanceof AssertionError) { + throw e; + } + expect(e).to.equal(error); + } + }); + it('should report a result fetch failed by close() as closed', async () => { const context = new ClientContextStub(); context.driver.getOperationStatusResp.operationState = TOperationState.FINISHED_STATE;