This reverts commit d041287a4f
.
This commit is contained in:
parent
d041287a4f
commit
7cc3dc8027
|
@ -72,7 +72,7 @@ type
|
||||||
connTable: HashSet[PeerID]
|
connTable: HashSet[PeerID]
|
||||||
forkId: ENRForkID
|
forkId: ENRForkID
|
||||||
rng*: ref BrHmacDrbgContext
|
rng*: ref BrHmacDrbgContext
|
||||||
peers*: Table[PeerID, Peer]
|
peers: Table[PeerID, Peer]
|
||||||
|
|
||||||
EthereumNode = Eth2Node # needed for the definitions in p2p_backends_helpers
|
EthereumNode = Eth2Node # needed for the definitions in p2p_backends_helpers
|
||||||
|
|
||||||
|
@ -101,8 +101,6 @@ type
|
||||||
requestQuota*: float
|
requestQuota*: float
|
||||||
lastReqTime*: Moment
|
lastReqTime*: Moment
|
||||||
connections*: int
|
connections*: int
|
||||||
enr*: Option[enr.Record]
|
|
||||||
direction*: PeerType
|
|
||||||
disconnectedFut: Future[void]
|
disconnectedFut: Future[void]
|
||||||
|
|
||||||
PeerAddr* = object
|
PeerAddr* = object
|
||||||
|
@ -243,9 +241,6 @@ const
|
||||||
SeenTablePenaltyError* = 60.minutes
|
SeenTablePenaltyError* = 60.minutes
|
||||||
## Period of time for peers which score below or equal to zero.
|
## Period of time for peers which score below or equal to zero.
|
||||||
|
|
||||||
ResolvePeerTimeout* = 1.minutes
|
|
||||||
## Maximum time allowed for peer resolve process.
|
|
||||||
|
|
||||||
template neterr(kindParam: Eth2NetworkingErrorKind): auto =
|
template neterr(kindParam: Eth2NetworkingErrorKind): auto =
|
||||||
err(type(result), Eth2NetworkingError(kind: kindParam))
|
err(type(result), Eth2NetworkingError(kind: kindParam))
|
||||||
|
|
||||||
|
@ -268,18 +263,6 @@ declarePublicCounter nbc_timeout_dials,
|
||||||
declarePublicGauge nbc_peers,
|
declarePublicGauge nbc_peers,
|
||||||
"Number of active libp2p peers"
|
"Number of active libp2p peers"
|
||||||
|
|
||||||
declarePublicCounter nbc_successful_discoveries,
|
|
||||||
"Number of successfull discoveries"
|
|
||||||
|
|
||||||
declarePublicCounter nbc_failed_discoveries,
|
|
||||||
"Number of failed discoveries"
|
|
||||||
|
|
||||||
const delayBuckets = [1.0, 5.0, 10.0, 20.0, 40.0, 60.0]
|
|
||||||
|
|
||||||
declareHistogram nbc_resolve_time,
|
|
||||||
"Time(s) used while resolving peer information",
|
|
||||||
buckets = delayBuckets
|
|
||||||
|
|
||||||
const
|
const
|
||||||
snappy_implementation {.strdefine.} = "libp2p"
|
snappy_implementation {.strdefine.} = "libp2p"
|
||||||
|
|
||||||
|
@ -335,10 +318,6 @@ proc getPeer*(node: Eth2Node, peerId: PeerID): Peer =
|
||||||
let peer = Peer.init(node, PeerInfo.init(peerId))
|
let peer = Peer.init(node, PeerInfo.init(peerId))
|
||||||
return node.peers.mGetOrPut(peerId, peer)
|
return node.peers.mGetOrPut(peerId, peer)
|
||||||
|
|
||||||
proc resetPeer*(node: Eth2Node, peerId: PeerID) =
|
|
||||||
let peer = Peer.init(node, PeerInfo.init(peerId))
|
|
||||||
node.peers[peerId] = peer
|
|
||||||
|
|
||||||
proc peerFromStream(network: Eth2Node, conn: Connection): Peer =
|
proc peerFromStream(network: Eth2Node, conn: Connection): Peer =
|
||||||
result = network.getPeer(conn.peerInfo.peerId)
|
result = network.getPeer(conn.peerInfo.peerId)
|
||||||
result.info = conn.peerInfo
|
result.info = conn.peerInfo
|
||||||
|
@ -347,6 +326,8 @@ proc getKey*(peer: Peer): PeerID {.inline.} =
|
||||||
peer.info.peerId
|
peer.info.peerId
|
||||||
|
|
||||||
proc getFuture*(peer: Peer): Future[void] {.inline.} =
|
proc getFuture*(peer: Peer): Future[void] {.inline.} =
|
||||||
|
if peer.disconnectedFut.isNil:
|
||||||
|
peer.disconnectedFut = newFuture[void]()
|
||||||
peer.disconnectedFut
|
peer.disconnectedFut
|
||||||
|
|
||||||
proc getScore*(a: Peer): int =
|
proc getScore*(a: Peer): int =
|
||||||
|
@ -458,6 +439,7 @@ proc disconnect*(peer: Peer, reason: DisconnectionReason,
|
||||||
SeenTablePenaltyError
|
SeenTablePenaltyError
|
||||||
peer.network.addSeen(peer.info.peerId, seenTime)
|
peer.network.addSeen(peer.info.peerId, seenTime)
|
||||||
await peer.network.switch.disconnect(peer.info.peerId)
|
await peer.network.switch.disconnect(peer.info.peerId)
|
||||||
|
peer.connectionState = Disconnected
|
||||||
except CatchableError:
|
except CatchableError:
|
||||||
# We do not care about exceptions in disconnection procedure.
|
# We do not care about exceptions in disconnection procedure.
|
||||||
trace "Exception while disconnecting peer", peer = peer.info.peerId,
|
trace "Exception while disconnecting peer", peer = peer.info.peerId,
|
||||||
|
@ -657,22 +639,6 @@ proc handleIncomingStream(network: Eth2Node,
|
||||||
|
|
||||||
let peer = peerFromStream(network, conn)
|
let peer = peerFromStream(network, conn)
|
||||||
try:
|
try:
|
||||||
case peer.connectionState
|
|
||||||
of Disconnecting, Disconnected, None:
|
|
||||||
# We got incoming stream request while disconnected or disconnecting.
|
|
||||||
warn "Got incoming request from disconnected peer", peer = peer,
|
|
||||||
message = msgName
|
|
||||||
await conn.closeWithEOF()
|
|
||||||
return
|
|
||||||
of Connecting:
|
|
||||||
# We got incoming stream request while handshake is not yet finished,
|
|
||||||
# TODO: We could check it here.
|
|
||||||
debug "Got incoming request from peer while in handshake", peer = peer,
|
|
||||||
msgName
|
|
||||||
of Connected:
|
|
||||||
# We got incoming stream from peer with proper connection state.
|
|
||||||
debug "Got incoming request from peer", peer = peer, msgName
|
|
||||||
|
|
||||||
template returnInvalidRequest(msg: ErrorMsg) =
|
template returnInvalidRequest(msg: ErrorMsg) =
|
||||||
peer.updateScore(PeerScoreInvalidRequest)
|
peer.updateScore(PeerScoreInvalidRequest)
|
||||||
await sendErrorResponse(peer, conn, InvalidRequest, msg)
|
await sendErrorResponse(peer, conn, InvalidRequest, msg)
|
||||||
|
@ -755,10 +721,8 @@ proc handleIncomingStream(network: Eth2Node,
|
||||||
await conn.closeWithEOF()
|
await conn.closeWithEOF()
|
||||||
discard network.peerPool.checkPeerScore(peer)
|
discard network.peerPool.checkPeerScore(peer)
|
||||||
|
|
||||||
proc toPeerAddr*(r: enr.TypedRecord,
|
proc toPeerAddr*(r: enr.TypedRecord):
|
||||||
proto: IpTransportProtocol): Result[PeerAddr, cstring] {.
|
Result[PeerAddr, cstring] {.raises: [Defect].} =
|
||||||
raises: [Defect].} =
|
|
||||||
|
|
||||||
if not r.secp256k1.isSome:
|
if not r.secp256k1.isSome:
|
||||||
return err("enr: no secp256k1 key in record")
|
return err("enr: no secp256k1 key in record")
|
||||||
|
|
||||||
|
@ -769,34 +733,18 @@ proc toPeerAddr*(r: enr.TypedRecord,
|
||||||
|
|
||||||
var addrs = newSeq[MultiAddress]()
|
var addrs = newSeq[MultiAddress]()
|
||||||
|
|
||||||
case proto
|
if r.ip.isSome and r.tcp.isSome:
|
||||||
of tcpProtocol:
|
let ip = ipv4(r.ip.get)
|
||||||
if r.ip.isSome and r.tcp.isSome:
|
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp.get)
|
||||||
let ip = ipv4(r.ip.get)
|
|
||||||
|
if r.ip6.isSome:
|
||||||
|
let ip = ipv6(r.ip6.get)
|
||||||
|
if r.tcp6.isSome:
|
||||||
|
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp6.get)
|
||||||
|
elif r.tcp.isSome:
|
||||||
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp.get)
|
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp.get)
|
||||||
|
else:
|
||||||
if r.ip6.isSome:
|
discard
|
||||||
let ip = ipv6(r.ip6.get)
|
|
||||||
if r.tcp6.isSome:
|
|
||||||
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp6.get)
|
|
||||||
elif r.tcp.isSome:
|
|
||||||
addrs.add MultiAddress.init(ip, tcpProtocol, Port r.tcp.get)
|
|
||||||
else:
|
|
||||||
discard
|
|
||||||
|
|
||||||
of udpProtocol:
|
|
||||||
if r.ip.isSome and r.udp.isSome:
|
|
||||||
let ip = ipv4(r.ip.get)
|
|
||||||
addrs.add MultiAddress.init(ip, udpProtocol, Port r.udp.get)
|
|
||||||
|
|
||||||
if r.ip6.isSome:
|
|
||||||
let ip = ipv6(r.ip6.get)
|
|
||||||
if r.udp6.isSome:
|
|
||||||
addrs.add MultiAddress.init(ip, udpProtocol, Port r.udp6.get)
|
|
||||||
elif r.udp.isSome:
|
|
||||||
addrs.add MultiAddress.init(ip, udpProtocol, Port r.udp.get)
|
|
||||||
else:
|
|
||||||
discard
|
|
||||||
|
|
||||||
if addrs.len == 0:
|
if addrs.len == 0:
|
||||||
return err("enr: no addresses in record")
|
return err("enr: no addresses in record")
|
||||||
|
@ -860,7 +808,7 @@ proc connectWorker(node: Eth2Node, index: int) {.async.} =
|
||||||
|
|
||||||
proc toPeerAddr(node: Node): Result[PeerAddr, cstring] {.raises: [Defect].} =
|
proc toPeerAddr(node: Node): Result[PeerAddr, cstring] {.raises: [Defect].} =
|
||||||
let nodeRecord = ? node.record.toTypedRecord()
|
let nodeRecord = ? node.record.toTypedRecord()
|
||||||
let peerAddr = ? nodeRecord.toPeerAddr(tcpProtocol)
|
let peerAddr = ? nodeRecord.toPeerAddr()
|
||||||
ok(peerAddr)
|
ok(peerAddr)
|
||||||
|
|
||||||
proc runDiscoveryLoop*(node: Eth2Node) {.async.} =
|
proc runDiscoveryLoop*(node: Eth2Node) {.async.} =
|
||||||
|
@ -924,120 +872,13 @@ proc getPersistentNetMetadata*(conf: BeaconNodeConf): Eth2Metadata =
|
||||||
else:
|
else:
|
||||||
result = Json.loadFile(metadataPath, Eth2Metadata)
|
result = Json.loadFile(metadataPath, Eth2Metadata)
|
||||||
|
|
||||||
proc resolvePeer(peer: Peer) {.async.} =
|
|
||||||
# Resolve task which performs searching of peer's public key and recovery of
|
|
||||||
# ENR using discovery5.
|
|
||||||
# This process could be stopped if one of the conditions is met:
|
|
||||||
# 1. ENR was successfully recovered.
|
|
||||||
# 2. Discovery5 failed to recover ENR.
|
|
||||||
# 3. Timeout ``ResolvePeerTimeout`` has exceeded .
|
|
||||||
# 4. Connection with the peer is lost.
|
|
||||||
logScope: peer = peer.info.peerId
|
|
||||||
let startTime = now(chronos.Moment)
|
|
||||||
let nodeId =
|
|
||||||
block:
|
|
||||||
var key: PublicKey
|
|
||||||
# `secp256k1` keys are always stored inside PeerID.
|
|
||||||
discard peer.info.peerId.extractPublicKey(key)
|
|
||||||
keys.PublicKey.fromRaw(key.skkey.getBytes()).get().toNodeId()
|
|
||||||
|
|
||||||
debug "Peer's ENR recovery task started", node_id = $nodeId
|
|
||||||
|
|
||||||
# This is "fast-path" for peers which was dialed. In this case discovery
|
|
||||||
# already has most recent ENR information about this peer.
|
|
||||||
let gnode = peer.network.discovery.getNode(nodeId)
|
|
||||||
if gnode.isSome():
|
|
||||||
peer.enr = some(gnode.get().record)
|
|
||||||
inc(nbc_successful_discoveries)
|
|
||||||
let delay = now(chronos.Moment) - startTime
|
|
||||||
nbc_resolve_time.observe(delay.toFloatSeconds())
|
|
||||||
debug "Peer's ENR recovered", delay = $delay
|
|
||||||
else:
|
|
||||||
let resolveFut = peer.network.discovery.resolve(nodeId)
|
|
||||||
let timeFut = sleepAsync(ResolvePeerTimeout)
|
|
||||||
|
|
||||||
discard await race(peer.disconnectedFut, resolveFut, timeFut)
|
|
||||||
|
|
||||||
if peer.disconnectedFut.finished():
|
|
||||||
# Peer is already disconnected
|
|
||||||
if not(timeFut.finished()):
|
|
||||||
timeFut.cancel()
|
|
||||||
if not(resolveFut.finished()):
|
|
||||||
await resolveFut.cancelAndWait()
|
|
||||||
# allFutures() will perform check for futures which are already finished,
|
|
||||||
# and we do not care about results anymore.
|
|
||||||
await allFutures(timeFut)
|
|
||||||
return
|
|
||||||
|
|
||||||
if resolveFut.finished():
|
|
||||||
if resolveFut.done():
|
|
||||||
let rnode = resolveFut.read()
|
|
||||||
if rnode.isSome():
|
|
||||||
peer.enr = some(rnode.get().record)
|
|
||||||
inc(nbc_successful_discoveries)
|
|
||||||
let delay = now(chronos.Moment) - startTime
|
|
||||||
nbc_resolve_time.observe(delay.toFloatSeconds())
|
|
||||||
debug "Peer's ENR recovered", delay = $delay
|
|
||||||
else:
|
|
||||||
inc(nbc_failed_discoveries)
|
|
||||||
debug "Discovery operation returns empty answer"
|
|
||||||
return
|
|
||||||
else:
|
|
||||||
inc(nbc_failed_discoveries)
|
|
||||||
debug "Discovery operation failed with an error",
|
|
||||||
error_name = resolveFut.error.name,
|
|
||||||
error_msg = resolveFut.error.msg
|
|
||||||
if not(timeFut.finished()):
|
|
||||||
await timeFut.cancelAndWait()
|
|
||||||
return
|
|
||||||
|
|
||||||
if timeFut.finished():
|
|
||||||
inc(nbc_failed_discoveries)
|
|
||||||
debug "Discovery operation exceeds timeout",
|
|
||||||
timeout = $ResolvePeerTimeout
|
|
||||||
if not(resolveFut.finished()):
|
|
||||||
await resolveFut.cancelAndWait()
|
|
||||||
return
|
|
||||||
|
|
||||||
proc handlePeer*(peer: Peer) {.async.} =
|
|
||||||
let res = peer.network.peerPool.addPeerNoWait(peer, peer.direction)
|
|
||||||
case res:
|
|
||||||
of PeerStatus.LowScoreError, PeerStatus.NoSpaceError:
|
|
||||||
# Peer has low score or we do not have enough space in PeerPool,
|
|
||||||
# we are going to disconnect it gracefully.
|
|
||||||
# Peer' state will be updated in connection event.
|
|
||||||
debug "Peer has low score or there no space in PeerPool",
|
|
||||||
peer = peer, reason = res
|
|
||||||
await peer.disconnect(FaultOrError)
|
|
||||||
of PeerStatus.DeadPeerError:
|
|
||||||
# Peer's lifetime future is finished, so its already dead,
|
|
||||||
# we do not need to perform gracefull disconect.
|
|
||||||
# Peer's state will be updated in connection event.
|
|
||||||
discard
|
|
||||||
of PeerStatus.DuplicateError:
|
|
||||||
# Peer is already present in PeerPool, we can't perform disconnect,
|
|
||||||
# because in such case we could kill both connections (connection
|
|
||||||
# which is present in PeerPool and new one).
|
|
||||||
# This is possible bug, because we could enter here only if number
|
|
||||||
# of `peer.connections == 1`, it means that Peer's lifetime is not
|
|
||||||
# tracked properly and we still not received `Disconnected` event.
|
|
||||||
warn "Peer is already present in PeerPool", peer = peer
|
|
||||||
of PeerStatus.Success:
|
|
||||||
# Peer was added to PeerPool.
|
|
||||||
peer.score = NewPeerScore
|
|
||||||
peer.connectionState = Connected
|
|
||||||
# We spawn task which will obtain ENR for this peer.
|
|
||||||
asyncSpawn resolvePeer(peer)
|
|
||||||
debug "Peer successfully connected", peer = peer,
|
|
||||||
connections = peer.connections
|
|
||||||
|
|
||||||
proc onConnEvent(node: Eth2Node, peerId: PeerID, event: ConnEvent) {.async.} =
|
proc onConnEvent(node: Eth2Node, peerId: PeerID, event: ConnEvent) {.async.} =
|
||||||
let peer = node.getPeer(peerId)
|
let peer = node.getPeer(peerId)
|
||||||
case event.kind
|
case event.kind
|
||||||
of ConnEventKind.Connected:
|
of ConnEventKind.Connected:
|
||||||
inc peer.connections
|
inc peer.connections
|
||||||
debug "Peer connection upgraded", peer = $peerId,
|
debug "Peer upgraded", peer = $peerId, connections = peer.connections
|
||||||
connections = peer.connections
|
|
||||||
if peer.connections == 1:
|
if peer.connections == 1:
|
||||||
# Libp2p may connect multiple times to the same peer - using different
|
# Libp2p may connect multiple times to the same peer - using different
|
||||||
# transports for both incoming and outgoing. For now, we'll count our
|
# transports for both incoming and outgoing. For now, we'll count our
|
||||||
|
@ -1048,58 +889,45 @@ proc onConnEvent(node: Eth2Node, peerId: PeerID, event: ConnEvent) {.async.} =
|
||||||
# * For peer limits, we might miscount the incoming vs outgoing quota
|
# * For peer limits, we might miscount the incoming vs outgoing quota
|
||||||
# * Protocol handshakes are wonky: we'll not necessarily use the newly
|
# * Protocol handshakes are wonky: we'll not necessarily use the newly
|
||||||
# connected transport - instead we'll just pick a random one!
|
# connected transport - instead we'll just pick a random one!
|
||||||
case peer.connectionState
|
|
||||||
of Disconnecting:
|
|
||||||
# We got connection with peer which we currently disconnecting. This
|
|
||||||
# situation should not be happened, because when we disconnecting
|
|
||||||
# we adding peer to `SeenTable`.
|
|
||||||
warn "Got connection attempt from peer that we are disconnecting",
|
|
||||||
peer = peerId
|
|
||||||
return
|
|
||||||
of None:
|
|
||||||
# We have established a connection with the new peer.
|
|
||||||
peer.connectionState = Connecting
|
|
||||||
of Disconnected:
|
|
||||||
# We have established a connection with the peer that we have seen
|
|
||||||
# before, so we perform reset peer in node.peers table.
|
|
||||||
node.resetPeer(peerId)
|
|
||||||
peer.connectionState = Connecting
|
|
||||||
of Connecting, Connected:
|
|
||||||
# This means that we got notification event from peer which we already
|
|
||||||
# connected or connecting right now. If this situation will happened,
|
|
||||||
# it means bug on `nim-libp2p` side.
|
|
||||||
warn "Got connection attempt from peer which we already connected",
|
|
||||||
peer = peerId
|
|
||||||
await peer.disconnect(FaultOrError)
|
|
||||||
return
|
|
||||||
|
|
||||||
# Store connection direction inside Peer object.
|
await performProtocolHandshakes(peer, event.incoming)
|
||||||
if event.incoming:
|
|
||||||
peer.direction = PeerType.Incoming
|
|
||||||
else:
|
|
||||||
peer.direction = PeerType.Outgoing
|
|
||||||
|
|
||||||
if peer.direction == PeerType.Outgoing:
|
# While performing the handshake, the peer might have been disconnected -
|
||||||
# We only perform handshake with outgoing peers, incoming peers should
|
# there's still a slim chance of a race condition here if a reconnect
|
||||||
# start handshake first, so it will be handled in handleIncomingStream.
|
# happens quickly
|
||||||
await performProtocolHandshakes(peer, event.incoming)
|
if peer.connections == 1:
|
||||||
|
let res =
|
||||||
|
if event.incoming:
|
||||||
|
node.peerPool.addPeerNoWait(peer, PeerType.Incoming)
|
||||||
|
else:
|
||||||
|
node.peerPool.addPeerNoWait(peer, PeerType.Outgoing)
|
||||||
|
|
||||||
|
case res:
|
||||||
|
of PeerStatus.LowScoreError, PeerStatus.NoSpaceError:
|
||||||
|
# Peer has low score or we do not have enough space in PeerPool,
|
||||||
|
# we are going to disconnect it gracefully.
|
||||||
|
await peer.disconnect(FaultOrError)
|
||||||
|
of PeerStatus.DeadPeerError:
|
||||||
|
# Peer's lifetime future is finished, so its already dead,
|
||||||
|
# we do not need to perform gracefull disconect.
|
||||||
|
discard
|
||||||
|
of PeerStatus.DuplicateError:
|
||||||
|
# Peer is already present in PeerPool, we can't perform disconnect,
|
||||||
|
# because in such case we could kill both connections (connection
|
||||||
|
# which is present in PeerPool and new one).
|
||||||
|
discard
|
||||||
|
of PeerStatus.Success:
|
||||||
|
# Peer was added to PeerPool.
|
||||||
|
discard
|
||||||
|
|
||||||
of ConnEventKind.Disconnected:
|
of ConnEventKind.Disconnected:
|
||||||
dec peer.connections
|
dec peer.connections
|
||||||
debug "Lost connection to peer", peer = peerId,
|
debug "Peer disconnected", peer = $peerId, connections = peer.connections
|
||||||
connections = peer.connections
|
|
||||||
if peer.connections == 0:
|
if peer.connections == 0:
|
||||||
debug "Peer disconnected", peer = $peerId, connections = peer.connections
|
|
||||||
let fut = peer.disconnectedFut
|
let fut = peer.disconnectedFut
|
||||||
if not(isNil(fut)):
|
if fut != nil:
|
||||||
fut.complete()
|
|
||||||
peer.disconnectedFut = nil
|
peer.disconnectedFut = nil
|
||||||
else:
|
fut.complete()
|
||||||
# TODO (cheatfate): This could be removed when bug will be fixed inside
|
|
||||||
# `nim-libp2p`.
|
|
||||||
debug "Got new event while peer is already disconnected",
|
|
||||||
peer = peerId, peer_state = peer.connectionState
|
|
||||||
peer.connectionState = Disconnected
|
|
||||||
|
|
||||||
proc init*(T: type Eth2Node, conf: BeaconNodeConf, enrForkId: ENRForkID,
|
proc init*(T: type Eth2Node, conf: BeaconNodeConf, enrForkId: ENRForkID,
|
||||||
switch: Switch, pubsub: PubSub, ip: Option[ValidIpAddress],
|
switch: Switch, pubsub: PubSub, ip: Option[ValidIpAddress],
|
||||||
|
@ -1189,7 +1017,7 @@ proc start*(node: Eth2Node) {.async.} =
|
||||||
for enr in node.discovery.bootstrapRecords:
|
for enr in node.discovery.bootstrapRecords:
|
||||||
let tr = enr.toTypedRecord()
|
let tr = enr.toTypedRecord()
|
||||||
if tr.isOk():
|
if tr.isOk():
|
||||||
let pa = tr.get().toPeerAddr(tcpProtocol)
|
let pa = tr.get().toPeerAddr()
|
||||||
if pa.isOk():
|
if pa.isOk():
|
||||||
await node.connQueue.addLast(pa.get())
|
await node.connQueue.addLast(pa.get())
|
||||||
|
|
||||||
|
@ -1208,20 +1036,17 @@ proc stop*(node: Eth2Node) {.async.} =
|
||||||
futureErrors = waitedFutures.filterIt(it.error != nil).mapIt(it.error.msg)
|
futureErrors = waitedFutures.filterIt(it.error != nil).mapIt(it.error.msg)
|
||||||
|
|
||||||
proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer =
|
proc init*(T: type Peer, network: Eth2Node, info: PeerInfo): Peer =
|
||||||
let res = Peer(
|
new result
|
||||||
info: info,
|
result.info = info
|
||||||
network: network,
|
result.network = network
|
||||||
connectionState: ConnectionState.None,
|
result.connectionState = Connected
|
||||||
maxInactivityAllowed: 15.minutes, # TODO: Read this from the config
|
result.maxInactivityAllowed = 15.minutes # TODO: Read this from the config
|
||||||
lastReqTime: now(chronos.Moment),
|
result.lastReqTime = now(chronos.Moment)
|
||||||
disconnectedFut: newFuture[void]("peer.lifetime"),
|
newSeq result.protocolStates, allProtocols.len
|
||||||
protocolStates: newSeq[RootRef](len(allProtocols))
|
for i in 0 ..< allProtocols.len:
|
||||||
)
|
|
||||||
for i in 0 ..< len(allProtocols):
|
|
||||||
let proto = allProtocols[i]
|
let proto = allProtocols[i]
|
||||||
if not(isNil(proto.peerStateInitializer)):
|
if proto.peerStateInitializer != nil:
|
||||||
res.protocolStates[i] = proto.peerStateInitializer(res)
|
result.protocolStates[i] = proto.peerStateInitializer(result)
|
||||||
res
|
|
||||||
|
|
||||||
proc registerMsg(protocol: ProtocolInfo,
|
proc registerMsg(protocol: ProtocolInfo,
|
||||||
name: string,
|
name: string,
|
||||||
|
@ -1568,7 +1393,7 @@ proc createEth2Node*(rng: ref BrHmacDrbgContext,
|
||||||
rng = rng)
|
rng = rng)
|
||||||
|
|
||||||
let
|
let
|
||||||
params =
|
params =
|
||||||
block:
|
block:
|
||||||
var p = GossipSubParams.init()
|
var p = GossipSubParams.init()
|
||||||
# https://github.com/ethereum/eth2.0-specs/blob/v1.0.0/specs/phase0/p2p-interface.md#the-gossip-domain-gossipsub
|
# https://github.com/ethereum/eth2.0-specs/blob/v1.0.0/specs/phase0/p2p-interface.md#the-gossip-domain-gossipsub
|
||||||
|
@ -1585,9 +1410,9 @@ proc createEth2Node*(rng: ref BrHmacDrbgContext,
|
||||||
pubsub = GossipSub.init(
|
pubsub = GossipSub.init(
|
||||||
switch = switch,
|
switch = switch,
|
||||||
msgIdProvider = msgIdProvider,
|
msgIdProvider = msgIdProvider,
|
||||||
triggerSelf = true,
|
triggerSelf = true,
|
||||||
sign = false,
|
sign = false,
|
||||||
verifySignature = false,
|
verifySignature = false,
|
||||||
anonymize = true,
|
anonymize = true,
|
||||||
parameters = params).PubSub
|
parameters = params).PubSub
|
||||||
|
|
||||||
|
|
|
@ -1,10 +1,8 @@
|
||||||
import std/options,
|
import
|
||||||
chronicles,
|
chronicles,
|
||||||
json_rpc/[rpcserver, jsonmarshal],
|
json_rpc/[rpcserver, jsonmarshal],
|
||||||
eth/p2p/discoveryv5/enr,
|
|
||||||
libp2p/[multiaddress, multicodec],
|
../beacon_node_common, ../eth2_network,
|
||||||
nimcrypto/utils as ncrutils,
|
|
||||||
../beacon_node_common, ../eth2_network, ../sync_manager,
|
|
||||||
../peer_pool, ../version,
|
../peer_pool, ../version,
|
||||||
../spec/[datatypes, digest, presets],
|
../spec/[datatypes, digest, presets],
|
||||||
../spec/eth2_apis/callsigs_types
|
../spec/eth2_apis/callsigs_types
|
||||||
|
@ -17,230 +15,30 @@ type
|
||||||
template unimplemented() =
|
template unimplemented() =
|
||||||
raise (ref CatchableError)(msg: "Unimplemented")
|
raise (ref CatchableError)(msg: "Unimplemented")
|
||||||
|
|
||||||
proc validateState(state: Option[seq[string]]): Option[set[ConnectionState]] =
|
|
||||||
var res: set[ConnectionState]
|
|
||||||
if state.isSome():
|
|
||||||
let states = state.get()
|
|
||||||
for item in states:
|
|
||||||
case item
|
|
||||||
of "disconnected":
|
|
||||||
if ConnectionState.Disconnected notin res:
|
|
||||||
res.incl(ConnectionState.Disconnected)
|
|
||||||
else:
|
|
||||||
# `state` values should be unique
|
|
||||||
return none(set[ConnectionState])
|
|
||||||
of "connecting":
|
|
||||||
if ConnectionState.Disconnected notin res:
|
|
||||||
res.incl(ConnectionState.Connecting)
|
|
||||||
else:
|
|
||||||
# `state` values should be unique
|
|
||||||
return none(set[ConnectionState])
|
|
||||||
of "connected":
|
|
||||||
if ConnectionState.Connected notin res:
|
|
||||||
res.incl(ConnectionState.Connected)
|
|
||||||
else:
|
|
||||||
# `state` values should be unique
|
|
||||||
return none(set[ConnectionState])
|
|
||||||
of "disconnecting":
|
|
||||||
if ConnectionState.Disconnecting notin res:
|
|
||||||
res.incl(ConnectionState.Disconnecting)
|
|
||||||
else:
|
|
||||||
# `state` values should be unique
|
|
||||||
return none(set[ConnectionState])
|
|
||||||
else:
|
|
||||||
# Found incorrect `state` string value
|
|
||||||
return none(set[ConnectionState])
|
|
||||||
|
|
||||||
if res == {}:
|
|
||||||
res = {ConnectionState.Connecting, ConnectionState.Connected,
|
|
||||||
ConnectionState.Disconnecting, ConnectionState.Disconnected}
|
|
||||||
some(res)
|
|
||||||
|
|
||||||
proc validateDirection(direction: Option[seq[string]]): Option[set[PeerType]] =
|
|
||||||
var res: set[PeerType]
|
|
||||||
if direction.isSome():
|
|
||||||
let directions = direction.get()
|
|
||||||
for item in directions:
|
|
||||||
case item
|
|
||||||
of "inbound":
|
|
||||||
if PeerType.Incoming notin res:
|
|
||||||
res.incl(PeerType.Incoming)
|
|
||||||
else:
|
|
||||||
# `direction` values should be unique
|
|
||||||
return none(set[PeerType])
|
|
||||||
of "outbound":
|
|
||||||
if PeerType.Outgoing notin res:
|
|
||||||
res.incl(PeerType.Outgoing)
|
|
||||||
else:
|
|
||||||
# `direction` values should be unique
|
|
||||||
return none(set[PeerType])
|
|
||||||
else:
|
|
||||||
# Found incorrect `direction` string value
|
|
||||||
return none(set[PeerType])
|
|
||||||
|
|
||||||
if res == {}:
|
|
||||||
res = {PeerType.Incoming, PeerType.Outgoing}
|
|
||||||
some(res)
|
|
||||||
|
|
||||||
proc toString(state: ConnectionState): string =
|
|
||||||
case state
|
|
||||||
of ConnectionState.Disconnected:
|
|
||||||
"disconnected"
|
|
||||||
of ConnectionState.Connecting:
|
|
||||||
"connecting"
|
|
||||||
of ConnectionState.Connected:
|
|
||||||
"connected"
|
|
||||||
of ConnectionState.Disconnecting:
|
|
||||||
"disconnecting"
|
|
||||||
else:
|
|
||||||
""
|
|
||||||
|
|
||||||
proc toString(direction: PeerType): string =
|
|
||||||
case direction:
|
|
||||||
of PeerType.Incoming:
|
|
||||||
"inbound"
|
|
||||||
of PeerType.Outgoing:
|
|
||||||
"outbound"
|
|
||||||
|
|
||||||
proc getLastSeenAddress(info: PeerInfo): string =
|
|
||||||
# TODO (cheatfate): We need to provide filter here, which will be able to
|
|
||||||
# filter such multiaddresses like `/ip4/0.0.0.0` or local addresses or
|
|
||||||
# addresses with peer ids.
|
|
||||||
if len(info.addrs) > 0:
|
|
||||||
$info.addrs[len(info.addrs) - 1]
|
|
||||||
else:
|
|
||||||
""
|
|
||||||
|
|
||||||
proc getDiscoveryAddresses(node: BeaconNode): Option[seq[string]] =
|
|
||||||
let restr = node.network.enrRecord().toTypedRecord()
|
|
||||||
if restr.isErr():
|
|
||||||
return none[seq[string]]()
|
|
||||||
let respa = restr.get().toPeerAddr(udpProtocol)
|
|
||||||
if respa.isErr():
|
|
||||||
return none[seq[string]]()
|
|
||||||
let pa = respa.get()
|
|
||||||
let mpa = MultiAddress.init(multicodec("p2p"), pa.peerId)
|
|
||||||
if mpa.isErr():
|
|
||||||
return none[seq[string]]()
|
|
||||||
var addresses = newSeqOfCap[string](len(pa.addrs))
|
|
||||||
for item in pa.addrs:
|
|
||||||
let resa = concat(item, mpa.get())
|
|
||||||
if resa.isOk():
|
|
||||||
addresses.add($(resa.get()))
|
|
||||||
return some(addresses)
|
|
||||||
|
|
||||||
proc getP2PAddresses(node: BeaconNode): Option[seq[string]] =
|
|
||||||
let pinfo = node.network.switch.peerInfo
|
|
||||||
let mpa = MultiAddress.init(multicodec("p2p"), pinfo.peerId)
|
|
||||||
if mpa.isErr():
|
|
||||||
return none[seq[string]]()
|
|
||||||
var addresses = newSeqOfCap[string](len(pinfo.addrs))
|
|
||||||
for item in pinfo.addrs:
|
|
||||||
let resa = concat(item, mpa.get())
|
|
||||||
if resa.isOk():
|
|
||||||
addresses.add($(resa.get()))
|
|
||||||
return some(addresses)
|
|
||||||
|
|
||||||
proc installNodeApiHandlers*(rpcServer: RpcServer, node: BeaconNode) =
|
proc installNodeApiHandlers*(rpcServer: RpcServer, node: BeaconNode) =
|
||||||
rpcServer.rpc("get_v1_node_identity") do () -> NodeIdentityTuple:
|
rpcServer.rpc("get_v1_node_identity") do () -> NodeIdentityTuple:
|
||||||
let discoveryAddresses =
|
|
||||||
block:
|
|
||||||
let res = node.getDiscoveryAddresses()
|
|
||||||
if res.isSome():
|
|
||||||
res.get()
|
|
||||||
else:
|
|
||||||
newSeq[string](0)
|
|
||||||
|
|
||||||
let p2pAddresses =
|
|
||||||
block:
|
|
||||||
let res = node.getP2PAddresses()
|
|
||||||
if res.isSome():
|
|
||||||
res.get()
|
|
||||||
else:
|
|
||||||
newSeq[string]()
|
|
||||||
|
|
||||||
return (
|
return (
|
||||||
peer_id: $node.network.peerId(),
|
peer_id: node.network.peerId(),
|
||||||
enr: node.network.enrRecord().toUri(),
|
enr: node.network.enrRecord(),
|
||||||
p2p_addresses: p2pAddresses,
|
# TODO rest of fields
|
||||||
discovery_addresses: discoveryAddresses,
|
p2p_addresses: newSeq[MultiAddress](0),
|
||||||
metadata: (node.network.metadata.seq_number,
|
discovery_addresses: newSeq[MultiAddress](0),
|
||||||
"0x" & ncrutils.toHex(node.network.metadata.attnets.bytes))
|
metadata: (0'u64, "")
|
||||||
)
|
)
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_peers") do (state: Option[seq[string]],
|
rpcServer.rpc("get_v1_node_peers") do () -> JsonNode:
|
||||||
direction: Option[seq[string]]) -> seq[NodePeerTuple]:
|
unimplemented()
|
||||||
var res = newSeq[NodePeerTuple]()
|
|
||||||
let rstates = validateState(state)
|
|
||||||
if rstates.isNone():
|
|
||||||
raise newException(CatchableError, "Incorrect state parameter")
|
|
||||||
let rdirs = validateDirection(direction)
|
|
||||||
if rdirs.isNone():
|
|
||||||
raise newException(CatchableError, "Incorrect direction parameter")
|
|
||||||
let states = rstates.get()
|
|
||||||
let dirs = rdirs.get()
|
|
||||||
for item in node.network.peers.values():
|
|
||||||
if (item.connectionState in states) and (item.direction in dirs):
|
|
||||||
let peer = (
|
|
||||||
peer_id: $item.info.peerId,
|
|
||||||
enr: if item.enr.isSome(): item.enr.get().toUri() else: "",
|
|
||||||
last_seen_p2p_address: item.info.getLastSeenAddress(),
|
|
||||||
state: item.connectionState.toString(),
|
|
||||||
direction: item.direction.toString(),
|
|
||||||
agent: item.info.agentVersion, # Fields `agent` and `proto` are not
|
|
||||||
proto: item.info.protoVersion # part of specification.
|
|
||||||
)
|
|
||||||
res.add(peer)
|
|
||||||
return res
|
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_peer_count") do () -> NodePeerCountTuple:
|
rpcServer.rpc("get_v1_node_peers_peerId") do () -> JsonNode:
|
||||||
var res: NodePeerCountTuple
|
unimplemented()
|
||||||
for item in node.network.peers.values():
|
|
||||||
case item.connectionState
|
|
||||||
of Connecting:
|
|
||||||
inc(res.connecting)
|
|
||||||
of Connected:
|
|
||||||
inc(res.connected)
|
|
||||||
of Disconnecting:
|
|
||||||
inc(res.disconnecting)
|
|
||||||
of Disconnected:
|
|
||||||
inc(res.disconnected)
|
|
||||||
of ConnectionState.None:
|
|
||||||
discard
|
|
||||||
return res
|
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_peers_peerId") do (
|
|
||||||
peer_id: string) -> NodePeerTuple:
|
|
||||||
let pres = PeerID.init(peer_id)
|
|
||||||
if pres.isErr():
|
|
||||||
raise newException(CatchableError,
|
|
||||||
"The peer ID supplied could not be parsed")
|
|
||||||
let pid = pres.get()
|
|
||||||
let peer = node.network.peers.getOrDefault(pid)
|
|
||||||
if isNil(peer):
|
|
||||||
raise newException(CatchableError, "Peer not found")
|
|
||||||
|
|
||||||
return (
|
|
||||||
peer_id: $peer.info.peerId,
|
|
||||||
enr: if peer.enr.isSome(): peer.enr.get().toUri() else: "",
|
|
||||||
last_seen_p2p_address: peer.info.getLastSeenAddress(),
|
|
||||||
state: peer.connectionState.toString(),
|
|
||||||
direction: peer.direction.toString(),
|
|
||||||
agent: peer.info.agentVersion, # Fields `agent` and `proto` are not part
|
|
||||||
proto: peer.info.protoVersion # of specification
|
|
||||||
)
|
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_version") do () -> JsonNode:
|
rpcServer.rpc("get_v1_node_version") do () -> JsonNode:
|
||||||
return %*{"version": "Nimbus/" & fullVersionStr}
|
return %{
|
||||||
|
"version": "Nimbus/" & fullVersionStr
|
||||||
|
}
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_syncing") do () -> SyncInfo:
|
rpcServer.rpc("get_v1_node_syncing") do () -> JsonNode:
|
||||||
return node.syncManager.getInfo()
|
unimplemented()
|
||||||
|
|
||||||
rpcServer.rpc("get_v1_node_health") do () -> JsonNode:
|
rpcServer.rpc("get_v1_node_health") do () -> JsonNode:
|
||||||
# TODO: There currently no way to situation when we node has issues, so
|
unimplemented()
|
||||||
# its impossible to return HTTP ERROR 503 according to specification.
|
|
||||||
if node.syncManager.inProgress:
|
|
||||||
# We need to return HTTP ERROR 206 according to specification
|
|
||||||
return %*{"health": 206}
|
|
||||||
else:
|
|
||||||
return %*{"health": 200}
|
|
||||||
|
|
|
@ -5,7 +5,9 @@ import
|
||||||
# TODO for some reason "../[datatypes, digest, crypto]" results in "Error: cannot open file"
|
# TODO for some reason "../[datatypes, digest, crypto]" results in "Error: cannot open file"
|
||||||
../datatypes,
|
../datatypes,
|
||||||
../digest,
|
../digest,
|
||||||
../crypto
|
../crypto,
|
||||||
|
libp2p/[peerid, multiaddress],
|
||||||
|
eth/p2p/discoveryv5/enr
|
||||||
|
|
||||||
type
|
type
|
||||||
AttesterDuties* = tuple
|
AttesterDuties* = tuple
|
||||||
|
@ -44,23 +46,8 @@ type
|
||||||
header: SignedBeaconBlockHeader
|
header: SignedBeaconBlockHeader
|
||||||
|
|
||||||
NodeIdentityTuple* = tuple
|
NodeIdentityTuple* = tuple
|
||||||
peer_id: string
|
peer_id: PeerID
|
||||||
enr: string
|
enr: Record
|
||||||
p2p_addresses: seq[string]
|
p2p_addresses: seq[MultiAddress]
|
||||||
discovery_addresses: seq[string]
|
discovery_addresses: seq[MultiAddress]
|
||||||
metadata: tuple[seq_number: uint64, attnets: string]
|
metadata: tuple[seq_number: uint64, attnets: string]
|
||||||
|
|
||||||
NodePeerTuple* = tuple
|
|
||||||
peer_id: string
|
|
||||||
enr: string
|
|
||||||
last_seen_p2p_address: string
|
|
||||||
state: string
|
|
||||||
direction: string
|
|
||||||
agent: string # This is not part of specification
|
|
||||||
proto: string # This is not part of specification
|
|
||||||
|
|
||||||
NodePeerCountTuple* = tuple
|
|
||||||
disconnected: int
|
|
||||||
connecting: int
|
|
||||||
connected: int
|
|
||||||
disconnecting: int
|
|
||||||
|
|
|
@ -129,10 +129,6 @@ type
|
||||||
peer*: T
|
peer*: T
|
||||||
stamp*: chronos.Moment
|
stamp*: chronos.Moment
|
||||||
|
|
||||||
SyncInfo* = object
|
|
||||||
head_slot*: Slot
|
|
||||||
sync_distance*: int64
|
|
||||||
|
|
||||||
SyncManagerError* = object of CatchableError
|
SyncManagerError* = object of CatchableError
|
||||||
BeaconBlocksRes* = NetRes[seq[SignedBeaconBlock]]
|
BeaconBlocksRes* = NetRes[seq[SignedBeaconBlock]]
|
||||||
|
|
||||||
|
@ -1181,9 +1177,3 @@ proc syncLoop[A, B](man: SyncManager[A, B]) {.async.} =
|
||||||
proc start*[A, B](man: SyncManager[A, B]) =
|
proc start*[A, B](man: SyncManager[A, B]) =
|
||||||
## Starts SyncManager's main loop.
|
## Starts SyncManager's main loop.
|
||||||
man.syncFut = man.syncLoop()
|
man.syncFut = man.syncLoop()
|
||||||
|
|
||||||
proc getInfo*[A, B](man: SyncManager[A, B]): SyncInfo =
|
|
||||||
## Returns current synchronization information for RPC call.
|
|
||||||
let wallSlot = man.getLocalWallSlot()
|
|
||||||
let headSlot = man.getLocalHeadSlot()
|
|
||||||
SyncInfo(head_slot: headSlot, sync_distance: int64(wallSlot - headSlot))
|
|
||||||
|
|
|
@ -239,10 +239,6 @@ proc handleStatus(peer: Peer,
|
||||||
await peer.disconnect(IrrelevantNetwork)
|
await peer.disconnect(IrrelevantNetwork)
|
||||||
else:
|
else:
|
||||||
peer.setStatusMsg(theirStatus)
|
peer.setStatusMsg(theirStatus)
|
||||||
if peer.connectionState == Connecting:
|
|
||||||
# As soon as we get here it means that we passed handshake succesfully. So
|
|
||||||
# we can add this peer to PeerPool.
|
|
||||||
await peer.handlePeer()
|
|
||||||
|
|
||||||
proc initBeaconSync*(network: Eth2Node, chainDag: ChainDAGRef,
|
proc initBeaconSync*(network: Eth2Node, chainDag: ChainDAGRef,
|
||||||
forkDigest: ForkDigest) =
|
forkDigest: ForkDigest) =
|
||||||
|
|
|
@ -1 +1 @@
|
||||||
Subproject commit 8709ef9ed5e31e71a1c960237d2eb2c23a2adcd2
|
Subproject commit ac9b3e304f630a450efc996f47dc9e6133246a87
|
Loading…
Reference in New Issue