fix(NODE-7298): ensure commonWireVersion is computed from server maxWireVersion (#4805)

Co-authored-by: Christoph Rehbichler <c.rehbichler@appmea.com>
Co-authored-by: Sergey Zelenov <sergey.zelenov@mongodb.com>
Co-authored-by: Bailey Pearson <bailey.pearson@mongodb.com>

Author: 4 people

src/sdam/server_selection.ts

@@ -88,7 +88,7 @@ export function sameServerSelector(description?: ServerDescription): ServerSelec
  * server potentially for a write on a secondary.
  */
 export function secondaryWritableServerSelector(
-  wireVersion?: number,
+  wireVersion: number,
   readPreference?: ReadPreference
 ): ServerSelector {
   // If server version < 5.0, read preference always primary.

src/sdam/topology.ts

@@ -778,7 +778,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     return result;
   }
 
-  get commonWireVersion(): number | undefined {
+  get commonWireVersion(): number {
     return this.description.commonWireVersion;
   }
 

src/sdam/topology_description.ts

@@ -192,7 +192,7 @@ export class TopologyDescription {
 
     // update common wire version
     if (serverDescription.maxWireVersion !== 0) {
-      if (commonWireVersion == null) {
+      if (commonWireVersion === 0) {
         commonWireVersion = serverDescription.maxWireVersion;
       } else {
         commonWireVersion = Math.min(commonWireVersion, serverDescription.maxWireVersion);

test/integration/crud/aggregation.test.ts

@@ -870,4 +870,73 @@ describe('Aggregation', function () {
         .finally(() => client.close());
     }
   });
+
+  it(
+    'should perform aggregations with a write stage on secondary when readPreference is secondary',
+    {
+      metadata: { requires: { topology: 'replicaset', mongodb: '>=5.0' } },
+      async test() {
+        const databaseName = this.configuration.db;
+        const client = this.configuration.newClient(this.configuration.writeConcernMax(), {
+          maxPoolSize: 1,
+          monitorCommands: true
+        });
+
+        const events = [];
+        client.on('commandStarted', filterForCommands(['hello', 'aggregate'], events));
+
+        // Discover primary to be able to check the actual server address
+        await client.db('admin').command({ hello: 1 });
+        const [helloEvent] = events;
+        const primaryAddress = helloEvent.address;
+
+        // Clear events
+        events.length = 0;
+
+        const src = client.db(databaseName).collection('read_pref_src');
+        const outMerge = client.db(databaseName).collection('read_pref_merge_out');
+        const outOut = client.db(databaseName).collection('read_pref_out_out');
+
+        await Promise.all([src.deleteMany({}), outMerge.deleteMany({}), outOut.deleteMany({})]);
+        await src.insertMany([{ a: 1 }, { a: 2 }]);
+        await Promise.all([
+          src
+            .aggregate(
+              [
+                {
+                  $merge: {
+                    into: 'read_pref_merge_out',
+                    whenMatched: 'replace',
+                    whenNotMatched: 'insert'
+                  }
+                }
+              ],
+              { readPreference: 'secondary' }
+            )
+            .toArray(),
+          src
+            .aggregate(
+              [
+                {
+                  $out: 'read_pref_out_out'
+                }
+              ],
+              { readPreference: 'secondary' }
+            )
+            .toArray()
+        ]);
+
+        expect(events).to.have.length(2);
+        events.forEach(event => {
+          expect(event).to.have.property('commandName', 'aggregate');
+          expect(event.address).to.not.equal(primaryAddress);
+          expect(event).to.have.deep.nested.property('command.$readPreference', {
+            mode: 'secondary'
+          });
+        });
+
+        await client.close();
+      }
+    }
+  );
 });

test/unit/sdam/server_selection.test.ts

@@ -295,7 +295,7 @@ describe('server selection', function () {
           new ObjectId(),
           MIN_SECONDARY_WRITE_WIRE_VERSION
         );
-        const selector = secondaryWritableServerSelector();
+        const selector = secondaryWritableServerSelector(0);
         const servers = selector(
           topologyDescription,
           Array.from(serverDescriptions.values()),
@@ -401,7 +401,7 @@ describe('server selection', function () {
           new ObjectId(),
           MIN_SECONDARY_WRITE_WIRE_VERSION
         );
-        const selector = secondaryWritableServerSelector();
+        const selector = secondaryWritableServerSelector(0);
         const servers = selector(
           topologyDescription,
           Array.from(serverDescriptions.values()),
@@ -507,7 +507,7 @@ describe('server selection', function () {
           new ObjectId(),
           MIN_SECONDARY_WRITE_WIRE_VERSION
         );
-        const selector = secondaryWritableServerSelector();
+        const selector = secondaryWritableServerSelector(0);
         const servers = selector(
           topologyDescription,
           Array.from(serverDescriptions.values()),
@@ -564,7 +564,7 @@ describe('server selection', function () {
           new ObjectId(),
           MIN_SECONDARY_WRITE_WIRE_VERSION
         );
-        const selector = secondaryWritableServerSelector();
+        const selector = secondaryWritableServerSelector(0);
         const servers = selector(
           topologyDescription,
           Array.from(serverDescriptions.values()),
@@ -587,7 +587,7 @@ describe('server selection', function () {
           MIN_SECONDARY_WRITE_WIRE_VERSION,
           { localThresholdMS: 5 }
         );
-        const selector = secondaryWritableServerSelector();
+        const selector = secondaryWritableServerSelector(0);
         const servers = selector(
           topologyDescription,
           Array.from(serverDescriptions.values()),

test/unit/sdam/topology_description.test.ts

@@ -0,0 +1,63 @@
+import { expect } from 'chai';
+
+import { TopologyType } from '../../../src/sdam/common';
+import { ServerDescription } from '../../../src/sdam/server_description';
+import { TopologyDescription } from '../../../src/sdam/topology_description';
+
+describe('TopologyDescription', function () {
+  describe('#constructor', function () {
+    it('sets commonWireVersion to 0', function () {
+      const initial = new TopologyDescription(TopologyType.ReplicaSetWithPrimary);
+
+      expect(initial.commonWireVersion).to.equal(0);
+    });
+  });
+
+  describe('update()', function () {
+    it('initializes commonWireVersion from first non-zero maxWireVersion', function () {
+      const initial = new TopologyDescription(TopologyType.ReplicaSetWithPrimary);
+
+      const sd1 = new ServerDescription('a:27017', {
+        maxWireVersion: 25
+      });
+
+      const updated = initial.update(sd1);
+
+      expect(updated.commonWireVersion).to.equal(25);
+    });
+
+    it('tracks the minimum non-zero maxWireVersion across updates in commonWireVersion', function () {
+      const initial = new TopologyDescription(TopologyType.ReplicaSetWithPrimary);
+
+      const sd1 = new ServerDescription('a:27017', {
+        maxWireVersion: 25
+      });
+
+      const sd2 = new ServerDescription('b:27017', {
+        maxWireVersion: 21
+      });
+
+      let updated = initial.update(sd1);
+      updated = updated.update(sd2);
+
+      expect(updated.commonWireVersion).to.equal(21);
+    });
+
+    it('ignores servers with maxWireVersion === 0 when computing commonWireVersion', function () {
+      const initial = new TopologyDescription(TopologyType.ReplicaSetWithPrimary);
+
+      const sd1 = new ServerDescription('a:27017', {
+        maxWireVersion: 25
+      });
+
+      const sdUnknown = new ServerDescription('b:27017', {
+        maxWireVersion: 0
+      });
+
+      let updated = initial.update(sd1);
+      updated = updated.update(sdUnknown);
+
+      expect(updated.commonWireVersion).to.equal(25);
+    });
+  });
+});