feat: use error labels for retryable writes in legacy topologies

NODE-2379

Author: mbroadst

lib/core/topologies/mongos.js

@@ -13,10 +13,10 @@ const cloneOptions = require('./shared').cloneOptions;
 const SessionMixins = require('./shared').SessionMixins;
 const isRetryableWritesSupported = require('./shared').isRetryableWritesSupported;
 const relayEvents = require('../utils').relayEvents;
-const isRetryableError = require('../error').isRetryableError;
 const BSON = retrieveBSON();
 const getMMAPError = require('./shared').getMMAPError;
 const makeClientMetadata = require('../utils').makeClientMetadata;
+const legacyIsRetryableWriteError = require('./shared').legacyIsRetryableWriteError;
 
 /**
  * @fileOverview The **Mongos** class is a class that represents a Mongos Proxy topology and is
@@ -113,6 +113,18 @@ var Mongos = function(seedlist, options) {
   // Get replSet Id
   this.id = id++;
 
+  // deduplicate seedlist
+  if (Array.isArray(seedlist)) {
+    seedlist = seedlist.reduce((seeds, seed) => {
+      if (seeds.find(s => s.host === seed.host && s.port === seed.port)) {
+        return seeds;
+      }
+
+      seeds.push(seed);
+      return seeds;
+    }, []);
+  }
+
   // Internal state
   this.s = {
     options: Object.assign({ metadata: makeClientMetadata(options) }, options),
@@ -911,7 +923,7 @@ function executeWriteOperation(args, options, callback) {
 
   const handler = (err, result) => {
     if (!err) return callback(null, result);
-    if (!isRetryableError(err) || !willRetryWrite) {
+    if (!legacyIsRetryableWriteError(err, self) || !willRetryWrite) {
       err = getMMAPError(err);
       return callback(err);
     }
@@ -1107,7 +1119,7 @@ Mongos.prototype.command = function(ns, cmd, options, callback) {
 
   const cb = (err, result) => {
     if (!err) return callback(null, result);
-    if (!isRetryableError(err)) {
+    if (!legacyIsRetryableWriteError(err, self)) {
       return callback(err);
     }
 
@@ -1121,8 +1133,8 @@ Mongos.prototype.command = function(ns, cmd, options, callback) {
 
   // increment and assign txnNumber
   if (willRetryWrite) {
-    options.session.incrementTransactionNumber();
-    options.willRetryWrite = willRetryWrite;
+    clonedOptions.session.incrementTransactionNumber();
+    clonedOptions.willRetryWrite = willRetryWrite;
   }
 
   // Execute the command

lib/core/topologies/replset.js

@@ -15,11 +15,11 @@ const Interval = require('./shared').Interval;
 const SessionMixins = require('./shared').SessionMixins;
 const isRetryableWritesSupported = require('./shared').isRetryableWritesSupported;
 const relayEvents = require('../utils').relayEvents;
-const isRetryableError = require('../error').isRetryableError;
 const BSON = retrieveBSON();
 const calculateDurationInMs = require('../utils').calculateDurationInMs;
 const getMMAPError = require('./shared').getMMAPError;
 const makeClientMetadata = require('../utils').makeClientMetadata;
+const legacyIsRetryableWriteError = require('./shared').legacyIsRetryableWriteError;
 
 //
 // States
@@ -1202,7 +1202,7 @@ function executeWriteOperation(args, options, callback) {
 
   const handler = (err, result) => {
     if (!err) return callback(null, result);
-    if (!isRetryableError(err)) {
+    if (!legacyIsRetryableWriteError(err, self)) {
       err = getMMAPError(err);
       return callback(err);
     }
@@ -1365,7 +1365,7 @@ ReplSet.prototype.command = function(ns, cmd, options, callback) {
 
   const cb = (err, result) => {
     if (!err) return callback(null, result);
-    if (!isRetryableError(err)) {
+    if (!legacyIsRetryableWriteError(err, self)) {
       return callback(err);
     }
 

lib/core/topologies/shared.js

@@ -2,7 +2,9 @@
 const ReadPreference = require('./read_preference');
 const TopologyType = require('../sdam/common').TopologyType;
 const MongoError = require('../error').MongoError;
-
+const isRetryableWriteError = require('../error').isRetryableWriteError;
+const maxWireVersion = require('../utils').maxWireVersion;
+const MongoNetworkError = require('../error').MongoNetworkError;
 const MMAPv1_RETRY_WRITES_ERROR_CODE = 20;
 
 /**
@@ -409,18 +411,39 @@ function getMMAPError(err) {
   return newErr;
 }
 
-module.exports.SessionMixins = SessionMixins;
-module.exports.resolveClusterTime = resolveClusterTime;
-module.exports.inquireServerState = inquireServerState;
-module.exports.getTopologyType = getTopologyType;
-module.exports.emitServerDescriptionChanged = emitServerDescriptionChanged;
-module.exports.emitTopologyDescriptionChanged = emitTopologyDescriptionChanged;
-module.exports.cloneOptions = cloneOptions;
-module.exports.createCompressionInfo = createCompressionInfo;
-module.exports.clone = clone;
-module.exports.diff = diff;
-module.exports.Interval = Interval;
-module.exports.Timeout = Timeout;
-module.exports.isRetryableWritesSupported = isRetryableWritesSupported;
-module.exports.getMMAPError = getMMAPError;
-module.exports.topologyType = topologyType;
+// NOTE: only used for legacy topology types
+function legacyIsRetryableWriteError(err, topology) {
+  if (!(err instanceof MongoError)) {
+    return false;
+  }
+
+  // if pre-4.4 server, then add error label if its a retryable write error
+  if (
+    isRetryableWritesSupported(topology) &&
+    (err instanceof MongoNetworkError ||
+      (maxWireVersion(topology) < 9 && isRetryableWriteError(err)))
+  ) {
+    err.addErrorLabel('RetryableWriteError');
+  }
+
+  return err.hasErrorLabel('RetryableWriteError');
+}
+
+module.exports = {
+  SessionMixins,
+  resolveClusterTime,
+  inquireServerState,
+  getTopologyType,
+  emitServerDescriptionChanged,
+  emitTopologyDescriptionChanged,
+  cloneOptions,
+  createCompressionInfo,
+  clone,
+  diff,
+  Interval,
+  Timeout,
+  isRetryableWritesSupported,
+  getMMAPError,
+  topologyType,
+  legacyIsRetryableWriteError
+};