fix: treat a failed close of a finished operation as best effort

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - 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)
+- `IOperation.close()` no longer throws when the close request fails after the operation has finished (for example, once its results were read): the failure is logged as a warning and reported in the returned status's `info`, and the operation is still closed locally. Closing an operation that hasn't finished still throws on failure. (#480)
 - Reject non-scalar query parameter values at bind time with a `ParameterError` on both the Thrift and kernel backends, instead of stringifying them (which could leave a Thrift operation stuck before FINISHED). Newly rejected on the Thrift backend: arrays, plain objects, boxed primitives (e.g. `new String('x')`, `new Number(1)`), functions, symbols, and a `DBSQLParameter` that wraps another `DBSQLParameter`. Primitives, `Date`, `Int64`, and `DBSQLParameter` are still accepted.
 - `DBSQLLogger`: a `filepath` naming an existing directory now writes to `databricks-sql-nodejs.log` inside it, instead of silently producing no logs. (#565)
 - `DBSQLLogger`: if the log file can't be opened (including when its directory can't be created, which used to throw from the constructor), the logger now logs to the console only, leaves `transports.file` undefined, and emits a process warning with code `DBSQL_LOG_FILE_UNAVAILABLE`. Previously an unwritable file could silently stall all logging, including the console. (#565)

lib/DBSQLOperation.ts

@@ -55,6 +55,10 @@ export default class DBSQLOperation implements IOperation {
 
   private cancelled: boolean = false;
 
+  // Set once the backend reports the statement finished successfully, after
+  // which a failed close cannot leave it running server-side.
+  private executionFinished: boolean = false;
+
   public readonly _data?: unknown;
 
   private metadata?: ResultMetadata;
@@ -212,7 +216,8 @@ export default class DBSQLOperation implements IOperation {
   }
 
   /**
-   * Closes operation
+   * Closes operation. Once the operation has finished, a failed close request
+   * is logged and reported in the returned status's `info` instead of thrown.
    * @throws {StatusError}
    */
   public async close(): Promise<Status> {
@@ -223,7 +228,25 @@ export default class DBSQLOperation implements IOperation {
     if (this.closed || this.cancelled) {
       return Status.success();
     }
-    const result = await this.backend.close();
+    let result: Status;
+    try {
+      result = await this.backend.close();
+    } 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).
+      if (!this.executionFinished) {
+        throw error;
+      }
+      // Afterwards the close only releases server-side results, which closing
+      // the session also does: report the failure and finish the teardown.
+      // `StatusError` doesn't extend `Error`, so read `message` structurally.
+      const message = typeof error?.message === 'string' ? error.message : String(error);
+      this.context
+        .getLogger()
+        .log(LogLevel.warn, `Failed to close finished operation ${this.id}; it is closed locally. Cause: ${message}`);
+      this.emitErrorEvent(error);
+      result = Status.success([`Failed to close operation: ${message}`]);
+    }
     this.closed = true;
 
     // Emit statement.complete telemetry event
@@ -367,6 +390,7 @@ export default class DBSQLOperation implements IOperation {
     };
     try {
       await this.backend.waitUntilReady(backendOptions);
+      this.executionFinished = true;
     } catch (err) {
       // Reflect terminal states back into facade flags so subsequent calls
       // short-circuit via failIfClosed().

tests/unit/DBSQLOperation.test.ts

@@ -11,7 +11,8 @@ import {
   TTypeId,
 } from '../../thrift/TCLIService_types';
 import StatusError from '../../lib/errors/StatusError';
-import OperationStateError from '../../lib/errors/OperationStateError';
+import OperationStateError, { OperationStateErrorCode } from '../../lib/errors/OperationStateError';
+import { LogLevel } from '../../lib/contracts/IDBSQLLogger';
 import HiveDriverError from '../../lib/errors/HiveDriverError';
 import JsonResultHandler from '../../lib/result/JsonResultHandler';
 import ArrowResultConverter from '../../lib/result/ArrowResultConverter';
@@ -416,6 +417,108 @@ describe('DBSQLOperation', () => {
       await expectFailure(() => operation.finished());
       await expectFailure(() => operation.getSchema());
     });
+
+    describe('after the operation finished', () => {
+      // How the HTTP layer reports the 404 a teardown CloseOperation can get.
+      const closeFailure = () => new Error('Received a response with a bad HTTP status: 404');
+
+      function finishedOperationWithFailingClose(failure: () => Promise<unknown>) {
+        const context = new ClientContextStub();
+        const log = sinon.spy(context.logger, 'log');
+        const closeOperation = sinon.stub(context.driver, 'closeOperation').callsFake(failure as any);
+        // One page of rows, then an empty last page.
+        sinon
+          .stub(context.driver, 'fetchResults')
+          .onFirstCall()
+          .resolves(context.driver.fetchResultsResp)
+          .resolves({
+            status: { statusCode: TStatusCode.SUCCESS_STATUS },
+            hasMoreRows: false,
+            results: { startRowOffset: new Int64(3), rows: [], columns: [] },
+          });
+        const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context });
+        const onClose = sinon.stub();
+        operation.onClose = onClose;
+        return { context, log, closeOperation, operation, onClose };
+      }
+
+      it('should complete teardown when CloseOperation fails after results were consumed', async () => {
+        const { log, closeOperation, operation, onClose } = finishedOperationWithFailingClose(() =>
+          Promise.reject(closeFailure()),
+        );
+
+        expect(await operation.fetchAll()).to.have.lengthOf(3);
+
+        const status = await operation.close();
+
+        expect(closeOperation.calledOnce).to.be.true;
+        expect(status.isSuccess).to.be.true;
+        expect(status.info.join()).to.contain('404');
+        expect(operation['closed']).to.be.true;
+        expect(onClose.calledOnce).to.be.true;
+        expect(log.calledWith(LogLevel.warn, sinon.match(operation.id).and(sinon.match('404')))).to.be.true;
+
+        try {
+          await operation.fetchAll();
+          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(OperationStateErrorCode.Closed);
+        }
+
+        await operation.close();
+        expect(closeOperation.calledOnce).to.be.true;
+      });
+
+      it('should complete teardown when CloseOperation returns an error status after finished()', async () => {
+        const { operation, onClose } = finishedOperationWithFailingClose(async () => ({
+          status: { statusCode: TStatusCode.ERROR_STATUS, errorMessage: 'Invalid OperationHandle' },
+        }));
+
+        await operation.finished();
+        const status = await operation.close();
+
+        expect(status.isSuccess).to.be.true;
+        expect(status.info.join()).to.contain('Invalid OperationHandle');
+        expect(operation['closed']).to.be.true;
+        expect(onClose.calledOnce).to.be.true;
+      });
+
+      it('should not throw from an autoClose iterator when CloseOperation fails', async () => {
+        const { closeOperation, operation, onClose } = finishedOperationWithFailingClose(() =>
+          Promise.reject(closeFailure()),
+        );
+
+        const rows: Array<object> = [];
+        for await (const row of operation.iterateRows({ autoClose: true })) {
+          rows.push(row);
+        }
+
+        expect(rows).to.have.lengthOf(3);
+        expect(closeOperation.calledOnce).to.be.true;
+        expect(operation['closed']).to.be.true;
+        expect(onClose.calledOnce).to.be.true;
+      });
+
+      it('should still rethrow when the operation was not observed to finish', async () => {
+        const { closeOperation, operation, onClose } = finishedOperationWithFailingClose(() =>
+          Promise.reject(closeFailure()),
+        );
+
+        await expectFailure(() => operation.close());
+        expect(operation['closed']).to.be.false;
+        expect(onClose.called).to.be.false;
+
+        // Still open, so the close can be retried.
+        closeOperation.resolves({ status: { statusCode: TStatusCode.SUCCESS_STATUS } });
+        await operation.close();
+        expect(operation['closed']).to.be.true;
+        expect(onClose.calledOnce).to.be.true;
+      });
+    });
   });
 
   describe('finished', () => {
@@ -1211,6 +1314,22 @@ describe('DBSQLOperation', () => {
       expect(emitError.calledOnce).to.be.true;
     });
 
+    it('emits ERROR and statement-complete telemetry when a finished operation fails to close', async () => {
+      const { context, emitError } = makeContextWithEmitter();
+      const completeStatement = sinon.stub();
+      context.telemetryAggregator = { completeStatement } as any;
+      context.driver.closeOperationResp.status.statusCode = TStatusCode.ERROR_STATUS;
+      const operation = createOperationForTest({ handle: operationHandleStub({ hasResultSet: true }), context });
+
+      await operation.finished();
+      await operation.close();
+
+      expect(emitError.calledOnce).to.be.true;
+      expect(emitError.firstCall.args[0].errorName).to.match(/Status\s?Error/);
+      expect((context.telemetryEmitter!.emitStatementComplete as sinon.SinonStub).calledOnce).to.be.true;
+      expect(completeStatement.calledOnceWith(operation.id)).to.be.true;
+    });
+
     it('emits ERROR telemetry when getMetadata rejects', async () => {
       const { context, emitError } = makeContextWithEmitter();
       // Force the driver to fail metadata fetch

tests/unit/DBSQLSession.test.ts

@@ -724,6 +724,23 @@ describe('DBSQLSession', () => {
       expect(session['operations']['items'].size).to.eq(0);
     });
 
+    it('should close the session when a finished operation fails to close', async () => {
+      const context = new ClientContextStub();
+      const closeSession = sinon.spy(context.driver, 'closeSession');
+      sinon
+        .stub(context.driver, 'closeOperation')
+        .rejects(new Error('Received a response with a bad HTTP status: 404'));
+      const session = createSessionForTest({ handle: sessionHandleStub, context });
+      const operation = await session.executeStatement('SELECT * FROM table');
+      await operation.finished();
+
+      await session.close();
+
+      expect(closeSession.calledOnce).to.be.true;
+      expect(session['isOpen']).to.be.false;
+      expect(session['operations']['items'].size).to.eq(0);
+    });
+
     it('should reject all methods once closed', async () => {
       const session = createSessionForTest({ handle: sessionHandleStub, context: new ClientContextStub() });
       await session.close();

tests/unit/kernel/operation-lifecycle.test.ts

@@ -25,6 +25,9 @@
 
 import { expect } from 'chai';
 import sinon from 'sinon';
+import { tableFromArrays, tableToIPC } from 'apache-arrow';
+import DBSQLOperation from '../../../lib/DBSQLOperation';
+import ClientContextStub from '../.stubs/ClientContextStub';
 import { OperationStatus, OperationState } from '../../../lib/contracts/OperationStatus';
 import IClientContext from '../../../lib/contracts/IClientContext';
 import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
@@ -436,3 +439,59 @@ describe('KernelOperationBackend (lifecycle integration)', () => {
     expect(op.hasResultSet()).to.equal(true);
   });
 });
+
+describe('DBSQLOperation over KernelOperationBackend — close after the result was drained', () => {
+  function drainedOperation(close: () => Promise<void>) {
+    const ipcBytes = Buffer.from(tableToIPC(tableFromArrays({ x: Int32Array.from([1, 2]) }), 'stream'));
+    const batches = [ipcBytes];
+    const statement = {
+      cancel: sinon.stub().resolves(),
+      close: sinon.stub().callsFake(close),
+      schema: () => ({ ipcBytes }),
+      fetchNextBatch: async () => (batches.length > 0 ? { ipcBytes: batches.shift()! } : null),
+    };
+    const context = new ClientContextStub();
+    const log = sinon.spy(context.logger, 'log');
+    const backend = new KernelOperationBackend({ statement, context, id: 'kernel-op' });
+    const operation = new DBSQLOperation({ backend, context });
+    const onClose = sinon.stub();
+    operation.onClose = onClose;
+    return { statement, log, operation, onClose };
+  }
+
+  it('completes teardown when the CloseStatement RPC fails', async () => {
+    const { statement, log, operation, onClose } = drainedOperation(async () => {
+      throw new Error(`__databricks_error__:${JSON.stringify({ code: 'NotFound', message: 'HTTP 404 Not Found' })}`);
+    });
+
+    expect(await operation.fetchAll()).to.deep.equal([{ x: 1 }, { x: 2 }]);
+    const status = await operation.close();
+
+    expect(statement.close.calledOnce).to.equal(true);
+    expect(status.isSuccess).to.equal(true);
+    expect(onClose.calledOnce).to.equal(true);
+    expect(log.calledWith(LogLevel.warn, sinon.match('kernel-op').and(sinon.match('404')))).to.equal(true);
+
+    let thrown: unknown;
+    try {
+      await operation.fetchAll();
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(OperationStateError);
+    expect((thrown as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Closed);
+  });
+
+  it('closes normally when the RPC succeeds', async () => {
+    const { statement, log, operation, onClose } = drainedOperation(async () => undefined);
+
+    await operation.fetchAll();
+    const status = await operation.close();
+
+    expect(statement.close.calledOnce).to.equal(true);
+    expect(status.isSuccess).to.equal(true);
+    expect(status.info).to.deep.equal([]);
+    expect(onClose.calledOnce).to.equal(true);
+    expect(log.calledWith(LogLevel.warn)).to.equal(false);
+  });
+});