fix(NODE-4783): handle orphaned operation descriptions

src/cmap/connection.ts

@@ -17,6 +17,7 @@ import {
   MongoMissingDependencyError,
   MongoNetworkError,
   MongoNetworkTimeoutError,
+  MongoRuntimeError,
   MongoServerError,
   MongoWriteConcernError
 } from '../error';
@@ -68,6 +69,8 @@ const kAutoEncrypter = Symbol('autoEncrypter');
 /** @internal */
 const kDelayedTimeoutId = Symbol('delayedTimeoutId');
 
+const INVALID_QUEUE_SIZE = 'Connection internal queue contains more than 1 operation description';
+
 /** @internal */
 export interface CommandOptions extends BSONSerializeOptions {
   command?: boolean;
@@ -369,7 +372,29 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
     // always emit the message, in case we are streaming
     this.emit('message', message);
-    const operationDescription = this[kQueue].get(message.responseTo);
+    let operationDescription = this[kQueue].get(message.responseTo);
+
+    // Protect against multiplexing.
+    if (this[kQueue].size > 1) {
+      this.onError(new MongoRuntimeError(INVALID_QUEUE_SIZE));
+      return;
+    }
+
+    if (!operationDescription && this.isMonitoringConnection) {
+      // This is how we recover when the initial hello's requestId is not
+      // the responseTo when hello responses have been skipped:
+
+      // Get the first orphaned operation description.
+      const entry = this[kQueue].entries().next();
+      if (entry) {
+        const [requestId, orphaned]: [number, OperationDescription] = entry.value;
+        // If the orphaned operation description exists then set it.
+        operationDescription = orphaned;
+        // Remove the entry with the bad request id from the queue.
+        this[kQueue].delete(requestId);
+      }
+    }
+
     if (!operationDescription) {
       return;
     }
@@ -381,7 +406,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     // making the `responseTo` change on each response
     this[kQueue].delete(message.responseTo);
     if ('moreToCome' in message && message.moreToCome) {
-      // requeue the callback for next synthetic request
+      // If the operation description check above does find an orphaned
+      // description and sets the operationDescription then this line will put one
+      // back in the queue with the correct requestId and will resolve not being able
+      // to find the next one via the responseTo of the next streaming hello.
       this[kQueue].set(message.requestId, operationDescription);
     } else if (operationDescription.socketTimeoutOverride) {
       this[kStream].setTimeout(this.socketTimeoutMS);

test/tools/utils.ts

@@ -1,6 +1,7 @@
 import { EJSON } from 'bson';
 import * as BSON from 'bson';
 import { expect } from 'chai';
+import { Readable } from 'stream';
 import { setTimeout } from 'timers';
 import { inspect, promisify } from 'util';
 
@@ -354,6 +355,18 @@ export class TestBuilder {
   }
 }
 
+export function bufferToStream(buffer) {
+  const stream = new Readable();
+  if (Array.isArray(buffer)) {
+    buffer.forEach(b => stream.push(b));
+  } else {
+    stream.push(buffer);
+  }
+
+  stream.push(null);
+  return stream;
+}
+
 export function generateOpMsgBuffer(document: Document): Buffer {
   const header = Buffer.alloc(4 * 4 + 4);