sync and clean

Author: nbbeeken

src/cursor/abstract_cursor.ts

@@ -608,7 +608,7 @@ export abstract class AbstractCursor<
   abstract clone(): AbstractCursor<TSchema>;
 
   /** @internal */
-  abstract _initialize(
+  protected abstract _initialize(
     session: ClientSession | undefined,
     callback: Callback<ExecutionResult>
   ): void;

src/cursor/run_command_cursor.ts

@@ -1,20 +1,13 @@
 import type { BSONSerializeOptions, Document, Long } from '../bson';
 import type { Db } from '../db';
-import {
-  MongoCursorExhaustedError,
-  MongoError,
-  MongoRuntimeError,
-  MongoUnexpectedServerResponseError
-} from '../error';
-import type { MongoClient } from '../mongo_client';
-import { executeOperation } from '../operations/execute_operation';
+import { MongoAPIError, MongoUnexpectedServerResponseError } from '../error';
+import { executeOperation, ExecutionResult } from '../operations/execute_operation';
 import { GetMoreOperation } from '../operations/get_more';
-import { KillCursorsOperation } from '../operations/kill_cursors';
 import { RunCommandOperation } from '../operations/run_command';
-import { ReadPreference, ReadPreferenceLike } from '../read_preference';
-import type { Server } from '../sdam/server';
-import { ClientSession, maybeClearPinnedConnection } from '../sessions';
-import { List, MongoDBNamespace, ns } from '../utils';
+import type { ReadPreferenceLike } from '../read_preference';
+import type { ClientSession } from '../sessions';
+import { Callback, ns } from '../utils';
+import { AbstractCursor } from './abstract_cursor';
 
 /** @public */
 export type RunCommandCursorOptions = {
@@ -29,213 +22,74 @@ type RunCursorCommandResponse = {
 };
 
 /** @public */
-export class RunCommandCursor {
-  /** @internal */
-  client: MongoClient;
-  /** @internal */
-  db: Db;
-  /** @internal */
-  options: RunCommandCursorOptions;
-  /** @internal */
-  documents: List<Document>;
-  /** @internal */
-  getMoreOptions: { comment?: any; maxAwaitTimeMS?: number; batchSize?: number } = {};
-  /** @internal */
-  id: bigint | null = null;
-  /** @internal */
-  ns: MongoDBNamespace | null = null;
-  /** @internal */
-  server: Server | null = null;
-  /** @internal */
-  readPreference: ReadPreference;
-  /** @internal */
-  session: ClientSession;
-  /** @internal */
-  done = false;
-
-  readonly command: Readonly<Record<string, any>>;
-
-  set comment(comment: any) {
+export class RunCommandCursor extends AbstractCursor {
+  public readonly command: Readonly<Record<string, any>>;
+  public readonly getMoreOptions: {
+    comment?: any;
+    maxAwaitTimeMS?: number;
+    batchSize?: number;
+  } = {};
+
+  public setComment(comment: any) {
     this.getMoreOptions.comment = comment;
   }
 
-  get comment() {
-    return this.getMoreOptions.comment;
-  }
-
-  set maxTimeMS(maxAwaitTimeMS: number) {
-    this.getMoreOptions.maxAwaitTimeMS = maxAwaitTimeMS;
+  public setMaxTimeMS(maxTimeMS: number) {
+    this.getMoreOptions.maxAwaitTimeMS = maxTimeMS;
   }
 
-  get maxTimeMS() {
-    return this.getMoreOptions.maxAwaitTimeMS ?? 0;
-  }
-
-  set batchSize(batchSize: number) {
+  public setBatchSize(batchSize: number) {
     this.getMoreOptions.batchSize = batchSize;
   }
 
-  get batchSize() {
-    return this.getMoreOptions.batchSize ?? 0;
+  public clone(): never {
+    throw new MongoAPIError('RunCommandCursor cannot be cloned');
   }
 
   /** @internal */
-  constructor(
-    client: MongoClient,
-    db: Db,
-    command: Document | Map<string, any>,
-    options: RunCommandCursorOptions = {}
-  ) {
-    this.client = client;
-    this.db = db;
-    this.command =
-      Symbol.toStringTag in command && command[Symbol.toStringTag] === 'Map'
-        ? Object.freeze(Object.fromEntries(command.entries()))
-        : Object.freeze({ ...command });
-    this.options = options;
-    this.documents = new List();
-    this.readPreference = ReadPreference.fromOptions(options) ?? ReadPreference.primary;
-    this.session = options.session == null ? client.startSession({ owner: this }) : options.session;
-  }
+  private db: Db | undefined;
 
   /** @internal */
-  private static getIdFromResponse(response: { cursor: { id: number | Long | bigint } }): bigint {
-    return typeof response.cursor.id === 'number'
-      ? BigInt(response.cursor.id)
-      : typeof response.cursor.id === 'object'
-      ? response.cursor.id.toBigInt()
-      : response.cursor.id;
+  constructor(db: Db, command: Document, options: RunCommandCursorOptions = {}) {
+    super(db.s.client, ns(db.namespace), options);
+    this.db = db;
+    this.command = Object.freeze({ ...command });
   }
 
   /** @internal */
-  private async runCommand() {
+  protected _initialize(session: ClientSession, callback: Callback<ExecutionResult>) {
     const operation = new RunCommandOperation<RunCursorCommandResponse>(this.db, this.command, {
-      ...this.options,
-      session: this.session,
+      ...this.cursorOptions,
+      session: session,
       readPreference: this.readPreference
     });
-
-    try {
-      const commandResponse = await executeOperation(this.client, operation);
-
-      if (commandResponse.cursor == null) {
-        throw new MongoUnexpectedServerResponseError('command must return a cursor document');
-      }
-
-      this.id = RunCommandCursor.getIdFromResponse(commandResponse);
-      this.ns = ns(commandResponse.cursor.ns);
-      this.documents.pushMany(commandResponse.cursor.firstBatch);
-      this.server = operation.server;
-    } catch (error) {
-      await this.cleanup(error);
-      throw error;
-    }
-  }
-
-  /** @internal */
-  private async getMore() {
-    if (this.ns == null || this.id == null || this.server == null) {
-      throw new MongoRuntimeError('getMore cannot be invoked with null namespace, id, nor server');
-    }
-
-    try {
-      const getMoreResponse = await executeOperation(
-        this.client,
-        new GetMoreOperation(this.ns, this.id, this.server, {
-          ...this.options,
-          session: this.session,
-          readPreference: this.readPreference,
-          ...this.getMoreOptions
-        })
-      );
-
-      this.documents.pushMany(getMoreResponse.cursor.nextBatch);
-      this.id = RunCommandCursor.getIdFromResponse(getMoreResponse);
-
-      if (this.id === 0n) {
-        await this.close();
-      }
-    } catch (error) {
-      await this.cleanup(error);
-      throw error;
-    }
-  }
-
-  async *[Symbol.asyncIterator](): AsyncGenerator<Document> {
-    if (this.done) {
-      throw new MongoCursorExhaustedError('This cursor needs a nap');
-    }
-    try {
-      if (this.id == null) {
-        await this.runCommand();
-      }
-      while (this.documents.length !== 0 || this.id !== 0n) {
-        const doc = this.documents.shift();
-        if (doc != null) {
-          yield doc;
-        }
-        if (this.documents.length === 0 && this.id !== 0n) {
-          await this.getMore();
+    executeOperation(this.client, operation).then(
+      response => {
+        if (!response.cursor) {
+          callback(
+            new MongoUnexpectedServerResponseError('Expected server to respond with cursor')
+          );
+          return;
         }
-      }
-    } finally {
-      await this.close();
-    }
-  }
-
-  async next(): Promise<Document | null> {
-    if (this.done) {
-      throw new MongoCursorExhaustedError('This cursor needs a nap');
-    }
-
-    const result = await this[Symbol.asyncIterator]().next();
-
-    return result.done ? null : result.value;
-  }
-
-  async tryNext(): Promise<Document | null> {
-    if (this.done) {
-      throw new MongoCursorExhaustedError('This cursor needs a nap');
-    }
-
-    if (this.documents.length === 0 && this.id !== 0n) {
-      await this.getMore();
-    }
-
-    return this.documents.shift();
-  }
-
-  async toArray() {
-    const documents = new List();
-    for await (const doc of this) {
-      documents.push(doc);
-    }
-    return documents.toArray();
-  }
-
-  async close() {
-    if (this.done) {
-      return;
-    }
-    await this.cleanup().catch(() => null);
+        callback(undefined, {
+          server: operation.server,
+          session,
+          response
+        });
+      },
+      err => callback(err)
+    );
   }
 
   /** @internal */
-  async cleanup(error?: MongoError) {
-    if (this.done) {
-      return;
-    }
-    if (this.id != null && this.id !== 0n && this.ns != null && this.server != null) {
-      const options = this.session.hasEnded ? {} : { session: this.session };
-      const killCursorsOperation = new KillCursorsOperation(this.id, this.ns, this.server, options);
-      await executeOperation(this.client, killCursorsOperation).catch(() => null);
-    }
-    if (this.session.owner === this) {
-      await this.session.endSession().catch(() => null);
-    }
-    if (!this.session.inTransaction()) {
-      maybeClearPinnedConnection(this.session, { error });
-    }
-    this.done = true;
+  override _getMore(_batchSize: number, callback: Callback<Document>) {
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    const getMoreOperation = new GetMoreOperation(this.namespace, this.id!, this.server!, {
+      ...this.cursorOptions,
+      session: this.session,
+      ...this.getMoreOptions
+    });
+
+    executeOperation(this.client, getMoreOperation, callback);
   }
 }

src/db.ts

@@ -534,11 +534,8 @@ export class Db {
    * @param command - The command that will start a cursor on the server.
    * @param options - Configurations for running the command, bson options will apply to getMores
    */
-  runCursorCommand(
-    command: Document | Map<string, any>,
-    options?: RunCommandCursorOptions
-  ): RunCommandCursor {
-    return new RunCommandCursor(this.s.client, this, command, options);
+  runCursorCommand(command: Document, options?: RunCommandCursorOptions): RunCommandCursor {
+    return new RunCommandCursor(this, command, options);
   }
 }
 

test/spec/run-command/runCursorCommand.json

@@ -80,8 +80,7 @@
             "command": {
               "checkMetadataConsistency": 1
             }
-          },
-          "expectResult": []
+          }
         }
       ],
       "expectEvents": [

test/spec/run-command/runCursorCommand.yml

@@ -41,7 +41,6 @@ tests:
         arguments:
           commandName: checkMetadataConsistency
           command: { checkMetadataConsistency: 1 }
-        expectResult: []
     expectEvents:
       - client: *client
         eventType: command
@@ -255,6 +254,8 @@ tests:
     expectEvents:
       - client: *client
         eventType: command
+        # The getMore should receive an error here because we do not have the right kind of cursor
+        # So drivers should run a killCursors, but neither the error nor the killCursors command is relevant to this test
         ignoreExtraEvents: true
         events:
           - commandStartedEvent:

test/tools/unified-spec-runner/operations.ts

@@ -666,9 +666,9 @@ operations.set('runCursorCommand', async ({ entities, operation }: OperationFunc
     session: opts.session
   });
 
-  if (!Number.isNaN(+opts.batchSize)) cursor.batchSize = +opts.batchSize;
-  if (!Number.isNaN(+opts.maxTimeMS)) cursor.maxTimeMS = +opts.maxTimeMS;
-  if (opts.comment !== undefined) cursor.comment = opts.comment;
+  if (!Number.isNaN(+opts.batchSize)) cursor.setBatchSize(+opts.batchSize);
+  if (!Number.isNaN(+opts.maxTimeMS)) cursor.setMaxTimeMS(+opts.maxTimeMS);
+  if (opts.comment !== undefined) cursor.setComment(opts.comment);
 
   return cursor.toArray();
 });
@@ -683,14 +683,16 @@ operations.set(
       session: opts.session
     });
 
-    if (!Number.isNaN(+opts.batchSize)) cursor.batchSize = +opts.batchSize;
-    if (!Number.isNaN(+opts.maxTimeMS)) cursor.maxTimeMS = +opts.maxTimeMS;
-    if (opts.comment !== undefined) cursor.comment = opts.comment;
+    if (!Number.isNaN(+opts.batchSize)) cursor.setBatchSize(+opts.batchSize);
+    if (!Number.isNaN(+opts.maxTimeMS)) cursor.setMaxTimeMS(+opts.maxTimeMS);
+    if (opts.comment !== undefined) cursor.setComment(opts.comment);
 
     // The spec dictates that we create the cursor and force the find command
     // to execute, but the first document must still be returned for the first iteration.
-    const result = await cursor[Symbol.asyncIterator]().next();
-    if (!result.done) cursor.documents.unshift(result.value);
+    const result = await cursor.tryNext();
+    const kDocuments = getSymbolFrom(cursor, 'documents');
+    if (result) cursor[kDocuments].unshift(result);
+
     return cursor;
   }
 );