Compare commits

...

4 Commits

Author SHA1 Message Date
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
ae21299ade chore: release version v0.35.1 2021-12-03 16:24:14 +00:00
149120bebc chore: update contributors 2021-12-03 16:24:13 +00:00
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
11 changed files with 19 additions and 198 deletions

View File

@ -1,3 +1,12 @@
## [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)

View File

@ -415,14 +415,7 @@ const node = await Libp2p.create({
contentRouting: [delegatedContentRouting],
peerRouting: [delegatedPeerRouting],
},
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
}
}
peerId
})
```

View File

@ -1,6 +1,6 @@
{
"name": "libp2p",
"version": "0.35.0",
"version": "0.35.1",
"description": "JavaScript implementation of libp2p, a modular peer to peer network stack",
"leadMaintainer": "Jacob Heun <jacobheun@gmail.com>",
"main": "src/index.js",
@ -82,7 +82,6 @@
"dependencies": {
"@motrix/nat-api": "^0.3.1",
"@vascosantos/moving-average": "^1.1.0",
"abort-controller": "^3.0.0",
"abortable-iterator": "^3.0.0",
"aggregate-error": "^3.1.0",
"any-signal": "^2.1.1",
@ -136,7 +135,6 @@
"@chainsafe/libp2p-noise": "^4.0.0",
"@nodeutils/defaults-deep": "^1.1.0",
"@types/es6-promisify": "^6.0.0",
"@types/node": "^16.0.1",
"@types/node-forge": "^0.10.1",
"@types/varint": "^6.0.0",
"aegir": "^36.0.0",

View File

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

View File

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

View File

@ -384,7 +384,6 @@ class Libp2p extends EventEmitter {
this._isStarted = false
this.relay && this.relay.stop()
this.peerRouting.stop()
this._autodialler.stop()
await (this._dht && this._dht.stop())
@ -663,8 +662,6 @@ class Libp2p extends EventEmitter {
// Relay
this.relay && this.relay.start()
this.peerRouting.start()
}
/**

View File

@ -1,26 +1,17 @@
'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 {
storeAddresses,
uniquePeers,
requirePeers
} = require('./content-routing/utils')
const { TimeoutController } = require('timeout-abort-controller')
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
// @ts-ignore module with no types
} = require('set-delayed-interval')
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
* @property {RefreshManagerOptions} [refreshManager]
*/
class PeerRouting {
@ -54,42 +39,6 @@ class PeerRouting {
if (libp2p._dht && libp2p._config.dht.enabled) {
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 {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[] }>}
*/
async * getClosestPeers (key, options = { timeout: 30e3 }) {
@ -139,6 +89,10 @@ class PeerRouting {
throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
}
if (options.timeout) {
options.signal = new TimeoutController(options.timeout).signal
}
yield * pipe(
merge(
...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 { AbortError } = require('libp2p-interfaces/src/transport/errors')
const AbortController = require('abort-controller')
const AggregateError = require('aggregate-error')
const pDefer = require('p-defer')
const delay = require('delay')

View File

@ -6,9 +6,7 @@ const nock = require('nock')
const sinon = require('sinon')
const intoStream = require('into-stream')
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')
@ -453,109 +451,4 @@ describe('peer-routing', () => {
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

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

View File

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