feat: limit the number of cold calls we can do

README.md

@@ -61,8 +61,11 @@ const sw = new switch(peerInfo , peerBook [, options])
 
 If defined, `options` should be an object with the following keys and respective values:
 
-- `blacklistTTL`: - number of ms a peer should not be dialable to after it errors. Defaults to `120000`(120 seconds)
-- `maxParallelDials`: - number of concurrent dials the switch should allow. Defaults to `50`
+- `blacklistTTL`: - number of ms a peer should not be dialable to after it errors. Each successive blacklisting will increase the ttl from the base value. Defaults to 5 minutes
+- `blackListAttempts`: - number of blacklists before a peer
+is permanently blacklisted. Defaults to 5.
+- `maxParallelDials`: - number of concurrent dials the switch should allow. Defaults to `100`
+- `maxColdCalls`: - number of queued cold calls that are allowed. Defaults to `50`
 - `dialTimeout`: - number of ms a dial to a peer should be allowed to run. Defaults to `30000` (30 seconds)
 - `stats`: an object with the following keys and respective values:
   - `maxOldPeersRetention`: maximum old peers retention. For when peers disconnect and keeping the stats around in case they reconnect. Defaults to `100`.

src/constants.js

@@ -1,7 +1,9 @@
 'use strict'
 
 module.exports = {
-  BLACK_LIST_TTL: 120e3, // How long before an errored peer can be dialed again
+  BLACK_LIST_TTL: 5 * 60 * 1e3, // How long before an errored peer can be dialed again
+  BLACK_LIST_ATTEMPTS: 5, // Num of unsuccessful dials before a peer is permanently blacklisted
   DIAL_TIMEOUT: 30e3, // How long in ms a dial attempt is allowed to take
-  MAX_PARALLEL_DIALS: 50 // Maximum allowed concurrent dials
+  MAX_COLD_CALLS: 50, // How many dials w/o protocols that can be queued
+  MAX_PARALLEL_DIALS: 100 // Maximum allowed concurrent dials
 }

src/dialer/index.js

@@ -2,7 +2,12 @@
 
 const DialQueueManager = require('./queueManager')
 const getPeerInfo = require('../get-peer-info')
-const { MAX_PARALLEL_DIALS, BLACK_LIST_TTL } = require('../constants')
+const {
+  BLACK_LIST_ATTEMPTS,
+  BLACK_LIST_TTL,
+  MAX_COLD_CALLS,
+  MAX_PARALLEL_DIALS
+} = require('../constants')
 
 module.exports = function (_switch) {
   const dialQueueManager = new DialQueueManager(_switch)
@@ -74,7 +79,9 @@ module.exports = function (_switch) {
     dialFSM,
     abort,
     clearBlacklist,
+    BLACK_LIST_ATTEMPTS: isNaN(_switch._options.blackListAttempts) ? BLACK_LIST_ATTEMPTS : _switch._options.blackListAttempts,
     BLACK_LIST_TTL: isNaN(_switch._options.blacklistTTL) ? BLACK_LIST_TTL : _switch._options.blacklistTTL,
+    MAX_COLD_CALLS: isNaN(_switch._options.maxColdCalls) ? MAX_COLD_CALLS : _switch._options.maxColdCalls,
     MAX_PARALLEL_DIALS: isNaN(_switch._options.maxParallelDials) ? MAX_PARALLEL_DIALS : _switch._options.maxParallelDials
   }
 }

src/dialer/queue.js

@@ -70,6 +70,7 @@ class Queue {
     this.switch = _switch
     this._queue = []
     this.blackListed = null
+    this.blackListCount = 0
     this.isRunning = false
     this.onStopped = onStopped
   }
@@ -97,7 +98,7 @@ class Queue {
   isDialAllowed () {
     if (this.blackListed) {
       // If the blacklist ttl has passed, reset it
-      if (Date.now() - this.blackListed > this.switch.dialer.BLACK_LIST_TTL) {
+      if (Date.now() > this.blackListed) {
         this.blackListed = null
         return true
       }
@@ -146,10 +147,24 @@ class Queue {
 
   /**
    * Marks the queue as blacklisted. The queue will be immediately aborted.
+   * @returns {void}
    */
   blacklist () {
-    log('blacklisting queue for %s', this.id)
-    this.blackListed = Date.now()
+    this.blackListCount++
+
+    if (this.blackListCount >= this.switch.dialer.BLACK_LIST_ATTEMPTS) {
+      this.blackListed = Infinity
+      return
+    }
+
+    let ttl = this.switch.dialer.BLACK_LIST_TTL * Math.pow(this.blackListCount, 3)
+    const minTTL = ttl * 0.9
+    const maxTTL = ttl * 1.1
+
+    // Add a random jitter of 20% to the ttl
+    ttl = Math.floor(Math.random() * (maxTTL - minTTL) + minTTL)
+
+    this.blackListed = Date.now() + ttl
     this.abort()
   }
 
@@ -236,13 +251,16 @@ class Queue {
 
     // If we're not muxed yet, add listeners
     connectionFSM.once('muxed', () => {
+      this.blackListCount = 0 // reset blacklisting on good connections
       this.switch.connection.add(connectionFSM)
       queuedDial.connection = connectionFSM
       createConnectionWithProtocol(queuedDial)
       next()
     })
 
     connectionFSM.once('unmuxed', () => {
+      this.blackListCount = 0
+      this.switch.connection.add(connectionFSM)
       queuedDial.connection = connectionFSM
       createConnectionWithProtocol(queuedDial)
       next()

src/dialer/queueManager.js

@@ -2,6 +2,8 @@
 
 const once = require('once')
 const Queue = require('./queue')
+const { DIAL_ABORTED } = require('../errors')
+const nextTick = require('async/nextTick')
 const noop = () => {}
 
 class DialQueueManager {
@@ -11,6 +13,7 @@ class DialQueueManager {
    */
   constructor (_switch) {
     this._queue = new Set()
+    this._coldCallQueue = new Set()
     this._dialingQueues = new Set()
     this._queues = {}
     this.switch = _switch
@@ -34,7 +37,7 @@ class DialQueueManager {
   }
 
   /**
-   * Adds the `dialRequest` to the queue and ensures the queue is running
+   * Adds the `dialRequest` to the qMAX_COLD_CALLSe queue is running
    *
    * @param {DialRequest} dialRequest
    * @returns {void}
@@ -44,6 +47,11 @@ class DialQueueManager {
 
     // Add the dial to its respective queue
     const targetQueue = this.getQueue(peerInfo)
+    // If we have too many cold calls, abort the dial immediately
+    if (this._coldCallQueue.size >= this.switch.dialer.MAX_COLD_CALLS && !protocol) {
+      return nextTick(callback, DIAL_ABORTED())
+    }
+
     targetQueue.add(protocol, useFSM, callback)
 
     // If we're already connected to the peer, start the queue now
@@ -54,10 +62,23 @@ class DialQueueManager {
       return
     }
 
-    // Add the id to the general queue set if the queue isn't running
-    // and if the queue is allowed to dial
-    if (!targetQueue.isRunning && targetQueue.isDialAllowed()) {
-      this._queue.add(targetQueue.id)
+    // If dialing is not allowed, abort
+    if (!targetQueue.isDialAllowed()) {
+      return
+    }
+
+    // Add the id to its respective queue set if the queue isn't running
+    if (!targetQueue.isRunning) {
+      if (protocol) {
+        this._queue.add(targetQueue.id)
+        this._coldCallQueue.delete(targetQueue.id)
+      // Only add it to the cold queue if it's not in the normal queue
+      } else if (!this._queue.has(targetQueue.id)) {
+        this._coldCallQueue.add(targetQueue.id)
+      // The peer is already in the normal queue, abort the cold call
+      } else {
+        return nextTick(callback, DIAL_ABORTED())
+      }
     }
 
     this.run()
@@ -67,11 +88,21 @@ class DialQueueManager {
    * Will execute up to `MAX_PARALLEL_DIALS` dials
    */
   run () {
-    if (this._dialingQueues.size < this.switch.dialer.MAX_PARALLEL_DIALS && this._queue.size > 0) {
-      let nextQueue = this._queue.values().next()
-      if (nextQueue.done) return
+    if (this._dialingQueues.size < this.switch.dialer.MAX_PARALLEL_DIALS) {
+      let nextQueue = { done: true }
+      // Check the queue first and fall back to the cold call queue
+      if (this._queue.size > 0) {
+        nextQueue = this._queue.values().next()
+        this._queue.delete(nextQueue.value)
+      } else if (this._coldCallQueue.size > 0) {
+        nextQueue = this._coldCallQueue.values().next()
+        this._coldCallQueue.delete(nextQueue.value)
+      }
+
+      if (nextQueue.done) {
+        return
+      }
 
-      this._queue.delete(nextQueue.value)
       let targetQueue = this._queues[nextQueue.value]
       this._dialingQueues.add(targetQueue.id)
       targetQueue.start()
@@ -83,7 +114,9 @@ class DialQueueManager {
    * @param {PeerInfo} peerInfo
    */
   clearBlacklist (peerInfo) {
-    this.getQueue(peerInfo).blackListed = null
+    const queue = this.getQueue(peerInfo)
+    queue.blackListed = null
+    queue.blackListCount = 0
   }
 
   /**

test/dialer.spec.js

@@ -0,0 +1,139 @@
+/* eslint-env mocha */
+'use strict'
+
+const chai = require('chai')
+const dirtyChai = require('dirty-chai')
+const expect = chai.expect
+chai.use(require('chai-checkmark'))
+chai.use(dirtyChai)
+const sinon = require('sinon')
+
+const PeerBook = require('peer-book')
+const Queue = require('../src/dialer/queue')
+const QueueManager = require('../src/dialer/queueManager')
+const Switch = require('../src')
+
+const utils = require('./utils')
+const createInfos = utils.createInfos
+
+describe('dialer', () => {
+  let switchA
+
+  before((done) => createInfos(1, (err, infos) => {
+    expect(err).to.not.exist()
+
+    switchA = new Switch(infos[0], new PeerBook())
+
+    done()
+  }))
+
+  afterEach(() => {
+    sinon.restore()
+  })
+
+  describe('queue', () => {
+    it('should blacklist forever after 5 blacklists', () => {
+      const queue = new Queue('QM', switchA)
+      for (var i = 0; i < 4; i++) {
+        queue.blacklist()
+        expect(queue.blackListed).to.be.a('number')
+        expect(queue.blackListed).to.not.eql(Infinity)
+      }
+
+      queue.blacklist()
+      expect(queue.blackListed).to.eql(Infinity)
+    })
+  })
+
+  describe('queue manager', () => {
+    let queueManager
+    before(() => {
+      queueManager = new QueueManager(switchA)
+    })
+
+    it('should abort cold calls when the queue is full', (done) => {
+      sinon.stub(queueManager._coldCallQueue, 'size').value(switchA.dialer.MAX_COLD_CALLS)
+      const dialRequest = {
+        peerInfo: {
+          id: { toB58String: () => 'QmA' }
+        },
+        protocol: null,
+        useFSM: true,
+        callback: (err) => {
+          expect(err.code).to.eql('DIAL_ABORTED')
+          done()
+        }
+      }
+
+      queueManager.add(dialRequest)
+    })
+
+    it('should add a protocol dial to the normal queue', () => {
+      const dialRequest = {
+        peerInfo: {
+          id: { toB58String: () => 'QmA' },
+          isConnected: () => null
+        },
+        protocol: '/echo/1.0.0',
+        useFSM: true,
+        callback: () => {}
+      }
+
+      const runSpy = sinon.stub(queueManager, 'run')
+      const addSpy = sinon.stub(queueManager._queue, 'add')
+      const deleteSpy = sinon.stub(queueManager._coldCallQueue, 'delete')
+
+      queueManager.add(dialRequest)
+
+      expect(runSpy.called).to.eql(true)
+      expect(addSpy.called).to.eql(true)
+      expect(addSpy.getCall(0).args[0]).to.eql('QmA')
+      expect(deleteSpy.called).to.eql(true)
+      expect(deleteSpy.getCall(0).args[0]).to.eql('QmA')
+    })
+
+    it('should add a cold call to the cold call queue', () => {
+      const dialRequest = {
+        peerInfo: {
+          id: { toB58String: () => 'QmA' },
+          isConnected: () => null
+        },
+        protocol: null,
+        useFSM: true,
+        callback: () => {}
+      }
+
+      const runSpy = sinon.stub(queueManager, 'run')
+      const addSpy = sinon.stub(queueManager._coldCallQueue, 'add')
+
+      queueManager.add(dialRequest)
+
+      expect(runSpy.called).to.eql(true)
+      expect(addSpy.called).to.eql(true)
+      expect(addSpy.getCall(0).args[0]).to.eql('QmA')
+    })
+
+    it('should abort a cold call if it\'s in the normal queue', (done) => {
+      const dialRequest = {
+        peerInfo: {
+          id: { toB58String: () => 'QmA' },
+          isConnected: () => null
+        },
+        protocol: null,
+        useFSM: true,
+        callback: (err) => {
+          expect(runSpy.called).to.eql(false)
+          expect(hasSpy.called).to.eql(true)
+          expect(hasSpy.getCall(0).args[0]).to.eql('QmA')
+          expect(err.code).to.eql('DIAL_ABORTED')
+          done()
+        }
+      }
+
+      const runSpy = sinon.stub(queueManager, 'run')
+      const hasSpy = sinon.stub(queueManager._queue, 'has').returns(true)
+
+      queueManager.add(dialRequest)
+    })
+  })
+})