rollback extra changes

Author: nbbeeken

.evergreen/run-kms-servers.sh

@@ -2,6 +2,6 @@ cd ${DRIVERS_TOOLS}/.evergreen/csfle
 . ./activate_venv.sh
 # by default it always runs on port 5698
 ./kmstlsvenv/bin/python3 -u kms_kmip_server.py &
-./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/expired.pem --port 9000 &
-./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/wrong-host.pem --port 9001 &
-./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/server.pem --port 9002 --require_client_cert &
+./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/expired.pem --port 8000 &
+./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/wrong-host.pem --port 8001 &
+./kmstlsvenv/bin/python3 -u kms_http_server.py --ca_file ../x509gen/ca.pem --cert_file ../x509gen/server.pem --port 8002 --require_client_cert &

docs/errors.md

@@ -95,7 +95,11 @@ This class should **never** be directly instantiated.
 
 ### MongoUnexpectedServerResponseError
 
-xxx
+Intended for the scenario where the MongoDB returns an unexpected response in relation to some state the driver is in
+This error should **NOT** represent a response that couldn't be parsed due to errors in protocol formatting.
+
+Ex. Server selection results in a feature detection change, this is not a direct unexpected response, but if we've begun retrying an operation and serverSelection during that retry returns a server with a lower wireVersion than expected, we can no longer proceed with the retry.
+
 
 ### `MongoNetworkError`
 

src/change_stream.ts

@@ -2,7 +2,6 @@ import Denque = require('denque');
 import type { Readable } from 'stream';
 
 import type { Document, Timestamp } from './bson';
-import { MONGODB_WIRE_VERSION } from './cmap/wire_protocol/constants';
 import { Collection } from './collection';
 import {
   AbstractCursor,
@@ -463,10 +462,7 @@ export class ChangeStreamCursor<TSchema extends Document = Document> extends Abs
         const resumeKey =
           this.options.startAfter && !this.hasReceived ? 'startAfter' : 'resumeAfter';
         Reflect.set(result, resumeKey, this.resumeToken);
-      } else if (
-        this.startAtOperationTime &&
-        maxWireVersion(this.server) >= MONGODB_WIRE_VERSION.REPLICA_SET_TRANSACTIONS
-      ) {
+      } else if (this.startAtOperationTime && maxWireVersion(this.server) >= 7) {
         result.startAtOperationTime = this.startAtOperationTime;
       }
     }
@@ -517,7 +513,7 @@ export class ChangeStreamCursor<TSchema extends Document = Document> extends Abs
         this.startAtOperationTime == null &&
         this.resumeAfter == null &&
         this.startAfter == null &&
-        maxWireVersion(server) >= MONGODB_WIRE_VERSION.REPLICA_SET_TRANSACTIONS
+        maxWireVersion(server) >= 7
       ) {
         this.startAtOperationTime = response.operationTime;
       }

src/cmap/auth/mongo_credentials.ts

@@ -2,7 +2,6 @@
 import type { Document } from '../../bson';
 import { MongoAPIError, MongoMissingCredentialsError } from '../../error';
 import { emitWarningOnce } from '../../utils';
-import { MONGODB_WIRE_VERSION } from '../wire_protocol/constants';
 import { AUTH_MECHS_AUTH_SRC_EXTERNAL, AuthMechanism } from './providers';
 
 // https://github.com/mongodb/specifications/blob/master/source/auth/auth.rst
@@ -17,7 +16,7 @@ function getDefaultAuthMechanism(hello?: Document): AuthMechanism {
     }
 
     // Fallback to legacy selection method. If wire version >= 3, use scram-sha-1
-    if (hello.maxWireVersion >= MONGODB_WIRE_VERSION.RELEASE_2_7_7) {
+    if (hello.maxWireVersion >= 3) {
       return AuthMechanism.MONGODB_SCRAM_SHA1;
     }
   }

src/cmap/auth/mongodb_aws.ts

@@ -11,7 +11,6 @@ import {
   MongoRuntimeError
 } from '../../error';
 import { Callback, maxWireVersion, ns } from '../../utils';
-import { MONGODB_WIRE_VERSION } from '../wire_protocol/constants';
 import { AuthContext, AuthProvider } from './auth_provider';
 import { MongoCredentials } from './mongo_credentials';
 import { AuthMechanism } from './providers';
@@ -45,7 +44,7 @@ export class MongoDBAWS extends AuthProvider {
     }
     const { sign } = aws4;
 
-    if (maxWireVersion(connection) < MONGODB_WIRE_VERSION.RESUMABLE_INITIAL_SYNC) {
+    if (maxWireVersion(connection) < 9) {
       callback(
         new MongoCompatibilityError(
           'MONGODB-AWS authentication requires MongoDB version 4.4 or later'

src/cmap/connection.ts

@@ -53,7 +53,6 @@ import {
 import type { Stream } from './connect';
 import { MessageStream, OperationDescription } from './message_stream';
 import { StreamDescription, StreamDescriptionOptions } from './stream_description';
-import { MONGODB_WIRE_VERSION } from './wire_protocol/constants';
 import { applyCommonQueryOptions, getReadPreference, isSharded } from './wire_protocol/shared';
 
 /** @internal */
@@ -251,35 +250,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       /* ignore errors, listen to `close` instead */
     });
 
-    const eventHandlerFactory = (event: 'error' | 'close' | 'timeout') =>
-      [
-        event,
-        (error?: Error) => {
-          if (this.closed) {
-            return;
-          }
-          this.closed = true;
-
-          switch (event) {
-            case 'error':
-              this.onError(error);
-              break;
-            case 'close':
-              this.onClose();
-              break;
-            case 'timeout':
-              this.onTimeout();
-              break;
-          }
-
-          this[kQueue].clear();
-          this.emit(Connection.CLOSE);
-        }
-      ] as const;
-
-    this[kMessageStream].on(...eventHandlerFactory('error'));
-    stream.on(...eventHandlerFactory('close'));
-    stream.on(...eventHandlerFactory('timeout'));
+    this[kMessageStream].on('error', error => this.handleIssue({ destroy: error }));
+    stream.on('close', () => this.handleIssue({ isClose: true }));
+    stream.on('timeout', () => this.handleIssue({ isTimeout: true, destroy: true }));
 
     // hook the message stream up to the passed in stream
     stream.pipe(this[kMessageStream]);
@@ -335,29 +308,33 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     this[kLastUseTime] = now();
   }
 
-  onTimeout() {
-    this[kStream].destroy();
-    const beforeHandshake = {
-      beforeHandshake: this.hello == null
-    };
-    const msg = `connection ${this.id} to ${this.address} timed out`;
-
-    for (const [, op] of this[kQueue]) {
-      op.cb(new MongoNetworkTimeoutError(msg, beforeHandshake));
+  handleIssue(issue: { isTimeout?: boolean; isClose?: boolean; destroy?: boolean | Error }): void {
+    if (this.closed) {
+      return;
     }
-  }
 
-  onClose() {
-    for (const [, op] of this[kQueue]) {
-      op.cb(new MongoNetworkError(`connection ${this.id} to ${this.address} closed`));
+    if (issue.destroy) {
+      this[kStream].destroy(typeof issue.destroy === 'boolean' ? undefined : issue.destroy);
     }
-  }
 
-  onError(error?: Error) {
-    this[kStream].destroy(error);
+    this.closed = true;
+
     for (const [, op] of this[kQueue]) {
-      op.cb(error);
+      if (issue.isTimeout) {
+        op.cb(
+          new MongoNetworkTimeoutError(`connection ${this.id} to ${this.address} timed out`, {
+            beforeHandshake: this.hello == null
+          })
+        );
+      } else if (issue.isClose) {
+        op.cb(new MongoNetworkError(`connection ${this.id} to ${this.address} closed`));
+      } else {
+        op.cb(typeof issue.destroy === 'boolean' ? undefined : issue.destroy);
+      }
     }
+
+    this[kQueue].clear();
+    this.emit(Connection.CLOSE);
   }
 
   destroy(): void;
@@ -566,7 +543,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       return;
     }
 
-    if (wireVersion < MONGODB_WIRE_VERSION.FIND_COMMAND) {
+    if (wireVersion < 4) {
       const getMoreOp = new GetMore(ns.toString(), cursorId, { numberToReturn: options.batchSize });
       const queryOptions = applyCommonQueryOptions(
         {},
@@ -620,7 +597,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       throw new MongoRuntimeError(`Invalid list of cursor ids provided: ${cursorIds}`);
     }
 
-    if (maxWireVersion(this) < MONGODB_WIRE_VERSION.FIND_COMMAND) {
+    if (maxWireVersion(this) < 4) {
       try {
         write(
           this,
@@ -678,12 +655,12 @@ export class CryptoConnection extends Connection {
     }
 
     const serverWireVersion = maxWireVersion(this);
-    if (serverWireVersion === MONGODB_WIRE_VERSION.UNKNOWN) {
+    if (serverWireVersion === 0) {
       // This means the initial handshake hasn't happened yet
       return super.command(ns, cmd, options, callback);
     }
 
-    if (serverWireVersion < MONGODB_WIRE_VERSION.SHARDED_TRANSACTIONS) {
+    if (serverWireVersion < 8) {
       callback(
         new MongoCompatibilityError('Auto-encryption requires a minimum MongoDB version of 4.2')
       );
@@ -720,10 +697,7 @@ function supportsOpMsg(conn: Connection) {
     return false;
   }
 
-  return (
-    maxWireVersion(conn) >= MONGODB_WIRE_VERSION.SUPPORTS_OP_MSG &&
-    !description.__nodejs_mock_server__ // mock server does not support OP_MSG
-  );
+  return maxWireVersion(conn) >= 6 && !description.__nodejs_mock_server__;
 }
 
 function messageHandler(conn: Connection) {

src/cmap/errors.ts

@@ -3,9 +3,6 @@ import type { ConnectionPool } from './connection_pool';
 
 /**
  * An error indicating a connection pool is closed
- *
- * @public
- *
  * @category Error
  */
 export class PoolClosedError extends MongoDriverError {

src/cmap/wire_protocol/constants.ts

@@ -1,54 +1,7 @@
-/**
- * All known MongoDB Wire versions, used to determine basic feature support
- * - see the [server enum](https://github.com/mongodb/mongo/blob/fe4cf6134b16f102591053d6f4fe11e5cc0eb3ec/src/mongo/db/wire_version.h#L57)
- * - (note: the is a link to a commit so you can find the file, you should then check the main branch)
- */
-export const MONGODB_WIRE_VERSION = Object.freeze({
-  /** A helper wire version, 0 means no information about what is supported on the server is known */
-  UNKNOWN: 0,
-  /** Everything before we started tracking. */
-  RELEASE_2_4_AND_BEFORE: 0,
-  /** The aggregation command may now be requested to return cursors. */
-  AGG_RETURNS_CURSORS: 1,
-  /** insert, update, and delete batch command */
-  BATCH_COMMANDS: 2,
-  /** support SCRAM-SHA1, listIndexes, listCollections, new explain */
-  RELEASE_2_7_7: 3,
-  /** Support find and getMore commands, as well as OP_COMMAND in mongod (but not mongos). */
-  FIND_COMMAND: 4,
-  /** Supports all write commands take a write concern. */
-  COMMANDS_ACCEPT_WRITE_CONCERN: 5,
-  /** Supports the new OP_MSG wireprotocol (3.6+). */
-  SUPPORTS_OP_MSG: 6,
-  /** Supports replica set transactions (4.0+). */
-  REPLICA_SET_TRANSACTIONS: 7,
-  /** Supports sharded transactions (4.2+). */
-  SHARDED_TRANSACTIONS: 8,
-  /** Supports resumable initial sync (4.4+). */
-  RESUMABLE_INITIAL_SYNC: 9,
-  /** Supports features available from 4.7 and onwards. */
-  WIRE_VERSION_47: 10,
-  /** Supports features available from 4.8 and onwards. */
-  WIRE_VERSION_48: 11,
-  /**
-   * Supports features available from 4.9 and onwards.
-   * - EstimatedDocumentCountOperation can run as an aggregation
-   */
-  WIRE_VERSION_49: 12,
-  /**
-   * Supports features available from 5.0 and onwards.
-   * - Writes to secondaries $out/$merge
-   * - Snapshot reads
-   */
-  WIRE_VERSION_50: 13,
-  /** Supports features available from 5.1 and onwards. */
-  WIRE_VERSION_51: 14
-} as const);
-
 export const MIN_SUPPORTED_SERVER_VERSION = '3.6';
 export const MAX_SUPPORTED_SERVER_VERSION = '5.1';
-export const MIN_SUPPORTED_WIRE_VERSION = MONGODB_WIRE_VERSION.SUPPORTS_OP_MSG;
-export const MAX_SUPPORTED_WIRE_VERSION = MONGODB_WIRE_VERSION.WIRE_VERSION_51;
+export const MIN_SUPPORTED_WIRE_VERSION = 6;
+export const MAX_SUPPORTED_WIRE_VERSION = 14;
 export const OP_REPLY = 1;
 export const OP_UPDATE = 2001;
 export const OP_INSERT = 2002;

src/error.ts

@@ -1,5 +1,4 @@
 import type { Document } from './bson';
-import { MONGODB_WIRE_VERSION } from './cmap/wire_protocol/constants';
 import type { TopologyVersion } from './sdam/server_description';
 import type { TopologyDescription } from './sdam/topology_description';
 
@@ -101,8 +100,6 @@ export interface ErrorDescription extends Document {
   errInfo?: Document;
 }
 
-const kCreated = Symbol('created');
-
 /**
  * @public
  * @category Error
@@ -113,7 +110,6 @@ const kCreated = Symbol('created');
 export class MongoError extends Error {
   /** @internal */
   [kErrorLabels]: Set<string>;
-  [kCreated]: Date;
   /**
    * This is a number in MongoServerError and a string in MongoDriverError
    * @privateRemarks
@@ -128,7 +124,6 @@ export class MongoError extends Error {
     } else {
       super(message);
     }
-    this[kCreated] = new Date();
   }
 
   get name(): string {
@@ -756,7 +751,7 @@ export function isRetryableWriteError(error: MongoError, maxWireVersion: number)
     return false;
   }
 
-  if (maxWireVersion >= MONGODB_WIRE_VERSION.RESUMABLE_INITIAL_SYNC) {
+  if (maxWireVersion >= 9) {
     // After 4.4 the error label is the only source of truth for retry writes
     return error.hasErrorLabel(MONGODB_ERROR_LABELS.RetryableWriteError);
   } else if (error.hasErrorLabel(MONGODB_ERROR_LABELS.RetryableWriteError)) {
@@ -884,7 +879,7 @@ export function isResumableError(error?: MongoError, wireVersion?: number): bool
     return true;
   }
 
-  if (wireVersion != null && wireVersion >= MONGODB_WIRE_VERSION.RESUMABLE_INITIAL_SYNC) {
+  if (wireVersion != null && wireVersion >= 9) {
     // DRIVERS-1308: For 4.4 drivers running against 4.4 servers, drivers will add a special case to treat the CursorNotFound error code as resumable
     if (error && error instanceof MongoError && error.code === MONGODB_ERROR_CODES.CursorNotFound) {
       return true;

src/operations/add_user.ts

@@ -1,7 +1,6 @@
 import * as crypto from 'crypto';
 
 import type { Document } from '../bson';
-import { MONGODB_WIRE_VERSION } from '../cmap/wire_protocol/constants';
 import type { Db } from '../db';
 import { MongoInvalidArgumentError } from '../error';
 import type { Server } from '../sdam/server';
@@ -76,8 +75,7 @@ export class AddUserOperation extends CommandOperation<Document> {
       roles = Array.isArray(options.roles) ? options.roles : [options.roles];
     }
 
-    const digestPassword =
-      getTopology(db).lastHello().maxWireVersion >= MONGODB_WIRE_VERSION.REPLICA_SET_TRANSACTIONS;
+    const digestPassword = getTopology(db).lastHello().maxWireVersion >= 7;
 
     let userPassword = password;
 

src/operations/aggregate.ts

@@ -1,5 +1,4 @@
 import type { Document } from '../bson';
-import { MONGODB_WIRE_VERSION } from '../cmap/wire_protocol/constants';
 import { MongoInvalidArgumentError } from '../error';
 import type { Server } from '../sdam/server';
 import type { ClientSession } from '../sessions';
@@ -10,6 +9,7 @@ import { Aspect, defineAspects, Hint } from './operation';
 
 /** @internal */
 export const DB_AGGREGATE_COLLECTION = 1 as const;
+const MIN_WIRE_VERSION_$OUT_READ_CONCERN_SUPPORT = 8 as const;
 
 /** @public */
 export interface AggregateOptions extends CommandOperationOptions {
@@ -91,11 +91,11 @@ export class AggregateOperation<T = Document> extends CommandOperation<T> {
     const serverWireVersion = maxWireVersion(server);
     const command: Document = { aggregate: this.target, pipeline: this.pipeline };
 
-    if (this.hasWriteStage && serverWireVersion < MONGODB_WIRE_VERSION.SHARDED_TRANSACTIONS) {
+    if (this.hasWriteStage && serverWireVersion < MIN_WIRE_VERSION_$OUT_READ_CONCERN_SUPPORT) {
       this.readConcern = undefined;
     }
 
-    if (serverWireVersion >= MONGODB_WIRE_VERSION.COMMANDS_ACCEPT_WRITE_CONCERN) {
+    if (serverWireVersion >= 5) {
       if (this.hasWriteStage && this.writeConcern) {
         Object.assign(command, { writeConcern: this.writeConcern });
       }