[WIP] feat: handle stream muxer errors

src/dial.js

@@ -77,7 +77,10 @@ function dial (swtch) {
         // 3. update the peerInfo that is already stored in the conn
       }
 
-      openConnInMuxedConn(muxer, (conn) => {
+      openConnInMuxedConn(muxer, (err, conn) => {
+        if (err) {
+          return callback(err)
+        }
         protocolHandshake(conn, protocol, callback)
       })
     }
@@ -89,18 +92,20 @@ function dial (swtch) {
 
       const tKeys = swtch.availableTransports(pi)
 
+      const circuitEnabled = Boolean(swtch.transports[Circuit.tag])
       let circuitTried = false
       nextTransport(tKeys.shift())
 
       function nextTransport (key) {
         let transport = key
         if (!transport) {
-          if (circuitTried) {
-            return cb(new Error(`Circuit already tried!`))
+          if (!circuitEnabled) {
+            const msg = `Circuit not enabled and all transports failed to dial peer ${pi.id.toB58String()}!`
+            return cb(new Error(msg))
           }
 
-          if (!swtch.transports[Circuit.tag]) {
-            return cb(new Error(`Circuit not enabled!`))
+          if (circuitTried) {
+            return cb(new Error(`No available transports to dial peer ${pi.id.toB58String()}!`))
           }
 
           log(`Falling back to dialing over circuit`)
@@ -208,7 +213,13 @@ function dial (swtch) {
     }
 
     function openConnInMuxedConn (muxer, cb) {
-      cb(muxer.newStream())
+      muxer.newStream((err, conn) => {
+        if (err) {
+          return cb(err)
+        }
+
+        cb(null, conn)
+      })
     }
 
     function protocolHandshake (conn, protocol, cb) {

test/circuit-relay.node.js

@@ -88,7 +88,7 @@ describe(`circuit`, function () {
 
     swarmA.dial(swarmC._peerInfo, (err, conn) => {
       expect(err).to.exist()
-      expect(err).to.match(/Circuit already tried!/)
+      expect(err).to.match(/No available transports to dial peer/)
       expect(conn).to.not.exist()
       expect(dialSpyA.callCount).to.be.eql(1)
       done()

test/node.js

@@ -7,3 +7,4 @@ require('./swarm-no-muxing.node')
 require('./swarm-muxing.node')
 require('./circuit-relay.node')
 require('./limit-dialer.node')
+require('./stats.node')

test/stats.node.js

@@ -10,7 +10,7 @@ const each = require('async/each')
 const map = require('async/map')
 const series = require('async/series')
 const TCP = require('libp2p-tcp')
-const multiplex = require('libp2p-multiplex')
+const multiplex = require('libp2p-mplex')
 const pull = require('pull-stream')
 const secio = require('libp2p-secio')
 const PeerBook = require('peer-book')

test/stream-muxers.node.js

@@ -22,14 +22,15 @@ const Switch = require('../src')
 describe('Stream Multiplexing', () => {
   [
     multiplex,
-    spdy
+    spdy // TODO: do we still support this?
   ].forEach((sm) => describe(sm.multicodec, () => {
     let switchA
     let switchB
     let switchC
 
     before((done) => createInfos(3, (err, peerInfos) => {
       expect(err).to.not.exist()
+
       function maGen (port) { return `/ip4/127.0.0.1/tcp/${port}` }
 
       const peerA = peerInfos[0]
@@ -149,5 +150,22 @@ describe('Stream Multiplexing', () => {
         }, 500)
       })
     })
+
+    it('should fail graciously on dead stream mux', function (done) {
+      if (sm.multicodec === '/spdy/3.1.0') {
+        this.skip()
+      }
+
+      const id = switchA._peerInfo.id.toB58String()
+      const muxer = switchB.muxedConns[id].muxer
+      muxer.multiplex.destroyed = true // simulate a destroyed stream
+
+      switchB.dial(switchA._peerInfo, '/papaia/1.0.0', (err, conn) => {
+        expect(err).to.exist()
+        expect(err).to.match(/Error: Multiplexer is destroyed/)
+        muxer.multiplex.destroyed = false // re-enable stream to avoid hangs
+        done()
+      })
+    })
   }))
 })