diff --git a/CHANGELOG.md b/CHANGELOG.md index d3f3dc8b..38e9e593 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ - Upgrade the native kernel backend to 1.4.0. - **Breaking:** require Node.js versions containing the CVE-2026-111497 Brotli decoder fix: 22.23.4+, 24.22.1+, or 26.12.1+. Node.js 20, 23, and 25 are no longer supported because those end-of-life lines have no fix. - Return empty results with the standard schema from Thrift `getPrimaryKeys()` when a catalog, schema, or table is missing, instead of throwing. (#553) +- Deduplicate concurrent `DBSQLOperation.cancel()` calls so they share a single CancelOperation request and emit one error telemetry event. (#472) +- **Behavior change:** concurrent `DBSQLOperation.close()` calls now share a single CloseOperation request, and `close()` waits for any in-flight `cancel()` to settle first. If that cancel succeeds, `close()` resolves with success without sending CloseOperation; if it fails, `close()` proceeds normally. `close()` can therefore take as long as the pending cancel RPC, bounded by the retry policy (`retryMaxAttempts` / `retriesTimeout`) and `socketTimeout`. Likewise, `cancel()` waits for any in-flight `close()` to settle first: if that close succeeds, `cancel()` resolves with success without sending CancelOperation; if it fails, `cancel()` proceeds normally, so `cancel()` can take as long as the pending close RPC. (#472) - Update locked `follow-redirects` and `serialize-javascript` versions to remediate security findings. ## 2.4.0 diff --git a/lib/DBSQLOperation.ts b/lib/DBSQLOperation.ts index 1ec1922c..f6765b8c 100644 --- a/lib/DBSQLOperation.ts +++ b/lib/DBSQLOperation.ts @@ -55,6 +55,10 @@ export default class DBSQLOperation implements IOperation { private cancelled: boolean = false; + private cancelPromise?: Promise; + + private closePromise?: Promise; + public readonly _data?: unknown; private metadata?: ResultMetadata; @@ -194,32 +198,113 @@ export default class DBSQLOperation implements IOperation { } /** - * Cancels operation + * Cancels operation. + * + * Concurrent `cancel()` calls share a single CancelOperation request. If a + * `close()` is already in flight, `cancel()` first waits for it to settle + * (success or failure) so CancelOperation and CloseOperation never race and + * `onClose` fires only once. As a result, `cancel()` can take as long as the + * pending close RPC, bounded by the retry policy (`retryMaxAttempts` / + * `retriesTimeout`) and the per-request `socketTimeout`. If the close + * succeeds, `cancel()` resolves with success and sends no CancelOperation. + * If it fails, `cancel()` continues and sends CancelOperation as usual. * @throws {StatusError} */ public async cancel(): Promise { - return this.withErrorTelemetry(() => this.cancelInternal()); + return this.cancelInternal(); } private async cancelInternal(): Promise { if (this.closed || this.cancelled) { return Status.success(); } - const result = await this.backend.cancel(); - this.cancelled = true; - this.onClose?.(); - return result; + // If a close is in flight (and no cancel is already shared), wait for it + // to settle so we don't send CancelOperation alongside CloseOperation and + // fire onClose twice. If close fails, fall through and cancel normally. + if (!this.cancelPromise && this.closePromise) { + try { + await this.closePromise; + } catch { + // ignore - close() callers receive the error + } + if (this.closed || this.cancelled) { + return Status.success(); + } + } + // Deduplicate concurrent cancel() calls: share a single in-flight + // CancelOperation request between all callers (STATEMENT-022). + // Error telemetry is emitted inside the shared promise so one failed + // CancelOperation produces exactly one error event, regardless of how + // many callers are awaiting it. + if (!this.cancelPromise) { + const promise = this.withErrorTelemetry(async () => { + const result = await this.backend.cancel(); + this.cancelled = true; + this.onClose?.(); + return result; + }); + this.cancelPromise = promise; + // Allow a retry if cancel failed; on success `cancelled` short-circuits further calls + const reset = () => { + if (this.cancelPromise === promise) { + this.cancelPromise = undefined; + } + }; + promise.then(reset, reset); + } + return this.cancelPromise; } /** - * Closes operation + * Closes operation. + * + * Concurrent `close()` calls share a single CloseOperation request. If a + * `cancel()` is already in flight, `close()` first waits for it to settle + * (success or failure) before doing anything else, so CancelOperation and + * CloseOperation never race and `onClose` fires only once. As a result, + * `close()` can take as long as the pending cancel RPC. That RPC goes + * through the same retry policy as every other request, so the wait is + * bounded by `retryMaxAttempts` / `retriesTimeout` and the per-request + * `socketTimeout`. If the cancel succeeds, `close()` resolves with success + * and sends no CloseOperation. If it fails, `close()` continues and sends + * CloseOperation as usual. * @throws {StatusError} */ public async close(): Promise { - return this.withErrorTelemetry(() => this.closeInternal()); + return this.closeInternal(); } private async closeInternal(): Promise { + // Deduplicate concurrent close() calls: share a single in-flight + // CloseOperation request between all callers. Error telemetry is emitted + // inside the shared promise so one failed CloseOperation produces exactly + // one error event, regardless of how many callers are awaiting it. + if (this.closePromise) { + return this.closePromise; + } + const promise = this.withErrorTelemetry(() => this.doClose()); + this.closePromise = promise; + // Clear once settled so a failed close doesn't block a later cancel() + const reset = () => { + if (this.closePromise === promise) { + this.closePromise = undefined; + } + }; + promise.then(reset, reset); + return promise; + } + + private async doClose(): Promise { + // If a cancel is in flight, wait for it to settle so we don't race a + // CloseOperation against it (and fire onClose twice). A failed cancel is + // ignored here; close proceeds normally in that case. + if (this.cancelPromise) { + try { + await this.cancelPromise; + } catch { + // ignore - cancel() callers receive the error + } + } if (this.closed || this.cancelled) { return Status.success(); } diff --git a/tests/e2e/cancel_concurrent.test.ts b/tests/e2e/cancel_concurrent.test.ts new file mode 100644 index 00000000..5e977f54 --- /dev/null +++ b/tests/e2e/cancel_concurrent.test.ts @@ -0,0 +1,60 @@ +import { AssertionError, expect } from 'chai'; +import sinon from 'sinon'; +import { DBSQLClient } from '../../lib'; +import Status from '../../lib/dto/Status'; +import OperationStateError, { OperationStateErrorCode } from '../../lib/errors/OperationStateError'; + +import config from './utils/config'; + +// STATEMENT-022: concurrent Operation.cancel() calls must not deadlock or crash, +// and on Thrift must be deduplicated into exactly one CancelOperation RPC +// (parity with databricks-sql-go#419 and the shared driver conformance spec). +describe('Concurrent cancel (STATEMENT-022)', function () { + this.timeout(120 * 1000); + + it('three simultaneous cancel() calls: no deadlock, single CancelOperation [thrift]', async () => { + const client = new DBSQLClient(); + const connection = await client.connect({ + host: config.host, + path: config.path, + token: config.token, + }); + const session = await connection.openSession({ + initialCatalog: config.catalog, + initialSchema: config.schema, + }); + + // @ts-expect-error TS2339: Property context does not exist on type IDBSQLSession + const cancelSpy = sinon.spy(session.context.driver, 'cancelOperation'); + + try { + const operation = await session.executeStatement( + 'SELECT a.*, b.* FROM range(0, 100000) a CROSS JOIN range(0, 100000) b ORDER BY a.id DESC, b.id DESC', + { runAsync: true, maxRows: null }, + ); + + const results = await Promise.all([operation.cancel(), operation.cancel(), operation.cancel()]); + results.forEach((status) => { + expect(status).to.be.instanceOf(Status); + expect(status.isSuccess).to.be.true; + }); + + expect(cancelSpy.callCount).to.equal(1); + + try { + await operation.fetchChunk({ disableBuffering: true }); + expect.fail('fetchChunk should be rejected after cancel'); + } catch (error) { + if (error instanceof AssertionError) { + throw error; + } + expect(error).to.be.instanceOf(OperationStateError); + expect((error as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Canceled); + } + } finally { + cancelSpy.restore(); + await session.close(); + await client.close(); + } + }); +}); diff --git a/tests/unit/DBSQLOperation.test.ts b/tests/unit/DBSQLOperation.test.ts index f9195105..7ade003a 100644 --- a/tests/unit/DBSQLOperation.test.ts +++ b/tests/unit/DBSQLOperation.test.ts @@ -288,6 +288,105 @@ describe('DBSQLOperation', () => { } }); + it('should send a single CancelOperation for concurrent cancel() calls (STATEMENT-022)', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const results = await Promise.all([operation.cancel(), operation.cancel(), operation.cancel()]); + + expect(driver.cancelOperation.callCount).to.be.equal(1); + results.forEach((status) => expect(status.isSuccess).to.be.true); + expect(operation['cancelled']).to.be.true; + + await operation.cancel(); + expect(driver.cancelOperation.callCount).to.be.equal(1); + }); + + it('should propagate in-flight cancel failure to all concurrent callers and allow retry', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + context.driver.cancelOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const results = await Promise.allSettled([operation.cancel(), operation.cancel(), operation.cancel()]); + expect(driver.cancelOperation.callCount).to.be.equal(1); + results.forEach((r) => { + expect(r.status).to.equal('rejected'); + expect((r as PromiseRejectedResult).reason).to.be.instanceOf(StatusError); + }); + expect(operation['cancelled']).to.be.false; + + context.driver.cancelOperationResp.status.statusCode = TStatusCode.SUCCESS_STATUS; + await operation.cancel(); + expect(driver.cancelOperation.callCount).to.be.equal(2); + expect(operation['cancelled']).to.be.true; + }); + + it('should not send CloseOperation when close() is called while cancel is in flight', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + const onClose = sinon.spy(); + operation.onClose = onClose; + + const [cancelResult, closeResult] = await Promise.all([operation.cancel(), operation.close()]); + + expect(cancelResult.isSuccess).to.be.true; + expect(closeResult.isSuccess).to.be.true; + expect(driver.cancelOperation.callCount).to.be.equal(1); + expect(driver.closeOperation.callCount).to.be.equal(0); + expect(onClose.callCount).to.be.equal(1); + expect(operation['cancelled']).to.be.true; + }); + + it('should not send CancelOperation when cancel() is called while close is in flight', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + const onClose = sinon.spy(); + operation.onClose = onClose; + + const [closeResult, cancelResult] = await Promise.all([operation.close(), operation.cancel()]); + + expect(closeResult.isSuccess).to.be.true; + expect(cancelResult.isSuccess).to.be.true; + expect(driver.closeOperation.callCount).to.be.equal(1); + expect(driver.cancelOperation.callCount).to.be.equal(0); + expect(onClose.callCount).to.be.equal(1); + expect(operation['closed']).to.be.true; + }); + + it('should still cancel if an in-flight close fails', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const [closeResult, cancelResult] = await Promise.allSettled([operation.close(), operation.cancel()]); + + expect(closeResult.status).to.equal('rejected'); + expect(cancelResult.status).to.equal('fulfilled'); + expect(driver.closeOperation.callCount).to.be.equal(1); + expect(driver.cancelOperation.callCount).to.be.equal(1); + expect(operation['cancelled']).to.be.true; + }); + + it('should still close if an in-flight cancel fails', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + context.driver.cancelOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const [cancelResult, closeResult] = await Promise.allSettled([operation.cancel(), operation.close()]); + + expect(cancelResult.status).to.equal('rejected'); + expect(closeResult.status).to.equal('fulfilled'); + expect(driver.cancelOperation.callCount).to.be.equal(1); + expect(driver.closeOperation.callCount).to.be.equal(1); + expect(operation['closed']).to.be.true; + }); + it('should reject all methods once cancelled', async () => { const context = new ClientContextStub(); const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); @@ -319,6 +418,44 @@ describe('DBSQLOperation', () => { expect(operation['closed']).to.be.true; }); + it('should send a single CloseOperation for concurrent close() calls', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + const onClose = sinon.spy(); + operation.onClose = onClose; + + const results = await Promise.all([operation.close(), operation.close(), operation.close()]); + + expect(driver.closeOperation.callCount).to.be.equal(1); + expect(onClose.callCount).to.be.equal(1); + results.forEach((status) => expect(status.isSuccess).to.be.true); + expect(operation['closed']).to.be.true; + + await operation.close(); + expect(driver.closeOperation.callCount).to.be.equal(1); + }); + + it('should propagate in-flight close failure to all concurrent callers and allow retry', async () => { + const context = new ClientContextStub(); + const driver = sinon.spy(context.driver); + context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const results = await Promise.allSettled([operation.close(), operation.close(), operation.close()]); + expect(driver.closeOperation.callCount).to.be.equal(1); + results.forEach((r) => { + expect(r.status).to.equal('rejected'); + expect((r as PromiseRejectedResult).reason).to.be.instanceOf(StatusError); + }); + expect(operation['closed']).to.be.false; + + context.driver.closeOperationResp.status.statusCode = TStatusCode.SUCCESS_STATUS; + await operation.close(); + expect(driver.closeOperation.callCount).to.be.equal(2); + expect(operation['closed']).to.be.true; + }); + it('should return immediately if already closed', async () => { const context = new ClientContextStub(); const driver = sinon.spy(context.driver); @@ -1201,6 +1338,17 @@ describe('DBSQLOperation', () => { expect(emitError.firstCall.args[0].errorName).to.match(/Status\s?Error/); }); + it('emits a single ERROR telemetry event when concurrent cancel() calls share a failed request', async () => { + const { context, emitError } = makeContextWithEmitter(); + context.driver.cancelOperationResp.status.statusCode = TStatusCode.ERROR_STATUS; + const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context }); + + const results = await Promise.allSettled([operation.cancel(), operation.cancel(), operation.cancel()]); + + expect(results.every((r) => r.status === 'rejected')).to.be.true; + expect(emitError.calledOnce, 'emitError should fire once per failed CancelOperation').to.be.true; + }); + it('emits ERROR telemetry when close rejects', async () => { const { context, emitError } = makeContextWithEmitter(); context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS;