fluencelabs/js-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/js-libp2p synced 2025-07-31 16:32:01 +00:00
Code Issues Projects Releases Wiki Activity

fix: time out slow reads (#1227)

Browse Source 
There are a few places in the codebase where we send/receive data from the network without timeouts/abort controllers which means the user has to wait for the underlying socket to timeout which can take a long time depending on the platform, if at all.

This change ensures we can time out while running identify (both flavours), ping and fetch and adds tests to ensure there are no regressions.
This commit is contained in:
Alex Potsides
2022-05-25 18:15:21 +01:00
committed by GitHub
parent 5934b13cce
commit a1220d22f5
23 changed files with 1039 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/config.ts
View File

@@ -35,9 +35,6 @@ const DefaultConfig: Partial<Libp2pInit> = {
transportManager: {
faultTolerance: FaultTolerance.FATAL_ALL
},
host: {
agentVersion: AGENT_VERSION
},
metrics: {
enabled: false,
computeThrottleMaxQueueSize: 1000,
@@ -56,7 +53,6 @@ const DefaultConfig: Partial<Libp2pInit> = {
bootDelay: 10e3
}
},
protocolPrefix: 'ipfs',
nat: {
enabled: true,
ttl: 7200,
@@ -77,6 +73,19 @@ const DefaultConfig: Partial<Libp2pInit> = {
enabled: false,
maxListeners: 2
}
},
identify: {
protocolPrefix: 'ipfs',
host: {
agentVersion: AGENT_VERSION
},
timeout: 30000
},
ping: {
protocolPrefix: 'ipfs'
},
fetch: {
protocolPrefix: 'libp2p'
}
}

3
src/fetch/constants.ts
View File

@@ -1,3 +1,4 @@
// https://github.com/libp2p/specs/tree/master/fetch#wire-protocol
export const PROTOCOL = '/libp2p/fetch/0.0.1'
export const PROTOCOL_VERSION = '0.0.1'
export const PROTOCOL_NAME = 'fetch'

28
src/fetch/index.ts
View File

@@ -4,16 +4,19 @@ import { codes } from '../errors.js'
import * as lp from 'it-length-prefixed'
import { FetchRequest, FetchResponse } from './pb/proto.js'
import { handshake } from 'it-handshake'
import { PROTOCOL } from './constants.js'
import { PROTOCOL_NAME, PROTOCOL_VERSION } from './constants.js'
import type { PeerId } from '@libp2p/interfaces/peer-id'
import type { Startable } from '@libp2p/interfaces/startable'
import type { Stream } from '@libp2p/interfaces/connection'
import type { IncomingStreamData } from '@libp2p/interfaces/registrar'
import type { Components } from '@libp2p/interfaces/components'
import type { AbortOptions } from '@libp2p/interfaces'
import type { Duplex } from 'it-stream-types'
import { abortableDuplex } from 'abortable-iterator'
const log = logger('libp2p:fetch')
export interface FetchInit {
export interface FetchServiceInit {
protocolPrefix: string
}
@@ -33,15 +36,15 @@ export interface LookupFunction {
* by a fixed prefix that all keys that should be routed to that lookup function will start with.
*/
export class FetchService implements Startable {
public readonly protocol: string
private readonly components: Components
private readonly lookupFunctions: Map<string, LookupFunction>
private readonly protocol: string
private started: boolean
constructor (components: Components, init: FetchInit) {
constructor (components: Components, init: FetchServiceInit) {
this.started = false
this.components = components
this.protocol = PROTOCOL
this.protocol = `/${init.protocolPrefix ?? 'libp2p'}/${PROTOCOL_NAME}/${PROTOCOL_VERSION}`
this.lookupFunctions = new Map() // Maps key prefix to value lookup function
this.handleMessage = this.handleMessage.bind(this)
}
@@ -67,12 +70,19 @@ export class FetchService implements Startable {
/**
* Sends a request to fetch the value associated with the given key from the given peer
*/
async fetch (peer: PeerId, key: string): Promise<Uint8Array | null> {
async fetch (peer: PeerId, key: string, options: AbortOptions = {}): Promise<Uint8Array | null> {
log('dialing %s to %p', this.protocol, peer)
const connection = await this.components.getConnectionManager().openConnection(peer)
const { stream } = await connection.newStream([this.protocol])
const shake = handshake(stream)
const connection = await this.components.getConnectionManager().openConnection(peer, options)
const { stream } = await connection.newStream([this.protocol], options)
let source: Duplex<Uint8Array> = stream
// make stream abortable if AbortSignal passed
if (options.signal != null) {
source = abortableDuplex(stream, options.signal)
}
const shake = handshake(source)
// send message
shake.write(lp.encode.single(FetchRequest.encode({ identifier: key })).slice())

104
src/identify/index.ts
View File

@@ -2,8 +2,6 @@ import { logger } from '@libp2p/logger'
import errCode from 'err-code'
import * as lp from 'it-length-prefixed'
import { pipe } from 'it-pipe'
import all from 'it-all'
import take from 'it-take'
import drain from 'it-drain'
import first from 'it-first'
import { fromString as uint8ArrayFromString } from 'uint8arrays/from-string'
@@ -21,13 +19,19 @@ import {
} from './consts.js'
import { codes } from '../errors.js'
import type { IncomingStreamData } from '@libp2p/interfaces/registrar'
import type { Connection } from '@libp2p/interfaces/connection'
import type { Connection, Stream } from '@libp2p/interfaces/connection'
import type { Startable } from '@libp2p/interfaces/startable'
import { peerIdFromKeys } from '@libp2p/peer-id'
import type { Components } from '@libp2p/interfaces/components'
import { TimeoutController } from 'timeout-abort-controller'
import type { AbortOptions } from '@libp2p/interfaces'
import { abortableDuplex } from 'abortable-iterator'
import type { Duplex } from 'it-stream-types'
const log = logger('libp2p:identify')
const IDENTIFY_TIMEOUT = 30000
export interface HostProperties {
agentVersion: string
}
@@ -35,6 +39,7 @@ export interface HostProperties {
export interface IdentifyServiceInit {
protocolPrefix: string
host: HostProperties
timeout?: number
}
export class IdentifyService implements Startable {
@@ -46,11 +51,13 @@ export class IdentifyService implements Startable {
agentVersion: string
}
private readonly init: IdentifyServiceInit
private started: boolean
constructor (components: Components, init: IdentifyServiceInit) {
this.components = components
this.started = false
this.init = init
this.handleMessage = this.handleMessage.bind(this)
@@ -128,8 +135,17 @@ export class IdentifyService implements Startable {
const protocols = await this.components.getPeerStore().protoBook.get(this.components.getPeerId())
const pushes = connections.map(async connection => {
const timeoutController = new TimeoutController(this.init.timeout ?? IDENTIFY_TIMEOUT)
let stream: Stream | undefined
try {
const { stream } = await connection.newStream([this.identifyPushProtocolStr])
const data = await connection.newStream([this.identifyPushProtocolStr], {
signal: timeoutController.signal
})
stream = data.stream
// make stream abortable
const source: Duplex<Uint8Array> = abortableDuplex(stream, timeoutController.signal)
await pipe(
[Identify.encode({
@@ -138,12 +154,18 @@ export class IdentifyService implements Startable {
protocols
})],
lp.encode(),
stream,
source,
drain
)
} catch (err: any) {
// Just log errors
log.error('could not push identify update to peer', err)
} finally {
if (stream != null) {
stream.close()
}
timeoutController.clear()
}
})
@@ -175,31 +197,44 @@ export class IdentifyService implements Startable {
await this.push(connections)
}
async _identify (connection: Connection, options: AbortOptions = {}): Promise<Identify> {
const { stream } = await connection.newStream([this.identifyProtocolStr], options)
let source: Duplex<Uint8Array> = stream
// make stream abortable if AbortSignal passed
if (options.signal != null) {
source = abortableDuplex(stream, options.signal)
}
try {
const data = await pipe(
[],
source,
lp.decode(),
async (source) => await first(source)
)
if (data == null) {
throw errCode(new Error('No data could be retrieved'), codes.ERR_CONNECTION_ENDED)
}
try {
return Identify.decode(data)
} catch (err: any) {
throw errCode(err, codes.ERR_INVALID_MESSAGE)
}
} finally {
stream.close()
}
}
/**
* Requests the `Identify` message from peer associated with the given `connection`.
* If the identified peer does not match the `PeerId` associated with the connection,
* an error will be thrown.
*/
async identify (connection: Connection): Promise<void> {
const { stream } = await connection.newStream([this.identifyProtocolStr])
const [data] = await pipe(
[],
stream,
lp.decode(),
(source) => take(source, 1),
async (source) => await all(source)
)
if (data == null) {
throw errCode(new Error('No data could be retrieved'), codes.ERR_CONNECTION_ENDED)
}
let message: Identify
try {
message = Identify.decode(data)
} catch (err: any) {
throw errCode(err, codes.ERR_INVALID_MESSAGE)
}
async identify (connection: Connection, options: AbortOptions = {}): Promise<void> {
const message = await this._identify(connection, options)
const {
publicKey,
@@ -308,6 +343,8 @@ export class IdentifyService implements Startable {
*/
async _handleIdentify (data: IncomingStreamData) {
const { connection, stream } = data
const timeoutController = new TimeoutController(this.init.timeout ?? IDENTIFY_TIMEOUT)
try {
const publicKey = this.components.getPeerId().publicKey ?? new Uint8Array(0)
const peerData = await this.components.getPeerStore().get(this.components.getPeerId())
@@ -335,14 +372,20 @@ export class IdentifyService implements Startable {
protocols: peerData.protocols
})
// make stream abortable
const source: Duplex<Uint8Array> = abortableDuplex(stream, timeoutController.signal)
await pipe(
[message],
lp.encode(),
stream,
source,
drain
)
} catch (err: any) {
log.error('could not respond to identify request', err)
} finally {
stream.close()
timeoutController.clear()
}
}
@@ -351,12 +394,16 @@ export class IdentifyService implements Startable {
*/
async _handlePush (data: IncomingStreamData) {
const { connection, stream } = data
const timeoutController = new TimeoutController(this.init.timeout ?? IDENTIFY_TIMEOUT)
let message: Identify | undefined
try {
// make stream abortable
const source: Duplex<Uint8Array> = abortableDuplex(stream, timeoutController.signal)
const data = await pipe(
[],
stream,
source,
lp.decode(),
async (source) => await first(source)
)
@@ -366,6 +413,9 @@ export class IdentifyService implements Startable {
}
} catch (err: any) {
return log.error('received invalid message', err)
} finally {
stream.close()
timeoutController.clear()
}
if (message == null) {

13
src/index.ts
View File

@@ -4,7 +4,7 @@ import type { EventEmitter } from '@libp2p/interfaces/events'
import type { Startable } from '@libp2p/interfaces/startable'
import type { Multiaddr } from '@multiformats/multiaddr'
import type { FaultTolerance } from './transport-manager.js'
import type { HostProperties } from './identify/index.js'
import type { IdentifyServiceInit } from './identify/index.js'
import type { DualDHT } from '@libp2p/interfaces/dht'
import type { Datastore } from 'interface-datastore'
import type { PeerStore, PeerStoreInit } from '@libp2p/interfaces/peer-store'
@@ -24,6 +24,8 @@ import type { Metrics, MetricsInit } from '@libp2p/interfaces/metrics'
import type { PeerInfo } from '@libp2p/interfaces/peer-info'
import type { KeyChain } from './keychain/index.js'
import type { ConnectionManagerInit } from './connection-manager/index.js'
import type { PingServiceInit } from './ping/index.js'
import type { FetchServiceInit } from './fetch/index.js'
export interface PersistentPeerStoreOptions {
threshold?: number
@@ -95,7 +97,6 @@ export interface RefreshManagerConfig {
export interface Libp2pInit {
peerId: PeerId
host: HostProperties
addresses: AddressesConfig
connectionManager: ConnectionManagerInit
connectionGater: Partial<ConnectionGater>
@@ -105,9 +106,11 @@ export interface Libp2pInit {
peerStore: PeerStoreInit
peerRouting: PeerRoutingConfig
keychain: KeychainConfig
protocolPrefix: string
nat: NatManagerConfig
relay: RelayConfig
identify: IdentifyServiceInit
ping: PingServiceInit
fetch: FetchServiceInit
transports: Transport[]
streamMuxers?: StreamMuxerFactory[]
@@ -195,12 +198,12 @@ export interface Libp2p extends Startable, EventEmitter<Libp2pEvents> {
/**
* Pings the given peer in order to obtain the operation latency
*/
ping: (peer: Multiaddr |PeerId) => Promise<number>
ping: (peer: Multiaddr | PeerId, options?: AbortOptions) => Promise<number>
/**
* Sends a request to fetch the value associated with the given key from the given peer.
*/
fetch: (peer: PeerId | Multiaddr | string, key: string) => Promise<Uint8Array | null>
fetch: (peer: PeerId | Multiaddr | string, key: string, options?: AbortOptions) => Promise<Uint8Array | null>
/**
* Returns the public key for the passed PeerId. If the PeerId is of the 'RSA' type

21
src/libp2p.ts
View File

@@ -166,10 +166,7 @@ export class Libp2pNode extends EventEmitter<Libp2pEvents> implements Libp2p {
if (init.streamMuxers != null && init.streamMuxers.length > 0) {
// Add the identify service since we can multiplex
this.identifyService = new IdentifyService(this.components, {
protocolPrefix: init.protocolPrefix,
host: {
agentVersion: init.host.agentVersion
}
...init.identify
})
this.configureComponent(this.identifyService)
}
@@ -229,11 +226,11 @@ export class Libp2pNode extends EventEmitter<Libp2pEvents> implements Libp2p {
}
this.fetchService = this.configureComponent(new FetchService(this.components, {
protocolPrefix: init.protocolPrefix
...init.fetch
}))
this.pingService = this.configureComponent(new PingService(this.components, {
protocolPrefix: init.protocolPrefix
...init.ping
}))
const autoDialer = this.configureComponent(new AutoDialer(this.components, {
@@ -419,9 +416,9 @@ export class Libp2pNode extends EventEmitter<Libp2pEvents> implements Libp2p {
throw errCode(new Error('no protocols were provided to open a stream'), codes.ERR_INVALID_PROTOCOLS_FOR_STREAM)
}
const connection = await this.dial(peer)
const connection = await this.dial(peer, options)
return await connection.newStream(protocols)
return await connection.newStream(protocols, options)
}
getMultiaddrs (): Multiaddr[] {
@@ -473,24 +470,24 @@ export class Libp2pNode extends EventEmitter<Libp2pEvents> implements Libp2p {
throw errCode(new Error(`Node not responding with its public key: ${peer.toString()}`), codes.ERR_INVALID_RECORD)
}
async fetch (peer: PeerId | Multiaddr | string, key: string): Promise<Uint8Array | null> {
async fetch (peer: PeerId | Multiaddr | string, key: string, options: AbortOptions = {}): Promise<Uint8Array | null> {
const { id, multiaddrs } = getPeer(peer)
if (multiaddrs != null) {
await this.components.getPeerStore().addressBook.add(id, multiaddrs)
}
return await this.fetchService.fetch(id, key)
return await this.fetchService.fetch(id, key, options)
}
async ping (peer: PeerId | Multiaddr | string): Promise<number> {
async ping (peer: PeerId | Multiaddr | string, options: AbortOptions = {}): Promise<number> {
const { id, multiaddrs } = getPeer(peer)
if (multiaddrs.length > 0) {
await this.components.getPeerStore().addressBook.add(id, multiaddrs)
}
return await this.pingService.ping(id)
return await this.pingService.ping(id, options)
}
async handle (protocols: string | string[], handler: StreamHandler): Promise<void> {

40
src/ping/index.ts
View File

@@ -10,6 +10,9 @@ import type { IncomingStreamData } from '@libp2p/interfaces/registrar'
import type { PeerId } from '@libp2p/interfaces/peer-id'
import type { Startable } from '@libp2p/interfaces/startable'
import type { Components } from '@libp2p/interfaces/components'
import type { AbortOptions } from '@libp2p/interfaces'
import type { Duplex } from 'it-stream-types'
import { abortableDuplex } from 'abortable-iterator'
const log = logger('libp2p:ping')
@@ -18,8 +21,8 @@ export interface PingServiceInit {
}
export class PingService implements Startable {
public readonly protocol: string
private readonly components: Components
private readonly protocol: string
private started: boolean
constructor (components: Components, init: PingServiceInit) {
@@ -60,25 +63,36 @@ export class PingService implements Startable {
* @param {PeerId|Multiaddr} peer
* @returns {Promise<number>}
*/
async ping (peer: PeerId): Promise<number> {
async ping (peer: PeerId, options: AbortOptions = {}): Promise<number> {
log('dialing %s to %p', this.protocol, peer)
const connection = await this.components.getConnectionManager().openConnection(peer)
const { stream } = await connection.newStream([this.protocol])
const connection = await this.components.getConnectionManager().openConnection(peer, options)
const { stream } = await connection.newStream([this.protocol], options)
const start = Date.now()
const data = randomBytes(PING_LENGTH)
const result = await pipe(
[data],
stream,
async (source) => await first(source)
)
const end = Date.now()
let source: Duplex<Uint8Array> = stream
if (result == null || !uint8ArrayEquals(data, result)) {
throw errCode(new Error('Received wrong ping ack'), codes.ERR_WRONG_PING_ACK)
// make stream abortable if AbortSignal passed
if (options.signal != null) {
source = abortableDuplex(stream, options.signal)
}
return end - start
try {
const result = await pipe(
[data],
source,
async (source) => await first(source)
)
const end = Date.now()
if (result == null || !uint8ArrayEquals(data, result)) {
throw errCode(new Error('Received wrong ping ack'), codes.ERR_WRONG_PING_ACK)
}
return end - start
} finally {
stream.close()
}
}
}

12
src/upgrader.ts
View File

@@ -15,6 +15,7 @@ import type { PeerId } from '@libp2p/interfaces/peer-id'
import type { MultiaddrConnection, Upgrader, UpgraderEvents } from '@libp2p/interfaces/transport'
import type { Duplex } from 'it-stream-types'
import type { Components } from '@libp2p/interfaces/components'
import type { AbortOptions } from '@libp2p/interfaces'
const log = logger('libp2p:upgrader')
@@ -266,7 +267,7 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
} = opts
let muxer: StreamMuxer | undefined
let newStream: ((multicodecs: string[]) => Promise<ProtocolStream>) | undefined
let newStream: ((multicodecs: string[], options?: AbortOptions) => Promise<ProtocolStream>) | undefined
let connection: Connection // eslint-disable-line prefer-const
if (muxerFactory != null) {
@@ -308,7 +309,7 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
}
})
newStream = async (protocols: string[]): Promise<ProtocolStream> => {
newStream = async (protocols: string[], options: AbortOptions = {}): Promise<ProtocolStream> => {
if (muxer == null) {
throw errCode(new Error('Stream is not multiplexed'), codes.ERR_MUXER_UNAVAILABLE)
}
@@ -319,7 +320,7 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
const metrics = this.components.getMetrics()
try {
let { stream, protocol } = await mss.select(protocols)
let { stream, protocol } = await mss.select(protocols, options)
if (metrics != null) {
stream = metrics.trackStream({ stream, remotePeer, protocol })
@@ -328,6 +329,11 @@ export class DefaultUpgrader extends EventEmitter<UpgraderEvents> implements Upg
return { stream: { ...muxedStream, ...stream }, protocol }
} catch (err: any) {
log.error('could not create new stream', err)
if (err.code != null) {
throw err
}
throw errCode(err, codes.ERR_UNSUPPORTED_PROTOCOL)
}
}