feat: async peerstore (#62)

Refactors interfaces and classes used by `libp2p-interfaces` to use the async peer store from libp2p/js-libp2p#1058

BREAKING CHANGE: peerstore methods are now all async

package.json

@@ -27,7 +27,7 @@
     "release-major": "aegir release --no-build --type major -t node"
   },
   "engines": {
-    "node": ">=15.0.0"
+    "node": ">=16.0.0"
   },
   "bugs": {
     "url": "https://github.com/libp2p/js-libp2p-daemon/issues"
@@ -46,6 +46,7 @@
     "sinon": "^12.0.1"
   },
   "dependencies": {
+    "@achingbrain/libp2p-gossipsub": "^0.12.1",
     "@chainsafe/libp2p-noise": "^5.0.0",
     "debug": "^4.3.1",
     "it-all": "^1.0.6",
@@ -54,11 +55,10 @@
     "it-length-prefixed": "^5.0.2",
     "it-pipe": "^1.1.0",
     "it-pushable": "^1.4.0",
-    "libp2p": "^0.35.8",
+    "libp2p": "libp2p/js-libp2p#feat/async-peerstore",
     "libp2p-bootstrap": "^0.14.0",
-    "libp2p-floodsub": "^0.28.0",
-    "libp2p-gossipsub": "^0.12.1",
-    "libp2p-kad-dht": "^0.27.6",
+    "libp2p-floodsub": "^0.29.0",
+    "libp2p-kad-dht": "^0.28.3",
     "libp2p-mplex": "^0.10.0",
     "libp2p-tcp": "^0.17.1",
     "libp2p-websockets": "^0.16.1",

src/daemon.js

@@ -57,12 +57,12 @@ class Daemon {
    * @param {ConnectRequest} connectRequest
    * @returns {Promise<Connection>}
    */
-  connect (connectRequest) {
+  async connect (connectRequest) {
     const peer = connectRequest.connect.peer
     const addrs = connectRequest.connect.addrs.map((a) => new Multiaddr(a))
     const peerId = PeerId.createFromBytes(peer)
 
-    this.libp2p.peerStore.addressBook.set(peerId, addrs)
+    await this.libp2p.peerStore.addressBook.set(peerId, addrs)
     return this.libp2p.dial(peerId)
   }
 
@@ -189,10 +189,10 @@ class Daemon {
 
   handlePeerStoreRequest ({ peerStore }) {
     const peerStoreAction = {
-      [PeerstoreRequest.Type.GET_PROTOCOLS]: function * (daemon) {
+      [PeerstoreRequest.Type.GET_PROTOCOLS]: async function * (daemon) {
         try {
           const peerId = PeerId.createFromBytes(peerStore.id)
-          const peer = daemon.libp2p.peerStore.get(peerId)
+          const peer = await daemon.libp2p.peerStore.get(peerId)
           const protos = peer.protocols
           yield OkResponse({ peerStore: { protos } })
         } catch (err) {
@@ -436,16 +436,19 @@ class Daemon {
             }
             // Get a list of our current peers
             case Request.Type.LIST_PEERS: {
-              const peers = Array.from(daemon.libp2p.peerStore.peers.values()).map((peer) => {
+              const peers = []
+
+              for await (const peer of daemon.libp2p.peerStore.getPeers()) {
                 // TODO: conn mgr
                 const conn = daemon.libp2p.registrar.getConnection(peer.id)
                 const addr = conn.remoteAddr
 
-                return {
+                peers.push({
                   id: peer.id.toBytes(),
                   addrs: [addr ? addr.bytes : null]
-                }
-              })
+                })
+              }
+
               yield OkResponse({ peers })
               break
             }

src/libp2p.js

@@ -8,7 +8,7 @@ const MPLEX = require('libp2p-mplex')
 const { NOISE } = require('@chainsafe/libp2p-noise')
 const { KadDHT } = require('libp2p-kad-dht/src/kad-dht')
 const FloodSub = require('libp2p-floodsub')
-const GossipSub = require('libp2p-gossipsub')
+const GossipSub = require('@achingbrain/libp2p-gossipsub')
 const PeerID = require('peer-id')
 const { Multiaddr } = require('multiaddr')
 const fsPromises = require('fs').promises

test/daemon/dht.spec.js

@@ -29,6 +29,15 @@ const daemonAddr = isWindows
   ? new Multiaddr('/ip4/0.0.0.0/tcp/8080')
   : new Multiaddr(`/unix${path.resolve(os.tmpdir(), '/tmp/p2pd.sock')}`)
 
+function sameMultiAddrs (a, b) {
+  const set = new Set([
+    ...a.map(a => new Multiaddr(a)).map(a => a.toString()),
+    ...b.map(b => new Multiaddr(b)).map(b => b.toString())
+  ])
+
+  return (a.length === b.length) && (a.length === set.size)
+}
+
 describe('dht', () => {
   const cid = CID.parse('QmVzw6MPsF96TyXBSRs1ptLoVMWRv5FCYJZZGJSVB2Hp38')
   let daemon
@@ -107,9 +116,8 @@ describe('dht', () => {
     const response = Response.decode(await streamHandler.read())
     expect(response.type).to.eql(Response.Type.OK)
     expect(PeerId.createFromBytes(response.dht.peer.id).equals(libp2pPeer.peerId)).to.eql(true)
-    response.dht.peer.addrs.forEach((a, i) => {
-      expect((new Multiaddr(a)).equals(libp2pPeer.multiaddrs[i])).to.eql(true)
-    })
+    expect(sameMultiAddrs(response.dht.peer.addrs, libp2pPeer.multiaddrs)).to.be.true()
+
     streamHandler.close()
   })
 
@@ -213,9 +221,7 @@ describe('dht', () => {
         const response = DHTResponse.decode(message)
         expect(response.type).to.eql(DHTResponse.Type.VALUE)
         expect(PeerId.createFromBytes(response.peer.id).equals(libp2pPeer.peerId)).to.eql(true)
-        response.peer.addrs.forEach((a, i) => {
-          expect((new Multiaddr(a)).equals(libp2pPeer.multiaddrs[i])).to.eql(true)
-        })
+        expect(sameMultiAddrs(response.peer.addrs, libp2pPeer.multiaddrs)).to.be.true()
       },
       (message) => {
         const response = DHTResponse.decode(message)

test/daemon/peerstore.spec.js

@@ -96,11 +96,11 @@ describe('peerstore features', () => {
     expect(response.type).to.eql(Response.Type.OK)
     expect(response.peerStore.protos).to.eql([
       '/libp2p/circuit/relay/0.1.0',
+      '/ipfs/ping/1.0.0',
       '/ipfs/id/1.0.0',
       '/ipfs/id/push/1.0.0',
-      '/ipfs/ping/1.0.0',
       '/ipfs/kad/1.0.0'
-    ])
+    ].sort())
     streamHandler.close()
   })