chore: release version v0.28.4

* chore: add test for pubsub on reconnect

* chore: update interface dep

Co-authored-by: Jacob Heun <jacobheun@gmail.com>

CHANGELOG.md

@@ -1,3 +1,43 @@
+<a name="0.28.4"></a>
+## [0.28.4](https://github.com/libp2p/js-libp2p/compare/v0.28.3...v0.28.4) (2020-07-03)
+
+
+
+<a name="0.28.3"></a>
+## [0.28.3](https://github.com/libp2p/js-libp2p/compare/v0.28.2...v0.28.3) (2020-06-18)
+
+
+### Bug Fixes
+
+* catch pipe errors ([#678](https://github.com/libp2p/js-libp2p/issues/678)) ([a8219e6](https://github.com/libp2p/js-libp2p/commit/a8219e6))
+
+
+
+<a name="0.28.2"></a>
+## [0.28.2](https://github.com/libp2p/js-libp2p/compare/v0.28.1...v0.28.2) (2020-06-15)
+
+
+### Reverts
+
+* "fix: throw if no conn encryption module provided ([#665](https://github.com/libp2p/js-libp2p/issues/665))" ([b621fbd](https://github.com/libp2p/js-libp2p/commit/b621fbd))
+
+
+
+<a name="0.28.1"></a>
+## [0.28.1](https://github.com/libp2p/js-libp2p/compare/v0.28.0...v0.28.1) (2020-06-12)
+
+
+### Bug Fixes
+
+* throw if no conn encryption module provided ([#665](https://github.com/libp2p/js-libp2p/issues/665)) ([c038550](https://github.com/libp2p/js-libp2p/commit/c038550))
+
+
+### Features
+
+* add ConnectionManager#getAll ([8f680e2](https://github.com/libp2p/js-libp2p/commit/8f680e2))
+
+
+
 <a name="0.28.0"></a>
 # [0.28.0](https://github.com/libp2p/js-libp2p/compare/v0.28.0-rc.0...v0.28.0) (2020-06-05)
 

README.md

@@ -155,7 +155,7 @@ List of packages currently in existence for libp2p
 | [`libp2p-kad-dht`](//github.com/libp2p/js-libp2p-kad-dht) | [![npm](https://img.shields.io/npm/v/libp2p-kad-dht.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-kad-dht/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-kad-dht.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-kad-dht) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-kad-dht/master)](https://travis-ci.com/libp2p/js-libp2p-kad-dht) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-kad-dht/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-kad-dht) | [Vasco Santos](mailto:vasco.santos@moxy.studio) |
 | [`libp2p-mdns`](//github.com/libp2p/js-libp2p-mdns) | [![npm](https://img.shields.io/npm/v/libp2p-mdns.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-mdns/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-mdns.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-mdns) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-mdns/master)](https://travis-ci.com/libp2p/js-libp2p-mdns) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-mdns/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-mdns) | [Jacob Heun](mailto:jacobheun@gmail.com) |
 | [`libp2p-webrtc-star`](//github.com/libp2p/js-libp2p-webrtc-star) | [![npm](https://img.shields.io/npm/v/libp2p-webrtc-star.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-webrtc-star/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-webrtc-star.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-webrtc-star) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-webrtc-star/master)](https://travis-ci.com/libp2p/js-libp2p-webrtc-star) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-webrtc-star/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-webrtc-star) | [Vasco Santos](mailto:vasco.santos@moxy.studio) |
-| [`discv5`](//github.com/ChainSafe/discv5) | [![npm](https://img.shields.io/npm/v/discv5.svg?maxAge=86400&style=flat-square)](//github.com/ChainSafe/discv5/releases) | [![Deps](https://david-dm.org/ChainSafe/discv5.svg?style=flat-square)](https://david-dm.org/ChainSafe/discv5) | [![Travis CI](https://flat.badgen.net/travis/ChainSafe/discv5/master)](https://travis-ci.com/ChainSafe/discv5) | [![codecov](https://codecov.io/gh/ChainSafe/discv5/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/ChainSafe/discv5) | [Cayman Nava](mailto:caymannava@gmail.com) |
+| [`@chainsafe/discv5`](//github.com/ChainSafe/discv5) | [![npm](https://img.shields.io/npm/v/@chainsafe/discv5.svg?maxAge=86400&style=flat-square)](//github.com/ChainSafe/discv5/releases) | [![Deps](https://david-dm.org/ChainSafe/discv5.svg?style=flat-square)](https://david-dm.org/ChainSafe/discv5) | [![Travis CI](https://flat.badgen.net/travis/ChainSafe/discv5/master)](https://travis-ci.com/ChainSafe/discv5) | [![codecov](https://codecov.io/gh/ChainSafe/discv5/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/ChainSafe/discv5) | [Cayman Nava](mailto:caymannava@gmail.com) |
 | **content routing** |
 | [`libp2p-delegated-content-routing`](//github.com/libp2p/js-libp2p-delegated-content-routing) | [![npm](https://img.shields.io/npm/v/libp2p-delegated-content-routing.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-delegated-content-routing/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-delegated-content-routing.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-delegated-content-routing) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-delegated-content-routing/master)](https://travis-ci.com/libp2p/js-libp2p-delegated-content-routing) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-delegated-content-routing/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-delegated-content-routing) | [Jacob Heun](mailto:jacobheun@gmail.com) |
 | [`libp2p-kad-dht`](//github.com/libp2p/js-libp2p-kad-dht) | [![npm](https://img.shields.io/npm/v/libp2p-kad-dht.svg?maxAge=86400&style=flat-square)](//github.com/libp2p/js-libp2p-kad-dht/releases) | [![Deps](https://david-dm.org/libp2p/js-libp2p-kad-dht.svg?style=flat-square)](https://david-dm.org/libp2p/js-libp2p-kad-dht) | [![Travis CI](https://flat.badgen.net/travis/libp2p/js-libp2p-kad-dht/master)](https://travis-ci.com/libp2p/js-libp2p-kad-dht) | [![codecov](https://codecov.io/gh/libp2p/js-libp2p-kad-dht/branch/master/graph/badge.svg?style=flat-square)](https://codecov.io/gh/libp2p/js-libp2p-kad-dht) | [Vasco Santos](mailto:vasco.santos@moxy.studio) |

RELEASE.md

@@ -26,6 +26,7 @@
 - Documentation
   - [ ] Ensure that README.md is up to date
   - [ ] Ensure that all the examples run
+  - [ ] Ensure that [libp2p/docs](https://github.com/libp2p/docs) is updated
 - Communication
   - [ ] Create the release issue
   - [ ] Take a snapshot between of everyone that has contributed to this release (including its subdeps in IPFS, libp2p, IPLD and multiformats) using [`name-your-contributors`](https://www.npmjs.com/package/name-your-contributors). Generate a nice markdown list with [this script](https://gist.github.com/jacobheun/d2ff479ca991733c13cdcf688a1317e5)

doc/STREAMING_ITERABLES.md

@@ -161,4 +161,4 @@ const duplex = {
 [it-pipe]: https://github.com/alanshaw/it-pipe
 [it-pushable]: https://github.com/alanshaw/it-pushable
 [it-reader]: https://github.com/alanshaw/it-reader
-[streaming-iterables]: https://github.com/bustle/streaming-iterables
+[streaming-iterables]: https://github.com/reconbot/streaming-iterables

examples/libp2p-in-the-browser/README.md

@@ -11,26 +11,9 @@ cd ./examples/libp2p-in-the-browser
 npm install
 ```
 
-## Signaling Server
-
-This example uses the `libp2p-webrtc-star` module, which enables libp2p browser nodes to establish direct connections to one another via a central signaling server. For this example, we are using the signaling server that ships with `libp2p-webrtc-star`.
-
-You can start the server by running `npm run server`. This will start a signaling server locally on port `9090`. If you'd like to run a signaling server outside of this example, you can see instructions on how to do so in the [`libp2p-webrtc-star` README](https://github.com/libp2p/js-libp2p-webrtc-star).
-
-When you run the server, you should see output that looks something like this:
-
-```log
-$ npm run server
-
-> libp2p-in-browser@1.0.0 server
-> star-signal
-
-Listening on: http://0.0.0.0:9090
-```
-
 ## Running the examples
 
-Once you have started the signaling server, you can run the Parcel server.
+Start by running the Parcel server:
 
 ```
 npm start
@@ -53,3 +36,11 @@ This will compile the code and start a server listening on port [http://localhos
 Now, if you open a second browser tab to `http://localhost:1234`, you should discover your node from the previous tab. This is due to the fact that the `libp2p-webrtc-star` transport also acts as a Peer Discovery interface. Your node will be notified of any peer that connects to the same signaling server you are connected to. Once libp2p discovers this new peer, it will attempt to establish a direct WebRTC connection.
 
 **Note**: In the example we assign libp2p to `window.libp2p`, in case you would like to play around with the API directly in the browser. You can of course make changes to `index.js` and Parcel will automatically rebuild and reload the browser tabs.
+
+## Going to production?
+
+This example uses public `libp2p-webrtc-star` servers. These servers should be used for experimenting and demos, they **MUST** not be used in production as there is no guarantee on availability.
+
+You can see how to deploy your own signaling server in [libp2p/js-libp2p-webrtc-star/DEPLOYMENT.md](https://github.com/libp2p/js-libp2p-webrtc-star/blob/master/DEPLOYMENT.md).
+
+Once you have your own server running, you should add its listen address in your libp2p node configuration.

examples/libp2p-in-the-browser/index.js

@@ -14,7 +14,10 @@ document.addEventListener('DOMContentLoaded', async () => {
       // Add the signaling server address, along with our PeerId to our multiaddrs list
       // libp2p will automatically attempt to dial to the signaling server so that it can
       // receive inbound connections from other peers
-      listen: ['/ip4/0.0.0.0/tcp/9090/wss/p2p-webrtc-star']
+      listen: [
+        '/dns4/wrtc-star1.par.dwebops.pub/tcp/443/wss/p2p-webrtc-star',
+        '/dns4/wrtc-star2.sjc.dwebops.pub/tcp/443/wss/p2p-webrtc-star'
+      ]
     },
     modules: {
       transport: [Websockets, WebRTCStar],

examples/libp2p-in-the-browser/package.json

@@ -8,8 +8,7 @@
   ],
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1",
-    "start": "parcel index.html",
+    "start": "parcel index.html"
-    "server": "star-signal"
   },
   "keywords": [],
   "author": "",

package-list.json

@@ -33,7 +33,7 @@
     ["libp2p/js-libp2p-kad-dht", "libp2p-kad-dht"],
     ["libp2p/js-libp2p-mdns", "libp2p-mdns"],
     ["libp2p/js-libp2p-webrtc-star", "libp2p-webrtc-star"],
-    ["ChainSafe/discv5", "discv5"],
+    ["ChainSafe/discv5", "@chainsafe/discv5"],
 
     "content routing",
     ["libp2p/js-libp2p-delegated-content-routing", "libp2p-delegated-content-routing"],

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p",
-  "version": "0.28.0",
+  "version": "0.28.4",
   "description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
@@ -50,7 +50,7 @@
     "err-code": "^2.0.0",
     "events": "^3.1.0",
     "hashlru": "^2.3.0",
-    "interface-datastore": "^0.8.3",
+    "interface-datastore": "^1.0.4",
     "ipfs-utils": "^2.2.0",
     "it-all": "^1.0.1",
     "it-buffer": "^0.1.2",
@@ -59,7 +59,7 @@
     "it-pipe": "^1.1.0",
     "it-protocol-buffers": "^0.2.0",
     "libp2p-crypto": "^0.17.6",
-    "libp2p-interfaces": "^0.3.0",
+    "libp2p-interfaces": "^0.3.1",
     "libp2p-utils": "^0.1.2",
     "mafmt": "^7.0.0",
     "merge-options": "^2.0.0",
@@ -92,7 +92,7 @@
     "datastore-level": "^1.1.0",
     "delay": "^4.3.0",
     "dirty-chai": "^2.0.1",
-    "interop-libp2p": "libp2p/interop#chore/update-libp2p-daemon-with-peerstore",
+    "interop-libp2p": "^0.1.0",
     "ipfs-http-client": "^44.0.0",
     "it-concat": "^1.0.0",
     "it-pair": "^1.0.0",
@@ -125,12 +125,12 @@
     "Jacob Heun <jacobheun@gmail.com>",
     "Vasco Santos <vasco.santos@moxy.studio>",
     "Alan Shaw <alan@tableflip.io>",
+    "Cayman <caymannava@gmail.com>",
     "Pedro Teixeira <i@pgte.me>",
     "Friedel Ziegelmayer <dignifiedquire@gmail.com>",
     "Alex Potsides <alex@achingbrain.net>",
     "Maciej Krüger <mkg20001@gmail.com>",
     "Hugo Dias <mail@hugodias.me>",
-    "Cayman <caymannava@gmail.com>",
     "dirkmc <dirkmdev@gmail.com>",
     "Volker Mische <volker.mische@gmail.com>",
     "Richard Littauer <richard.littauer@gmail.com>",
@@ -141,7 +141,7 @@
     "Thomas Eizinger <thomas@eizinger.io>",
     "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com>",
     "Didrik Nordström <didrik@betamos.se>",
-    "Joel Gustafson <joelg@mit.edu>",
+    "Francis Gulotta <wizard@roborooter.com>",
     "Julien Bouquillon <contact@revolunet.com>",
     "Kevin Kwok <antimatter15@gmail.com>",
     "Florian-Merle <florian.david.merle@gmail.com>",
@@ -161,6 +161,7 @@
     "Chris Bratlien <chrisbratlien@gmail.com>",
     "isan_rivkin <isanrivkin@gmail.com>",
     "Henrique Dias <hacdias@gmail.com>",
-    "Irakli Gozalishvili <rfobic@gmail.com>"
+    "Irakli Gozalishvili <rfobic@gmail.com>",
+    "Joel Gustafson <joelg@mit.edu>"
   ]
 }

src/connection-manager/index.js

@@ -215,6 +215,19 @@ class ConnectionManager extends EventEmitter {
    * @returns {Connection}
    */
   get (peerId) {
+    const connections = this.getAll(peerId)
+    if (connections.length) {
+      return connections[0]
+    }
+    return null
+  }
+
+  /**
+   * Get all open connections with a peer.
+   * @param {PeerId} peerId
+   * @returns {Array<Connection>}
+   */
+  getAll (peerId) {
     if (!PeerId.isPeerId(peerId)) {
       throw errcode(new Error('peerId must be an instance of peer-id'), ERR_INVALID_PARAMETERS)
     }
@@ -222,11 +235,11 @@ class ConnectionManager extends EventEmitter {
     const id = peerId.toB58String()
     const connections = this.connections.get(id)
 
-    // Return the first, open connection
+    // Return all open connections
     if (connections) {
-      return connections.find(connection => connection.stat.status === 'open')
+      return connections.filter(connection => connection.stat.status === 'open')
     }
-    return null
+    return []
   }
 
   /**

src/identify/index.js

@@ -208,7 +208,7 @@ class IdentifyService {
    * @param {*} options.stream
    * @param {Connection} options.connection
    */
-  _handleIdentify ({ connection, stream }) {
+  async _handleIdentify ({ connection, stream }) {
     let publicKey = Buffer.alloc(0)
     if (this.peerId.pubKey) {
       publicKey = this.peerId.pubKey.bytes
@@ -223,12 +223,16 @@ class IdentifyService {
       protocols: Array.from(this._protocols.keys())
     })
 
-    pipe(
+    try {
-      [message],
+      await pipe(
-      lp.encode(),
+        [message],
-      stream,
+        lp.encode(),
-      consume
+        stream,
-    )
+        consume
+      )
+    } catch (err) {
+      log.error('could not respond to identify request', err)
+    }
   }
 
   /**
@@ -239,17 +243,16 @@ class IdentifyService {
    * @param {Connection} options.connection
    */
   async _handlePush ({ connection, stream }) {
-    const [data] = await pipe(
-      [],
-      stream,
-      lp.decode(),
-      take(1),
-      toBuffer,
-      collect
-    )
-
     let message
     try {
+      const [data] = await pipe(
+        [],
+        stream,
+        lp.decode(),
+        take(1),
+        toBuffer,
+        collect
+      )
       message = Message.decode(data)
     } catch (err) {
       return log.error('received invalid message', err)

src/metrics/index.js

@@ -215,7 +215,7 @@ class Metrics {
 
     const _sink = stream.sink
     stream.sink = source => {
-      pipe(
+      return pipe(
         source,
         tap(chunk => metrics._onMessage({
           remotePeer,

src/peer-store/persistent/index.js

@@ -96,7 +96,9 @@ class PersistentPeerStore extends PeerStore {
 
     if (this._dirtyPeers.size >= this.threshold) {
       // Commit current data
-      this._commitData()
+      this._commitData().catch(err => {
+        log.error('error committing data', err)
+      })
     }
   }
 
@@ -120,7 +122,9 @@ class PersistentPeerStore extends PeerStore {
 
     if (this._dirtyPeers.size >= this.threshold) {
       // Commit current data
-      this._commitData()
+      this._commitData().catch(err => {
+        log.error('error committing data', err)
+      })
     }
   }
 

src/pnet/index.js

@@ -17,7 +17,7 @@ const handshake = require('it-handshake')
 const { NONCE_LENGTH } = require('./key-generator')
 const debug = require('debug')
 const log = debug('libp2p:pnet')
-log.err = debug('libp2p:pnet:err')
+log.error = debug('libp2p:pnet:err')
 
 /**
  * Takes a Private Shared Key (psk) and provides a `protect` method
@@ -69,7 +69,7 @@ class Protector {
       // Decrypt all inbound traffic
       createUnboxStream(remoteNonce, this.psk),
       external
-    )
+    ).catch(log.error)
 
     return internal
   }

src/upgrader.js

@@ -258,7 +258,7 @@ class Upgrader {
       }
 
       // Pipe all data through the muxer
-      pipe(upgradedConn, muxer, upgradedConn)
+      pipe(upgradedConn, muxer, upgradedConn).catch(log.error)
     }
 
     const _timeline = maConn.timeline

test/peer-store/persisted-peer-store.spec.js

@@ -5,7 +5,6 @@ const chai = require('chai')
 chai.use(require('dirty-chai'))
 const { expect } = chai
 const sinon = require('sinon')
-
 const PeerStore = require('../../src/peer-store/persistent')
 const multiaddr = require('multiaddr')
 const { MemoryDatastore } = require('interface-datastore')
@@ -62,6 +61,7 @@ describe('Persisted PeerStore', () => {
       const protocols = ['/ping/1.0.0']
       const spyDirty = sinon.spy(peerStore, '_addDirtyPeer')
       const spyDs = sinon.spy(datastore, 'batch')
+      const commitSpy = sinon.spy(peerStore, '_commitData')
 
       await peerStore.start()
 
@@ -71,12 +71,18 @@ describe('Persisted PeerStore', () => {
       expect(spyDirty).to.have.property('callCount', 1) // Address
       expect(spyDs).to.have.property('callCount', 1)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // ProtoBook
       peerStore.protoBook.set(peer, protocols)
 
       expect(spyDirty).to.have.property('callCount', 2) // Protocol
       expect(spyDs).to.have.property('callCount', 2)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // Should have three peer records stored in the datastore
       const queryParams = {
         prefix: '/peers/'
@@ -98,6 +104,7 @@ describe('Persisted PeerStore', () => {
     it('should load content to the peerStore when restart but not put in datastore again', async () => {
       const spyDs = sinon.spy(datastore, 'batch')
       const peers = await peerUtils.createPeerId({ number: 2 })
+      const commitSpy = sinon.spy(peerStore, '_commitData')
       const multiaddrs = [
         multiaddr('/ip4/156.10.1.22/tcp/1000'),
         multiaddr('/ip4/156.10.1.23/tcp/1000')
@@ -110,17 +117,29 @@ describe('Persisted PeerStore', () => {
       peerStore.addressBook.set(peers[0], [multiaddrs[0]])
       peerStore.addressBook.set(peers[1], [multiaddrs[1]])
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // KeyBook
       peerStore.keyBook.set(peers[0], peers[0].pubKey)
       peerStore.keyBook.set(peers[1], peers[1].pubKey)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // ProtoBook
       peerStore.protoBook.set(peers[0], protocols)
       peerStore.protoBook.set(peers[1], protocols)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // MetadataBook
       peerStore.metadataBook.set(peers[0], 'location', Buffer.from('earth'))
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       expect(spyDs).to.have.property('callCount', 7) // 2 Address + 2 Key + 2 Proto + 1 Metadata
       expect(peerStore.peers.size).to.equal(2)
 
@@ -148,6 +167,7 @@ describe('Persisted PeerStore', () => {
       const [peer] = await peerUtils.createPeerId()
       const multiaddrs = [multiaddr('/ip4/156.10.1.22/tcp/1000')]
       const protocols = ['/ping/1.0.0']
+      const commitSpy = sinon.spy(peerStore, '_commitData')
 
       await peerStore.start()
 
@@ -158,6 +178,9 @@ describe('Persisted PeerStore', () => {
       // MetadataBook
       peerStore.metadataBook.set(peer, 'location', Buffer.from('earth'))
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       const spyDs = sinon.spy(datastore, 'batch')
       const spyAddressBook = sinon.spy(peerStore.addressBook, 'delete')
       const spyKeyBook = sinon.spy(peerStore.keyBook, 'delete')
@@ -166,6 +189,10 @@ describe('Persisted PeerStore', () => {
 
       // Delete from PeerStore
       peerStore.delete(peer)
+
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       await peerStore.stop()
 
       expect(spyAddressBook).to.have.property('callCount', 1)
@@ -197,6 +224,7 @@ describe('Persisted PeerStore', () => {
       const spyDirty = sinon.spy(peerStore, '_addDirtyPeer')
       const spyDirtyMetadata = sinon.spy(peerStore, '_addDirtyPeerMetadata')
       const spyDs = sinon.spy(datastore, 'batch')
+      const commitSpy = sinon.spy(peerStore, '_commitData')
 
       const peers = await peerUtils.createPeerId({ number: 2 })
 
@@ -213,9 +241,15 @@ describe('Persisted PeerStore', () => {
       peerStore.protoBook.set(peers[0], protocols)
       peerStore.metadataBook.set(peers[0], 'location', Buffer.from('earth'))
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // Remove data from the same Peer
       peerStore.addressBook.delete(peers[0])
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       expect(spyDirty).to.have.property('callCount', 3) // 2 AddrBook ops, 1 ProtoBook op
       expect(spyDirtyMetadata).to.have.property('callCount', 1) // 1 MetadataBook op
       expect(peerStore._dirtyPeers.size).to.equal(1)
@@ -231,6 +265,9 @@ describe('Persisted PeerStore', () => {
       // Add data for second book
       peerStore.addressBook.set(peers[1], multiaddrs)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       expect(spyDirty).to.have.property('callCount', 4)
       expect(spyDirtyMetadata).to.have.property('callCount', 1)
       expect(spyDs).to.have.property('callCount', 1)
@@ -333,6 +370,7 @@ describe('libp2p.peerStore (Persisted)', () => {
     })
 
     it('should load content to the peerStore when a new node is started with the same datastore', async () => {
+      const commitSpy = sinon.spy(libp2p.peerStore, '_commitData')
       const peers = await peerUtils.createPeerId({ number: 2 })
       const multiaddrs = [
         multiaddr('/ip4/156.10.1.22/tcp/1000'),
@@ -346,10 +384,16 @@ describe('libp2p.peerStore (Persisted)', () => {
       libp2p.peerStore.addressBook.set(peers[0], [multiaddrs[0]])
       libp2p.peerStore.addressBook.set(peers[1], [multiaddrs[1]])
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       // ProtoBook
       libp2p.peerStore.protoBook.set(peers[0], protocols)
       libp2p.peerStore.protoBook.set(peers[1], protocols)
 
+      // let batch commit complete
+      await Promise.all(commitSpy.returnValues)
+
       expect(libp2p.peerStore.peers.size).to.equal(2)
 
       await libp2p.stop()

test/pubsub/operation.node.js

@@ -189,4 +189,91 @@ describe('Pubsub subsystem operates correctly', () => {
       await defer.promise
     })
   })
+
+  describe('pubsub started after disconnect and connect', () => {
+    beforeEach(async () => {
+      libp2p = await create(mergeOptions(subsystemOptions, {
+        peerId,
+        addresses: {
+          listen: [listenAddr]
+        }
+      }))
+
+      remoteLibp2p = await create(mergeOptions(subsystemOptions, {
+        peerId: remotePeerId,
+        addresses: {
+          listen: [remoteListenAddr]
+        },
+        config: {
+          pubsub: {
+            enabled: true
+          }
+        }
+      }))
+
+      await libp2p.start()
+      await remoteLibp2p.start()
+
+      libp2p.peerStore.addressBook.set(remotePeerId, remoteLibp2p.multiaddrs)
+    })
+
+    afterEach(() => Promise.all([
+      libp2p && libp2p.stop(),
+      remoteLibp2p && remoteLibp2p.stop()
+    ]))
+
+    afterEach(() => {
+      sinon.restore()
+    })
+
+    it('should receive pubsub messages', async () => {
+      const topic = 'test-topic'
+      const data = 'hey!'
+      const libp2pId = libp2p.peerId.toB58String()
+
+      let counter = 0
+      const defer1 = pDefer()
+      const defer2 = pDefer()
+      const handler = (msg) => {
+        expect(msg.data.toString()).to.equal(data)
+        counter++
+        counter === 1 ? defer1.resolve() : defer2.resolve()
+      }
+
+      await libp2p.dial(remotePeerId)
+
+      let subscribedTopics = libp2p.pubsub.getTopics()
+      expect(subscribedTopics).to.not.include(topic)
+
+      libp2p.pubsub.subscribe(topic, handler)
+
+      subscribedTopics = libp2p.pubsub.getTopics()
+      expect(subscribedTopics).to.include(topic)
+
+      // wait for remoteLibp2p to know about libp2p subscription
+      await pWaitFor(() => {
+        const subscribedPeers = remoteLibp2p.pubsub.getSubscribers(topic)
+        return subscribedPeers.includes(libp2pId)
+      })
+      remoteLibp2p.pubsub.publish(topic, data)
+
+      await defer1.promise
+
+      await remoteLibp2p.stop()
+      await remoteLibp2p.start()
+
+      libp2p.peerStore.addressBook.set(remotePeerId, remoteLibp2p.multiaddrs)
+      await libp2p.dial(remotePeerId)
+
+      // wait for remoteLibp2p to know about libp2p subscription
+      await pWaitFor(() => {
+        const subscribedPeers = remoteLibp2p.pubsub.getSubscribers(topic)
+        return subscribedPeers.includes(libp2pId)
+      })
+
+      remoteLibp2p.pubsub.publish(topic, data)
+
+      await defer2.promise
+    })
+  })
 })