wip: testing

src/client-side-encryption/auto_encrypter.ts

@@ -11,6 +11,7 @@ import { kDecorateResult } from '../constants';
 import { getMongoDBClientEncryption } from '../deps';
 import { MongoRuntimeError } from '../error';
 import { MongoClient, type MongoClientOptions } from '../mongo_client';
+import { type Abortable } from '../mongo_types';
 import { MongoDBCollectionNamespace } from '../utils';
 import { autoSelectSocketOptions } from './client_encryption';
 import * as cryptoCallbacks from './crypto_callbacks';
@@ -372,8 +373,10 @@ export class AutoEncrypter {
   async encrypt(
     ns: string,
     cmd: Document,
-    options: CommandOptions = {}
+    options: CommandOptions & Abortable = {}
   ): Promise<Document | Uint8Array> {
+    options.signal?.throwIfAborted();
+
     if (this._bypassEncryption) {
       // If `bypassAutoEncryption` has been specified, don't encrypt
       return cmd;
@@ -407,7 +410,12 @@ export class AutoEncrypter {
   /**
    * Decrypt a command response
    */
-  async decrypt(response: Uint8Array, options: CommandOptions = {}): Promise<Uint8Array> {
+  async decrypt(
+    response: Uint8Array,
+    options: CommandOptions & Abortable = {}
+  ): Promise<Uint8Array> {
+    options.signal?.throwIfAborted();
+
     const context = this._mongocrypt.makeDecryptionContext(response);
 
     context.id = this._contextCounter++;

src/client-side-encryption/state_machine.ts

@@ -206,6 +206,7 @@ export class StateMachine {
     let result: Uint8Array | null = null;
 
     while (context.state !== MONGOCRYPT_CTX_DONE && context.state !== MONGOCRYPT_CTX_ERROR) {
+      options.signal?.throwIfAborted();
       debug(`[context#${context.id}] ${stateToString.get(context.state) || context.state}`);
 
       switch (context.state) {

src/cmap/connection.ts

@@ -474,7 +474,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         );
       }
 
-      for await (const response of this.readMany({ timeoutContext: options.timeoutContext })) {
+      for await (const response of this.readMany({
+        timeoutContext: options.timeoutContext,
+        signal: options.signal
+      })) {
         this.socket.setTimeout(0);
         const bson = response.parse();
 

src/cmap/wire_protocol/on_data.ts

@@ -24,6 +24,8 @@ export function onData(
   emitter: EventEmitter,
   { timeoutContext, signal }: { timeoutContext?: TimeoutContext } & Abortable
 ) {
+  signal?.throwIfAborted();
+
   // Setup pending events and pending promise lists
   /**
    * When the caller has not yet called .next(), we store the

src/utils.ts

@@ -1351,6 +1351,8 @@ export const randomBytes = promisify(crypto.randomBytes);
  * @param name - An event name to wait for
  */
 export async function once<T>(ee: EventEmitter, name: string, options?: Abortable): Promise<T> {
+  options?.signal?.throwIfAborted();
+
   const { promise, resolve, reject } = promiseWithResolvers<T>();
   const onEvent = (data: T) => resolve(data);
   const onError = (error: Error) => reject(error);