fix: protocols change event

I was spelunking last night and found that pubsub was constantly having it's `onConnect` handler fired from the multicodec topology.

It was because bitswap was calling `dialProtocol` to connect to a peer. `dialProtocol` makes a `PeerInfo` out of the multiaddr it's given but it has no `protocols` i.e. `[]`. This is passed to `peerStore.update()` where we compare `[]` with an array of populated protocols. Obviously there is no intersection here so `change:protocols` was being emitted, even though no protocols were added or removed.

This logic needs to be improved and tested properly but I just wanted to send a PR to document my findings.

.travis.yml

@@ -39,5 +39,12 @@ jobs:
       script:
         - npx aegir test -t browser -t webworker -- --browsers FirefoxHeadless
 
+    - stage: test
+      name: interop
+      script:
+        - cd node_modules/interop-libp2p
+        - npm install
+        - LIBP2P_JS=${TRAVIS_BUILD_DIR}/src/index.js npx aegir test -t node --bail
+
 notifications:
   email: false

CHANGELOG.md

@@ -1,3 +1,13 @@
+<a name="0.27.2"></a>
+## [0.27.2](https://github.com/libp2p/js-libp2p/compare/v0.27.1...v0.27.2) (2020-02-05)
+
+
+### Bug Fixes
+
+* ensure identify streams are closed ([#551](https://github.com/libp2p/js-libp2p/issues/551)) ([f662fdc](https://github.com/libp2p/js-libp2p/commit/f662fdc))
+
+
+
 <a name="0.27.1"></a>
 ## [0.27.1](https://github.com/libp2p/js-libp2p/compare/v0.27.0...v0.27.1) (2020-02-03)
 

README.md

@@ -5,10 +5,10 @@
 <h3 align="center">The JavaScript implementation of the libp2p Networking Stack.</h3>
 
 <p align="center">
-  <a href="http://ipn.io"><img src="https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square" /></a>
+  <a href="http://protocol.ai"><img src="https://img.shields.io/badge/made%20by-Protocol%20Labs-blue.svg?style=flat-square" /></a>
   <a href="http://libp2p.io/"><img src="https://img.shields.io/badge/project-libp2p-yellow.svg?style=flat-square" /></a>
   <a href="http://webchat.freenode.net/?channels=%23libp2p"><img src="https://img.shields.io/badge/freenode-%23libp2p-yellow.svg?style=flat-square" /></a>
-  <a href="https://riot.permaweb.io/#/room/#libp2p:permaweb.io"><img src="https://img.shields.io/badge/matrix-%23libp2p%3Apermaweb.io-blue.svg?style=flat-square" /> </a>
+  <a href="https://riot.im/app/#/room/#libp2p:matrix.org"><img src="https://img.shields.io/badge/matrix-%23libp2p%3Apermaweb.io-blue.svg?style=flat-square" /> </a>
   <a href="https://discord.gg/66KBrm2"><img src="https://img.shields.io/discord/475789330380488707?color=blueviolet&label=discord&style=flat-square" /></a>
   <a href="https://discuss.libp2p.io"><img src="https://img.shields.io/discourse/https/discuss.libp2p.io/posts.svg" /></a>
 </p>

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p",
-  "version": "0.27.1",
+  "version": "0.27.2",
   "description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
@@ -47,7 +47,7 @@
     "bignumber.js": "^9.0.0",
     "class-is": "^1.1.0",
     "debug": "^4.1.1",
-    "err-code": "^1.1.2",
+    "err-code": "^2.0.0",
     "hashlru": "^2.3.0",
     "it-all": "^1.0.1",
     "it-buffer": "^0.1.1",
@@ -76,7 +76,7 @@
   },
   "devDependencies": {
     "@nodeutils/defaults-deep": "^1.1.0",
-    "abortable-iterator": "^2.1.0",
+    "abortable-iterator": "^3.0.0",
     "aegir": "^20.5.1",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
@@ -86,6 +86,7 @@
     "it-concat": "^1.0.0",
     "it-pair": "^1.0.0",
     "it-pushable": "^1.4.0",
+    "interop-libp2p": "~0.0.1",
     "libp2p-bootstrap": "^0.10.3",
     "libp2p-delegated-content-routing": "^0.4.1",
     "libp2p-delegated-peer-routing": "^0.4.0",
@@ -98,7 +99,7 @@
     "libp2p-tcp": "^0.14.1",
     "libp2p-webrtc-star": "^0.17.0",
     "libp2p-websockets": "^0.13.1",
-    "nock": "^10.0.6",
+    "nock": "^11.7.2",
     "p-defer": "^3.0.0",
     "p-times": "^2.1.0",
     "p-wait-for": "^3.1.0",
@@ -154,6 +155,7 @@
     "Yusef Napora <yusef@napora.org>",
     "Zane Starr <zcstarr@gmail.com>",
     "a1300 <a1300@users.noreply.github.com>",
+    "dependabot-preview[bot] <27856297+dependabot-preview[bot]@users.noreply.github.com>",
     "dirkmc <dirkmdev@gmail.com>",
     "ebinks <elizabethjbinks@gmail.com>",
     "greenkeeperio-bot <support@greenkeeper.io>",

pdd/README.md

@@ -1,3 +0,0 @@
-# PDD Test Stories Implementation
-
-> Implementation of the Compliance tests from https://github.com/libp2p/interop

pdd/package.json

@@ -1,20 +0,0 @@
-{
-  "name": "pdd-impl",
-  "version": "0.0.0",
-  "description": "PDD Test Stories implementation",
-  "repository": {
-    "type": "git",
-    "url": "  "
-  },
-  "keywords": [
-    "PDD",
-    "libp2p"
-  ],
-  "author": "David Dias <daviddias@ipfs.io>",
-  "license": "MIT",
-  "dependencies": {
-    "libp2p": "file:./..",
-    "libp2p-interop": "github:libp2p/interop#master",
-    "tape": "^4.8.0"
-  }
-}

pdd/pdd-the-ipfs-bundle--story-1--peer-a.js

@@ -1,104 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const TCP = require('libp2p-tcp')
-const WebSockets = require('libp2p-websockets')
-const SECIO = require('libp2p-secio')
-const Multiplex = require('libp2p-multiplex')
-const Railing = require('libp2p-railing')
-const MulticastDNS = require('libp2p-mdns')
-const KadDHT = require('libp2p-kad-dht')
-const PeerInfo = require('peer-info')
-const pull = require('pull-stream')
-const waterfall = require('async/waterfall')
-const series = require('async/series')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class IPFSBundle extends libp2p {
-  constructor (peerInfo, options) {
-    options = Object.assign({ bootstrap: [] }, options)
-
-    const modules = {
-      transport: [
-        new TCP(),
-        new WebSockets()
-      ],
-      connection: {
-        muxer: [
-          Multiplex
-        ],
-        crypto: [
-          SECIO
-        ]
-      },
-      discovery: [
-        new MulticastDNS(peerInfo, 'ipfs.local'),
-        new Railing(options.bootstrap)
-      ],
-      DHT: KadDHT
-    }
-
-    super(modules, peerInfo, undefined, options)
-  }
-}
-
-test('story 1 - peerA', (t) => {
-  t.plan(10)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerA, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10000')
-      node = new IPFSBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node successfully')
-    t.ok(node.isStarted(), 'PeerA is Running')
-
-    const peerBAddr = `/ip4/127.0.0.1/tcp/10001/p2p/${PeerB.id}`
-
-    node.handle('/time/1.0.0', (protocol, conn) => {
-      pull(
-        pull.values([Date.now().toString()]),
-        conn,
-        pull.onEnd((err) => {
-          t.ifErr(err)
-          t.pass('Sent time successfully')
-        })
-      )
-    })
-
-    series([
-      (cb) => setTimeout(cb, 5 * 1000), // time to run both scripts
-      (cb) => node.ping(peerBAddr, (err, p) => {
-        t.ifErr(err, 'initiated Ping to PeerB')
-        p.once('error', (err) => t.ifErr(err, 'Ping should not fail'))
-        p.once('ping', (time) => {
-          t.pass('ping PeerB successfully')
-          p.stop()
-          cb()
-        })
-      }),
-      (cb) => node.dial(peerBAddr, '/echo/1.0.0', (err, conn) => {
-        t.ifErr(err, 'dial successful')
-
-        const data = Buffer.from('Hey')
-
-        pull(
-          pull.values([data]),
-          conn,
-          pull.collect((err, values) => {
-            t.ifErr(err, 'Received echo back')
-            t.deepEqual(values[0], data)
-            cb()
-          })
-        )
-      }),
-      (cb) => setTimeout(cb, 2 * 1000) // time to both finish
-    ], () => node.stop((err) => t.ifErr(err, 'PeerA has stopped')))
-  })
-})

pdd/pdd-the-ipfs-bundle--story-1--peer-b.js

@@ -1,98 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const TCP = require('libp2p-tcp')
-const WebSockets = require('libp2p-websockets')
-const SECIO = require('libp2p-secio')
-const Multiplex = require('libp2p-multiplex')
-const Railing = require('libp2p-railing')
-const MulticastDNS = require('libp2p-mdns')
-const KadDHT = require('libp2p-kad-dht')
-const PeerInfo = require('peer-info')
-const pull = require('pull-stream')
-const waterfall = require('async/waterfall')
-const series = require('async/series')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class IPFSBundle extends libp2p {
-  constructor (peerInfo, options) {
-    options = Object.assign({ bootstrap: [] }, options)
-
-    const modules = {
-      transport: [
-        new TCP(),
-        new WebSockets()
-      ],
-      connection: {
-        muxer: [
-          Multiplex
-        ],
-        crypto: [
-          SECIO
-        ]
-      },
-      discovery: [
-        new MulticastDNS(peerInfo, 'ipfs.local'),
-        new Railing(options.bootstrap)
-      ],
-      DHT: KadDHT
-    }
-
-    super(modules, peerInfo, undefined, options)
-  }
-}
-
-test('story 1 - peerA', (t) => {
-  t.plan(8)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerB, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10001')
-      node = new IPFSBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node successfully')
-    t.ok(node.isStarted(), 'PeerB is Running')
-
-    const peerAAddr = `/ip4/127.0.0.1/tcp/10000/p2p/${PeerA.id}`
-
-    node.handle('/echo/1.0.0', (protocol, conn) => {
-      pull(
-        conn,
-        conn,
-        pull.onEnd((err) => t.ifErr(err, 'echo was successful'))
-      )
-    })
-
-    series([
-      (cb) => setTimeout(cb, 5 * 1000), // time to run both scripts
-      (cb) => node.ping(peerAAddr, (err, p) => {
-        t.ifErr(err, 'initiated Ping to PeerA')
-        p.once('error', (err) => t.ifErr(err, 'Ping should not fail'))
-        p.once('ping', (time) => {
-          t.pass('ping PeerA successfully')
-          p.stop()
-          cb()
-        })
-      }),
-      (cb) => node.dial(peerAAddr, '/time/1.0.0', (err, conn) => {
-        t.ifErr(err, 'dial successful')
-
-        pull(
-          pull.values([]),
-          conn,
-          pull.collect((err, values) => {
-            t.ifErr(err, 'Received time')
-            cb()
-          })
-        )
-      }),
-      (cb) => setTimeout(cb, 2 * 1000) // time to both finish
-    ], () => node.stop((err) => t.ifErr(err, 'PeerB has stopped')))
-  })
-})

pdd/pdd-transport--story-1--peer-a.js

@@ -1,54 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const TCP = require('libp2p-tcp')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const pull = require('pull-stream')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new TCP()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 1 - peerA', (t) => {
-  t.plan(6)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerA, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10000')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node')
-    t.ok(node.isStarted(), 'PeerA is running')
-
-    const PeerBAddr = `/ip4/127.0.0.1/tcp/10001/p2p/${PeerB.id}`
-
-    node.dial(PeerBAddr, '/echo/1.0.0', (err, conn) => {
-      t.ifErr(err, 'dial successful')
-
-      const data = Buffer.from('Heey')
-
-      pull(
-        pull.values([data]),
-        conn,
-        pull.collect((err, values) => {
-          t.ifErr(err, 'Received echo back')
-          t.deepEqual(values[0], data)
-          node.stop((err) => t.ifErr(err, 'PeerA has stopped'))
-        })
-      )
-    })
-  })
-})

pdd/pdd-transport--story-1--peer-b.js

@@ -1,49 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const TCP = require('libp2p-tcp')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const pull = require('pull-stream')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new TCP()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 1 - peerB', (t) => {
-  t.plan(5)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerB, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10001')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err)
-    t.ok(node.isStarted(), 'PeerB is running')
-
-    node.handle('/echo/1.0.0', (protocol, conn) => {
-      pull(
-        conn,
-        conn,
-        pull.onEnd((err) => {
-          t.ifErr(err)
-          t.pass('Received End of Connection')
-          node.stop((err) => {
-            t.ifErr(err, 'PeerB has stopped')
-          })
-        })
-      )
-    })
-  })
-})

pdd/pdd-transport--story-2--peer-a.js

@@ -1,54 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const WebSockets = require('libp2p-websockets')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const pull = require('pull-stream')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new WebSockets()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 2 - peerA', (t) => {
-  t.plan(6)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerA, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10000/ws')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node')
-    t.ok(node.isStarted(), 'PeerA is running')
-
-    const PeerBAddr = `/ip4/127.0.0.1/tcp/10001/p2p/${PeerB.id}`
-
-    node.dial(PeerBAddr, '/echo/1.0.0', (err, conn) => {
-      t.ifErr(err, 'dial successful')
-
-      const data = Buffer.from('Heey')
-
-      pull(
-        pull.values([data]),
-        conn,
-        pull.collect((err, values) => {
-          t.ifErr(err, 'Received echo back')
-          t.deepEqual(values[0], data)
-          node.stop((err) => t.ifErr(err, 'PeerA has stopped'))
-        })
-      )
-    })
-  })
-})

pdd/pdd-transport--story-2--peer-b.js

@@ -1,49 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const WebSockets = require('libp2p-websockets')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const pull = require('pull-stream')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new WebSockets()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 2 - peerB', (t) => {
-  t.plan(5)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerB, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10001/ws')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err)
-    t.ok(node.isStarted(), 'PeerB is running')
-
-    node.handle('/echo/1.0.0', (protocol, conn) => {
-      pull(
-        conn,
-        pull.through(v => v, err => {
-          t.ifErr(err)
-          t.pass('Received End of Connection')
-          node.stop((err) => {
-            t.ifErr(err, 'PeerB has stopped')
-          })
-        }),
-        conn
-      )
-    })
-  })
-})

pdd/pdd-transport--story-3--peer-a.js

@@ -1,42 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const TCP = require('libp2p-tcp')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new TCP()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 3 - peerA', (t) => {
-  t.plan(4)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerA, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10000')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node')
-    t.ok(node.isStarted(), 'PeerA is running')
-
-    const PeerBAddr = `/ip4/127.0.0.1/tcp/10001/ws/p2p/${PeerB.id}`
-
-    setTimeout(() => node.dial(PeerBAddr, '/echo/1.0.0', (err, conn) => {
-      t.ok(err, 'dial failed')
-      node.stop((err) => t.ifErr(err, 'PeerA has stopped'))
-    }), 1000)
-  })
-})

pdd/pdd-transport--story-3--peer-b.js

@@ -1,42 +0,0 @@
-'use strict'
-
-const test = require('tape')
-const libp2p = require('libp2p')
-const WebSockets = require('libp2p-websockets')
-const PeerInfo = require('peer-info')
-const waterfall = require('async/waterfall')
-const PeerA = require('libp2p-interop/peer-a.json')
-const PeerB = require('libp2p-interop/peer-b.json')
-
-class MyBundle extends libp2p {
-  constructor (peerInfo) {
-    const modules = {
-      transport: [new WebSockets()]
-    }
-    super(modules, peerInfo)
-  }
-}
-
-test('story 3 - peerB', (t) => {
-  t.plan(4)
-  let node
-
-  waterfall([
-    (cb) => PeerInfo.create(PeerB, cb),
-    (peerInfo, cb) => {
-      peerInfo.multiaddrs.add('/ip4/127.0.0.1/tcp/10000/ws')
-      node = new MyBundle(peerInfo)
-      node.start(cb)
-    }
-  ], (err) => {
-    t.ifErr(err, 'created Node')
-    t.ok(node.isStarted(), 'PeerA is running')
-
-    const PeerAAddr = `/ip4/127.0.0.1/tcp/10000/ws/p2p/${PeerA.id}`
-
-    setTimeout(() => node.dial(PeerAAddr, '/echo/1.0.0', (err, conn) => {
-      t.ok(err, 'dial failed')
-      node.stop((err) => t.ifErr(err, 'PeerA has stopped'))
-    }), 1000)
-  })
-})

src/identify/index.js

@@ -4,7 +4,7 @@ const debug = require('debug')
 const pb = require('it-protocol-buffers')
 const lp = require('it-length-prefixed')
 const pipe = require('it-pipe')
-const { collect, take } = require('streaming-iterables')
+const { collect, take, consume } = require('streaming-iterables')
 
 const PeerInfo = require('peer-info')
 const PeerId = require('peer-id')
@@ -114,7 +114,8 @@ class IdentifyService {
             protocols: Array.from(this._protocols.keys())
           }],
           pb.encode(Message),
-          stream
+          stream,
+          consume
         )
       } catch (err) {
         // Just log errors
@@ -153,6 +154,7 @@ class IdentifyService {
   async identify (connection) {
     const { stream } = await connection.newStream(MULTICODEC_IDENTIFY)
     const [data] = await pipe(
+      [],
       stream,
       lp.decode(),
       take(1),
@@ -242,7 +244,8 @@ class IdentifyService {
     pipe(
       [message],
       lp.encode(),
-      stream
+      stream,
+      consume
     )
   }
 
@@ -255,6 +258,7 @@ class IdentifyService {
    */
   async _handlePush ({ connection, stream }) {
     const [data] = await pipe(
+      [],
       stream,
       lp.decode(),
       take(1),

src/peer-store/index.js

@@ -132,19 +132,16 @@ class PeerStore extends EventEmitter {
         multiaddrs: recorded.multiaddrs.toArray()
       })
     }
-
+    
-    // Update protocols
+    let isProtocolsChanged = false
-    // TODO: better track added and removed protocols
+    for (const protocol of peerInfo.protocols) {
-    const protocolsIntersection = new Set(
+      if (!recorded.protocols.has(protocol)) {
-      [...recorded.protocols].filter((p) => peerInfo.protocols.has(p))
+        isProtocolsChanged = true
-    )
-
-    if (protocolsIntersection.size !== peerInfo.protocols.size ||
-      protocolsIntersection.size !== recorded.protocols.size) {
-      for (const protocol of peerInfo.protocols) {
         recorded.protocols.add(protocol)
       }
+    }
 
+    if (isProtocolsChanged) {
       this.emit('change:protocols', {
         peerInfo: recorded,
         protocols: Array.from(recorded.protocols)

src/upgrader.js

@@ -231,7 +231,7 @@ class Upgrader {
             const { stream, protocol } = await mss.handle(Array.from(this.protocols.keys()))
             log('%s: incoming stream opened on %s', direction, protocol)
             if (this.metrics) this.metrics.trackStream({ stream, remotePeer, protocol })
-            connection.addStream(stream, protocol)
+            connection.addStream(muxedStream, { protocol })
             this._onStream({ connection, stream, protocol })
           } catch (err) {
             log.error(err)

test/identify/index.spec.js

@@ -12,6 +12,7 @@ const PeerId = require('peer-id')
 const PeerInfo = require('peer-info')
 const duplexPair = require('it-pair/duplex')
 const multiaddr = require('multiaddr')
+const pWaitFor = require('p-wait-for')
 
 const { codes: Errors } = require('../../src/errors')
 const { IdentifyService, multicodecs } = require('../../src/identify')
@@ -203,16 +204,17 @@ describe('Identify', () => {
       })
 
       sinon.spy(libp2p.identifyService, 'identify')
-      sinon.spy(libp2p.peerStore, 'replace')
+      const peerStoreSpy = sinon.spy(libp2p.peerStore, 'replace')
 
       const connection = await libp2p.dialer.connectToPeer(remoteAddr)
       expect(connection).to.exist()
-      // Wait for nextTick to trigger the identify call
-      await delay(1)
-      expect(libp2p.identifyService.identify.callCount).to.equal(1)
-      await libp2p.identifyService.identify.firstCall.returnValue
 
-      expect(libp2p.peerStore.replace.callCount).to.equal(1)
+      // Wait for peer store to be updated
+      await pWaitFor(() => peerStoreSpy.callCount === 1)
+      expect(libp2p.identifyService.identify.callCount).to.equal(1)
+
+      // The connection should have no open streams
+      expect(connection.streams).to.have.length(0)
       await connection.close()
     })
 
@@ -247,6 +249,9 @@ describe('Identify', () => {
         const results = await call.returnValue
         expect(results.length).to.equal(1)
       }
+
+      // Verify the streams close
+      await pWaitFor(() => connection.streams.length === 0)
     })
   })
 })