[coverage] Conformance findings: STATEMENT-022 (#472)

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

lib/DBSQLOperation.ts

@@ -55,6 +55,10 @@ export default class DBSQLOperation implements IOperation {
 
   private cancelled: boolean = false;
 
+  private cancelPromise?: Promise<Status>;
+
+  private closePromise?: Promise<Status>;
+
   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<Status> {
-    return this.withErrorTelemetry(() => this.cancelInternal());
+    return this.cancelInternal();
   }
 
   private async cancelInternal(): Promise<Status> {
     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<Status> {
-    return this.withErrorTelemetry(() => this.closeInternal());
+    return this.closeInternal();
   }
 
   private async closeInternal(): Promise<Status> {
+    // 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<Status> {
+    // 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();
     }

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

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;