CASSNODEJS-14: Multiple Node Status Updates - All but One being Dropped

lib/control-connection.js

@@ -99,8 +99,8 @@ class ControlConnection extends events.EventEmitter {
 
     // Timeout used for delayed handling of topology changes
     this._topologyChangeTimeout = null;
-    // Timeout used for delayed handling of node status changes
-    this._nodeStatusChangeTimeout = null;
+    // Map of host endpoint strings to their pending status change timeouts
+    this._nodeStatusChangeTimers = new Map();
 
     if (context && context.borrowHostConnection) {
       this._borrowHostConnection = context.borrowHostConnection;
@@ -598,14 +598,23 @@ class ControlConnection extends events.EventEmitter {
         self.log('warning', 'Received status change event but host was not found: ' + addressToTranslate);
         return;
       }
+
+      const existingTimer = self._nodeStatusChangeTimers.get(endPoint);
+      if (existingTimer) {
+        clearTimeout(existingTimer);
+        self._nodeStatusChangeTimers.delete(endPoint);
+      }
+
       const distance = self._profileManager.getDistance(host);
       if (event.up) {
         if (distance === types.distance.ignored) {
           return host.setUp(true);
         }
-        clearTimeout(self._nodeStatusChangeTimeout);
         // Waits a couple of seconds before marking it as UP
-        self._nodeStatusChangeTimeout = setTimeout(() => host.checkIsUp(), newNodeDelay);
+        self._nodeStatusChangeTimers.set(endPoint, setTimeout(() => {
+          self._nodeStatusChangeTimers.delete(endPoint);
+          host.checkIsUp();
+        }, newNodeDelay));
         return;
       }
       // marked as down
@@ -998,7 +1007,10 @@ class ControlConnection extends events.EventEmitter {
     this.emit('newConnection', new errors.DriverError('ControlConnection is being shutdown'));
     // Cancel timers
     clearTimeout(this._topologyChangeTimeout);
-    clearTimeout(this._nodeStatusChangeTimeout);
+    for (const timer of this._nodeStatusChangeTimers.values()) {
+      clearTimeout(timer);
+    }
+    this._nodeStatusChangeTimers.clear();
   }
 
   /**

test/unit/control-connection-tests.js

@@ -449,6 +449,49 @@ describe('ControlConnection', function () {
     });
   });
 
+  describe('#_nodeStatusChangeHandler()', function () {
+    it('should manage pending timers correctly across UP and DOWN events', function () {
+      const options = clientOptions.extend({}, helper.baseOptions);
+      const cc = newInstance(options);
+      cc._addressTranslator = { translate: (addr, port, cb) => cb(`${addr}:${port}`) };
+      cc._profileManager.getDistance = () => types.distance.local;
+      cc.log = helper.noop;
+
+      const host = new Host('127.0.0.1:9042', 1, options);
+      let checkIsUpCalled = false;
+      host.checkIsUp = () => { checkIsUpCalled = true; };
+      let setUpCalled = false;
+      host.setUp = () => { setUpCalled = true; };
+      cc.hosts.set('127.0.0.1:9042', host);
+
+      const makeEvent = (up) => ({ inet: { address: { toString: () => '127.0.0.1' } }, up });
+
+      // UP event should schedule timer w/o immediately calling checkIsUp
+      cc._nodeStatusChangeHandler(makeEvent(true));
+      assert.strictEqual(cc._nodeStatusChangeTimers.size, 1);
+      assert.ok(cc._nodeStatusChangeTimers.has('127.0.0.1:9042'));
+      assert.ok(!checkIsUpCalled);
+
+      // second UP event should replace first timer
+      const firstTimer = cc._nodeStatusChangeTimers.get('127.0.0.1:9042');
+      cc._nodeStatusChangeHandler(makeEvent(true));
+      assert.strictEqual(cc._nodeStatusChangeTimers.size, 1);
+      assert.notStrictEqual(cc._nodeStatusChangeTimers.get('127.0.0.1:9042'), firstTimer);
+
+      // DOWN event should cancel pending UP timer
+      cc._nodeStatusChangeHandler(makeEvent(false));
+      assert.strictEqual(cc._nodeStatusChangeTimers.size, 0);
+
+      // Ignored hosts should call setUp immediately without any timer
+      cc._profileManager.getDistance = () => types.distance.ignored;
+      cc._nodeStatusChangeHandler(makeEvent(true));
+      assert.strictEqual(cc._nodeStatusChangeTimers.size, 0);
+      assert.ok(setUpCalled);
+
+      cc.shutdown();
+    });
+  });
+
   describe('#refresh()', function () {
     it('should schedule reconnection when it cant borrow a connection', async () => {
       const state = {};