[SEA-NodeJS] Address code-review findings on async/TLS/options (#413)

msrathore-db · msrathore-db · commit 368945a5eba5 · 2026-06-03T07:06:42.000Z
Code-review #413 (81/100). Validated each against the code + a live warehouse: - F1 (HIGH): the async poll loop threw plain HiveDriverError for server-driven Cancelled/Closed/Unknown. The DBSQLOperation facade only mirrors its cancelled/closed flags when `err instanceof OperationStateError` (and OperationStateError extends HiveDriverError, not the reverse), so a server-side cancel/close/admin-kill left the facade desynced. Now throws OperationStateError(Canceled/Closed/Unknown) — matching the Thrift backend. The Failed branch still surfaces the kernel SQL-error envelope via awaitResult. - F2 (MED): the server-Cancelled test asserted only instanceOf(HiveDriverError), which passes for both the correct and incorrect type — it couldn't catch F1. Now asserts instanceOf(OperationStateError) + errorCode, plus a new Closed test. - F3 (MED): queryTimeout was forwarded to submitStatement but the kernel ignores queryTimeoutSecs on submit (always wait_timeout=0s), so the documented public option was a silent no-op, and the poll loop had no client-side deadline (a stalled Running statement polled forever). Now enforced client-side: the poll loop tracks a deadline, best-effort cancels the statement on expiry, and throws OperationStateError(Timeout) — matching Thrift's server TIMEDOUT outcome. Stopped forwarding the ignored queryTimeoutSecs to the napi options. Validated live: a 2s timeout interrupts a slow cross-join with TIMEOUT. - F4 (LOW): customCaCert PEM string check now requires the END marker too (a truncated/headerless cert no longer passes), consistent with the Buffer path. - F5 (LOW): SeaAuth reads the SEA-only fields (checkServerCertificate / customCaCert / maxConnections) through `InternalConnectionOptions` instead of ad-hoc inline casts, so a typo'd key fails to compile. - F6 (LOW): corrected the poll-loop comment — the prior text justified polling by an incorrect "blocking awaitResult holds the mutex and queues cancel" claim; cancel() is documented lock-free. The real rationale (real-time status to the progress callback + cancel observed between ticks) is now stated. Co-authored-by: Isaac Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>
diff --git a/lib/sea/SeaAuth.ts b/lib/sea/SeaAuth.ts
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 import { ConnectionOptions } from '../contracts/IDBSQLClient';
+import { InternalConnectionOptions } from '../contracts/InternalConnectionOptions';
 import AuthenticationError from '../errors/AuthenticationError';
 import HiveDriverError from '../errors/HiveDriverError';
 
@@ -181,10 +182,10 @@ const MAX_U32 = 0xffffffff;
  * (for strings) lacks a PEM certificate header.
  */
 export function buildSeaTlsOptions(options: ConnectionOptions): SeaTlsOptions {
-  const { checkServerCertificate, customCaCert } = options as {
-    checkServerCertificate?: boolean;
-    customCaCert?: Buffer | string;
-  };
+  // Read the SEA-only fields through the purpose-built internal options type
+  // rather than an ad-hoc inline cast, so the shape can't silently drift from
+  // its declaration and a typo'd key fails to compile.
+  const { checkServerCertificate, customCaCert } = options as ConnectionOptions & InternalConnectionOptions;
 
   const tls: SeaTlsOptions = {};
 
@@ -194,10 +195,17 @@ export function buildSeaTlsOptions(options: ConnectionOptions): SeaTlsOptions {
 
   if (customCaCert !== undefined) {
     if (typeof customCaCert === 'string') {
-      if (!customCaCert.includes('-----BEGIN CERTIFICATE-----')) {
+      // Light PEM sanity check — require both the BEGIN and END markers so a
+      // truncated/headerless cert is rejected here rather than surfacing as an
+      // opaque kernel TLS error. Full parsing is deferred to the kernel.
+      if (
+        !customCaCert.includes('-----BEGIN CERTIFICATE-----') ||
+        !customCaCert.includes('-----END CERTIFICATE-----')
+      ) {
         throw new HiveDriverError(
           'SEA backend: `customCaCert` string does not look like a PEM certificate ' +
-            "(missing '-----BEGIN CERTIFICATE-----'). Pass PEM text or a Buffer of PEM bytes.",
+            "(missing the '-----BEGIN CERTIFICATE-----' / '-----END CERTIFICATE-----' markers). " +
+            'Pass PEM text or a Buffer of PEM bytes.',
         );
       }
       tls.customCaCert = Buffer.from(customCaCert, 'utf8');
@@ -293,7 +301,7 @@ export function buildSeaConnectionOptions(options: ConnectionOptions): SeaNative
   // SEA-only pool sizing; read via cast to match how this function reads the
   // other SEA-specific options (TLS) — they live on the internal options
   // surface, not the published public `ConnectionOptions` `.d.ts`.
-  const { maxConnections } = options as { maxConnections?: number };
+  const { maxConnections } = options as ConnectionOptions & InternalConnectionOptions;
   if (maxConnections !== undefined) {
     if (!Number.isInteger(maxConnections) || maxConnections < 1) {
       throw new HiveDriverError(`SEA backend: \`maxConnections\` must be a positive integer; got ${maxConnections}.`);
diff --git a/lib/sea/SeaOperationBackend.ts b/lib/sea/SeaOperationBackend.ts
@@ -45,6 +45,7 @@ import IClientContext from '../contracts/IClientContext';
 import { LogLevel } from '../contracts/IDBSQLLogger';
 import Status from '../dto/Status';
 import HiveDriverError from '../errors/HiveDriverError';
+import OperationStateError, { OperationStateErrorCode } from '../errors/OperationStateError';
 import ArrowResultConverter from '../result/ArrowResultConverter';
 import ResultSlicer from '../result/ResultSlicer';
 import SeaResultsProvider from './SeaResultsProvider';
@@ -121,6 +122,15 @@ export interface SeaOperationBackendOptions {
    * handle exposes one, else a fresh UUIDv4.
    */
   id?: string;
+  /**
+   * Client-side query timeout in whole seconds (the public `queryTimeout`).
+   * The kernel ignores `queryTimeoutSecs` on the async submit path
+   * (`submitStatement` always sends `wait_timeout=0s`), so the JS poll loop
+   * enforces it as a deadline — on expiry it best-effort cancels the statement
+   * and throws `OperationStateError(Timeout)`, matching the Thrift path's
+   * server-side TIMEDOUT outcome. Omitted ⇒ no client-side deadline.
+   */
+  queryTimeoutSecs?: number;
 }
 
 export default class SeaOperationBackend implements IOperationBackend {
@@ -154,7 +164,11 @@ export default class SeaOperationBackend implements IOperationBackend {
   // already-terminal statement. Drives both fetch and result-metadata.
   private fetchHandlePromise?: Promise<SeaFetchHandle>;
 
-  constructor({ asyncStatement, statement, context, id }: SeaOperationBackendOptions) {
+  // Client-side query-timeout deadline in ms (the public `queryTimeout`),
+  // undefined when unset. Enforced in the async poll loop.
+  private readonly queryTimeoutMs?: number;
+
+  constructor({ asyncStatement, statement, context, id, queryTimeoutSecs }: SeaOperationBackendOptions) {
     if ((asyncStatement === undefined) === (statement === undefined)) {
       throw new HiveDriverError('SeaOperationBackend: exactly one of `asyncStatement` or `statement` must be provided');
     }
@@ -163,6 +177,7 @@ export default class SeaOperationBackend implements IOperationBackend {
     this.lifecycleHandle = (asyncStatement ?? statement) as SeaStatementHandle;
     this.context = context;
     this._id = id ?? asyncStatement?.statementId ?? statement?.statementId ?? uuidv4();
+    this.queryTimeoutMs = queryTimeoutSecs !== undefined && queryTimeoutSecs > 0 ? queryTimeoutSecs * 1000 : undefined;
   }
 
   public get id(): string {
@@ -329,20 +344,31 @@ export default class SeaOperationBackend implements IOperationBackend {
   // ---------------------------------------------------------------------------
 
   /**
-   * Poll the kernel `AsyncStatement` to a terminal state, mirroring the Thrift
-   * backend's `waitUntilReady` loop (100ms cadence). Polling `status()` rather
-   * than awaiting `awaitResult()` directly is deliberate: a blocking
-   * `awaitResult()` holds the kernel statement mutex for the whole query and
-   * would queue a concurrent `cancel()` behind it, whereas the poll loop
-   * releases the mutex between ticks so `cancel()` stays responsive. On
-   * success it materialises the result handle (so the first fetch is free);
-   * on a bad terminal state it surfaces the real kernel error.
+   * Poll the kernel `AsyncStatement` to a terminal state on a fixed 100ms
+   * cadence, mirroring the Thrift backend's `waitUntilReady` loop. We poll
+   * `status()` (a cheap GetStatementStatus RPC) rather than awaiting
+   * `awaitResult()` directly so that `status()` reports the real
+   * Pending/Running/Succeeded state to a progress callback each tick, and so a
+   * JS-initiated `cancel()`/`close()` is observed between ticks via
+   * `failIfNotActive`. On success it materialises the result handle (so the
+   * first fetch is free); on a server-driven terminal state it throws the typed
+   * error the `IOperationBackend` contract requires.
+   *
+   * Terminal errors are thrown as `OperationStateError` (NOT plain
+   * `HiveDriverError`) for Cancelled/Closed/Unknown, because the DBSQLOperation
+   * facade only mirrors its `cancelled`/`closed` flags when
+   * `err instanceof OperationStateError` — exactly as the Thrift backend does.
+   * The Failed branch surfaces the kernel's typed SQL-error envelope via
+   * `awaitResult()`.
    */
   private async waitUntilReadyAsync(options?: IOperationBackendWaitOptions): Promise<void> {
     // Already materialised → terminal-and-ready, nothing to wait for.
     if (this.fetchHandlePromise) {
       return;
     }
+    // Client-side timeout deadline: the kernel ignores queryTimeoutSecs on the
+    // async submit path, so we enforce the public `queryTimeout` here.
+    const deadline = this.queryTimeoutMs !== undefined ? Date.now() + this.queryTimeoutMs : undefined;
     for (;;) {
       // A JS-initiated cancel/close short-circuits before the next poll.
       failIfNotActive(this.lifecycle);
@@ -373,11 +399,20 @@ export default class SeaOperationBackend implements IOperationBackend {
           await this.throwAsyncError();
           break;
         case OperationState.Cancelled:
-          throw new HiveDriverError(`SEA operation ${this._id} was cancelled server-side.`);
+          throw new OperationStateError(OperationStateErrorCode.Canceled);
         case OperationState.Closed:
-          throw new HiveDriverError(`SEA operation ${this._id} was closed before it produced a result.`);
+          throw new OperationStateError(OperationStateErrorCode.Closed);
         default:
-          throw new HiveDriverError(`SEA operation ${this._id} reached an unexpected state: ${state}.`);
+          throw new OperationStateError(OperationStateErrorCode.Unknown);
+      }
+
+      // Still Pending/Running — enforce the client-side timeout before sleeping.
+      if (deadline !== undefined && Date.now() >= deadline) {
+        // Best-effort server-side cancel so the statement doesn't keep running
+        // after we stop waiting; never mask the timeout with a cancel failure.
+        // eslint-disable-next-line no-await-in-loop
+        await this.cancel().catch(() => undefined);
+        throw new OperationStateError(OperationStateErrorCode.Timeout);
       }
 
       // eslint-disable-next-line no-await-in-loop
diff --git a/lib/sea/SeaSessionBackend.ts b/lib/sea/SeaSessionBackend.ts
@@ -169,7 +169,15 @@ export default class SeaSessionBackend implements ISessionBackend {
     } catch (err) {
       throw this.logAndMapError('executeStatement', err);
     }
-    return new SeaOperationBackend({ asyncStatement: asyncStatement!, context: this.context });
+    // `queryTimeout` is enforced client-side by the operation backend's poll
+    // loop: the kernel ignores `queryTimeoutSecs` on the async submit path
+    // (`submitStatement` always sends `wait_timeout=0s`), so we do NOT forward
+    // it to the napi options — passing it there would be a silent no-op.
+    return new SeaOperationBackend({
+      asyncStatement: asyncStatement!,
+      context: this.context,
+      queryTimeoutSecs: options.queryTimeout !== undefined ? Number(options.queryTimeout) : undefined,
+    });
   }
 
   /**
@@ -195,11 +203,9 @@ export default class SeaSessionBackend implements ISessionBackend {
     if (namedParams !== undefined) {
       execOptions.namedParams = namedParams;
     }
-    // JDBC `setQueryTimeout` is whole seconds; the kernel's `queryTimeoutSecs`
-    // (SEA wait timeout) is the native equivalent. The SEA wire caps it at 50s.
-    if (options.queryTimeout !== undefined) {
-      execOptions.queryTimeoutSecs = Number(options.queryTimeout);
-    }
+    // `queryTimeout` is intentionally NOT forwarded here — the kernel ignores
+    // `queryTimeoutSecs` on `submitStatement`, so it is enforced client-side by
+    // the operation backend's poll-loop deadline instead (see executeStatement).
     if (options.rowLimit !== undefined) {
       execOptions.rowLimit = Number(options.rowLimit);
     }
diff --git a/tests/unit/sea/execution.test.ts b/tests/unit/sea/execution.test.ts
@@ -22,6 +22,7 @@ import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientCont
 import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
 import HiveDriverError from '../../../lib/errors/HiveDriverError';
 import ParameterError from '../../../lib/errors/ParameterError';
+import OperationStateError, { OperationStateErrorCode } from '../../../lib/errors/OperationStateError';
 import { ConnectionOptions } from '../../../lib/contracts/IDBSQLClient';
 import { OperationState } from '../../../lib/contracts/OperationStatus';
 
@@ -515,11 +516,14 @@ describe('SeaSessionBackend', () => {
     expect((thrown as Error).message).to.equal('Driver does not support both ordinal and named parameters.');
   });
 
-  it('executeStatement forwards queryTimeout as queryTimeoutSecs', async () => {
+  it('executeStatement does NOT forward queryTimeout to submit (kernel ignores it; enforced client-side)', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);
     await session.executeStatement('SELECT 1', { queryTimeout: 30 });
-    expect((connection.lastOptions as { queryTimeoutSecs?: number }).queryTimeoutSecs).to.equal(30);
+    // queryTimeout alone must not produce napi submit options — the kernel
+    // ignores queryTimeoutSecs on submitStatement, so it's enforced client-side
+    // by the operation backend's poll deadline instead (covered below).
+    expect((connection.lastOptions as { queryTimeoutSecs?: number } | undefined)?.queryTimeoutSecs).to.equal(undefined);
   });
 
   it('executeStatement forwards rowLimit', async () => {
@@ -734,9 +738,9 @@ describe('SeaOperationBackend', () => {
 });
 
 describe('SeaOperationBackend — async (submitStatement) path', () => {
-  const makeAsyncOp = (asyncStatement: FakeAsyncStatement) =>
+  const makeAsyncOp = (asyncStatement: FakeAsyncStatement, queryTimeoutSecs?: number) =>
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    new SeaOperationBackend({ asyncStatement: asyncStatement as any, context: makeContext() });
+    new SeaOperationBackend({ asyncStatement: asyncStatement as any, context: makeContext(), queryTimeoutSecs });
 
   it('rejects when neither asyncStatement nor statement is provided', () => {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -794,16 +798,51 @@ describe('SeaOperationBackend — async (submitStatement) path', () => {
     expect((thrown as Error).message).to.match(/TABLE_OR_VIEW_NOT_FOUND/);
   });
 
-  it('waitUntilReady() throws on a server-side Cancelled statement', async () => {
+  // A server-driven terminal state MUST throw OperationStateError (not a plain
+  // HiveDriverError) so the DBSQLOperation facade — which only mirrors its
+  // cancelled/closed flags when `err instanceof OperationStateError` — stays in
+  // sync. Asserting the subclass + errorCode is what catches a regression to
+  // the bare HiveDriverError (which would pass an `instanceOf HiveDriverError`
+  // check since OperationStateError extends it).
+  it('waitUntilReady() throws OperationStateError(Canceled) on a server-side Cancelled statement', async () => {
     const op = makeAsyncOp(new FakeAsyncStatement('Cancelled'));
     let thrown: unknown;
     try {
       await op.waitUntilReady();
     } catch (err) {
       thrown = err;
     }
-    expect(thrown).to.be.instanceOf(HiveDriverError);
-    expect((thrown as Error).message).to.match(/cancelled/i);
+    expect(thrown).to.be.instanceOf(OperationStateError);
+    expect((thrown as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Canceled);
+  });
+
+  it('waitUntilReady() throws OperationStateError(Closed) on a server-side Closed statement', async () => {
+    const op = makeAsyncOp(new FakeAsyncStatement('Closed'));
+    let thrown: unknown;
+    try {
+      await op.waitUntilReady();
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(OperationStateError);
+    expect((thrown as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Closed);
+  });
+
+  it('waitUntilReady() enforces queryTimeout client-side: throws Timeout and cancels a stuck Running statement', async function timeoutTest() {
+    // eslint-disable-next-line no-invalid-this
+    this.timeout(5000);
+    const stmt = new FakeAsyncStatement('Running'); // never reaches a terminal state
+    const op = makeAsyncOp(stmt, 0.05); // 50ms client-side deadline
+    let thrown: unknown;
+    try {
+      await op.waitUntilReady();
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(OperationStateError);
+    expect((thrown as OperationStateError).errorCode).to.equal(OperationStateErrorCode.Timeout);
+    // Best-effort server-side cancel fired so the statement doesn't keep running.
+    expect(stmt.cancelled).to.equal(true);
   });
 
   it('cancel() forwards to the async statement and short-circuits a subsequent poll', async () => {
