fix: improve connection tracking and closing (#291)

* chore: update deps

* fix: check we have a proper transport before filtering addresses

* fix: improve connection close on stop

* fix: improve stat stopping

* test: fix stats test

* fix: improve tracking of open connections

* chore: remove log

* fix: stats stop in browser

chore: fix linting and browser tests

* fix: remove uneeded set peer info

* fix: abort the base connection on close

* fix: catch edge cases of dialTimeout calling back twice

* fix: close all connections instead of checking peerbook peers

* test: update dial fsm test waits

* test: make parallel dial tests deterministic

fix: improve logic around disconnecting

fix: remove duplicate event handling logic

* chore: fix lint

* test: improve test reliability

package.json

@@ -40,20 +40,20 @@
  "npm": ">=3.0.0"
  },
  "devDependencies": {
- "aegir": "^17.0.1",
+ "aegir": "^17.1.1",
  "chai": "^4.2.0",
  "chai-checkmark": "^1.0.1",
  "dirty-chai": "^2.0.1",
  "libp2p-mplex": "~0.8.4",
  "libp2p-pnet": "~0.1.0",
  "libp2p-secio": "~0.10.1",
- "libp2p-spdy": "~0.13.0",
+ "libp2p-spdy": "~0.13.1",
  "libp2p-tcp": "~0.13.0",
- "libp2p-webrtc-star": "~0.15.5",
+ "libp2p-webrtc-star": "~0.15.6",
  "libp2p-websockets": "~0.12.0",
- "peer-book": "~0.8.0",
- "portfinder": "^1.0.19",
- "sinon": "^7.1.1",
+ "peer-book": "~0.9.0",
+ "portfinder": "^1.0.20",
+ "sinon": "^7.2.0",
  "webrtcsupport": "^2.2.0"
  },
  "dependencies": {
@@ -63,18 +63,18 @@
  "debug": "^4.1.0",
  "err-code": "^1.1.2",
  "fsm-event": "^2.1.0",
- "hashlru": "^2.2.1",
- "interface-connection": "~0.3.2",
+ "hashlru": "^2.3.0",
+ "interface-connection": "~0.3.3",
  "ip-address": "^5.8.9",
- "libp2p-circuit": "~0.3.0",
+ "libp2p-circuit": "~0.3.1",
  "libp2p-identify": "~0.7.2",
  "lodash.includes": "^4.3.0",
  "moving-average": "^1.0.0",
- "multiaddr": "^5.0.2",
+ "multiaddr": "^6.0.0",
  "multistream-select": "~0.14.3",
  "once": "^1.4.0",
  "peer-id": "~0.12.0",
- "peer-info": "~0.14.1",
+ "peer-info": "~0.15.0",
  "pull-stream": "^3.6.9",
  "retimer": "^2.0.0"
  },

src/connection/incoming.js

@@ -20,7 +20,9 @@ class IncomingConnectionFSM extends BaseConnection {
  this.msListener = new multistream.Listener()
 
  this._state = FSM('DIALED', {
- DISCONNECTED: { },
+ DISCONNECTED: {
+ disconnect: 'DISCONNECTED'
+ },
  DIALED: { // Base connection to peer established
  privatize: 'PRIVATIZING',
  encrypt: 'ENCRYPTING'

src/connection/index.js

@@ -1,7 +1,6 @@
 'use strict'
 
 const FSM = require('fsm-event')
-const setImmediate = require('async/setImmediate')
 const Circuit = require('libp2p-circuit')
 const multistream = require('multistream-select')
 const withIs = require('class-is')
@@ -15,6 +14,8 @@ const Errors = require('../errors')
  * @property {Switch} _switch Our switch instance
  * @property {PeerInfo} peerInfo The PeerInfo of the peer to dial
  * @property {Muxer} muxer Optional - A muxed connection
+ * @property {Connection} conn Optional - The base connection
+ * @property {string} type Optional - identify the connection as incoming or outgoing. Defaults to out.
  */
 
 /**
@@ -29,16 +30,16 @@ class ConnectionFSM extends BaseConnection {
  * @param {ConnectionOptions} param0
  * @constructor
  */
- constructor ({ _switch, peerInfo, muxer }) {
+ constructor ({ _switch, peerInfo, muxer, conn, type = 'out' }) {
  super({
  _switch,
- name: `out:${_switch._peerInfo.id.toB58String().slice(0, 8)}`
+ name: `${type}:${_switch._peerInfo.id.toB58String().slice(0, 8)}`
  })
 
  this.theirPeerInfo = peerInfo
  this.theirB58Id = this.theirPeerInfo.id.toB58String()
 
- this.conn = null // The base connection
+ this.conn = conn // The base connection
  this.muxer = muxer // The upgraded/muxed connection
 
  let startState = 'DISCONNECTED'
@@ -114,6 +115,7 @@ class ConnectionFSM extends BaseConnection {
  this._state.on('UPGRADING', () => this._onUpgrading())
  this._state.on('MUXED', () => {
  this.log(`successfully muxed connection to ${this.theirB58Id}`)
+ delete this.switch.conns[this.theirB58Id]
  this.emit('muxed', this.muxer)
  })
  this._state.on('CONNECTED', () => {
@@ -166,7 +168,6 @@ class ConnectionFSM extends BaseConnection {
  })
  }
 
- this.conn.setPeerInfo(this.theirPeerInfo)
  this._protocolHandshake(protocol, this.conn, callback)
  }
 
@@ -266,14 +267,22 @@ class ConnectionFSM extends BaseConnection {
  this.muxer.end()
  }
 
- delete this.switch.muxedConns[this.theirB58Id]
+ this.switch.connection.remove(this)
+
  delete this.switch.conns[this.theirB58Id]
  delete this.muxer
- delete this.conn
-
- this._state('done')
 
- setImmediate(() => this.switch.emit('peer-mux-closed', this.theirPeerInfo))
+ // If we have the base connection, abort it
+ if (this.conn) {
+ this.conn.source(true, () => {
+ this._state('done')
+ this.switch.emit('peer-mux-closed', this.theirPeerInfo)
+ delete this.conn
+ })
+ } else {
+ this._state('done')
+ this.switch.emit('peer-mux-closed', this.theirPeerInfo)
+ }
  }
 
  /**
@@ -352,7 +361,8 @@ class ConnectionFSM extends BaseConnection {
  const conn = observeConnection(null, key, _conn, this.switch.observer)
 
  this.muxer = this.switch.muxers[key].dialer(conn)
- this.switch.muxedConns[this.theirB58Id] = this
+ // this.switch.muxedConns[this.theirB58Id] = this
+ this.switch.connection.add(this)
 
  this.muxer.once('close', () => {
  this.close()
@@ -365,7 +375,7 @@ class ConnectionFSM extends BaseConnection {
  this.switch.protocolMuxer(null)(conn)
  })
 
- setImmediate(() => this.switch.emit('peer-mux-established', this.theirPeerInfo))
+ this.switch.emit('peer-mux-established', this.theirPeerInfo)
 
  this._didUpgrade(null)
  })

src/connection/manager.js

@@ -6,7 +6,6 @@ const waterfall = require('async/waterfall')
 const debug = require('debug')
 const log = debug('libp2p:switch:conn-manager')
 const once = require('once')
-const setImmediate = require('async/setImmediate')
 const ConnectionFSM = require('../connection')
 
 const Circuit = require('libp2p-circuit')
@@ -20,6 +19,92 @@ const plaintext = require('../plaintext')
 class ConnectionManager {
  constructor (_switch) {
  this.switch = _switch
+ this.connections = {}
+ }
+
+ /**
+ * Adds the connection for tracking if it's not already added
+ * @private
+ * @param {ConnectionFSM} connection
+ * @returns {void}
+ */
+ add (connection) {
+ this.connections[connection.theirB58Id] = this.connections[connection.theirB58Id] || []
+ // Only add it if it's not there
+ if (!this.get(connection)) {
+ this.connections[connection.theirB58Id].push(connection)
+ }
+ }
+
+ /**
+ * Gets the connection from the list if it exists
+ * @private
+ * @param {ConnectionFSM} connection
+ * @returns {ConnectionFSM|null} The found connection or null
+ */
+ get (connection) {
+ if (!this.connections[connection.theirB58Id]) return null
+
+ for (let i = 0; i < this.connections[connection.theirB58Id].length; i++) {
+ if (this.connections[connection.theirB58Id][i] === connection) {
+ return this.connections[connection.theirB58Id][i]
+ }
+ }
+ return null
+ }
+
+ /**
+ * Gets a connection associated with the given peer
+ * @private
+ * @param {string} peerId The peers id
+ * @returns {ConnectionFSM|null} The found connection or null
+ */
+ getOne (peerId) {
+ if (this.connections[peerId]) {
+ // TODO: Maybe select the best?
+ return this.connections[peerId][0]
+ }
+ return null
+ }
+
+ /**
+ * Removes the connection from tracking
+ * @private
+ * @param {ConnectionFSM} connection The connection to remove
+ * @returns {void}
+ */
+ remove (connection) {
+ if (!this.connections[connection.theirB58Id]) return
+
+ for (let i = 0; i < this.connections[connection.theirB58Id].length; i++) {
+ if (this.connections[connection.theirB58Id][i] === connection) {
+ this.connections[connection.theirB58Id].splice(i, 1)
+ return
+ }
+ }
+ }
+
+ /**
+ * Returns all connections being tracked
+ * @private
+ * @returns {ConnectionFSM[]}
+ */
+ getAll () {
+ let connections = []
+ for (const conns of Object.values(this.connections)) {
+ connections = [...connections, ...conns]
+ }
+ return connections
+ }
+
+ /**
+ * Returns all connections being tracked for a given peer id
+ * @private
+ * @param {string} peerId Stringified peer id
+ * @returns {ConnectionFSM[]}
+ */
+ getAllById (peerId) {
+ return this.connections[peerId] || []
  }
 
  /**
@@ -70,9 +155,6 @@ class ConnectionManager {
  ], (err, peerInfo) => {
  if (err) {
  return muxedConn.end(() => {
- if (peerInfo) {
- setImmediate(() => this.switch.emit('peer-mux-closed', peerInfo))
- }
  callback(err, null)
  })
  }
@@ -91,11 +173,14 @@ class ConnectionManager {
  }
  const b58Str = peerInfo.id.toB58String()
 
- this.switch.muxedConns[b58Str] = new ConnectionFSM({
+ const connection = new ConnectionFSM({
  _switch: this.switch,
  peerInfo,
- muxer: muxedConn
+ muxer: muxedConn,
+ conn: conn,
+ type: 'inc'
  })
+ this.switch.connection.add(connection)
 
  if (peerInfo.multiaddrs.size > 0) {
  // with incomming conn and through identify, going to pick one
@@ -111,14 +196,10 @@ class ConnectionManager {
  peerInfo = this.switch._peerBook.put(peerInfo)
 
  muxedConn.once('close', () => {
- delete this.switch.muxedConns[b58Str]
- peerInfo.disconnect()
- peerInfo = this.switch._peerBook.put(peerInfo)
- log(`closed connection to ${b58Str}`)
- setImmediate(() => this.switch.emit('peer-mux-closed', peerInfo))
+ connection.close()
  })
 
- setImmediate(() => this.switch.emit('peer-mux-established', peerInfo))
+ this.switch.emit('peer-mux-established', peerInfo)
  })
  })
  }

src/dialer.js

@@ -54,13 +54,14 @@ function dial (_switch, returnFSM) {
 
  log(`dialing to ${b58Id.slice(0, 8)} with protocol ${protocol || 'unknown'}`)
 
- let connection = _switch.muxedConns[b58Id] || _switch.conns[b58Id]
+ let connection = _switch.connection.getOne(b58Id)
 
  if (!ConnectionFSM.isConnectionFSM(connection)) {
  connection = new ConnectionFSM({
  _switch,
  peerInfo,
- muxer: _switch.muxedConns[b58Id] || null
+ muxer: null,
+ conn: null
  })
  connection.once('error', (err) => callback(err))
  connection.once('connected', () => connection.protect())