chore: auto relay multiaddr update push

src/circuit/auto-relay.js

@@ -143,8 +143,7 @@ class AutoRelay {
 
     try {
       await this._transportManager.listen([multiaddr(listenAddr)])
-      // TODO: push announce multiaddrs update
-      // await this._libp2p.identifyService.pushToPeerStore()
+      // Announce multiaddrs will update on listen success by TransportManager event being triggered
     } catch (err) {
       log.error(err)
       this._listenRelays.delete(id)

src/circuit/listener.js

@@ -20,8 +20,8 @@ module.exports = (libp2p) => {
     const deleted = listeningAddrs.delete(connection.remotePeer.toB58String())
 
     if (deleted) {
-      // TODO push announce multiaddrs update
-      // libp2p.identifyService.pushToPeerStore()
+      // Announce listen addresses change
+      listener.emit('close')
     }
   })
 

src/identify/index.js

@@ -63,12 +63,6 @@ class IdentifyService {
      */
     this.connectionManager = libp2p.connectionManager
 
-    this.connectionManager.on('peer:connect', (connection) => {
-      const peerId = connection.remotePeer
-
-      this.identify(connection, peerId).catch(log.error)
-    })
-
     /**
      * @property {PeerId}
      */
@@ -82,6 +76,19 @@ class IdentifyService {
     this._protocols = protocols
 
     this.handleMessage = this.handleMessage.bind(this)
+
+    this.connectionManager.on('peer:connect', (connection) => {
+      const peerId = connection.remotePeer
+
+      this.identify(connection, peerId).catch(log.error)
+    })
+
+    // When self multiaddrs change, trigger identify-push
+    this.peerStore.on('change:multiaddrs', ({ peerId }) => {
+      if (peerId.toString() === this.peerId.toString()) {
+        this.pushToPeerStore()
+      }
+    })
   }
 
   /**
@@ -90,7 +97,7 @@ class IdentifyService {
    * @returns {Promise<void>}
    */
   async push (connections) {
-    const signedPeerRecord = await this._getSelfPeerRecord()
+    const signedPeerRecord = await this.peerStore.addressBook.getRawEnvelope(this.peerId)
     const listenAddrs = this._libp2p.multiaddrs.map((ma) => ma.bytes)
     const protocols = Array.from(this._protocols.keys())
 
@@ -239,7 +246,7 @@ class IdentifyService {
       publicKey = this.peerId.pubKey.bytes
     }
 
-    const signedPeerRecord = await this._getSelfPeerRecord()
+    const signedPeerRecord = await this.peerStore.addressBook.getRawEnvelope(this.peerId)
 
     const message = Message.encode({
       protocolVersion: PROTOCOL_VERSION,
@@ -308,33 +315,6 @@ class IdentifyService {
     // Update the protocols
     this.peerStore.protoBook.set(id, message.protocols)
   }
-
-  /**
-   * Get self signed peer record raw envelope.
-   * @return {Uint8Array}
-   */
-  async _getSelfPeerRecord () {
-    const selfSignedPeerRecord = this.peerStore.addressBook.getRawEnvelope(this.peerId)
-
-    // TODO: support invalidation when dynamic multiaddrs are supported
-    if (selfSignedPeerRecord) {
-      return selfSignedPeerRecord
-    }
-
-    try {
-      const peerRecord = new PeerRecord({
-        peerId: this.peerId,
-        multiaddrs: this._libp2p.multiaddrs
-      })
-      const envelope = await Envelope.seal(peerRecord, this.peerId)
-      this.peerStore.addressBook.consumePeerRecord(envelope)
-
-      return this.peerStore.addressBook.getRawEnvelope(this.peerId)
-    } catch (err) {
-      log.error('failed to get self peer record')
-    }
-    return null
-  }
 }
 
 module.exports.IdentifyService = IdentifyService

src/index.js

@@ -247,6 +247,7 @@ class Libp2p extends EventEmitter {
     log('libp2p is stopping')
 
     try {
+      this._isStarted = false
       for (const service of this._discovery.values()) {
         service.removeListener('peer', this._onDiscoveryPeer)
       }
@@ -274,7 +275,6 @@ class Libp2p extends EventEmitter {
         this.emit('error', err)
       }
     }
-    this._isStarted = false
     log('libp2p has stopped')
   }
 

src/transport-manager.js

@@ -7,6 +7,9 @@ const debug = require('debug')
 const log = debug('libp2p:transports')
 log.error = debug('libp2p:transports:error')
 
+const Envelope = require('./record/envelope')
+const PeerRecord = require('./record/peer-record')
+
 class TransportManager {
   /**
    * @constructor
@@ -62,6 +65,8 @@ class TransportManager {
       log('closing listeners for %s', key)
       while (listeners.length) {
         const listener = listeners.pop()
+        listener.removeAllListeners('listening')
+        listener.removeAllListeners('close')
         tasks.push(listener.close())
       }
     }
@@ -150,6 +155,10 @@ class TransportManager {
         const listener = transport.createListener({}, this.onConnection)
         this._listeners.get(key).push(listener)
 
+        // Track listen/close events
+        listener.on('listening', () => this._createSelfPeerRecord())
+        listener.on('close', () => this._createSelfPeerRecord())
+
         // We need to attempt to listen on everything
         tasks.push(listener.listen(addr))
       }
@@ -194,6 +203,8 @@ class TransportManager {
     if (this._listeners.has(key)) {
       // Close any running listeners
       for (const listener of this._listeners.get(key)) {
+        listener.removeAllListeners('listening')
+        listener.removeAllListeners('close')
         await listener.close()
       }
     }
@@ -215,6 +226,26 @@ class TransportManager {
 
     await Promise.all(tasks)
   }
+
+  /**
+   * Create self signed peer record raw envelope.
+   * @return {Uint8Array}
+   */
+  async _createSelfPeerRecord () {
+    try {
+      const peerRecord = new PeerRecord({
+        peerId: this.libp2p.peerId,
+        multiaddrs: this.libp2p.multiaddrs
+      })
+      const envelope = await Envelope.seal(peerRecord, this.libp2p.peerId)
+      this.libp2p.peerStore.addressBook.consumePeerRecord(envelope)
+
+      return this.libp2p.peerStore.addressBook.getRawEnvelope(this.libp2p.peerId)
+    } catch (err) {
+      log.error('failed to get self peer record')
+    }
+    return null
+  }
 }
 
 /**

test/dialing/direct.spec.js

@@ -354,7 +354,6 @@ describe('Dialing (direct, WebSockets)', () => {
       const connection = await libp2p.dial(remoteAddr)
       expect(connection).to.exist()
 
-      sinon.spy(libp2p.peerStore.addressBook, 'consumePeerRecord')
       sinon.spy(libp2p.peerStore.protoBook, 'set')
 
       // Wait for onConnection to be called
@@ -363,8 +362,6 @@ describe('Dialing (direct, WebSockets)', () => {
       expect(libp2p.identifyService.identify.callCount).to.equal(1)
       await libp2p.identifyService.identify.firstCall.returnValue
 
-      // Self + New peer
-      expect(libp2p.peerStore.addressBook.consumePeerRecord.callCount).to.equal(2)
       expect(libp2p.peerStore.protoBook.set.callCount).to.equal(1)
     })
 

test/identify/index.spec.js

@@ -20,6 +20,7 @@ const { IdentifyService, multicodecs } = require('../../src/identify')
 const Peers = require('../fixtures/peers')
 const Libp2p = require('../../src')
 const Envelope = require('../../src/record/envelope')
+const PeerRecord = require('../../src/record/peer-record')
 const PeerStore = require('../../src/peer-store')
 const baseOptions = require('../utils/base-options.browser')
 const pkg = require('../../package.json')
@@ -78,6 +79,9 @@ describe('Identify', () => {
     sinon.spy(localIdentify.peerStore.addressBook, 'consumePeerRecord')
     sinon.spy(localIdentify.peerStore.protoBook, 'set')
 
+    // Transport Manager creates signed peer record
+    await _createSelfPeerRecord(remoteIdentify._libp2p)
+
     // Run identify
     await Promise.all([
       localIdentify.identify(localConnectionMock),
@@ -239,6 +243,10 @@ describe('Identify', () => {
       sinon.spy(remoteIdentify.peerStore.addressBook, 'consumePeerRecord')
       sinon.spy(remoteIdentify.peerStore.protoBook, 'set')
 
+      // Transport Manager creates signed peer record
+      await _createSelfPeerRecord(localIdentify._libp2p)
+      await _createSelfPeerRecord(remoteIdentify._libp2p)
+
       // Run identify
       await Promise.all([
         localIdentify.push([localConnectionMock]),
@@ -249,7 +257,7 @@ describe('Identify', () => {
         })
       ])
 
-      expect(remoteIdentify.peerStore.addressBook.consumePeerRecord.callCount).to.equal(1)
+      expect(remoteIdentify.peerStore.addressBook.consumePeerRecord.callCount).to.equal(2)
       expect(remoteIdentify.peerStore.protoBook.set.callCount).to.equal(1)
 
       const addresses = localIdentify.peerStore.addressBook.get(localPeer)
@@ -359,8 +367,8 @@ describe('Identify', () => {
       expect(connection).to.exist()
 
       // Wait for peer store to be updated
-      // Dialer._createDialTarget (add), Identify (consume), Create self (consume)
-      await pWaitFor(() => peerStoreSpyConsumeRecord.callCount === 2 && peerStoreSpyAdd.callCount === 1)
+      // Dialer._createDialTarget (add), Identify (consume)
+      await pWaitFor(() => peerStoreSpyConsumeRecord.callCount === 1 && peerStoreSpyAdd.callCount === 1)
       expect(libp2p.identifyService.identify.callCount).to.equal(1)
 
       // The connection should have no open streams
@@ -404,5 +412,53 @@ describe('Identify', () => {
       // Verify the streams close
       await pWaitFor(() => connection.streams.length === 0)
     })
+
+    it('should push multiaddr updates to an already connected peer', async () => {
+      libp2p = new Libp2p({
+        ...baseOptions,
+        peerId
+      })
+
+      await libp2p.start()
+
+      sinon.spy(libp2p.identifyService, 'identify')
+      sinon.spy(libp2p.identifyService, 'push')
+
+      const connection = await libp2p.dialer.connectToPeer(remoteAddr)
+      expect(connection).to.exist()
+      // Wait for nextTick to trigger the identify call
+      await delay(1)
+
+      // Wait for identify to finish
+      await libp2p.identifyService.identify.firstCall.returnValue
+      sinon.stub(libp2p, 'isStarted').returns(true)
+
+      libp2p.peerStore.addressBook.add(libp2p.peerId, [multiaddr('/ip4/180.0.0.1/tcp/15001/ws')])
+
+      // Verify the remote peer is notified of change
+      expect(libp2p.identifyService.push.callCount).to.equal(1)
+      for (const call of libp2p.identifyService.push.getCalls()) {
+        const [connections] = call.args
+        expect(connections.length).to.equal(1)
+        expect(connections[0].remotePeer.toB58String()).to.equal(remoteAddr.getPeerId())
+        const results = await call.returnValue
+        expect(results.length).to.equal(1)
+      }
+
+      // Verify the streams close
+      await pWaitFor(() => connection.streams.length === 0)
+    })
   })
 })
+
+// Self peer record creating on Transport Manager simulation
+const _createSelfPeerRecord = async (libp2p) => {
+  try {
+    const peerRecord = new PeerRecord({
+      peerId: libp2p.peerId,
+      multiaddrs: libp2p.multiaddrs
+    })
+    const envelope = await Envelope.seal(peerRecord, libp2p.peerId)
+    libp2p.peerStore.addressBook.consumePeerRecord(envelope)
+  } catch (_) {}
+}

test/relay/auto-relay.node.js

@@ -259,6 +259,9 @@ describe('auto-relay', () => {
     it('should not listen on a relayed address if peer disconnects', async () => {
       const originalMultiaddrs1Length = relayLibp2p1.multiaddrs.length
 
+      // Spy if identify push is fired on adding/removing listen addr
+      sinon.spy(relayLibp2p1.identifyService, 'pushToPeerStore')
+
       // Discover one relay and connect
       relayLibp2p1.peerStore.addressBook.add(relayLibp2p2.peerId, relayLibp2p2.multiaddrs)
       await relayLibp2p1.dial(relayLibp2p2.peerId)
@@ -268,16 +271,18 @@ describe('auto-relay', () => {
       expect(autoRelay1._listenRelays.size).to.equal(1)
       expect(relayLibp2p1.multiaddrs[originalMultiaddrs1Length].getPeerId()).to.eql(relayLibp2p2.peerId.toB58String())
 
-      // Spy if identify push is fired
-      sinon.spy(relayLibp2p1.identifyService, 'pushToPeerStore')
+      // Identify push for adding listen relay multiaddr
+      expect(relayLibp2p1.identifyService.pushToPeerStore.callCount).to.equal(1)
 
       // Disconnect from peer used for relay
       await relayLibp2p1.hangUp(relayLibp2p2.peerId)
 
       // Wait for removed listening on the relay
       await pWaitFor(() => relayLibp2p1.multiaddrs.length === originalMultiaddrs1Length)
       expect(autoRelay1._listenRelays.size).to.equal(0)
-      // TODO: identify-push expect(relayLibp2p1.identifyService.pushToPeerStore.callCount).to.equal(1)
+
+      // Identify push for removing listen relay multiaddr
+      expect(relayLibp2p1.identifyService.pushToPeerStore.callCount).to.equal(2)
     })
 
     it('should try to listen on other connected peers relayed address if one used relay disconnects', async () => {

test/transports/transport-manager.node.js

@@ -4,24 +4,34 @@
 const chai = require('chai')
 chai.use(require('dirty-chai'))
 const { expect } = chai
+const sinon = require('sinon')
 
 const AddressManager = require('../../src/address-manager')
 const TransportManager = require('../../src/transport-manager')
+const PeerStore = require('../../src/peer-store')
 const Transport = require('libp2p-tcp')
+const PeerId = require('peer-id')
 const multiaddr = require('multiaddr')
 const mockUpgrader = require('../utils/mockUpgrader')
+const Peers = require('../fixtures/peers')
 const addrs = [
   multiaddr('/ip4/127.0.0.1/tcp/0'),
   multiaddr('/ip4/127.0.0.1/tcp/0')
 ]
 
 describe('Transport Manager (TCP)', () => {
   let tm
+  let localPeer
+
+  before(async () => {
+    localPeer = await PeerId.createFromJSON(Peers[0])
+  })
 
   before(() => {
     tm = new TransportManager({
       libp2p: {
-        addressManager: new AddressManager({ listen: addrs })
+        addressManager: new AddressManager({ listen: addrs }),
+        PeerStore: new PeerStore({ peerId: localPeer })
       },
       upgrader: mockUpgrader,
       onConnection: () => {}
@@ -40,10 +50,16 @@ describe('Transport Manager (TCP)', () => {
   })
 
   it('should be able to listen', async () => {
+    sinon.spy(tm, '_createSelfPeerRecord')
+
     tm.add(Transport.prototype[Symbol.toStringTag], Transport)
     await tm.listen(addrs)
     expect(tm._listeners).to.have.key(Transport.prototype[Symbol.toStringTag])
     expect(tm._listeners.get(Transport.prototype[Symbol.toStringTag])).to.have.length(addrs.length)
+
+    // Created Self Peer record on new listen address
+    expect(tm._createSelfPeerRecord.callCount).to.equal(addrs.length)
+
     // Ephemeral ip addresses may result in multiple listeners
     expect(tm.getAddrs().length).to.equal(addrs.length)
     await tm.close()