NODEJS-681: ControlConnection Concurrent Read and Write on .host and .connection

lib/control-connection.js

@@ -490,7 +490,9 @@ class ControlConnection extends events.EventEmitter {
       await this._registerToConnectionEvents();
     } catch (err) {
       this.log('error', 'ControlConnection failed to retrieve topology and keyspaces information', err);
-      this._triedHosts[this.connection.endpoint] = err;
+      if (this.connection != null) {
+        this._triedHosts[this.connection.endpoint] = err;
+      }
 
       if (err.isSocketError && this.host) {
         this.host.removeFromPool(this.connection);
@@ -501,10 +503,12 @@ class ControlConnection extends events.EventEmitter {
     }
 
     this._reconnectionSchedule = this._reconnectionPolicy.newSchedule();
-    this._setHealthListeners(this.host, this.connection);
-    this.emit('newConnection', null, this.connection, this.host);
-
-    this.log('info', `ControlConnection connected to ${this.connection.endpointFriendlyName} and up to date`);
+    if (this.host != null && this.connection != null) {
+      this._setHealthListeners(this.host, this.connection);
+      this.emit('newConnection', null, this.connection, this.host);
+
+      this.log('info', `ControlConnection connected to ${this.connection.endpointFriendlyName} and up to date`);
+    } // else, someone else started a _refresh or _initializeConnection, ignoring the current result
   }
 
   /**

test/integration/short/control-connection-tests.js

@@ -151,6 +151,27 @@ describe('ControlConnection', function () {
       assert.strictEqual(cc.hosts.length, 1);
     });
 
+    it('should not break when refreshing concurrently', async () => {
+      const cc = newInstance();
+      cc.options.policies.loadBalancing = new policies.loadBalancing.RoundRobinPolicy();
+      disposeAfter(cc);
+
+      await cc.init();
+      await new Promise(r => cc.options.policies.loadBalancing.init(null, cc.hosts, r));
+
+      /** @type {Array<Promise>} */
+      const refreshPromises = [];
+      // randomly emit 200 times cc._refresh
+      for (let i = 0; i < 200; i++) {
+        refreshPromises.push(cc._refresh());
+        // interval of random number
+        await helper.delayAsync(Math.floor(Math.random() * 200));
+      }
+      Promise.all(refreshPromises);
+      assert.ok(cc.host);
+      assert.ok(cc.connection);
+    });
+
     it('should reconnect when host used goes down', async () => {
       const options = clientOptions.extend(
         utils.extend({ pooling: helper.getPoolingOptions(1, 1, 500) }, helper.baseOptions));

test/test-helper.js

@@ -1441,10 +1441,10 @@ helper.ads._spawnAndWait = function(processName, params, cb) {
     cb = utils.noop;
   };
 
-  // If process hasn't completed in 10 seconds.
+  // If process hasn't completed in 20 seconds.
   timeout = setTimeout(function() {
     callbackOnce(new Error("Timed out while waiting for " + processName + " to complete."));
-  }, 10000);
+  }, 20000);
 
   const p = spawn(processName, params, {
     env: Object.assign({ KRB5_CONFIG: this.getKrb5ConfigPath()}, process.env)