test(NODE-6858): add tests for ServerSelectionError

Author: tadjik1

test/integration/change-streams/change_stream.test.ts

@@ -14,6 +14,7 @@ import {
   type CommandStartedEvent,
   type Db,
   isHello,
+  LEGACY_HELLO_COMMAND,
   Long,
   MongoAPIError,
   MongoChangeStreamError,
@@ -1818,7 +1819,7 @@ describe('Change Streams', function () {
   });
 });
 
-describe('ChangeStream resumability', function () {
+describe.only('ChangeStream resumability', function () {
   let client: MongoClient;
   let collection: Collection;
   let changeStream: ChangeStream;
@@ -1832,30 +1833,30 @@ describe('ChangeStream resumability', function () {
   };
 
   const resumableErrorCodes = [
-    { error: 'HostUnreachable', code: 6, message: 'host unreachable' },
-    { error: 'HostNotFound', code: 7, message: 'hot not found' },
-    { error: 'NetworkTimeout', code: 89, message: 'network timeout' },
-    { error: 'ShutdownInProgress', code: 91, message: 'shutdown in progress' },
-    { error: 'PrimarySteppedDown', code: 189, message: 'primary stepped down' },
-    { error: 'ExceededTimeLimit', code: 262, message: 'operation exceeded time limit' },
-    { error: 'SocketException', code: 9001, message: 'socket exception' },
-    { error: 'NotWritablePrimary', code: 10107, message: 'not writable primary' },
-    { error: 'InterruptedAtShutdown', code: 11600, message: 'interrupted at shutdown' },
-    {
-      error: 'InterruptedDueToReplStateChange',
-      code: 11602,
-      message: 'interrupted due to state change'
-    },
-    { error: 'NotPrimaryNoSecondaryOk', code: 13435, message: 'not primary and no secondary ok' },
-    { error: 'StaleShardVersion', code: 63, message: 'stale shard version' },
-    { error: 'StaleEpoch', code: 150, message: 'stale epoch' },
-    { error: 'RetryChangeStream', code: 234, message: 'retry change stream' },
-    {
-      error: 'FailedToSatisfyReadPreference',
-      code: 133,
-      message: 'failed to satisfy read preference'
-    },
-    { error: 'CursorNotFound', code: 43, message: 'cursor not found' }
+    { error: 'HostUnreachable', code: 6, message: 'host unreachable' }
+    // { error: 'HostNotFound', code: 7, message: 'hot not found' },
+    // { error: 'NetworkTimeout', code: 89, message: 'network timeout' },
+    // { error: 'ShutdownInProgress', code: 91, message: 'shutdown in progress' },
+    // { error: 'PrimarySteppedDown', code: 189, message: 'primary stepped down' },
+    // { error: 'ExceededTimeLimit', code: 262, message: 'operation exceeded time limit' },
+    // { error: 'SocketException', code: 9001, message: 'socket exception' },
+    // { error: 'NotWritablePrimary', code: 10107, message: 'not writable primary' },
+    // { error: 'InterruptedAtShutdown', code: 11600, message: 'interrupted at shutdown' },
+    // {
+    //   error: 'InterruptedDueToReplStateChange',
+    //   code: 11602,
+    //   message: 'interrupted due to state change'
+    // },
+    // { error: 'NotPrimaryNoSecondaryOk', code: 13435, message: 'not primary and no secondary ok' },
+    // { error: 'StaleShardVersion', code: 63, message: 'stale shard version' },
+    // { error: 'StaleEpoch', code: 150, message: 'stale epoch' },
+    // { error: 'RetryChangeStream', code: 234, message: 'retry change stream' },
+    // {
+    //   error: 'FailedToSatisfyReadPreference',
+    //   code: 133,
+    //   message: 'failed to satisfy read preference'
+    // },
+    // { error: 'CursorNotFound', code: 43, message: 'cursor not found' }
   ];
 
   beforeEach(function () {
@@ -2045,6 +2046,65 @@ describe('ChangeStream resumability', function () {
           expect(changeStream.closed).to.be.true;
         });
       });
+
+      context.only('when the error is not a server error', function () {
+        let client1: MongoClient;
+        let client2: MongoClient;
+
+        beforeEach(async function () {
+          client1 = this.configuration.newClient(
+            {},
+            { serverSelectionTimeoutMS: 1000, appName: 'client-errors' }
+          );
+          client2 = this.configuration.newClient();
+
+          collection = client1.db('client-errors').collection('test');
+        });
+
+        afterEach(async function () {
+          await client2.db('admin').command({
+            configureFailPoint: 'failCommand',
+            mode: 'off',
+            data: { appName: 'client-errors' }
+          } as FailCommandFailPoint);
+
+          await client1?.close();
+          await client2?.close();
+        });
+
+        it(
+          'should resume on ServerSelectionError',
+          { requires: { topology: '!single' } },
+          async function () {
+            changeStream = collection.watch([]);
+            await initIteratorMode(changeStream);
+
+            await collection.insertOne({ a: 1 });
+
+            await client2.db('admin').command({
+              configureFailPoint: 'failCommand',
+              mode: 'alwaysOn',
+              data: {
+                failCommands: ['ping', 'hello', LEGACY_HELLO_COMMAND],
+                closeConnection: true,
+                handshakeCommands: true,
+                failInternalCommands: true,
+                appName: 'client-errors'
+              }
+            } as FailCommandFailPoint);
+            await client2
+              .db('admin')
+              .command({ replSetFreeze: 0 }, { readPreference: ReadPreference.secondary });
+            await client2
+              .db('admin')
+              .command({ replSetStepDown: 15, secondaryCatchUpPeriodSecs: 10, force: true });
+            // await sleep(15_000);
+
+            const change = await changeStream.next();
+            expect(change).to.containSubset({ operationType: 'insert', fullDocument: { a: 1 } });
+          }
+        );
+      });
     });
 
     context('#hasNext', function () {