fix: pass peer id to onPeerConnect (#234)

The persistent peer store in libp2p 0.28 means sometimes we know about peers when we start up, so we try to start bitswap for them when we start up if we have an open connection already.

The connection object has a remotePeer property that should be used instead of passing the whole thing to `this._onPeerConnect`.

Adds a test for the above.

Fixes ipfs/js-ipfs#3182

Author: achingbrain

src/network.js

@@ -53,7 +53,7 @@ class Network {
     for (const peer of this.libp2p.peerStore.peers.values()) {
       const conn = this.libp2p.connectionManager.get(peer.id)
 
-      conn && this._onPeerConnect(conn)
+      conn && this._onPeerConnect(conn.remotePeer)
     }
   }
 

test/network/network.node.js

@@ -5,75 +5,63 @@ const { expect, assert } = require('aegir/utils/chai')
 const lp = require('it-length-prefixed')
 const pipe = require('it-pipe')
 const pDefer = require('p-defer')
-const pWaitFor = require('p-wait-for')
 const createLibp2pNode = require('../utils/create-libp2p-node')
 const makeBlock = require('../utils/make-block')
 const Network = require('../../src/network')
 const Message = require('../../src/types/message')
 
+function createBitswapMock () {
+  return {
+    _receiveMessage: async () => {},
+    _receiveError: async () => {},
+    _onPeerConnected: async () => {},
+    _onPeerDisconnected: async () => {}
+  }
+}
+
 describe('network', () => {
   let p2pA
   let networkA
+  let bitswapMockA
 
   let p2pB
   let networkB
+  let bitswapMockB
 
   let p2pC
   let networkC
+  let bitswapMockC
 
   let blocks
 
-  before(async () => {
+  beforeEach(async () => {
     [p2pA, p2pB, p2pC] = await Promise.all([
       createLibp2pNode(),
       createLibp2pNode(),
       createLibp2pNode()
     ])
     blocks = await makeBlock(2)
-  })
 
-  after(() => {
-    p2pA.stop()
-    p2pB.stop()
-    p2pC.stop()
-  })
-
-  const bitswapMockA = {
-    _receiveMessage: async () => {},
-    _receiveError: async () => {},
-    _onPeerConnected: async () => {},
-    _onPeerDisconnected: async () => {}
-  }
+    bitswapMockA = createBitswapMock()
+    bitswapMockB = createBitswapMock()
+    bitswapMockC = createBitswapMock()
 
-  const bitswapMockB = {
-    _receiveMessage: async () => {},
-    _receiveError: async () => {},
-    _onPeerConnected: async () => {},
-    _onPeerDisconnected: async () => {}
-  }
-
-  const bitswapMockC = {
-    _receiveMessage: async () => {},
-    _receiveError: async () => {},
-    _onPeerConnected: async () => {},
-    _onPeerDisconnected: async () => {}
-  }
-
-  it('instantiate the network obj', () => {
     networkA = new Network(p2pA, bitswapMockA)
     networkB = new Network(p2pB, bitswapMockB)
     // only bitswap100
     networkC = new Network(p2pC, bitswapMockC, { b100Only: true })
 
-    expect(networkA).to.exist()
-    expect(networkB).to.exist()
-    expect(networkC).to.exist()
-
     networkA.start()
     networkB.start()
     networkC.start()
   })
 
+  afterEach(() => {
+    p2pA.stop()
+    p2pB.stop()
+    p2pC.stop()
+  })
+
   it('connectTo fail', async () => {
     try {
       await networkA.connectTo(p2pB.peerId)
@@ -84,31 +72,70 @@ describe('network', () => {
   })
 
   it('onPeerConnected success', async () => {
-    var counter = 0
+    const p2pAConnected = pDefer()
+    const p2pBConnected = pDefer()
 
     bitswapMockA._onPeerConnected = (peerId) => {
       expect(peerId.toB58String()).to.equal(p2pB.peerId.toB58String())
-      counter++
+      p2pBConnected.resolve()
     }
 
     bitswapMockB._onPeerConnected = (peerId) => {
       expect(peerId.toB58String()).to.equal(p2pA.peerId.toB58String())
-      counter++
+      p2pAConnected.resolve()
     }
 
     const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}`
     await p2pA.dial(ma)
 
-    await pWaitFor(() => counter >= 2)
-    bitswapMockA._onPeerConnected = () => {}
-    bitswapMockB._onPeerConnected = () => {}
+    await Promise.all([
+      p2pAConnected,
+      p2pBConnected
+    ])
   })
 
   it('connectTo success', async () => {
     const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}`
     await networkA.connectTo(ma)
   })
 
+  it('sets up peer handlers for previously connected peers', async () => {
+    let p2pAConnected = pDefer()
+    let p2pBConnected = pDefer()
+
+    bitswapMockA._onPeerConnected = (peerId) => {
+      expect(peerId.toB58String()).to.equal(p2pB.peerId.toB58String())
+      p2pBConnected.resolve()
+    }
+
+    bitswapMockB._onPeerConnected = (peerId) => {
+      expect(peerId.toB58String()).to.equal(p2pA.peerId.toB58String())
+      p2pAConnected.resolve()
+    }
+
+    const ma = `${p2pB.multiaddrs[0]}/p2p/${p2pB.peerId.toB58String()}`
+    await p2pA.dial(ma)
+
+    await Promise.all([
+      p2pAConnected,
+      p2pBConnected
+    ])
+
+    networkA.stop()
+    networkB.stop()
+
+    p2pAConnected = pDefer()
+    p2pBConnected = pDefer()
+
+    networkA.start()
+    networkB.start()
+
+    await Promise.all([
+      p2pAConnected,
+      p2pBConnected
+    ])
+  })
+
   const versions = [{
     num: '1.0.0', serialize: (msg) => msg.serializeToBitswap100()
   }, {
@@ -159,6 +186,10 @@ describe('network', () => {
     msg.addBlock(b1)
     msg.addBlock(b2)
 
+    // In a real network scenario, peers will be discovered and their addresses
+    // will be added to the addressBook before bitswap kicks in
+    p2pA.peerStore.addressBook.set(p2pB.peerId, p2pB.multiaddrs)
+
     bitswapMockB._receiveMessage = async (peerId, msgReceived) => { // eslint-disable-line require-await
       expect(msg).to.eql(msgReceived)
       bitswapMockB._receiveMessage = async () => {}
@@ -189,6 +220,10 @@ describe('network', () => {
     msg.addBlock(b1)
     msg.addBlock(b2)
 
+    // In a real network scenario, peers will be discovered and their addresses
+    // will be added to the addressBook before bitswap kicks in
+    p2pA.peerStore.addressBook.set(p2pC.peerId, p2pC.multiaddrs)
+
     bitswapMockC._receiveMessage = async (peerId, msgReceived) => { // eslint-disable-line require-await
       expect(msg).to.eql(msgReceived)
       bitswapMockC._receiveMessage = async () => {}