fix(NODE-6412): read stale response from previously timed out connection (#4273)

Author: nbbeeken

src/cmap/connection.ts

@@ -752,9 +752,12 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       }
     } catch (readError) {
       if (TimeoutError.is(readError)) {
-        throw new MongoOperationTimeoutError(
+        const error = new MongoOperationTimeoutError(
           `Timed out during socket read (${readError.duration}ms)`
         );
+        this.dataEvents = null;
+        this.onError(error);
+        throw error;
       }
       throw readError;
     } finally {

test/integration/client-side-operations-timeout/client_side_operations_timeout.spec.test.ts

@@ -48,6 +48,12 @@ describe('CSOT spec tests', function () {
 
   runUnifiedSuite(specs, (test, configuration) => {
     const sessionCSOTTests = ['timeoutMS applied to withTransaction'];
+    if (
+      configuration.topologyType === 'LoadBalanced' &&
+      test.description === 'timeoutMS is refreshed for close'
+    ) {
+      return 'LoadBalanced cannot refresh timeoutMS and run expected killCursors because pinned connection has been closed by the timeout';
+    }
     if (
       sessionCSOTTests.includes(test.description) &&
       configuration.topologyType === 'ReplicaSetWithPrimary' &&

test/integration/client-side-operations-timeout/node_csot.test.ts

@@ -1116,4 +1116,50 @@ describe('CSOT driver tests', metadata, () => {
       );
     });
   });
+
+  describe('Connection after timeout', { requires: { mongodb: '>=4.4' } }, function () {
+    let client: MongoClient;
+
+    beforeEach(async function () {
+      client = this.configuration.newClient({ timeoutMS: 500 });
+
+      const failpoint: FailPoint = {
+        configureFailPoint: 'failCommand',
+        mode: {
+          times: 1
+        },
+        data: {
+          failCommands: ['insert'],
+          blockConnection: true,
+          blockTimeMS: 700
+        }
+      };
+
+      await client.db('admin').command(failpoint);
+    });
+
+    afterEach(async function () {
+      await client.close();
+    });
+
+    it('closes so pending messages are not read by another operation', async function () {
+      const cmap = [];
+      client.on('connectionCheckedOut', ev => cmap.push(ev));
+      client.on('connectionClosed', ev => cmap.push(ev));
+
+      const error = await client
+        .db('socket')
+        .collection('closes')
+        .insertOne({})
+        .catch(error => error);
+
+      expect(error).to.be.instanceOf(MongoOperationTimeoutError);
+      expect(cmap).to.have.lengthOf(2);
+
+      const [checkedOut, closed] = cmap;
+      expect(checkedOut).to.have.property('name', 'connectionCheckedOut');
+      expect(closed).to.have.property('name', 'connectionClosed');
+      expect(checkedOut).to.have.property('connectionId', closed.connectionId);
+    });
+  });
 });