feat(NODE-6856): end file system operations on client close

Author: nbbeeken

src/client-side-encryption/auto_encrypter.ts

@@ -409,7 +409,7 @@ export class AutoEncrypter {
     context.ns = ns;
     context.document = cmd;
 
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       promoteValues: false,
       promoteLongs: false,
       proxyOptions: this._proxyOptions,
@@ -436,7 +436,7 @@ export class AutoEncrypter {
 
     context.id = this._contextCounter++;
 
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       ...options,
       proxyOptions: this._proxyOptions,
       tlsOptions: this._tlsOptions,

src/client-side-encryption/client_encryption.ts

@@ -226,7 +226,7 @@ export class ClientEncryption {
       keyMaterial
     });
 
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       proxyOptions: this._proxyOptions,
       tlsOptions: this._tlsOptions,
       socketOptions: autoSelectSocketOptions(this._client.s.options)
@@ -295,7 +295,7 @@ export class ClientEncryption {
     }
     const filterBson = serialize(filter);
     const context = this._mongoCrypt.makeRewrapManyDataKeyContext(filterBson, keyEncryptionKeyBson);
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       proxyOptions: this._proxyOptions,
       tlsOptions: this._tlsOptions,
       socketOptions: autoSelectSocketOptions(this._client.s.options)
@@ -699,7 +699,7 @@ export class ClientEncryption {
     const valueBuffer = serialize({ v: value });
     const context = this._mongoCrypt.makeExplicitDecryptionContext(valueBuffer);
 
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       proxyOptions: this._proxyOptions,
       tlsOptions: this._tlsOptions,
       socketOptions: autoSelectSocketOptions(this._client.s.options)
@@ -783,7 +783,7 @@ export class ClientEncryption {
     }
 
     const valueBuffer = serialize({ v: value });
-    const stateMachine = new StateMachine({
+    const stateMachine = new StateMachine(this, {
       proxyOptions: this._proxyOptions,
       tlsOptions: this._tlsOptions,
       socketOptions: autoSelectSocketOptions(this._client.s.options)

src/client-side-encryption/state_machine.ts

@@ -1,8 +1,8 @@
-import * as fs from 'fs/promises';
 import { type MongoCryptContext, type MongoCryptKMSRequest } from 'mongodb-client-encryption';
 import * as net from 'net';
 import * as tls from 'tls';
 
+import { type AutoEncrypter } from '..';
 import {
   type BSONSerializeOptions,
   deserialize,
@@ -25,7 +25,7 @@ import {
   MongoDBCollectionNamespace,
   promiseWithResolvers
 } from '../utils';
-import { autoSelectSocketOptions, type DataKey } from './client_encryption';
+import { autoSelectSocketOptions, type ClientEncryption, type DataKey } from './client_encryption';
 import { MongoCryptError } from './errors';
 import { type MongocryptdManager } from './mongocryptd_manager';
 import { type KMSProviders } from './providers';
@@ -186,10 +186,15 @@ export type StateMachineOptions = {
  */
 // TODO(DRIVERS-2671): clarify CSOT behavior for FLE APIs
 export class StateMachine {
+  private parent: AutoEncrypter | ClientEncryption;
+
   constructor(
+    parent: AutoEncrypter | ClientEncryption,
     private options: StateMachineOptions,
     private bsonOptions = pluckBSONSerializeOptions(options)
-  ) {}
+  ) {
+    this.parent = parent;
+  }
 
   /**
    * Executes the state machine according to the specification
@@ -524,11 +529,11 @@ export class StateMachine {
     options: tls.ConnectionOptions
   ): Promise<void> {
     if (tlsOptions.tlsCertificateKeyFile) {
-      const cert = await fs.readFile(tlsOptions.tlsCertificateKeyFile);
+      const cert = await this.parent._client.io.fs.readFile(tlsOptions.tlsCertificateKeyFile);
       options.cert = options.key = cert;
     }
     if (tlsOptions.tlsCAFile) {
-      options.ca = await fs.readFile(tlsOptions.tlsCAFile);
+      options.ca = await this.parent._client.io.fs.readFile(tlsOptions.tlsCAFile);
     }
     if (tlsOptions.tlsCertificateKeyFilePassword) {
       options.passphrase = tlsOptions.tlsCertificateKeyFilePassword;

src/cmap/auth/mongodb_oidc.ts

@@ -1,5 +1,6 @@
 import type { Document } from '../../bson';
 import { MongoInvalidArgumentError, MongoMissingCredentialsError } from '../../error';
+import { type MongoClient } from '../../mongo_client';
 import type { HandshakeDocument } from '../connect';
 import type { Connection } from '../connection';
 import { type AuthContext, AuthProvider } from './auth_provider';
@@ -115,11 +116,11 @@ export interface Workflow {
 }
 
 /** @internal */
-export const OIDC_WORKFLOWS: Map<EnvironmentName, () => Workflow> = new Map();
-OIDC_WORKFLOWS.set('test', () => new TokenMachineWorkflow(new TokenCache()));
-OIDC_WORKFLOWS.set('azure', () => new AzureMachineWorkflow(new TokenCache()));
-OIDC_WORKFLOWS.set('gcp', () => new GCPMachineWorkflow(new TokenCache()));
-OIDC_WORKFLOWS.set('k8s', () => new K8SMachineWorkflow(new TokenCache()));
+export const OIDC_WORKFLOWS: Map<EnvironmentName, (client: MongoClient) => Workflow> = new Map();
+OIDC_WORKFLOWS.set('test', client => new TokenMachineWorkflow(client, new TokenCache()));
+OIDC_WORKFLOWS.set('azure', client => new AzureMachineWorkflow(client, new TokenCache()));
+OIDC_WORKFLOWS.set('gcp', client => new GCPMachineWorkflow(client, new TokenCache()));
+OIDC_WORKFLOWS.set('k8s', client => new K8SMachineWorkflow(client, new TokenCache()));
 
 /**
  * OIDC auth provider.

src/cmap/auth/mongodb_oidc/azure_machine_workflow.ts

@@ -3,7 +3,6 @@ import { MongoAzureError } from '../../../error';
 import { get } from '../../../utils';
 import type { MongoCredentials } from '../mongo_credentials';
 import { type AccessToken, MachineWorkflow } from './machine_workflow';
-import { type TokenCache } from './token_cache';
 
 /** Azure request headers. */
 const AZURE_HEADERS = Object.freeze({ Metadata: 'true', Accept: 'application/json' });
@@ -22,13 +21,6 @@ const TOKEN_RESOURCE_MISSING_ERROR =
  * @internal
  */
 export class AzureMachineWorkflow extends MachineWorkflow {
-  /**
-   * Instantiate the machine workflow.
-   */
-  constructor(cache: TokenCache) {
-    super(cache);
-  }
-
   /**
    * Get the token from the environment.
    */

src/cmap/auth/mongodb_oidc/gcp_machine_workflow.ts

@@ -2,7 +2,6 @@ import { MongoGCPError } from '../../../error';
 import { get } from '../../../utils';
 import { type MongoCredentials } from '../mongo_credentials';
 import { type AccessToken, MachineWorkflow } from './machine_workflow';
-import { type TokenCache } from './token_cache';
 
 /** GCP base URL. */
 const GCP_BASE_URL =
@@ -16,13 +15,6 @@ const TOKEN_RESOURCE_MISSING_ERROR =
   'TOKEN_RESOURCE must be set in the auth mechanism properties when ENVIRONMENT is gcp.';
 
 export class GCPMachineWorkflow extends MachineWorkflow {
-  /**
-   * Instantiate the machine workflow.
-   */
-  constructor(cache: TokenCache) {
-    super(cache);
-  }
-
   /**
    * Get the token from the environment.
    */

src/cmap/auth/mongodb_oidc/k8s_machine_workflow.ts

@@ -1,7 +1,6 @@
-import { readFile } from 'fs/promises';
-
+import { type MongoClient } from '../../../mongo_client';
+import { type MongoCredentials } from '../mongo_credentials';
 import { type AccessToken, MachineWorkflow } from './machine_workflow';
-import { type TokenCache } from './token_cache';
 
 /** The fallback file name */
 const FALLBACK_FILENAME = '/var/run/secrets/kubernetes.io/serviceaccount/token';
@@ -13,17 +12,10 @@ const AZURE_FILENAME = 'AZURE_FEDERATED_TOKEN_FILE';
 const AWS_FILENAME = 'AWS_WEB_IDENTITY_TOKEN_FILE';
 
 export class K8SMachineWorkflow extends MachineWorkflow {
-  /**
-   * Instantiate the machine workflow.
-   */
-  constructor(cache: TokenCache) {
-    super(cache);
-  }
-
   /**
    * Get the token from the environment.
    */
-  async getToken(): Promise<AccessToken> {
+  async getToken(_credentials: MongoCredentials, client: MongoClient): Promise<AccessToken> {
     let filename: string;
     if (process.env[AZURE_FILENAME]) {
       filename = process.env[AZURE_FILENAME];
@@ -32,7 +24,7 @@ export class K8SMachineWorkflow extends MachineWorkflow {
     } else {
       filename = FALLBACK_FILENAME;
     }
-    const token = await readFile(filename, 'utf8');
+    const token = await client.io.fs.readFile(filename, { encoding: 'utf8' });
     return { access_token: token };
   }
 }

src/cmap/auth/mongodb_oidc/machine_workflow.ts

@@ -1,6 +1,7 @@
 import { setTimeout } from 'timers/promises';
 
 import { type Document } from '../../../bson';
+import { type MongoClient } from '../../../mongo_client';
 import { ns } from '../../../utils';
 import type { Connection } from '../../connection';
 import type { MongoCredentials } from '../mongo_credentials';
@@ -21,7 +22,10 @@ export interface AccessToken {
 }
 
 /** @internal */
-export type OIDCTokenFunction = (credentials: MongoCredentials) => Promise<AccessToken>;
+export type OIDCTokenFunction = (
+  credentials: MongoCredentials,
+  client: MongoClient
+) => Promise<AccessToken>;
 
 /**
  * Common behaviour for OIDC machine workflows.
@@ -31,11 +35,13 @@ export abstract class MachineWorkflow implements Workflow {
   cache: TokenCache;
   callback: OIDCTokenFunction;
   lastExecutionTime: number;
+  client: MongoClient;
 
   /**
    * Instantiate the machine workflow.
    */
-  constructor(cache: TokenCache) {
+  constructor(client: MongoClient, cache: TokenCache) {
+    this.client = client;
     this.cache = cache;
     this.callback = this.withLock(this.getToken.bind(this));
     this.lastExecutionTime = Date.now() - THROTTLE_MS;
@@ -101,7 +107,7 @@ export abstract class MachineWorkflow implements Workflow {
       }
       return token;
     } else {
-      const token = await this.callback(credentials);
+      const token = await this.callback(credentials, connection.client);
       this.cache.put({ accessToken: token.access_token, expiresInSeconds: token.expires_in });
       // Put the access token on the connection as well.
       connection.accessToken = token.access_token;
@@ -129,7 +135,7 @@ export abstract class MachineWorkflow implements Workflow {
             await setTimeout(THROTTLE_MS - difference);
           }
           this.lastExecutionTime = Date.now();
-          return await callback(credentials);
+          return await callback(credentials, this.client);
         });
       return await lock;
     };
@@ -138,5 +144,5 @@ export abstract class MachineWorkflow implements Workflow {
   /**
    * Get the token from the environment or endpoint.
    */
-  abstract getToken(credentials: MongoCredentials): Promise<AccessToken>;
+  abstract getToken(credentials: MongoCredentials, client: MongoClient): Promise<AccessToken>;
 }

src/cmap/auth/mongodb_oidc/token_machine_workflow.ts

@@ -1,8 +1,7 @@
-import * as fs from 'fs';
-
 import { MongoAWSError } from '../../../error';
+import { type MongoClient } from '../../../mongo_client';
+import { type MongoCredentials } from '../mongo_credentials';
 import { type AccessToken, MachineWorkflow } from './machine_workflow';
-import { type TokenCache } from './token_cache';
 
 /** Error for when the token is missing in the environment. */
 const TOKEN_MISSING_ERROR = 'OIDC_TOKEN_FILE must be set in the environment.';
@@ -13,22 +12,15 @@ const TOKEN_MISSING_ERROR = 'OIDC_TOKEN_FILE must be set in the environment.';
  * @internal
  */
 export class TokenMachineWorkflow extends MachineWorkflow {
-  /**
-   * Instantiate the machine workflow.
-   */
-  constructor(cache: TokenCache) {
-    super(cache);
-  }
-
   /**
    * Get the token from the environment.
    */
-  async getToken(): Promise<AccessToken> {
+  async getToken(_: MongoCredentials, client: MongoClient): Promise<AccessToken> {
     const tokenFile = process.env.OIDC_TOKEN_FILE;
     if (!tokenFile) {
       throw new MongoAWSError(TOKEN_MISSING_ERROR);
     }
-    const token = await fs.promises.readFile(tokenFile, 'utf8');
+    const token = await client.io.fs.readFile(tokenFile, { encoding: 'utf8' });
     return { access_token: token };
   }
 }

src/cmap/connect.ts

@@ -16,6 +16,7 @@ import {
   MongoRuntimeError,
   needsRetryableWriteLabel
 } from '../error';
+import { type Monitor, type RTTPinger } from '../sdam/monitor';
 import { HostAddress, ns, promiseWithResolvers } from '../utils';
 import { AuthContext } from './auth/auth_provider';
 import { AuthMechanism } from './auth/providers';
@@ -25,6 +26,7 @@ import {
   type ConnectionOptions,
   CryptoConnection
 } from './connection';
+import { type ConnectionPool } from './connection_pool';
 import {
   MAX_SUPPORTED_SERVER_VERSION,
   MAX_SUPPORTED_WIRE_VERSION,
@@ -35,11 +37,14 @@ import {
 /** @public */
 export type Stream = Socket | TLSSocket;
 
-export async function connect(options: ConnectionOptions): Promise<Connection> {
+export async function connect(
+  parent: Monitor | RTTPinger | ConnectionPool,
+  options: ConnectionOptions
+): Promise<Connection> {
   let connection: Connection | null = null;
   try {
     const socket = await makeSocket(options);
-    connection = makeConnection(options, socket);
+    connection = makeConnection(parent, options, socket);
     await performInitialHandshake(connection, options);
     return connection;
   } catch (error) {
@@ -48,13 +53,17 @@ export async function connect(options: ConnectionOptions): Promise<Connection> {
   }
 }
 
-export function makeConnection(options: ConnectionOptions, socket: Stream): Connection {
+export function makeConnection(
+  parent: Monitor | RTTPinger | ConnectionPool,
+  options: ConnectionOptions,
+  socket: Stream
+): Connection {
   let ConnectionType = options.connectionType ?? Connection;
   if (options.autoEncrypter) {
     ConnectionType = CryptoConnection;
   }
 
-  return new ConnectionType(socket, options);
+  return new ConnectionType(parent, socket, options);
 }
 
 function checkSupportedServer(hello: Document, options: ConnectionOptions) {