Compare commits

...

8 Commits

Author SHA1 Message Date
achingbrain
c7e923a812 fix: remove peer routing search-for-self
The peer routing module starts a recurring process that searches for
peers close to our peer id.

This makes the DHT module query the network for peers.  Thing is the
DHT module is already doing this because periodically searching for
peers close to us is in the DHT spec so this ends up making redundant
queries.

This PR removes the recurring task configured by the peer routing module.
2021-12-06 19:15:20 +00:00
achingbrain
ae21299ade chore: release version v0.35.1 2021-12-03 16:24:14 +00:00
achingbrain
149120bebc chore: update contributors 2021-12-03 16:24:13 +00:00
Alex Potsides
91c2ec9856 fix: do not let closest peers run forever (#1047)
The DHT takes a `signal` not a timeout so if a timeout is passed,
create a `TimeoutController` that will abort the query after the
timeout.
2021-12-03 15:47:30 +00:00
achingbrain
6d0ac819f1 chore: release version v0.35.0 2021-12-02 10:44:07 +00:00
achingbrain
15a0b1dbf2 chore: update contributors 2021-12-02 10:44:07 +00:00
Alex Potsides
9cbf36fcb5 chore: update peer id and libp2p crypto (#1042)
BREAKING CHANGE: requires node 15+
2021-12-02 10:11:23 +00:00
Alex Potsides
3a9d5f64d9 fix: stop dht before connection manager (#1041)
Stop the dht before the connection manager, otherwise in-flight eviction pings fail and we move on to the next one when we should just abort them all.

Also pulls in the fix from #1039 and splits the auto-dialler out from the connection manager as during shutdown it can get into a weird state where it's simultaneously killing and creating connections so stop auto-dialling things before we cause connections to dip below the low watermark by killing existing connections.

Fixes: https://github.com/ipfs/js-ipfs/issues/3923
2021-11-30 18:07:57 +00:00
17 changed files with 203 additions and 279 deletions

View File

@@ -14,7 +14,7 @@ jobs:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
- uses: actions/setup-node@v2 - uses: actions/setup-node@v2
with: with:
node-version: 14 node-version: 16
- run: npm install - run: npm install
- run: npx aegir lint - run: npx aegir lint
- run: npx aegir build - run: npx aegir build
@@ -29,7 +29,7 @@ jobs:
strategy: strategy:
matrix: matrix:
os: [windows-latest, ubuntu-latest, macos-latest] os: [windows-latest, ubuntu-latest, macos-latest]
node: [14, 16] node: [16]
fail-fast: true fail-fast: true
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
@@ -44,6 +44,9 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
- uses: actions/setup-node@v2
with:
node-version: lts/*
- run: npm install - run: npm install
- run: npx aegir test -t browser -t webworker --bail - run: npx aegir test -t browser -t webworker --bail
test-firefox: test-firefox:
@@ -51,6 +54,9 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
- uses: actions/setup-node@v2
with:
node-version: lts/*
- run: npm install - run: npm install
- run: npx aegir test -t browser -t webworker --bail -- --browser firefox - run: npx aegir test -t browser -t webworker --bail -- --browser firefox
test-ts: test-ts:
@@ -58,6 +64,9 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
- uses: actions/setup-node@v2
with:
node-version: lts/*
- run: npm install - run: npm install
- run: npm run test:ts - run: npm run test:ts
test-interop: test-interop:
@@ -65,5 +74,8 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v2 - uses: actions/checkout@v2
- uses: actions/setup-node@v2
with:
node-version: lts/*
- run: npm install - run: npm install
- run: npm run test:interop -- --bail -- --exit - run: npm run test:interop -- --bail -- --exit

View File

@@ -1,3 +1,31 @@
## [0.35.1](https://github.com/libp2p/js-libp2p/compare/v0.35.0...v0.35.1) (2021-12-03)
### Bug Fixes
* do not let closest peers run forever ([#1047](https://github.com/libp2p/js-libp2p/issues/1047)) ([91c2ec9](https://github.com/libp2p/js-libp2p/commit/91c2ec9856a3e972b7b2c9c4d9a4eda1d431c7ef))
# [0.35.0](https://github.com/libp2p/js-libp2p/compare/v0.34.0...v0.35.0) (2021-12-02)
### Bug Fixes
* stop dht before connection manager ([#1041](https://github.com/libp2p/js-libp2p/issues/1041)) ([3a9d5f6](https://github.com/libp2p/js-libp2p/commit/3a9d5f64d96719ebb4d3b083c4f5832db4fa0816)), closes [#1039](https://github.com/libp2p/js-libp2p/issues/1039)
### chore
* update peer id and libp2p crypto ([#1042](https://github.com/libp2p/js-libp2p/issues/1042)) ([9cbf36f](https://github.com/libp2p/js-libp2p/commit/9cbf36fcb54099e6fed35ceccc4a2376f0926c1f))
### BREAKING CHANGES
* requires node 15+
# [0.34.0](https://github.com/libp2p/js-libp2p/compare/v0.33.0...v0.34.0) (2021-11-25) # [0.34.0](https://github.com/libp2p/js-libp2p/compare/v0.33.0...v0.34.0) (2021-11-25)

View File

@@ -415,14 +415,7 @@ const node = await Libp2p.create({
contentRouting: [delegatedContentRouting], contentRouting: [delegatedContentRouting],
peerRouting: [delegatedPeerRouting], peerRouting: [delegatedPeerRouting],
}, },
peerId, peerId
peerRouting: { // Peer routing configuration
refreshManager: { // Refresh known and connected closest peers
enabled: true, // Should find the closest peers.
interval: 6e5, // Interval for getting the new for closest peers of 10min
bootDelay: 10e3 // Delay for the initial query for closest peers
}
}
}) })
``` ```

View File

@@ -24,7 +24,7 @@
"libp2p-mplex": "^0.10.4", "libp2p-mplex": "^0.10.4",
"@chainsafe/libp2p-noise": "^4.1.0", "@chainsafe/libp2p-noise": "^4.1.0",
"libp2p-webrtc-direct": "^0.7.0", "libp2p-webrtc-direct": "^0.7.0",
"peer-id": "^0.15.4" "peer-id": "^0.16.0"
}, },
"browser": { "browser": {
"ipfs": "ipfs/dist/index.min.js" "ipfs": "ipfs/dist/index.min.js"

View File

@@ -1,6 +1,6 @@
{ {
"name": "libp2p", "name": "libp2p",
"version": "0.34.0", "version": "0.35.1",
"description": "JavaScript implementation of libp2p, a modular peer to peer network stack", "description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
"leadMaintainer": "Jacob Heun <jacobheun@gmail.com>", "leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
"main": "src/index.js", "main": "src/index.js",
@@ -66,7 +66,7 @@
"homepage": "https://libp2p.io", "homepage": "https://libp2p.io",
"license": "MIT", "license": "MIT",
"engines": { "engines": {
"node": ">=14.0.0" "node": ">=15.0.0"
}, },
"browser": { "browser": {
"@motrix/nat-api": false "@motrix/nat-api": false
@@ -80,10 +80,9 @@
] ]
}, },
"dependencies": { "dependencies": {
"abortable-iterator": "^3.0.0",
"@motrix/nat-api": "^0.3.1", "@motrix/nat-api": "^0.3.1",
"@vascosantos/moving-average": "^1.1.0", "@vascosantos/moving-average": "^1.1.0",
"abort-controller": "^3.0.0", "abortable-iterator": "^3.0.0",
"aggregate-error": "^3.1.0", "aggregate-error": "^3.1.0",
"any-signal": "^2.1.1", "any-signal": "^2.1.1",
"bignumber.js": "^9.0.1", "bignumber.js": "^9.0.1",
@@ -105,8 +104,8 @@
"it-merge": "^1.0.0", "it-merge": "^1.0.0",
"it-pipe": "^1.1.0", "it-pipe": "^1.1.0",
"it-take": "^1.0.0", "it-take": "^1.0.0",
"libp2p-crypto": "^0.20.0", "libp2p-crypto": "^0.21.0",
"libp2p-interfaces": "^1.3.1", "libp2p-interfaces": "^2.0.1",
"libp2p-utils": "^0.4.0", "libp2p-utils": "^0.4.0",
"mafmt": "^10.0.0", "mafmt": "^10.0.0",
"merge-options": "^3.0.4", "merge-options": "^3.0.4",
@@ -119,14 +118,14 @@
"p-fifo": "^1.0.0", "p-fifo": "^1.0.0",
"p-retry": "^4.4.0", "p-retry": "^4.4.0",
"p-settle": "^4.1.1", "p-settle": "^4.1.1",
"peer-id": "^0.15.4", "peer-id": "^0.16.0",
"private-ip": "^2.1.0", "private-ip": "^2.1.0",
"protobufjs": "^6.10.2", "protobufjs": "^6.10.2",
"retimer": "^3.0.0", "retimer": "^3.0.0",
"sanitize-filename": "^1.6.3", "sanitize-filename": "^1.6.3",
"set-delayed-interval": "^1.0.0", "set-delayed-interval": "^1.0.0",
"streaming-iterables": "^6.0.0", "streaming-iterables": "^6.0.0",
"timeout-abort-controller": "^1.1.1", "timeout-abort-controller": "^2.0.0",
"uint8arrays": "^3.0.0", "uint8arrays": "^3.0.0",
"varint": "^6.0.0", "varint": "^6.0.0",
"wherearewe": "^1.0.0", "wherearewe": "^1.0.0",
@@ -136,7 +135,6 @@
"@chainsafe/libp2p-noise": "^4.0.0", "@chainsafe/libp2p-noise": "^4.0.0",
"@nodeutils/defaults-deep": "^1.1.0", "@nodeutils/defaults-deep": "^1.1.0",
"@types/es6-promisify": "^6.0.0", "@types/es6-promisify": "^6.0.0",
"@types/node": "^16.0.1",
"@types/node-forge": "^0.10.1", "@types/node-forge": "^0.10.1",
"@types/varint": "^6.0.0", "@types/varint": "^6.0.0",
"aegir": "^36.0.0", "aegir": "^36.0.0",
@@ -149,25 +147,25 @@
"it-pair": "^1.0.0", "it-pair": "^1.0.0",
"it-pushable": "^1.4.0", "it-pushable": "^1.4.0",
"libp2p": ".", "libp2p": ".",
"libp2p-bootstrap": "^0.13.0", "libp2p-bootstrap": "^0.14.0",
"libp2p-delegated-content-routing": "^0.11.0", "libp2p-delegated-content-routing": "^0.11.0",
"libp2p-delegated-peer-routing": "^0.10.0", "libp2p-delegated-peer-routing": "^0.11.0",
"libp2p-floodsub": "^0.27.0", "libp2p-floodsub": "^0.27.0",
"libp2p-gossipsub": "^0.11.0", "libp2p-gossipsub": "^0.11.0",
"libp2p-interfaces-compliance-tests": "^1.2.1", "libp2p-interfaces-compliance-tests": "^2.0.1",
"libp2p-interop": "^0.5.0", "libp2p-interop": "^0.5.0",
"libp2p-kad-dht": "^0.26.6", "libp2p-kad-dht": "^0.27.1",
"libp2p-mdns": "^0.17.0", "libp2p-mdns": "^0.18.0",
"libp2p-mplex": "^0.10.1", "libp2p-mplex": "^0.10.1",
"libp2p-tcp": "^0.17.0", "libp2p-tcp": "^0.17.0",
"libp2p-webrtc-star": "^0.24.0", "libp2p-webrtc-star": "^0.25.0",
"libp2p-websockets": "^0.16.0", "libp2p-websockets": "^0.16.0",
"nock": "^13.0.3", "nock": "^13.0.3",
"p-defer": "^3.0.0", "p-defer": "^3.0.0",
"p-times": "^3.0.0", "p-times": "^3.0.0",
"p-wait-for": "^3.2.0", "p-wait-for": "^3.2.0",
"rimraf": "^3.0.2", "rimraf": "^3.0.2",
"sinon": "^11.1.1", "sinon": "^12.0.1",
"util": "^0.12.3" "util": "^0.12.3"
}, },
"contributors": [ "contributors": [
@@ -181,24 +179,23 @@
"Friedel Ziegelmayer <dignifiedquire@gmail.com>", "Friedel Ziegelmayer <dignifiedquire@gmail.com>",
"Maciej Krüger <mkg20001@gmail.com>", "Maciej Krüger <mkg20001@gmail.com>",
"Hugo Dias <mail@hugodias.me>", "Hugo Dias <mail@hugodias.me>",
"Chris Dostert <chrisdostert@users.noreply.github.com>",
"dirkmc <dirkmdev@gmail.com>", "dirkmc <dirkmdev@gmail.com>",
"Chris Dostert <chrisdostert@users.noreply.github.com>",
"Volker Mische <volker.mische@gmail.com>", "Volker Mische <volker.mische@gmail.com>",
"zeim839 <50573884+zeim839@users.noreply.github.com>", "zeim839 <50573884+zeim839@users.noreply.github.com>",
"Richard Littauer <richard.littauer@gmail.com>", "Richard Littauer <richard.littauer@gmail.com>",
"a1300 <matthias-knopp@gmx.net>", "a1300 <matthias-knopp@gmx.net>",
"Ryan Bell <ryan@piing.net>", "Ryan Bell <ryan@piing.net>",
"Elven <mon.samuel@qq.com>",
"ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com>", "ᴠɪᴄᴛᴏʀ ʙᴊᴇʟᴋʜᴏʟᴍ <victorbjelkholm@gmail.com>",
"Thomas Eizinger <thomas@eizinger.io>", "Giovanni T. Parra <fiatjaf@gmail.com>",
"acolytec3 <17355484+acolytec3@users.noreply.github.com>",
"Franck Royer <franck@royer.one>",
"Elven <mon.samuel@qq.com>",
"Robert Kiel <robert.kiel@hoprnet.org>", "Robert Kiel <robert.kiel@hoprnet.org>",
"Andrew Nesbitt <andrewnez@gmail.com>", "Andrew Nesbitt <andrewnez@gmail.com>",
"Samlior <samlior@foxmail.com>", "Samlior <samlior@foxmail.com>",
"acolytec3 <17355484+acolytec3@users.noreply.github.com>", "Thomas Eizinger <thomas@eizinger.io>",
"Franck Royer <franck@royer.one>",
"Giovanni T. Parra <fiatjaf@gmail.com>",
"Didrik Nordström <didrik.nordstrom@gmail.com>", "Didrik Nordström <didrik.nordstrom@gmail.com>",
"RasmusErik Voel Jensen <github@solsort.com>",
"Smite Chow <xiaopengyou@live.com>", "Smite Chow <xiaopengyou@live.com>",
"Soeren <nikorpoulsen@gmail.com>", "Soeren <nikorpoulsen@gmail.com>",
"Sönke Hahn <soenkehahn@gmail.com>", "Sönke Hahn <soenkehahn@gmail.com>",
@@ -247,6 +244,7 @@
"Michael Burns <5170+mburns@users.noreply.github.com>", "Michael Burns <5170+mburns@users.noreply.github.com>",
"Miguel Mota <miguelmota2@gmail.com>", "Miguel Mota <miguelmota2@gmail.com>",
"Nuno Nogueira <nunofmn@gmail.com>", "Nuno Nogueira <nunofmn@gmail.com>",
"Philipp Muens <raute1337@gmx.de>" "Philipp Muens <raute1337@gmx.de>",
"RasmusErik Voel Jensen <github@solsort.com>"
] ]
} }

View File

@@ -49,24 +49,11 @@ const DefaultConfig = {
persistence: false, persistence: false,
threshold: 5 threshold: 5
}, },
peerRouting: {
refreshManager: {
enabled: true,
interval: 6e5,
bootDelay: 10e3
}
},
config: { config: {
protocolPrefix: 'ipfs', protocolPrefix: 'ipfs',
dht: { dht: {
enabled: false, enabled: false,
kBucketSize: 20, kBucketSize: 20
randomWalk: {
enabled: false, // disabled waiting for https://github.com/libp2p/js-libp2p-kad-dht/issues/86
queriesPerPeriod: 1,
interval: 300e3,
timeout: 10e3
}
}, },
nat: { nat: {
enabled: true, enabled: true,

View File

@@ -0,0 +1,118 @@
'use strict'
const debug = require('debug')
const mergeOptions = require('merge-options')
// @ts-ignore retimer does not have types
const retimer = require('retimer')
const log = Object.assign(debug('libp2p:connection-manager:auto-dialler'), {
error: debug('libp2p:connection-manager:auto-dialler:err')
})
const defaultOptions = {
enabled: true,
minConnections: 0,
autoDialInterval: 10000
}
/**
* @typedef {import('../index')} Libp2p
* @typedef {import('libp2p-interfaces/src/connection').Connection} Connection
*/
/**
* @typedef {Object} AutoDiallerOptions
* @property {boolean} [enabled = true] - Should preemptively guarantee connections are above the low watermark
* @property {number} [minConnections = 0] - The minimum number of connections to avoid pruning
* @property {number} [autoDialInterval = 10000] - How often, in milliseconds, it should preemptively guarantee connections are above the low watermark
*/
class AutoDialler {
/**
* Proactively tries to connect to known peers stored in the PeerStore.
* It will keep the number of connections below the upper limit and sort
* the peers to connect based on wether we know their keys and protocols.
*
* @class
* @param {Libp2p} libp2p
* @param {AutoDiallerOptions} options
*/
constructor (libp2p, options = {}) {
this._options = mergeOptions.call({ ignoreUndefined: true }, defaultOptions, options)
this._libp2p = libp2p
this._running = false
this._autoDialTimeout = null
this._autoDial = this._autoDial.bind(this)
log('options: %j', this._options)
}
/**
* Starts the auto dialer
*/
start () {
if (!this._options.enabled) {
log('not enabled')
return
}
this._running = true
this._autoDial()
log('started')
}
/**
* Stops the auto dialler
*/
async stop () {
if (!this._options.enabled) {
log('not enabled')
return
}
this._running = false
this._autoDialTimeout && this._autoDialTimeout.clear()
log('stopped')
}
async _autoDial () {
const minConnections = this._options.minConnections
// Already has enough connections
if (this._libp2p.connections.size >= minConnections) {
this._autoDialTimeout = retimer(this._autoDial, this._options.autoDialInterval)
return
}
// Sort peers on wether we know protocols of public keys for them
const peers = Array.from(this._libp2p.peerStore.peers.values())
.sort((a, b) => {
if (b.protocols && b.protocols.length && (!a.protocols || !a.protocols.length)) {
return 1
} else if (b.id.pubKey && !a.id.pubKey) {
return 1
}
return -1
})
for (let i = 0; this._running && i < peers.length && this._libp2p.connections.size < minConnections; i++) {
if (!this._libp2p.connectionManager.get(peers[i].id)) {
log('connecting to a peerStore stored peer %s', peers[i].id.toB58String())
try {
await this._libp2p.dialer.connectToPeer(peers[i].id)
} catch (/** @type {any} */ err) {
log.error('could not connect to peerStore stored peer', err)
}
}
}
// Connection Manager was stopped
if (!this._running) {
return
}
this._autoDialTimeout = retimer(this._autoDial, this._options.autoDialInterval)
}
}
module.exports = AutoDialler

View File

@@ -94,9 +94,7 @@ class ConnectionManager extends EventEmitter {
this._started = false this._started = false
this._timer = null this._timer = null
this._autoDialTimeout = null
this._checkMetrics = this._checkMetrics.bind(this) this._checkMetrics = this._checkMetrics.bind(this)
this._autoDial = this._autoDial.bind(this)
this._latencyMonitor = new LatencyMonitor({ this._latencyMonitor = new LatencyMonitor({
latencyCheckIntervalMs: this._options.pollInterval, latencyCheckIntervalMs: this._options.pollInterval,
@@ -128,8 +126,6 @@ class ConnectionManager extends EventEmitter {
this._started = true this._started = true
log('started') log('started')
this._options.autoDial && this._autoDial()
} }
/** /**
@@ -138,7 +134,6 @@ class ConnectionManager extends EventEmitter {
* @async * @async
*/ */
async stop () { async stop () {
this._autoDialTimeout && this._autoDialTimeout.clear()
this._timer && this._timer.clear() this._timer && this._timer.clear()
this._latencyMonitor.removeListener('data', this._onLatencyMeasure) this._latencyMonitor.removeListener('data', this._onLatencyMeasure)
@@ -312,53 +307,6 @@ class ConnectionManager extends EventEmitter {
} }
} }
/**
* Proactively tries to connect to known peers stored in the PeerStore.
* It will keep the number of connections below the upper limit and sort
* the peers to connect based on wether we know their keys and protocols.
*
* @async
* @private
*/
async _autoDial () {
const minConnections = this._options.minConnections
// Already has enough connections
if (this.size >= minConnections) {
this._autoDialTimeout = retimer(this._autoDial, this._options.autoDialInterval)
return
}
// Sort peers on wether we know protocols of public keys for them
const peers = Array.from(this._libp2p.peerStore.peers.values())
.sort((a, b) => {
if (b.protocols && b.protocols.length && (!a.protocols || !a.protocols.length)) {
return 1
} else if (b.id.pubKey && !a.id.pubKey) {
return 1
}
return -1
})
for (let i = 0; i < peers.length && this.size < minConnections; i++) {
if (!this.get(peers[i].id)) {
log('connecting to a peerStore stored peer %s', peers[i].id.toB58String())
try {
await this._libp2p.dialer.connectToPeer(peers[i].id)
// Connection Manager was stopped
if (!this._started) {
return
}
} catch (/** @type {any} */ err) {
log.error('could not connect to peerStore stored peer', err)
}
}
}
this._autoDialTimeout = retimer(this._autoDial, this._options.autoDialInterval)
}
/** /**
* If we have more connections than our maximum, close a connection * If we have more connections than our maximum, close a connection
* to the lowest valued peer. * to the lowest valued peer.

View File

@@ -1,7 +1,6 @@
'use strict' 'use strict'
const errCode = require('err-code') const errCode = require('err-code')
const AbortController = require('abort-controller').default
const { anySignal } = require('any-signal') const { anySignal } = require('any-signal')
// @ts-ignore p-fifo does not export types // @ts-ignore p-fifo does not export types
const FIFO = require('p-fifo') const FIFO = require('p-fifo')

View File

@@ -6,8 +6,7 @@ const log = Object.assign(debug('libp2p:dialer'), {
}) })
const errCode = require('err-code') const errCode = require('err-code')
const { Multiaddr } = require('multiaddr') const { Multiaddr } = require('multiaddr')
// @ts-ignore timeout-abourt-controles does not export types const { TimeoutController } = require('timeout-abort-controller')
const TimeoutController = require('timeout-abort-controller')
const { AbortError } = require('abortable-iterator') const { AbortError } = require('abortable-iterator')
const { anySignal } = require('any-signal') const { anySignal } = require('any-signal')

View File

@@ -18,6 +18,7 @@ const { codes, messages } = require('./errors')
const AddressManager = require('./address-manager') const AddressManager = require('./address-manager')
const ConnectionManager = require('./connection-manager') const ConnectionManager = require('./connection-manager')
const AutoDialler = require('./connection-manager/auto-dialler')
const Circuit = require('./circuit/transport') const Circuit = require('./circuit/transport')
const Relay = require('./circuit') const Relay = require('./circuit')
const Dialer = require('./dialer') const Dialer = require('./dialer')
@@ -193,9 +194,13 @@ class Libp2p extends EventEmitter {
// Create the Connection Manager // Create the Connection Manager
this.connectionManager = new ConnectionManager(this, { this.connectionManager = new ConnectionManager(this, {
autoDial: this._config.peerDiscovery.autoDial,
...this._options.connectionManager ...this._options.connectionManager
}) })
this._autodialler = new AutoDialler(this, {
enabled: this._config.peerDiscovery.autoDial,
minConnections: this._options.connectionManager.minConnections,
autoDialInterval: this._options.connectionManager.autoDialInterval
})
// Create Metrics // Create Metrics
if (this._options.metrics.enabled) { if (this._options.metrics.enabled) {
@@ -379,7 +384,8 @@ class Libp2p extends EventEmitter {
this._isStarted = false this._isStarted = false
this.relay && this.relay.stop() this.relay && this.relay.stop()
this.peerRouting.stop() this._autodialler.stop()
await (this._dht && this._dht.stop())
for (const service of this._discovery.values()) { for (const service of this._discovery.values()) {
service.removeListener('peer', this._onDiscoveryPeer) service.removeListener('peer', this._onDiscoveryPeer)
@@ -394,7 +400,6 @@ class Libp2p extends EventEmitter {
await Promise.all([ await Promise.all([
this.pubsub && this.pubsub.stop(), this.pubsub && this.pubsub.stop(),
this._dht && this._dht.stop(),
this.metrics && this.metrics.stop() this.metrics && this.metrics.stop()
]) ])
@@ -650,14 +655,13 @@ class Libp2p extends EventEmitter {
} }
this.connectionManager.start() this.connectionManager.start()
this._autodialler.start()
// Peer discovery // Peer discovery
await this._setupPeerDiscovery() await this._setupPeerDiscovery()
// Relay // Relay
this.relay && this.relay.start() this.relay && this.relay.start()
this.peerRouting.start()
} }
/** /**

View File

@@ -1,26 +1,17 @@
'use strict' 'use strict'
const debug = require('debug')
const log = Object.assign(debug('libp2p:peer-routing'), {
error: debug('libp2p:peer-routing:err')
})
const errCode = require('err-code') const errCode = require('err-code')
const { const {
storeAddresses, storeAddresses,
uniquePeers, uniquePeers,
requirePeers requirePeers
} = require('./content-routing/utils') } = require('./content-routing/utils')
const { TimeoutController } = require('timeout-abort-controller')
const merge = require('it-merge') const merge = require('it-merge')
const { pipe } = require('it-pipe') const { pipe } = require('it-pipe')
const first = require('it-first') const first = require('it-first')
const drain = require('it-drain')
const filter = require('it-filter') const filter = require('it-filter')
const {
setDelayedInterval,
clearDelayedInterval
// @ts-ignore module with no types
} = require('set-delayed-interval')
const { DHTPeerRouting } = require('./dht/dht-peer-routing') const { DHTPeerRouting } = require('./dht/dht-peer-routing')
/** /**
@@ -30,13 +21,7 @@ const { DHTPeerRouting } = require('./dht/dht-peer-routing')
*/ */
/** /**
* @typedef {Object} RefreshManagerOptions
* @property {boolean} [enabled = true] - Whether to enable the Refresh manager
* @property {number} [bootDelay = 6e5] - Boot delay to start the Refresh Manager (in ms)
* @property {number} [interval = 10e3] - Interval between each Refresh Manager run (in ms)
*
* @typedef {Object} PeerRoutingOptions * @typedef {Object} PeerRoutingOptions
* @property {RefreshManagerOptions} [refreshManager]
*/ */
class PeerRouting { class PeerRouting {
@@ -54,42 +39,6 @@ class PeerRouting {
if (libp2p._dht && libp2p._config.dht.enabled) { if (libp2p._dht && libp2p._config.dht.enabled) {
this._routers.push(new DHTPeerRouting(libp2p._dht)) this._routers.push(new DHTPeerRouting(libp2p._dht))
} }
this._refreshManagerOptions = libp2p._options.peerRouting.refreshManager
this._findClosestPeersTask = this._findClosestPeersTask.bind(this)
}
/**
* Start peer routing service.
*/
start () {
if (!this._routers.length || this._timeoutId || !this._refreshManagerOptions.enabled) {
return
}
this._timeoutId = setDelayedInterval(
this._findClosestPeersTask, this._refreshManagerOptions.interval, this._refreshManagerOptions.bootDelay
)
}
/**
* Recurrent task to find closest peers and add their addresses to the Address Book.
*/
async _findClosestPeersTask () {
try {
// nb getClosestPeers adds the addresses to the address book
await drain(this.getClosestPeers(this._peerId.id))
} catch (/** @type {any} */ err) {
log.error(err)
}
}
/**
* Stop peer routing service.
*/
stop () {
clearDelayedInterval(this._timeoutId)
} }
/** /**
@@ -131,7 +80,8 @@ class PeerRouting {
* *
* @param {Uint8Array} key - A CID like key * @param {Uint8Array} key - A CID like key
* @param {Object} [options] * @param {Object} [options]
* @param {number} [options.timeout=30e3] - How long the query can take. * @param {number} [options.timeout=30e3] - How long the query can take
* @param {AbortSignal} [options.signal] - An AbortSignal to abort the request
* @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>} * @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
*/ */
async * getClosestPeers (key, options = { timeout: 30e3 }) { async * getClosestPeers (key, options = { timeout: 30e3 }) {
@@ -139,6 +89,10 @@ class PeerRouting {
throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE') throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
} }
if (options.timeout) {
options.signal = new TimeoutController(options.timeout).signal
}
yield * pipe( yield * pipe(
merge( merge(
...this._routers.map(router => router.getClosestPeers(key, options)) ...this._routers.map(router => router.getClosestPeers(key, options))

View File

@@ -5,7 +5,6 @@ const { expect } = require('aegir/utils/chai')
const sinon = require('sinon') const sinon = require('sinon')
const { AbortError } = require('libp2p-interfaces/src/transport/errors') const { AbortError } = require('libp2p-interfaces/src/transport/errors')
const AbortController = require('abort-controller')
const AggregateError = require('aggregate-error') const AggregateError = require('aggregate-error')
const pDefer = require('p-defer') const pDefer = require('p-defer')
const delay = require('delay') const delay = require('delay')

View File

@@ -6,9 +6,7 @@ const nock = require('nock')
const sinon = require('sinon') const sinon = require('sinon')
const intoStream = require('into-stream') const intoStream = require('into-stream')
const delay = require('delay')
const pDefer = require('p-defer') const pDefer = require('p-defer')
const pWaitFor = require('p-wait-for')
const mergeOptions = require('merge-options') const mergeOptions = require('merge-options')
const drain = require('it-drain') const drain = require('it-drain')
const all = require('it-all') const all = require('it-all')
@@ -453,109 +451,4 @@ describe('peer-routing', () => {
expect(peers).to.be.an('array').with.a.lengthOf(1).that.deep.equals(results) expect(peers).to.be.an('array').with.a.lengthOf(1).that.deep.equals(results)
}) })
}) })
describe('peer routing refresh manager service', () => {
let node
let peerIds
before(async () => {
peerIds = await peerUtils.createPeerId({ number: 2 })
})
afterEach(() => {
sinon.restore()
return node && node.stop()
})
it('should be enabled and start by default', async () => {
const results = [
{ id: peerIds[0], multiaddrs: [new Multiaddr('/ip4/30.0.0.1/tcp/2000')] },
{ id: peerIds[1], multiaddrs: [new Multiaddr('/ip4/32.0.0.1/tcp/2000')] }
]
;[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
bootDelay: 100
}
}
}),
started: false
})
sinon.spy(node.peerStore.addressBook, 'add')
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield results[0]
yield results[1]
})
await node.start()
await pWaitFor(() => node._dht.getClosestPeers.callCount === 1)
await pWaitFor(() => node.peerStore.addressBook.add.callCount === results.length)
const call0 = node.peerStore.addressBook.add.getCall(0)
expect(call0.args[0].equals(results[0].id))
call0.args[1].forEach((m, index) => {
expect(m.equals(results[0].multiaddrs[index]))
})
const call1 = node.peerStore.addressBook.add.getCall(1)
expect(call1.args[0].equals(results[1].id))
call0.args[1].forEach((m, index) => {
expect(m.equals(results[1].multiaddrs[index]))
})
})
it('should support being disabled', async () => {
[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
bootDelay: 100,
enabled: false
}
}
}),
started: false
})
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield
throw new Error('should not be called')
})
await node.start()
await delay(100)
expect(node._dht.getClosestPeers.callCount === 0)
})
it('should start and run recurrently on interval', async () => {
[node] = await peerUtils.createPeer({
config: mergeOptions(routingOptions, {
peerRouting: {
refreshManager: {
interval: 500,
bootDelay: 200
}
}
}),
started: false
})
sinon.stub(node._dht, 'getClosestPeers').callsFake(function * () {
yield { id: peerIds[0], multiaddrs: [new Multiaddr('/ip4/30.0.0.1/tcp/2000')] }
})
await node.start()
await delay(300)
expect(node._dht.getClosestPeers.callCount).to.eql(1)
await delay(500)
expect(node._dht.getClosestPeers.callCount).to.eql(2)
})
})
}) })

View File

@@ -16,7 +16,7 @@
"libp2p-record": "^0.10.4", "libp2p-record": "^0.10.4",
"libp2p-tcp": "^0.17.1", "libp2p-tcp": "^0.17.1",
"libp2p-websockets": "^0.16.1", "libp2p-websockets": "^0.16.1",
"peer-id": "^0.15.4" "peer-id": "^0.16.0"
}, },
"scripts": { "scripts": {
"build": "npx tsc", "build": "npx tsc",

View File

@@ -62,13 +62,6 @@ async function main() {
contentRouting: [delegatedContentRouting], contentRouting: [delegatedContentRouting],
peerRouting: [delegatedPeerRouting] peerRouting: [delegatedPeerRouting]
}, },
peerRouting: {
refreshManager: {
enabled: true,
interval: 1000,
bootDelay: 11111
}
},
dialer: { dialer: {
maxParallelDials: 100, maxParallelDials: 100,
maxDialsPerPeer: 4, maxDialsPerPeer: 4,

View File

@@ -2,7 +2,6 @@
const duplexPair = require('it-pair/duplex') const duplexPair = require('it-pair/duplex')
const abortable = require('abortable-iterator') const abortable = require('abortable-iterator')
const AbortController = require('abort-controller')
/** /**
* Returns both sides of a mocked MultiaddrConnection * Returns both sides of a mocked MultiaddrConnection