feat(NODE-3083): support aggregate writes on secondaries

Author: durran

src/operations/aggregate.ts

@@ -1,5 +1,4 @@
 import { CommandOperation, CommandOperationOptions, CollationOptions } from './command';
-import { ReadPreference } from '../read_preference';
 import { MongoInvalidArgumentError } from '../error';
 import { maxWireVersion, MongoDBNamespace } from '../utils';
 import { Aspect, defineAspects, Hint } from './operation';
@@ -65,7 +64,7 @@ export class AggregateOperation<T = Document> extends CommandOperation<T> {
     }
 
     if (this.hasWriteStage) {
-      this.readPreference = ReadPreference.primary;
+      this.trySecondaryWrite = true;
     }
 
     if (this.explain && this.writeConcern) {

src/operations/execute_operation.ts

@@ -17,6 +17,7 @@ import type { Topology } from '../sdam/topology';
 import type { ClientSession } from '../sessions';
 import type { Document } from '../bson';
 import { supportsRetryableWrites } from '../utils';
+import { secondaryWritableServerSelector } from '../sdam/server_selection';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
 const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
@@ -181,8 +182,18 @@ function executeWithServerSelection(
       return;
     }
 
+    let selector;
+
+    // If operation should try to write to secondary use the custom server selector
+    // otherwise provide the read preference.
+    if (operation.trySecondaryWrite) {
+      selector = secondaryWritableServerSelector(topology.commonWireVersion, readPreference);
+    } else {
+      selector = readPreference;
+    }
+
     // select a new server, and attempt to retry the operation
-    topology.selectServer(readPreference, serverSelectionOptions, (e?: any, server?: any) => {
+    topology.selectServer(selector, serverSelectionOptions, (e?: any, server?: any) => {
       if (
         e ||
         (operation.hasAspect(Aspect.READ_OPERATION) && !supportsRetryableReads(server)) ||

src/operations/operation.ts

@@ -49,6 +49,7 @@ export abstract class AbstractOperation<TResult = any> {
   readPreference: ReadPreference;
   server!: Server;
   bypassPinningCheck: boolean;
+  trySecondaryWrite = false;
 
   // BSON serialization options
   bsonOptions?: BSONSerializeOptions;

src/sdam/server_selection.ts

@@ -8,6 +8,9 @@ import type { ServerDescription, TagSet } from './server_description';
 const IDLE_WRITE_PERIOD = 10000;
 const SMALLEST_MAX_STALENESS_SECONDS = 90;
 
+//  Minimum version to try writes on secondaries.
+export const MIN_SECONDARY_WRITE_WIRE_VERSION = 13;
+
 /** @public */
 export type ServerSelector = (
   topologyDescription: TopologyDescription,
@@ -28,6 +31,24 @@ export function writableServerSelector(): ServerSelector {
     );
 }
 
+/**
+ * Returns a server selector that uses a read preference to select a
+ * server potentially for a write on a secondary.
+ */
+export function secondaryWritableServerSelector(
+  wireVersion?: number,
+  readPreference?: ReadPreference
+): ServerSelector {
+  // If server version < 5.0, read preference always primary.
+  // If server version >= 5.0...
+  // - If read preference is supplied, use that.
+  // - If no read preference is supplied, use primary.
+  if (!readPreference || (wireVersion && wireVersion < MIN_SECONDARY_WRITE_WIRE_VERSION)) {
+    return readPreferenceServerSelector(ReadPreference.primary);
+  }
+  return readPreferenceServerSelector(readPreference);
+}
+
 /**
  * Reduces the passed in array of servers by the rules of the "Max Staleness" specification
  * found here: https://github.com/mongodb/specifications/blob/master/source/max-staleness/max-staleness.rst

src/sdam/topology.ts

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

test/functional/crud_spec.test.js

@@ -424,15 +424,11 @@ describe('CRUD spec v1', function () {
   }
 });
 
-// TODO: Unskip when implementing NODE-3083.
-const SKIP = ['aggregate-write-readPreference', 'db-aggregate-write-readPreference'];
-
 describe('CRUD unified', function () {
   for (const crudSpecTest of loadSpecTests('crud/unified')) {
     expect(crudSpecTest).to.exist;
     const testDescription = String(crudSpecTest.description);
-    const spec = SKIP.includes(testDescription) ? context.skip : context;
-    spec(testDescription, function () {
+    context(testDescription, function () {
       for (const test of crudSpecTest.tests) {
         it(String(test.description), {
           metadata: { sessions: { skipLeakTests: true } },

test/unit/operations/aggregate.test.js

@@ -0,0 +1,42 @@
+'use strict';
+
+const { expect } = require('chai');
+const { AggregateOperation } = require('../../../src/operations/aggregate');
+
+describe('AggregateOperation', function () {
+  const db = 'test';
+
+  describe('#constructor', function () {
+    context('when out is in the options', function () {
+      const operation = new AggregateOperation(db, [], { out: 'test', dbName: db });
+
+      it('sets trySecondaryWrite to true', function () {
+        expect(operation.trySecondaryWrite).to.be.true;
+      });
+    });
+
+    context('when $out is the last stage', function () {
+      const operation = new AggregateOperation(db, [{ $out: 'test' }], { dbName: db });
+
+      it('sets trySecondaryWrite to true', function () {
+        expect(operation.trySecondaryWrite).to.be.true;
+      });
+    });
+
+    context('when $merge is the last stage', function () {
+      const operation = new AggregateOperation(db, [{ $merge: { into: 'test' } }], { dbName: db });
+
+      it('sets trySecondaryWrite to true', function () {
+        expect(operation.trySecondaryWrite).to.be.true;
+      });
+    });
+
+    context('when no writable stages', function () {
+      const operation = new AggregateOperation(db, [], { dbName: db });
+
+      it('sets trySecondaryWrite to false', function () {
+        expect(operation.trySecondaryWrite).to.be.false;
+      });
+    });
+  });
+});

test/unit/sdam/server_selection.test.js

@@ -0,0 +1,94 @@
+'use strict';
+
+const { expect } = require('chai');
+const { ObjectId } = require('../../../src/bson');
+const { ReadPreference } = require('../../../src/read_preference');
+const {
+  secondaryWritableServerSelector,
+  MIN_SECONDARY_WRITE_WIRE_VERSION
+} = require('../../../src/sdam/server_selection');
+const { ServerDescription } = require('../../../src/sdam/server_description');
+const { TopologyDescription } = require('../../../src/sdam/topology_description');
+const { TopologyType } = require('../../../src/sdam/common');
+
+describe('ServerSelector', function () {
+  describe('#secondaryWritableServerSelector', function () {
+    const primary = new ServerDescription('127.0.0.1:27017', {
+      setName: 'test',
+      isWritablePrimary: true,
+      ok: 1
+    });
+    const secondary = new ServerDescription('127.0.0.1:27018', {
+      setName: 'test',
+      secondary: true,
+      ok: 1
+    });
+    const serverDescriptions = new Map();
+    serverDescriptions.set('127.0.0.1:27017', primary);
+    serverDescriptions.set('127.0.0.1:27018', secondary);
+
+    context('when the common server version is >= 5.0', function () {
+      const topologyDescription = new TopologyDescription(
+        TopologyType.ReplicaSetWithPrimary,
+        serverDescriptions,
+        'test',
+        MIN_SECONDARY_WRITE_WIRE_VERSION,
+        new ObjectId(),
+        MIN_SECONDARY_WRITE_WIRE_VERSION
+      );
+
+      context('when a read preference is provided', function () {
+        const selector = secondaryWritableServerSelector(
+          MIN_SECONDARY_WRITE_WIRE_VERSION,
+          ReadPreference.secondary
+        );
+        const server = selector(topologyDescription, Array.from(serverDescriptions.values()));
+
+        it('uses the provided read preference', function () {
+          expect(server).to.deep.equal([secondary]);
+        });
+      });
+
+      context('when a read preference is not provided', function () {
+        const selector = secondaryWritableServerSelector(MIN_SECONDARY_WRITE_WIRE_VERSION);
+        const server = selector(topologyDescription, Array.from(serverDescriptions.values()));
+
+        it('selects a primary', function () {
+          expect(server).to.deep.equal([primary]);
+        });
+      });
+    });
+
+    context('when the common server version is < 5.0', function () {
+      const topologyDescription = new TopologyDescription(
+        TopologyType.ReplicaSetWithPrimary,
+        serverDescriptions,
+        'test',
+        MIN_SECONDARY_WRITE_WIRE_VERSION - 1,
+        new ObjectId(),
+        MIN_SECONDARY_WRITE_WIRE_VERSION - 1
+      );
+
+      context('when a read preference is provided', function () {
+        const selector = secondaryWritableServerSelector(
+          MIN_SECONDARY_WRITE_WIRE_VERSION - 1,
+          ReadPreference.secondary
+        );
+        const server = selector(topologyDescription, Array.from(serverDescriptions.values()));
+
+        it('selects a primary', function () {
+          expect(server).to.deep.equal([primary]);
+        });
+      });
+
+      context('when read preference is not provided', function () {
+        const selector = secondaryWritableServerSelector(MIN_SECONDARY_WRITE_WIRE_VERSION - 1);
+        const server = selector(topologyDescription, Array.from(serverDescriptions.values()));
+
+        it('selects a primary', function () {
+          expect(server).to.deep.equal([primary]);
+        });
+      });
+    });
+  });
+});