chore: release version v0.30.2

* fix: store provider multiaddrs during find providers

Changes the behaviour of `libp2p.contentRouting.findProviders` to store
the multiaddrs reported by the routers before yielding results to
the caller, so when they try to dial the provider, the multiaddrs are
already in the peer store's address book.

Also dedupes providers reported by routers but keeps all of the addresses
reported, even for duplicates.

Also, also fixes a performance bug where the previous implementation would
wait for any router to completely finish finding providers before sending
any results to the caller.  It'll now yield results as they come in which
makes it much, much faster.

.github/workflows/main.yml

@@ -65,3 +65,38 @@ jobs:
       - uses: actions/checkout@v2
       - run: yarn
       - run: cd examples && yarn && npm run test -- auto-relay
+  test-chat-example:
+    needs: check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: yarn
+      - run: cd examples && yarn && npm run test -- chat
+  test-connection-encryption-example:
+    needs: check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: yarn
+      - run: cd examples && yarn && npm run test -- connection-encryption
+  test-echo-example:
+    needs: check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: yarn
+      - run: cd examples && yarn && npm run test -- echo 
+  test-libp2p-in-the-browser-example:
+    needs: check
+    runs-on: macos-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: yarn
+      - run: cd examples && yarn && npm run test -- libp2p-in-the-browser
+  test-discovery-mechanisms-example:
+    needs: check
+    runs-on: macos-latest
+    steps:
+      - uses: actions/checkout@v2
+      - run: yarn
+      - run: cd examples && yarn && npm run test -- discovery-mechanisms

CHANGELOG.md

@@ -1,3 +1,21 @@
+## [0.30.2](https://github.com/libp2p/js-libp2p/compare/v0.30.1...v0.30.2) (2021-01-21)
+
+
+### Bug Fixes
+
+* store multiaddrs during content and peer routing queries ([#865](https://github.com/libp2p/js-libp2p/issues/865)) ([45c3367](https://github.com/libp2p/js-libp2p/commit/45c33675a7412c66d0fd4e113ef8506077b6f492))
+
+
+
+## [0.30.1](https://github.com/libp2p/js-libp2p/compare/v0.30.0...v0.30.1) (2021-01-18)
+
+
+### Bug Fixes
+
+* event emitter types with local types ([#864](https://github.com/libp2p/js-libp2p/issues/864)) ([6c41e30](https://github.com/libp2p/js-libp2p/commit/6c41e3045608bcae8061d20501be5751dad8157a))
+
+
+
 # [0.30.0](https://github.com/libp2p/js-libp2p/compare/v0.29.4...v0.30.0) (2020-12-16)
 
 

README.md

@@ -16,8 +16,8 @@
 </p>
 
 <p align="center">
-  <a href="https://travis-ci.com/libp2p/js-libp2p"><img src="https://flat.badgen.net/travis/libp2p/js-libp2p" /></a>
-  <a href="https://codecov.io/gh/libp2p/js-libp2p"><img src="https://img.shields.io/codecov/c/github/ipfs/js-ipfs-multipart/master.svg?style=flat-square"></a>
+  <a href="https://github.com/libp2p/js-libp2p/actions?query=branch%3Amaster+workflow%3Aci+"><img src="https://img.shields.io/github/workflow/status/libp2p/js-libp2p/ci?label=ci&style=flat-square" /></a>
+  <a href="https://codecov.io/gh/libp2p/js-libp2p"><img src="https://img.shields.io/codecov/c/github/libp2p/js-libp2p/master.svg?style=flat-square"></a>
   <a href="https://bundlephobia.com/result?p=ipfsd-ctl"><img src="https://flat.badgen.net/bundlephobia/minzip/ipfsd-ctl"></a>
   <br>
   <a href="https://david-dm.org/libp2p/js-libp2p"><img src="https://david-dm.org/libp2p/js-libp2p.svg?style=flat-square" /></a>

doc/CONFIGURATION.md

@@ -261,13 +261,14 @@ const TCP = require('libp2p-tcp')
 const MPLEX = require('libp2p-mplex')
 const { NOISE } = require('libp2p-noise')
 const MulticastDNS = require('libp2p-mdns')
+const Bootstrap = require('libp2p-bootstrap')
 
 const node = await Libp2p.create({
   modules: {
     transport: [TCP],
     streamMuxer: [MPLEX],
     connEncryption: [NOISE],
-    peerDiscovery: [MulticastDNS]
+    peerDiscovery: [MulticastDNS, Bootstrap]
   },
   config: {
     peerDiscovery: {
@@ -277,6 +278,15 @@ const node = await Libp2p.create({
       [MulticastDNS.tag]: {
         interval: 1000,
         enabled: true
+      },
+      [Bootstrap.tag:] {
+        list: [ // A list of bootstrap peers to connect to starting up the node
+          "/ip4/104.131.131.82/tcp/4001/ipfs/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ",
+          "/dnsaddr/bootstrap.libp2p.io/ipfs/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
+          "/dnsaddr/bootstrap.libp2p.io/ipfs/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
+        ],
+        interval: 2000,
+        enabled: true
       }
       // .. other discovery module options.
     }

examples/chat/src/dialer.js

@@ -3,7 +3,7 @@
 
 const PeerId = require('peer-id')
 const multiaddr = require('multiaddr')
-const createLibp2p = require('./libp2p-bundle')
+const createLibp2p = require('./libp2p')
 const { stdinToStream, streamToConsole } = require('./stream')
 
 async function run() {

examples/chat/src/listener.js

@@ -2,7 +2,7 @@
 /* eslint-disable no-console */
 
 const PeerId = require('peer-id')
-const createLibp2p = require('./libp2p-bundle.js')
+const createLibp2p = require('./libp2p.js')
 const { stdinToStream, streamToConsole } = require('./stream')
 
 async function run() {

examples/chat/test.js

@@ -0,0 +1,77 @@
+'use strict'
+
+const path = require('path')
+const execa = require('execa')
+const pDefer = require('p-defer')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+function startProcess(name) {
+  return execa('node', [path.join(__dirname, name)], {
+    cwd: path.resolve(__dirname),
+    all: true
+  })
+}
+
+async function test () {
+  const message = 'test message'
+  let listenerOutput = ''
+  let dialerOutput = ''
+
+  let isListening = false
+  let messageSent = false
+  const listenerReady = pDefer()
+  const dialerReady = pDefer()
+  const messageReceived = pDefer()
+
+  // Step 1 process
+  process.stdout.write('node listener.js\n')
+  const listenerProc = startProcess('src/listener.js')
+  listenerProc.all.on('data', async (data) => {
+    process.stdout.write(data)
+
+    listenerOutput += uint8ArrayToString(data)
+
+    if (!isListening && listenerOutput.includes('Listener ready, listening on')) {
+      listenerReady.resolve()
+      isListening = true
+    } else if (isListening && listenerOutput.includes(message)) {
+      messageReceived.resolve()
+    }
+  })
+
+  await listenerReady.promise
+  process.stdout.write('==================================================================\n')
+
+  // Step 2 process
+  process.stdout.write('node dialer.js\n')
+  const dialerProc = startProcess('src/dialer.js')
+  dialerProc.all.on('data', async (data) => {
+    process.stdout.write(data)
+    dialerOutput += uint8ArrayToString(data)
+
+    if (!messageSent && dialerOutput.includes('Type a message and see what happens')) {
+      dialerReady.resolve()
+      dialerProc.stdin.write(message)
+      dialerProc.stdin.write('\n')
+      messageSent = true
+    }
+  })
+
+  await dialerReady.promise
+  process.stdout.write('==================================================================\n')
+  await messageReceived.promise
+  process.stdout.write('chat message received\n')
+
+  listenerProc.kill()
+  dialerProc.kill()
+  await Promise.all([
+    listenerProc,
+    dialerProc
+  ]).catch((err) => {
+    if (err.signal !== 'SIGTERM') {
+      throw err
+    }
+  })
+}
+
+module.exports = test

examples/connection-encryption/1.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const Libp2p = require('../../')
+const Libp2p = require('../..')
 const TCP = require('libp2p-tcp')
 const Mplex = require('libp2p-mplex')
 const { NOISE } = require('libp2p-noise')

examples/connection-encryption/README.md

@@ -1,4 +1,4 @@
-# Encrypted Communications
+# Connection Encryption
 
 libp2p can leverage the encrypted communications from the transports it uses (i.e WebRTC). To ensure that every connection is encrypted, independently of how it was set up, libp2p also supports a set of modules that encrypt every communication established.
 

examples/connection-encryption/test.js

@@ -0,0 +1,30 @@
+'use strict'
+
+const path = require('path')
+const execa = require('execa')
+const pDefer = require('p-defer')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+async function test () {
+  const messageReceived = pDefer()
+  process.stdout.write('1.js\n')
+
+  const proc = execa('node', [path.join(__dirname, '1.js')], {
+    cwd: path.resolve(__dirname),
+    all: true
+  })
+
+  proc.all.on('data', async (data) => {
+    process.stdout.write(data)
+
+    const s = uint8ArrayToString(data)
+    if (s.includes('This information is sent out encrypted to the other peer')) {
+      messageReceived.resolve()
+    }
+  })
+
+  await messageReceived.promise
+  proc.kill()
+}
+
+module.exports = test

examples/discovery-mechanisms/1.js

@@ -7,15 +7,7 @@ const Mplex = require('libp2p-mplex')
 const { NOISE } = require('libp2p-noise')
 const Bootstrap = require('libp2p-bootstrap')
 
-// Find this list at: https://github.com/ipfs/js-ipfs/blob/master/src/core/runtime/config-nodejs.json
-const bootstrapers = [
-  '/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ',
-  '/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN',
-  '/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb',
-  '/dnsaddr/bootstrap.libp2p.io/p2p/QmZa1sAxajnQjVM8WjWXoMbmPd7NsWhfKsPkErzpm9wGkp',
-  '/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa',
-  '/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt'
-]
+const bootstrapers = require('./bootstrapers')
 
 ;(async () => {
   const node = await Libp2p.create({

examples/discovery-mechanisms/bootstrapers.js

@@ -0,0 +1,13 @@
+'use strict'
+
+// Find this list at: https://github.com/ipfs/js-ipfs/blob/master/src/core/runtime/config-nodejs.json
+const bootstrapers = [
+  '/ip4/104.131.131.82/tcp/4001/p2p/QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ',
+  '/dnsaddr/bootstrap.libp2p.io/p2p/QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN',
+  '/dnsaddr/bootstrap.libp2p.io/p2p/QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb',
+  '/dnsaddr/bootstrap.libp2p.io/p2p/QmZa1sAxajnQjVM8WjWXoMbmPd7NsWhfKsPkErzpm9wGkp',
+  '/dnsaddr/bootstrap.libp2p.io/p2p/QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa',
+  '/dnsaddr/bootstrap.libp2p.io/p2p/QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt'
+]
+
+module.exports = bootstrapers

examples/discovery-mechanisms/test-1.js

@@ -0,0 +1,42 @@
+'use strict'
+
+const path = require('path')
+const execa = require('execa')
+const pWaitFor = require('p-wait-for')
+const uint8ArrayToString = require('uint8arrays/to-string')
+const bootstrapers = require('./bootstrapers')
+
+const discoveredCopy = 'Discovered:'
+const connectedCopy = 'Connection established to:'
+
+async function test () {
+  const discoveredNodes = []
+  const connectedNodes = []
+
+  process.stdout.write('1.js\n')
+
+  const proc = execa('node', [path.join(__dirname, '1.js')], {
+    cwd: path.resolve(__dirname),
+    all: true
+  })
+
+  proc.all.on('data', async (data) => {
+    process.stdout.write(data)
+    const line = uint8ArrayToString(data)
+
+    // Discovered or Connected
+    if (line.includes(discoveredCopy)) {
+      const id = line.trim().split(discoveredCopy)[1]
+      discoveredNodes.push(id)
+    } else if (line.includes(connectedCopy)) {
+      const id = line.trim().split(connectedCopy)[1]
+      connectedNodes.push(id)
+    }
+  })
+
+  await pWaitFor(() => discoveredNodes.length === bootstrapers.length && connectedNodes.length === bootstrapers.length)
+
+  proc.kill()
+}
+
+module.exports = test

examples/discovery-mechanisms/test-2.js

@@ -0,0 +1,35 @@
+'use strict'
+
+const path = require('path')
+const execa = require('execa')
+const pWaitFor = require('p-wait-for')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+const discoveredCopy = 'Discovered:'
+
+async function test() {
+  const discoveredNodes = []
+
+  process.stdout.write('2.js\n')
+
+  const proc = execa('node', [path.join(__dirname, '2.js')], {
+    cwd: path.resolve(__dirname),
+    all: true
+  })
+
+  proc.all.on('data', async (data) => {
+    process.stdout.write(data)
+    const line = uint8ArrayToString(data)
+
+    if (line.includes(discoveredCopy)) {
+      const id = line.trim().split(discoveredCopy)[1]
+      discoveredNodes.push(id)
+    }
+  })
+
+  await pWaitFor(() => discoveredNodes.length === 2)
+
+  proc.kill()
+}
+
+module.exports = test

examples/discovery-mechanisms/test.js

@@ -0,0 +1,11 @@
+'use strict'
+
+const test1 = require('./test-1')
+const test2 = require('./test-2')
+
+async function test () {
+  await test1()
+  await test2()
+}
+
+module.exports = test

examples/echo/src/dialer.js

@@ -6,7 +6,7 @@
  */
 
 const PeerId = require('peer-id')
-const createLibp2p = require('./libp2p-bundle')
+const createLibp2p = require('./libp2p')
 const pipe = require('it-pipe')
 
 async function run() {

examples/echo/src/listener.js

@@ -6,7 +6,7 @@
  */
 
 const PeerId = require('peer-id')
-const createLibp2p = require('./libp2p-bundle')
+const createLibp2p = require('./libp2p')
 const pipe = require('it-pipe')
 
 async function run() {

examples/echo/test.js

@@ -0,0 +1,61 @@
+'use strict'
+
+const path = require('path')
+const execa = require('execa')
+const pDefer = require('p-defer')
+const uint8ArrayToString = require('uint8arrays/to-string')
+
+function startProcess(name) {
+  return execa('node', [path.join(__dirname, name)], {
+    cwd: path.resolve(__dirname),
+    all: true
+  })
+}
+
+async function test () {
+  const listenerReady = pDefer()
+  const messageReceived = pDefer()
+
+  // Step 1 process
+  process.stdout.write('node listener.js\n')
+  const listenerProc = startProcess('src/listener.js')
+  listenerProc.all.on('data', async (data) => {
+    process.stdout.write(data)
+    const s = uint8ArrayToString(data)
+
+    if (s.includes('Listener ready, listening on:')) {
+      listenerReady.resolve()
+    }
+  })
+
+  await listenerReady.promise
+  process.stdout.write('==================================================================\n')
+
+  // Step 2 process
+  process.stdout.write('node dialer.js\n')
+  const dialerProc = startProcess('src/dialer.js')
+  dialerProc.all.on('data', async (data) => {
+    process.stdout.write(data)
+    const s = uint8ArrayToString(data)
+
+    if (s.includes('received echo:')) {
+      messageReceived.resolve()
+    }
+  })
+
+  await messageReceived.promise
+  process.stdout.write('echo message received\n')
+
+  listenerProc.kill()
+  dialerProc.kill()
+  await Promise.all([
+    listenerProc,
+    dialerProc
+  ]).catch((err) => {
+    if (err.signal !== 'SIGTERM') {
+      throw err
+    }
+  })
+}
+
+module.exports = test

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

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

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

@@ -0,0 +1,52 @@
+'use strict'
+
+const execa = require('execa')
+const { chromium } = require('playwright');
+
+async function run() {
+  let url = ''
+  const proc = execa('parcel', ['./index.html'], {
+    preferLocal: true,
+    localDir: __dirname,
+    cwd: __dirname,
+    all: true
+  })
+
+  proc.all.on('data', async (chunk) => {
+    /**@type {string} */
+    const out = chunk.toString()
+
+    if (out.includes('Server running at')) {
+      url = out.replace('Server running at ', '')
+    }
+
+    if (out.includes('✨  Built in ')) {
+      try {
+        const browser = await chromium.launch();
+        const page = await browser.newPage();
+        await page.goto(url);
+        await page.waitForFunction(selector => document.querySelector(selector).innerText === 'libp2p started!', '#status')
+        await page.waitForFunction(
+          selector => {
+            const text = document.querySelector(selector).innerText
+            return text.includes('libp2p id is') &&
+              text.includes('Found peer') &&
+              text.includes('Connected to')
+          },
+          '#output',
+          { timeout: 5000 }
+        )
+        await browser.close();
+
+      } catch (err) {
+        console.error(err)
+        process.exit(1)
+      } finally {
+        proc.cancel()
+      }
+    }
+  })
+
+}
+
+module.exports = run

examples/package.json

@@ -12,5 +12,8 @@
     "fs-extra": "^8.1.0",
     "p-defer": "^3.0.0",
     "which": "^2.0.1"
+  },
+  "devDependencies": {
+    "playwright": "^1.7.1"
   }
 }

examples/test.js

@@ -22,7 +22,6 @@ async function testExample (dir) {
   await installDeps(dir)
   await build(dir)
   await runTest(dir)
-  // TODO: add browser test setup
 }
 
 async function installDeps (dir) {
@@ -89,7 +88,7 @@ async function runTest (dir) {
     return
   }
 
-  const runTest = require(testFile)
+  const test = require(testFile)
 
-  await runTest()
-}
+  await test()
+}

package.json

@@ -1,6 +1,6 @@
 {
   "name": "libp2p",
-  "version": "0.30.0",
+  "version": "0.30.2",
   "description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
   "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
   "main": "src/index.js",
@@ -65,10 +65,16 @@
     "ipfs-utils": "^5.0.1",
     "it-all": "^1.0.1",
     "it-buffer": "^0.1.2",
+    "it-drain": "^1.0.3",
+    "it-filter": "^1.0.1",
+    "it-first": "^1.0.4",
     "it-handshake": "^1.0.1",
     "it-length-prefixed": "^3.0.1",
+    "it-map": "^1.0.4",
+    "it-merge": "1.0.0",
     "it-pipe": "^1.1.0",
     "it-protocol-buffers": "^0.2.0",
+    "it-take": "1.0.0",
     "libp2p-crypto": "^0.18.0",
     "libp2p-interfaces": "^0.8.1",
     "libp2p-utils": "^0.2.2",
@@ -150,11 +156,11 @@
     "Andrew Nesbitt <andrewnez@gmail.com>",
     "Giovanni T. Parra <fiatjaf@gmail.com>",
     "Ryan Bell <ryan@piing.net>",
+    "Samlior <samlior@foxmail.com>",
     "Thomas Eizinger <thomas@eizinger.io>",
     "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com>",
     "Didrik Nordström <didrik@betamos.se>",
     "Irakli Gozalishvili <rfobic@gmail.com>",
-    "Ethan Lam <elmemphis2000@gmail.com>",
     "Joel Gustafson <joelg@mit.edu>",
     "Julien Bouquillon <contact@revolunet.com>",
     "Kevin Kwok <antimatter15@gmail.com>",
@@ -162,24 +168,24 @@
     "Dmitriy Ryajov <dryajov@gmail.com>",
     "RasmusErik Voel Jensen <github@solsort.com>",
     "Diogo Silva <fsdiogo@gmail.com>",
-    "Samlior <samlior@foxmail.com>",
+    "isan_rivkin <isanrivkin@gmail.com>",
     "Smite Chow <xiaopengyou@live.com>",
     "Soeren <nikorpoulsen@gmail.com>",
     "Sönke Hahn <soenkehahn@gmail.com>",
-    "robertkiel <robert.kiel@validitylabs.org>",
     "Tiago Alves <alvesjtiago@gmail.com>",
     "Daijiro Wachi <daijiro.wachi@gmail.com>",
     "Yusef Napora <yusef@napora.org>",
     "Zane Starr <zcstarr@gmail.com>",
+    "robertkiel <robert.kiel@validitylabs.org>",
     "Cindy Wu <ciindy.wu@gmail.com>",
     "Chris Bratlien <chrisbratlien@gmail.com>",
-    "ebinks <elizabethjbinks@gmail.com>",
-    "Bernd Strehl <bernd.strehl@gmail.com>",
     "Florian-Merle <florian.david.merle@gmail.com>",
     "Francis Gulotta <wizard@roborooter.com>",
     "Felipe Martins <felipebrasil93@gmail.com>",
-    "isan_rivkin <isanrivkin@gmail.com>",
+    "ebinks <elizabethjbinks@gmail.com>",
     "Henrique Dias <hacdias@gmail.com>",
-    "Fei Liu <liu.feiwood@gmail.com>"
+    "Bernd Strehl <bernd.strehl@gmail.com>",
+    "Fei Liu <liu.feiwood@gmail.com>",
+    "Ethan Lam <elmemphis2000@gmail.com>"
   ]
 }

src/connection-manager/index.js

@@ -10,7 +10,9 @@ const mergeOptions = require('merge-options')
 const LatencyMonitor = require('./latency-monitor')
 const retimer = require('retimer')
 
-const { EventEmitter } = require('events')
+/** @typedef {import('../types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 
 const PeerId = require('peer-id')
 
@@ -158,7 +160,7 @@ class ConnectionManager extends EventEmitter {
       }
     }
 
-    await tasks
+    await Promise.all(tasks)
     this.connections.clear()
   }
 

src/connection-manager/latency-monitor.js

@@ -7,7 +7,9 @@
 
 /* global window */
 const globalThis = require('ipfs-utils/src/globalthis')
-const { EventEmitter } = require('events')
+/** @typedef {import('../types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 const VisibilityChangeEmitter = require('./visibility-change-emitter')
 const debug = require('debug')('latency-monitor:LatencyMonitor')
 

src/connection-manager/visibility-change-emitter.js

@@ -6,7 +6,9 @@
  */
 'use strict'
 
-const { EventEmitter } = require('events')
+/** @typedef {import('../types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 
 const debug = require('debug')('latency-monitor:VisibilityChangeEmitter')
 

src/content-routing/index.js

@@ -1,10 +1,16 @@
 'use strict'
 
 const errCode = require('err-code')
-const { messages, codes } = require('./errors')
+const { messages, codes } = require('../errors')
+const {
+  storeAddresses,
+  uniquePeers,
+  requirePeers,
+  maybeLimitSource
+} = require('./utils')
 
-const all = require('it-all')
-const pAny = require('p-any')
+const merge = require('it-merge')
+const { pipe } = require('it-pipe')
 
 /**
  * @typedef {import('peer-id')} PeerId
@@ -21,22 +27,21 @@ const pAny = require('p-any')
 class ContentRouting {
   /**
    * @class
-   * @param {import('./')} libp2p
+   * @param {import('..')} libp2p
    */
   constructor (libp2p) {
     this.libp2p = libp2p
     this.routers = libp2p._modules.contentRouting || []
     this.dht = libp2p._dht
 
-    // If we have the dht, make it first
+    // If we have the dht, add it to the available content routers
     if (this.dht) {
-      this.routers.unshift(this.dht)
+      this.routers.push(this.dht)
     }
   }
 
   /**
-   * Iterates over all content routers in series to find providers of the given key.
-   * Once a content router succeeds, iteration will stop.
+   * Iterates over all content routers in parallel to find providers of the given key.
    *
    * @param {CID} key - The CID key of the content to find
    * @param {object} [options]
@@ -44,25 +49,20 @@ class ContentRouting {
    * @param {number} [options.maxNumProviders] - maximum number of providers to find
    * @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
    */
-  async * findProviders (key, options) {
+  async * findProviders (key, options = {}) {
     if (!this.routers.length) {
       throw errCode(new Error('No content this.routers available'), 'NO_ROUTERS_AVAILABLE')
     }
 
-    const result = await pAny(
-      this.routers.map(async (router) => {
-        const provs = await all(router.findProviders(key, options))
-
-        if (!provs || !provs.length) {
-          throw errCode(new Error('not found'), 'NOT_FOUND')
-        }
-        return provs
-      })
+    yield * pipe(
+      merge(
+        ...this.routers.map(router => router.findProviders(key, options))
+      ),
+      (source) => storeAddresses(source, this.libp2p.peerStore),
+      (source) => uniquePeers(source),
+      (source) => maybeLimitSource(source, options.maxNumProviders),
+      (source) => requirePeers(source)
     )
-
-    for (const peer of result) {
-      yield peer
-    }
   }
 
   /**

src/content-routing/utils.js

@@ -0,0 +1,89 @@
+'use strict'
+
+const errCode = require('err-code')
+const filter = require('it-filter')
+const map = require('it-map')
+const take = require('it-take')
+
+/**
+ * @typedef {import('peer-id')} PeerId
+ * @typedef {import('multiaddr')} Multiaddr
+ */
+
+/**
+ * Store the multiaddrs from every peer in the passed peer store
+ *
+ * @param {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>} source
+ * @param {import('../peer-store')} peerStore
+ */
+function storeAddresses (source, peerStore) {
+  return map(source, (peer) => {
+    // ensure we have the addresses for a given peer
+    peerStore.addressBook.add(peer.id, peer.multiaddrs)
+
+    return peer
+  })
+}
+
+/**
+ * Filter peers by unique peer id
+ *
+ * @param {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>} source
+ */
+function uniquePeers (source) {
+  /** @type Set<string> */
+  const seen = new Set()
+
+  return filter(source, (peer) => {
+    // dedupe by peer id
+    if (seen.has(peer.id.toString())) {
+      return false
+    }
+
+    seen.add(peer.id.toString())
+
+    return true
+  })
+}
+
+/**
+ * Require at least `min` peers to be yielded from `source`
+ *
+ * @param {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>} source
+ * @param {number} min
+ */
+async function * requirePeers (source, min = 1) {
+  let seen = 0
+
+  for await (const peer of source) {
+    seen++
+
+    yield peer
+  }
+
+  if (seen < min) {
+    throw errCode(new Error('not found'), 'NOT_FOUND')
+  }
+}
+
+/**
+ * If `max` is passed, only take that number of peers from the source
+ * otherwise take all the peers
+ *
+ * @param {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>} source
+ * @param {number} [max]
+ */
+function maybeLimitSource (source, max) {
+  if (max) {
+    return take(source, max)
+  }
+
+  return source
+}
+
+module.exports = {
+  storeAddresses,
+  uniquePeers,
+  requirePeers,
+  maybeLimitSource
+}

src/index.js

@@ -4,7 +4,9 @@ const debug = require('debug')
 const log = Object.assign(debug('libp2p'), {
   error: debug('libp2p:err')
 })
-const { EventEmitter } = require('events')
+/** @typedef {import('./types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 const globalThis = require('ipfs-utils/src/globalthis')
 
 const errCode = require('err-code')

src/metrics/stats.js

@@ -1,7 +1,9 @@
 // @ts-nocheck
 'use strict'
 
-const { EventEmitter } = require('events')
+/** @typedef {import('../types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 const Big = require('bignumber.js')
 const MovingAverage = require('moving-average')
 const retimer = require('retimer')

src/peer-routing.js

@@ -5,16 +5,24 @@ const log = Object.assign(debug('libp2p:peer-routing'), {
   error: debug('libp2p:peer-routing:err')
 })
 const errCode = require('err-code')
+const {
+  storeAddresses,
+  uniquePeers,
+  requirePeers
+} = require('./content-routing/utils')
 
-const all = require('it-all')
-const pAny = require('p-any')
+const merge = require('it-merge')
+const { pipe } = require('it-pipe')
+const first = require('it-first')
+const drain = require('it-drain')
+const filter = require('it-filter')
 const {
   setDelayedInterval,
   clearDelayedInterval
 } = require('set-delayed-interval')
+const PeerId = require('peer-id')
 
 /**
- * @typedef {import('peer-id')} PeerId
  * @typedef {import('multiaddr')} Multiaddr
  */
 class PeerRouting {
@@ -27,9 +35,9 @@ class PeerRouting {
     this._peerStore = libp2p.peerStore
     this._routers = libp2p._modules.peerRouting || []
 
-    // If we have the dht, make it first
+    // If we have the dht, add it to the available peer routers
     if (libp2p._dht) {
-      this._routers.unshift(libp2p._dht)
+      this._routers.push(libp2p._dht)
     }
 
     this._refreshManagerOptions = libp2p._options.peerRouting.refreshManager
@@ -55,9 +63,8 @@ class PeerRouting {
    */
   async _findClosestPeersTask () {
     try {
-      for await (const { id, multiaddrs } of this.getClosestPeers(this._peerId.id)) {
-        this._peerStore.addressBook.add(id, multiaddrs)
-      }
+      // nb getClosestPeers adds the addresses to the address book
+      await drain(this.getClosestPeers(this._peerId.id))
     } catch (err) {
       log.error(err)
     }
@@ -71,7 +78,7 @@ class PeerRouting {
   }
 
   /**
-   * Iterates over all peer routers in series to find the given peer.
+   * Iterates over all peer routers in parallel to find the given peer.
    *
    * @param {PeerId} id - The id of the peer to find
    * @param {object} [options]
@@ -83,16 +90,20 @@ class PeerRouting {
       throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
     }
 
-    return pAny(this._routers.map(async (router) => {
-      const result = await router.findPeer(id, options)
+    const output = await pipe(
+      merge(
+        ...this._routers.map(router => [router.findPeer(id, options)])
+      ),
+      (source) => filter(source, Boolean),
+      (source) => storeAddresses(source, this._peerStore),
+      (source) => first(source)
+    )
 
-      // If we don't have a result, we need to provide an error to keep trying
-      if (!result || Object.keys(result).length === 0) {
-        throw errCode(new Error('not found'), 'NOT_FOUND')
-      }
+    if (output) {
+      return output
+    }
 
-      return result
-    }))
+    throw errCode(new Error('not found'), 'NOT_FOUND')
   }
 
   /**
@@ -108,20 +119,14 @@ class PeerRouting {
       throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
     }
 
-    const result = await pAny(
-      this._routers.map(async (router) => {
-        const peers = await all(router.getClosestPeers(key, options))
-
-        if (!peers || !peers.length) {
-          throw errCode(new Error('not found'), 'NOT_FOUND')
-        }
-        return peers
-      })
+    yield * pipe(
+      merge(
+        ...this._routers.map(router => router.getClosestPeers(key, options))
+      ),
+      (source) => storeAddresses(source, this._peerStore),
+      (source) => uniquePeers(source),
+      (source) => requirePeers(source)
     )
-
-    for (const peer of result) {
-      yield peer
-    }
   }
 }
 

src/peer-store/index.js

@@ -2,7 +2,9 @@
 
 const errcode = require('err-code')
 
-const { EventEmitter } = require('events')
+/** @typedef {import('../types').EventEmitterFactory} Events */
+/** @type Events */
+const EventEmitter = require('events')
 const PeerId = require('peer-id')
 
 const AddressBook = require('./address-book')

src/types.ts

@@ -82,3 +82,22 @@ export type CircuitMessageProto = {
     CAN_HOP: CAN_HOP
   }
 }
+
+export interface EventEmitterFactory {
+  new(): EventEmitter;
+}
+
+export interface EventEmitter {
+  addListener(event: string | symbol, listener: (...args: any[]) => void);
+  on(event: string | symbol, listener: (...args: any[]) => void);
+  once(event: string | symbol, listener: (...args: any[]) => void);
+  removeListener(event: string | symbol, listener: (...args: any[]) => void);
+  off(event: string | symbol, listener: (...args: any[]) => void);
+  removeAllListeners(event?: string | symbol);
+  setMaxListeners(n: number);
+  getMaxListeners(): number;
+  listeners(event: string | symbol): Function[]; // eslint-disable-line @typescript-eslint/ban-types
+  rawListeners(event: string | symbol): Function[]; // eslint-disable-line @typescript-eslint/ban-types
+  emit(event: string | symbol, ...args: any[]): boolean;
+  listenerCount(event: string | symbol): number;
+}

test/connection-manager/index.node.js

@@ -3,6 +3,7 @@
 
 const { expect } = require('aegir/utils/chai')
 const sinon = require('sinon')
+const { CLOSED } = require('libp2p-interfaces/src/connection/status')
 
 const delay = require('delay')
 const pWaitFor = require('p-wait-for')
@@ -268,5 +269,40 @@ describe('libp2p.connections', () => {
 
       await libp2p.stop()
     })
+
+    it('should be closed status once immediately stopping', async () => {
+      const [libp2p] = await peerUtils.createPeer({
+        config: {
+          peerId: peerIds[0],
+          addresses: {
+            listen: ['/ip4/127.0.0.1/tcp/15003/ws']
+          },
+          modules: baseOptions.modules
+        }
+      })
+      const [remoteLibp2p] = await peerUtils.createPeer({
+        config: {
+          peerId: peerIds[1],
+          addresses: {
+            listen: ['/ip4/127.0.0.1/tcp/15004/ws']
+          },
+          modules: baseOptions.modules
+        }
+      })
+
+      libp2p.peerStore.addressBook.set(remoteLibp2p.peerId, remoteLibp2p.multiaddrs)
+      await libp2p.dial(remoteLibp2p.peerId)
+
+      const totalConns = Array.from(libp2p.connections.values())
+      expect(totalConns.length).to.eql(1)
+      const conns = totalConns[0]
+      expect(conns.length).to.eql(1)
+      const conn = conns[0]
+
+      await libp2p.stop()
+      expect(conn.stat.status).to.eql(CLOSED)
+
+      await remoteLibp2p.stop()
+    })
   })
 })

test/content-routing/content-routing.node.js

@@ -12,6 +12,8 @@ const CID = require('cids')
 const ipfsHttpClient = require('ipfs-http-client')
 const DelegatedContentRouter = require('libp2p-delegated-content-routing')
 const multiaddr = require('multiaddr')
+const drain = require('it-drain')
+const all = require('it-all')
 
 const peerUtils = require('../utils/creators/peer')
 const { baseOptions, routingOptions } = require('./utils')
@@ -78,10 +80,14 @@ describe('content-routing', () => {
 
     it('should use the nodes dht to find providers', async () => {
       const deferred = pDefer()
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
 
       sinon.stub(nodes[0]._dht, 'findProviders').callsFake(function * () {
         deferred.resolve()
-        yield
+        yield {
+          id: providerPeerId,
+          multiaddrs: []
+        }
       })
 
       await nodes[0].contentRouting.findProviders().next()
@@ -138,10 +144,14 @@ describe('content-routing', () => {
 
     it('should use the delegate router to find providers', async () => {
       const deferred = pDefer()
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
 
       sinon.stub(delegate, 'findProviders').callsFake(function * () {
         deferred.resolve()
-        yield
+        yield {
+          id: providerPeerId,
+          multiaddrs: []
+        }
       })
 
       await node.contentRouting.findProviders().next()
@@ -251,6 +261,110 @@ describe('content-routing', () => {
 
     afterEach(() => node.stop())
 
+    it('should store the multiaddrs of a peer', async () => {
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const result = {
+        id: providerPeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/49320')
+        ]
+      }
+
+      sinon.stub(node._dht, 'findProviders').callsFake(function * () {})
+      sinon.stub(delegate, 'findProviders').callsFake(function * () {
+        yield result
+      })
+
+      expect(node.peerStore.addressBook.get(providerPeerId)).to.not.be.ok()
+
+      await drain(node.contentRouting.findProviders('a cid'))
+
+      expect(node.peerStore.addressBook.get(providerPeerId)).to.deep.include({
+        isCertified: false,
+        multiaddr: result.multiaddrs[0]
+      })
+    })
+
+    it('should not wait for routing findProviders to finish before returning results', async () => {
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const result = {
+        id: providerPeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/49320')
+        ]
+      }
+
+      const defer = pDefer()
+
+      sinon.stub(node._dht, 'findProviders').callsFake(async function * () { // eslint-disable-line require-yield
+        await defer.promise
+      })
+      sinon.stub(delegate, 'findProviders').callsFake(async function * () {
+        yield result
+
+        await defer.promise
+      })
+
+      for await (const provider of node.contentRouting.findProviders('a cid')) {
+        expect(provider.id).to.deep.equal(providerPeerId)
+        defer.resolve()
+      }
+    })
+
+    it('should dedupe results', async () => {
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const result = {
+        id: providerPeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/49320')
+        ]
+      }
+
+      sinon.stub(node._dht, 'findProviders').callsFake(async function * () {
+        yield result
+      })
+      sinon.stub(delegate, 'findProviders').callsFake(async function * () {
+        yield result
+      })
+
+      const results = await all(node.contentRouting.findProviders('a cid'))
+
+      expect(results).to.be.an('array').with.lengthOf(1).that.deep.equals([result])
+    })
+
+    it('should combine multiaddrs when different addresses are returned by different content routers', async () => {
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const result1 = {
+        id: providerPeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/49320')
+        ]
+      }
+      const result2 = {
+        id: providerPeerId,
+        multiaddrs: [
+          multiaddr('/ip4/213.213.213.213/tcp/2344')
+        ]
+      }
+
+      sinon.stub(node._dht, 'findProviders').callsFake(async function * () {
+        yield result1
+      })
+      sinon.stub(delegate, 'findProviders').callsFake(async function * () {
+        yield result2
+      })
+
+      await drain(node.contentRouting.findProviders('a cid'))
+
+      expect(node.peerStore.addressBook.get(providerPeerId)).to.deep.include({
+        isCertified: false,
+        multiaddr: result1.multiaddrs[0]
+      }).and.to.deep.include({
+        isCertified: false,
+        multiaddr: result2.multiaddrs[0]
+      })
+    })
+
     it('should use both the dht and delegate router to provide', async () => {
       const dhtDeferred = pDefer()
       const delegatedDeferred = pDefer()
@@ -271,15 +385,18 @@ describe('content-routing', () => {
       ])
     })
 
-    it('should only use the dht if it finds providers', async () => {
-      const results = [true]
+    it('should use the dht if the delegate fails to find providers', async () => {
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = [{
+        id: providerPeerId,
+        multiaddrs: []
+      }]
 
       sinon.stub(node._dht, 'findProviders').callsFake(function * () {
         yield results[0]
       })
 
       sinon.stub(delegate, 'findProviders').callsFake(function * () { // eslint-disable-line require-yield
-        throw new Error('the delegate should not have been called')
       })
 
       const providers = []
@@ -292,7 +409,11 @@ describe('content-routing', () => {
     })
 
     it('should use the delegate if the dht fails to find providers', async () => {
-      const results = [true]
+      const [providerPeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = [{
+        id: providerPeerId,
+        multiaddrs: []
+      }]
 
       sinon.stub(node._dht, 'findProviders').callsFake(function * () {})
 

test/peer-routing/peer-routing.node.js

@@ -10,6 +10,8 @@ const delay = require('delay')
 const pDefer = require('p-defer')
 const pWaitFor = require('p-wait-for')
 const mergeOptions = require('merge-options')
+const drain = require('it-drain')
+const all = require('it-all')
 
 const ipfsHttpClient = require('ipfs-http-client')
 const DelegatedPeerRouter = require('libp2p-delegated-peer-routing')
@@ -82,10 +84,14 @@ describe('peer-routing', () => {
 
     it('should use the nodes dht to get the closest peers', async () => {
       const deferred = pDefer()
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
 
       sinon.stub(nodes[0]._dht, 'getClosestPeers').callsFake(function * () {
         deferred.resolve()
-        yield
+        yield {
+          id: remotePeerId,
+          multiaddrs: []
+        }
       })
 
       await nodes[0].peerRouting.getClosestPeers().next()
@@ -128,10 +134,14 @@ describe('peer-routing', () => {
 
     it('should use the delegate router to find peers', async () => {
       const deferred = pDefer()
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
 
       sinon.stub(delegate, 'findPeer').callsFake(() => {
         deferred.resolve()
-        return 'fake peer-id'
+        return {
+          id: remotePeerId,
+          multiaddrs: []
+        }
       })
 
       await node.peerRouting.findPeer()
@@ -140,10 +150,14 @@ describe('peer-routing', () => {
 
     it('should use the delegate router to get the closest peers', async () => {
       const deferred = pDefer()
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
 
       sinon.stub(delegate, 'getClosestPeers').callsFake(function * () {
         deferred.resolve()
-        yield
+        yield {
+          id: remotePeerId,
+          multiaddrs: []
+        }
       })
 
       await node.peerRouting.getClosestPeers().next()
@@ -152,7 +166,7 @@ describe('peer-routing', () => {
     })
 
     it('should be able to find a peer', async () => {
-      const peerKey = 'QmTp9VkYvnHyrqKQuFPiuZkiX9gPcqj6x5LJ1rmWuSySnL'
+      const peerKey = PeerId.createFromB58String('QmTp9VkYvnHyrqKQuFPiuZkiX9gPcqj6x5LJ1rmWuSySnL')
       const mockApi = nock('http://0.0.0.0:60197')
         .post('/api/v0/dht/findpeer')
         .query(true)
@@ -277,55 +291,93 @@ describe('peer-routing', () => {
 
     afterEach(() => node.stop())
 
-    it('should only use the dht if it finds the peer', async () => {
-      const dhtDeferred = pDefer()
-
-      sinon.stub(node._dht, 'findPeer').callsFake(() => {
-        dhtDeferred.resolve()
-        return { id: node.peerId }
-      })
-      sinon.stub(delegate, 'findPeer').callsFake(() => {
-        throw new Error('the delegate should not have been called')
-      })
-
-      await node.peerRouting.findPeer('a peer id')
-      await dhtDeferred.promise
-    })
-
     it('should use the delegate if the dht fails to find the peer', async () => {
-      const results = [true]
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = {
+        id: remotePeerId,
+        multiaddrs: []
+      }
 
       sinon.stub(node._dht, 'findPeer').callsFake(() => {})
       sinon.stub(delegate, 'findPeer').callsFake(() => {
         return results
       })
 
-      const peer = await node.peerRouting.findPeer('a peer id')
+      const peer = await node.peerRouting.findPeer(remotePeerId)
       expect(peer).to.eql(results)
     })
 
-    it('should only use the dht if it gets the closest peers', async () => {
-      const results = [true]
-
-      sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
-        yield results[0]
-      })
-
-      sinon.stub(delegate, 'getClosestPeers').callsFake(function * () { // eslint-disable-line require-yield
-        throw new Error('the delegate should not have been called')
-      })
-
-      const closest = []
-      for await (const peer of node.peerRouting.getClosestPeers('a cid')) {
-        closest.push(peer)
+    it('should not wait for the dht to return if the delegate does first', async () => {
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = {
+        id: remotePeerId,
+        multiaddrs: []
       }
 
-      expect(closest).to.have.length.above(0)
-      expect(closest).to.eql(results)
+      const defer = pDefer()
+
+      sinon.stub(node._dht, 'findPeer').callsFake(async () => {
+        await defer.promise
+      })
+      sinon.stub(delegate, 'findPeer').callsFake(() => {
+        return results
+      })
+
+      const peer = await node.peerRouting.findPeer(remotePeerId)
+      expect(peer).to.eql(results)
+
+      defer.resolve()
+    })
+
+    it('should not wait for the delegate to return if the dht does first', async () => {
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = {
+        id: remotePeerId,
+        multiaddrs: []
+      }
+
+      const defer = pDefer()
+
+      sinon.stub(node._dht, 'findPeer').callsFake(() => {
+        return results
+      })
+      sinon.stub(delegate, 'findPeer').callsFake(async () => {
+        await defer.promise
+      })
+
+      const peer = await node.peerRouting.findPeer(remotePeerId)
+      expect(peer).to.eql(results)
+
+      defer.resolve()
+    })
+
+    it('should store the addresses of the found peer', async () => {
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = {
+        id: remotePeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/38982')
+        ]
+      }
+
+      const spy = sinon.spy(node.peerStore.addressBook, 'add')
+
+      sinon.stub(node._dht, 'findPeer').callsFake(() => {
+        return results
+      })
+      sinon.stub(delegate, 'findPeer').callsFake(() => {})
+
+      await node.peerRouting.findPeer(remotePeerId)
+
+      expect(spy.calledWith(results.id, results.multiaddrs)).to.be.true()
     })
 
     it('should use the delegate if the dht fails to get the closest peer', async () => {
-      const results = [true]
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = [{
+        id: remotePeerId,
+        multiaddrs: []
+      }]
 
       sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () { })
 
@@ -333,14 +385,55 @@ describe('peer-routing', () => {
         yield results[0]
       })
 
-      const closest = []
-      for await (const peer of node.peerRouting.getClosestPeers('a cid')) {
-        closest.push(peer)
-      }
+      const closest = await all(node.peerRouting.getClosestPeers('a cid'))
 
       expect(closest).to.have.length.above(0)
       expect(closest).to.eql(results)
     })
+
+    it('should store the addresses of the closest peer', async () => {
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const result = {
+        id: remotePeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/38982')
+        ]
+      }
+
+      const spy = sinon.spy(node.peerStore.addressBook, 'add')
+
+      sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () { })
+
+      sinon.stub(delegate, 'getClosestPeers').callsFake(function * () {
+        yield result
+      })
+
+      await drain(node.peerRouting.getClosestPeers('a cid'))
+
+      expect(spy.calledWith(result.id, result.multiaddrs)).to.be.true()
+    })
+
+    it('should dedupe closest peers', async () => {
+      const [remotePeerId] = await peerUtils.createPeerId({ fixture: false })
+      const results = [{
+        id: remotePeerId,
+        multiaddrs: [
+          multiaddr('/ip4/123.123.123.123/tcp/38982')
+        ]
+      }]
+
+      sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
+        yield * results
+      })
+
+      sinon.stub(delegate, 'getClosestPeers').callsFake(function * () {
+        yield * results
+      })
+
+      const peers = await all(node.peerRouting.getClosestPeers('a cid'))
+
+      expect(peers).to.be.an('array').with.a.lengthOf(1).that.deep.equals(results)
+    })
   })
 
   describe('peer routing refresh manager service', () => {