1782 lines
63 KiB
Nim
1782 lines
63 KiB
Nim
# Fluffy
|
|
# Copyright (c) 2021-2024 Status Research & Development GmbH
|
|
# Licensed and distributed under either of
|
|
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
|
|
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
|
|
# at your option. This file may not be copied, modified, or distributed except according to those terms.
|
|
|
|
## Implementation of the Portal wire protocol as specified at:
|
|
## https://github.com/ethereum/portal-network-specs/blob/master/portal-wire-protocol.md
|
|
|
|
{.push raises: [].}
|
|
|
|
import
|
|
std/[sequtils, sets, algorithm, tables],
|
|
stew/[byteutils, leb128, endians2],
|
|
results,
|
|
chronicles,
|
|
chronos,
|
|
nimcrypto/hash,
|
|
bearssl,
|
|
ssz_serialization,
|
|
metrics,
|
|
faststreams,
|
|
eth/rlp,
|
|
eth/p2p/discoveryv5/
|
|
[protocol, node, enr, routing_table, random2, nodes_verification, lru],
|
|
"."/[portal_stream, portal_protocol_config],
|
|
./messages
|
|
|
|
export messages, routing_table, protocol
|
|
|
|
declareCounter portal_message_requests_incoming,
|
|
"Portal wire protocol incoming message requests",
|
|
labels = ["protocol_id", "message_type"]
|
|
declareCounter portal_message_decoding_failures,
|
|
"Portal wire protocol message decoding failures", labels = ["protocol_id"]
|
|
declareCounter portal_message_requests_outgoing,
|
|
"Portal wire protocol outgoing message requests",
|
|
labels = ["protocol_id", "message_type"]
|
|
declareCounter portal_message_response_incoming,
|
|
"Portal wire protocol incoming message responses",
|
|
labels = ["protocol_id", "message_type"]
|
|
|
|
const requestBuckets = [1.0, 3.0, 5.0, 7.0, 9.0, Inf]
|
|
declareHistogram portal_lookup_node_requests,
|
|
"Portal wire protocol amount of requests per node lookup",
|
|
labels = ["protocol_id"],
|
|
buckets = requestBuckets
|
|
declareHistogram portal_lookup_content_requests,
|
|
"Portal wire protocol amount of requests per node lookup",
|
|
labels = ["protocol_id"],
|
|
buckets = requestBuckets
|
|
declareCounter portal_lookup_content_failures,
|
|
"Portal wire protocol content lookup failures", labels = ["protocol_id"]
|
|
|
|
const contentKeysBuckets = [0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, Inf]
|
|
declareHistogram portal_content_keys_offered,
|
|
"Portal wire protocol amount of content keys per offer message send",
|
|
labels = ["protocol_id"],
|
|
buckets = contentKeysBuckets
|
|
declareHistogram portal_content_keys_accepted,
|
|
"Portal wire protocol amount of content keys per accept message received",
|
|
labels = ["protocol_id"],
|
|
buckets = contentKeysBuckets
|
|
declareCounter portal_gossip_offers_successful,
|
|
"Portal wire protocol successful content offers from neighborhood gossip",
|
|
labels = ["protocol_id"]
|
|
declareCounter portal_gossip_offers_failed,
|
|
"Portal wire protocol failed content offers from neighborhood gossip",
|
|
labels = ["protocol_id"]
|
|
declareCounter portal_gossip_with_lookup,
|
|
"Portal wire protocol neighborhood gossip that required a node lookup",
|
|
labels = ["protocol_id"]
|
|
declareCounter portal_gossip_without_lookup,
|
|
"Portal wire protocol neighborhood gossip that did not require a node lookup",
|
|
labels = ["protocol_id"]
|
|
|
|
# Note: These metrics are to get some idea on how many enrs are send on average.
|
|
# Relevant issue: https://github.com/ethereum/portal-network-specs/issues/136
|
|
const enrsBuckets = [0.0, 1.0, 3.0, 5.0, 8.0, 9.0, Inf]
|
|
declareHistogram portal_nodes_enrs_packed,
|
|
"Portal wire protocol amount of enrs packed in a nodes message",
|
|
labels = ["protocol_id"],
|
|
buckets = enrsBuckets
|
|
# This one will currently hit the max numbers because all neighbours are send,
|
|
# not only the ones closer to the content.
|
|
declareHistogram portal_content_enrs_packed,
|
|
"Portal wire protocol amount of enrs packed in a content message",
|
|
labels = ["protocol_id"],
|
|
buckets = enrsBuckets
|
|
|
|
const distanceBuckets = [
|
|
float64 239, 240, 241, 242, 243, 244, 245, 246, 247, 248, 249, 250, 251, 252, 253,
|
|
254, 255, 256,
|
|
]
|
|
declareHistogram portal_find_content_log_distance,
|
|
"Portal wire protocol logarithmic distance of requested content",
|
|
labels = ["protocol_id"],
|
|
buckets = distanceBuckets
|
|
|
|
declareHistogram portal_offer_log_distance,
|
|
"Portal wire protocol logarithmic distance of offered content",
|
|
labels = ["protocol_id"],
|
|
buckets = distanceBuckets
|
|
|
|
logScope:
|
|
topics = "portal_wire"
|
|
|
|
const
|
|
alpha = 3 ## Kademlia concurrency factor
|
|
enrsResultLimit* = 32 ## Maximum amount of ENRs in the total Nodes messages
|
|
## that will be processed
|
|
refreshInterval = 5.minutes ## Interval of launching a random query to
|
|
## refresh the routing table.
|
|
revalidateMax = 10000 ## Revalidation of a peer is done between 0 and this
|
|
## value in milliseconds
|
|
initialLookups = 1 ## Amount of lookups done when populating the routing table
|
|
|
|
# TalkResp message is a response message so the session is established and a
|
|
# regular discv5 packet is assumed for size calculation.
|
|
# Regular message = IV + header + message
|
|
# talkResp message = rlp: [request-id, response]
|
|
talkRespOverhead =
|
|
16 + # IV size
|
|
55 + # header size
|
|
1 + # talkResp msg id
|
|
3 + # rlp encoding outer list, max length will be encoded in 2 bytes
|
|
9 + # request id (max = 8) + 1 byte from rlp encoding byte string
|
|
3 + # rlp encoding response byte string, max length in 2 bytes
|
|
16 # HMAC
|
|
|
|
# These are the concurrent offers per Portal wire protocol that is running.
|
|
# Using the `offerQueue` allows for limiting the amount of offers send and
|
|
# thus how many streams can be started.
|
|
# TODO:
|
|
# More thought needs to go into this as it is currently on a per network
|
|
# basis. Keep it simple like that? Or limit it better at the stream transport
|
|
# level? In the latter case, this might still need to be checked/blocked at
|
|
# the very start of sending the offer, because blocking/waiting too long
|
|
# between the received accept message and actually starting the stream and
|
|
# sending data could give issues due to timeouts on the other side.
|
|
# And then there are still limits to be applied also for FindContent and the
|
|
# incoming directions.
|
|
concurrentOffers = 50
|
|
|
|
type
|
|
ToContentIdHandler* =
|
|
proc(contentKey: ContentKeyByteList): results.Opt[ContentId] {.raises: [], gcsafe.}
|
|
|
|
DbGetHandler* = proc(
|
|
contentKey: ContentKeyByteList, contentId: ContentId
|
|
): results.Opt[seq[byte]] {.raises: [], gcsafe.}
|
|
|
|
DbStoreHandler* = proc(
|
|
contentKey: ContentKeyByteList, contentId: ContentId, content: seq[byte]
|
|
) {.raises: [], gcsafe.}
|
|
|
|
DbRadiusHandler* = proc(): UInt256 {.raises: [], gcsafe.}
|
|
|
|
PortalProtocolId* = array[2, byte]
|
|
|
|
RadiusCache* = LRUCache[NodeId, UInt256]
|
|
|
|
ContentKV* = object
|
|
contentKey*: ContentKeyByteList
|
|
content*: seq[byte]
|
|
|
|
OfferRequestType = enum
|
|
Direct
|
|
Database
|
|
|
|
OfferRequest = object
|
|
dst: Node
|
|
case kind: OfferRequestType
|
|
of Direct:
|
|
contentList: List[ContentKV, contentKeysLimit]
|
|
of Database:
|
|
contentKeys: ContentKeysList
|
|
|
|
PortalProtocol* = ref object of TalkProtocol
|
|
protocolId*: PortalProtocolId
|
|
routingTable*: RoutingTable
|
|
baseProtocol*: protocol.Protocol
|
|
toContentId*: ToContentIdHandler
|
|
dbGet*: DbGetHandler
|
|
dbPut*: DbStoreHandler
|
|
dataRadius*: DbRadiusHandler
|
|
bootstrapRecords*: seq[Record]
|
|
lastLookup: chronos.Moment
|
|
refreshLoop: Future[void]
|
|
revalidateLoop: Future[void]
|
|
stream*: PortalStream
|
|
radiusCache: RadiusCache
|
|
offerQueue: AsyncQueue[OfferRequest]
|
|
offerWorkers: seq[Future[void]]
|
|
disablePoke: bool
|
|
pingTimings: Table[NodeId, chronos.Moment]
|
|
|
|
PortalResult*[T] = Result[T, string]
|
|
|
|
FoundContentKind* = enum
|
|
Nodes
|
|
Content
|
|
|
|
FoundContent* = object
|
|
src*: Node
|
|
case kind*: FoundContentKind
|
|
of Content:
|
|
content*: seq[byte]
|
|
utpTransfer*: bool
|
|
of Nodes:
|
|
nodes*: seq[Node]
|
|
|
|
ContentLookupResult* = object
|
|
content*: seq[byte]
|
|
utpTransfer*: bool
|
|
# List of nodes which do not have requested content, and for which
|
|
# content is in their range
|
|
nodesInterestedInContent*: seq[Node]
|
|
|
|
TraceResponse* = object
|
|
durationMs*: int64
|
|
respondedWith*: seq[NodeId]
|
|
|
|
NodeMetadata* = object
|
|
enr*: Record
|
|
distance*: UInt256
|
|
|
|
TraceObject* = object
|
|
origin*: NodeId
|
|
targetId: UInt256
|
|
receivedFrom*: Opt[NodeId]
|
|
responses*: Table[string, TraceResponse]
|
|
metadata*: Table[string, NodeMetadata]
|
|
cancelled*: seq[NodeId]
|
|
startedAtMs*: int64
|
|
|
|
TraceContentLookupResult* = object
|
|
content*: Opt[seq[byte]]
|
|
utpTransfer*: bool
|
|
trace*: TraceObject
|
|
|
|
func init*(T: type ContentKV, contentKey: ContentKeyByteList, content: seq[byte]): T =
|
|
ContentKV(contentKey: contentKey, content: content)
|
|
|
|
func init*(
|
|
T: type ContentLookupResult,
|
|
content: seq[byte],
|
|
utpTransfer: bool,
|
|
nodesInterestedInContent: seq[Node],
|
|
): T =
|
|
ContentLookupResult(
|
|
content: content,
|
|
utpTransfer: utpTransfer,
|
|
nodesInterestedInContent: nodesInterestedInContent,
|
|
)
|
|
|
|
func getProtocolId*(
|
|
network: PortalNetwork, subnetwork: PortalSubnetwork
|
|
): PortalProtocolId =
|
|
const portalPrefix = byte(0x50)
|
|
|
|
case network
|
|
of PortalNetwork.none, PortalNetwork.mainnet:
|
|
case subnetwork
|
|
of PortalSubnetwork.state:
|
|
[portalPrefix, 0x0A]
|
|
of PortalSubnetwork.history:
|
|
[portalPrefix, 0x0B]
|
|
of PortalSubnetwork.beacon:
|
|
[portalPrefix, 0x0C]
|
|
of PortalSubnetwork.transactionIndex:
|
|
[portalPrefix, 0x0D]
|
|
of PortalSubnetwork.verkleState:
|
|
[portalPrefix, 0x0E]
|
|
of PortalSubnetwork.transactionGossip:
|
|
[portalPrefix, 0x0F]
|
|
of PortalNetwork.angelfood:
|
|
case subnetwork
|
|
of PortalSubnetwork.state:
|
|
[portalPrefix, 0x4A]
|
|
of PortalSubnetwork.history:
|
|
[portalPrefix, 0x4B]
|
|
of PortalSubnetwork.beacon:
|
|
[portalPrefix, 0x4C]
|
|
of PortalSubnetwork.transactionIndex:
|
|
[portalPrefix, 0x4D]
|
|
of PortalSubnetwork.verkleState:
|
|
[portalPrefix, 0x4E]
|
|
of PortalSubnetwork.transactionGossip:
|
|
[portalPrefix, 0x4F]
|
|
|
|
func `$`(id: PortalProtocolId): string =
|
|
id.toHex()
|
|
|
|
proc addNode*(p: PortalProtocol, node: Node): NodeStatus =
|
|
p.routingTable.addNode(node)
|
|
|
|
proc addNode*(p: PortalProtocol, r: Record): bool =
|
|
p.addNode(Node.fromRecord(r)) == Added
|
|
|
|
func getNode*(p: PortalProtocol, id: NodeId): Opt[Node] =
|
|
p.routingTable.getNode(id)
|
|
|
|
func localNode*(p: PortalProtocol): Node =
|
|
p.baseProtocol.localNode
|
|
|
|
func neighbours*(p: PortalProtocol, id: NodeId, seenOnly = false): seq[Node] =
|
|
p.routingTable.neighbours(id = id, seenOnly = seenOnly)
|
|
|
|
func distance(p: PortalProtocol, a, b: NodeId): UInt256 =
|
|
p.routingTable.distance(a, b)
|
|
|
|
func logDistance(p: PortalProtocol, a, b: NodeId): uint16 =
|
|
p.routingTable.logDistance(a, b)
|
|
|
|
func inRange(
|
|
p: PortalProtocol, nodeId: NodeId, nodeRadius: UInt256, contentId: ContentId
|
|
): bool =
|
|
let distance = p.distance(nodeId, contentId)
|
|
distance <= nodeRadius
|
|
|
|
proc inRange*(p: PortalProtocol, contentId: ContentId): bool =
|
|
p.inRange(p.localNode.id, p.dataRadius(), contentId)
|
|
|
|
func truncateEnrs(
|
|
nodes: seq[Node], maxSize: int, enrOverhead: int
|
|
): List[ByteList[2048], 32] =
|
|
var enrs: List[ByteList[2048], 32]
|
|
var totalSize = 0
|
|
for n in nodes:
|
|
let enr = ByteList[2048].init(n.record.raw)
|
|
if totalSize + enr.len() + enrOverhead <= maxSize:
|
|
let res = enrs.add(enr)
|
|
# With max payload of discv5 and the sizes of ENRs this should not occur.
|
|
doAssert(res, "32 limit will not be reached")
|
|
totalSize = totalSize + enr.len() + enrOverhead
|
|
else:
|
|
break
|
|
|
|
enrs
|
|
|
|
proc handlePing(p: PortalProtocol, ping: PingMessage, srcId: NodeId): seq[byte] =
|
|
# TODO: This should become custom per Portal Network
|
|
# TODO: Need to think about the effect of malicious actor sending lots of
|
|
# pings from different nodes to clear the LRU.
|
|
let customPayloadDecoded =
|
|
try:
|
|
SSZ.decode(ping.customPayload.asSeq(), CustomPayload)
|
|
except SerializationError:
|
|
# invalid custom payload, send empty back
|
|
return @[]
|
|
p.radiusCache.put(srcId, customPayloadDecoded.dataRadius)
|
|
|
|
let customPayload = CustomPayload(dataRadius: p.dataRadius())
|
|
let p = PongMessage(
|
|
enrSeq: p.localNode.record.seqNum,
|
|
customPayload: ByteList[2048](SSZ.encode(customPayload)),
|
|
)
|
|
|
|
encodeMessage(p)
|
|
|
|
proc handleFindNodes(p: PortalProtocol, fn: FindNodesMessage): seq[byte] =
|
|
if fn.distances.len == 0:
|
|
let enrs = List[ByteList[2048], 32](@[])
|
|
encodeMessage(NodesMessage(total: 1, enrs: enrs))
|
|
elif fn.distances.contains(0):
|
|
# A request for our own record.
|
|
let enr = ByteList[2048](rlp.encode(p.localNode.record))
|
|
encodeMessage(NodesMessage(total: 1, enrs: List[ByteList[2048], 32](@[enr])))
|
|
else:
|
|
let distances = fn.distances.asSeq()
|
|
if distances.all(
|
|
proc(x: uint16): bool =
|
|
return x <= 256
|
|
):
|
|
let nodes = p.routingTable.neighboursAtDistances(distances, seenOnly = true)
|
|
|
|
# TODO: Total amount of messages is set fixed to 1 for now, else we would
|
|
# need to either move the send of the talkresp messages here, or allow for
|
|
# returning multiple messages.
|
|
# On the long run, it might just be better to use a stream in these cases?
|
|
# Size calculation is done to truncate the ENR results in order to not go
|
|
# over the discv5 packet size limits. ENRs are sorted so the closest nodes
|
|
# will still be passed.
|
|
const
|
|
nodesOverhead = 1 + 1 + 4 # msg id + total + container offset
|
|
maxPayloadSize = maxDiscv5PacketSize - talkRespOverhead - nodesOverhead
|
|
enrOverhead = 4 # per added ENR, 4 bytes offset overhead
|
|
|
|
let enrs = truncateEnrs(nodes, maxPayloadSize, enrOverhead)
|
|
portal_nodes_enrs_packed.observe(enrs.len().int64, labelValues = [$p.protocolId])
|
|
|
|
encodeMessage(NodesMessage(total: 1, enrs: enrs))
|
|
else:
|
|
# invalid request, send empty back
|
|
let enrs = List[ByteList[2048], 32](@[])
|
|
encodeMessage(NodesMessage(total: 1, enrs: enrs))
|
|
|
|
proc handleFindContent(
|
|
p: PortalProtocol, fc: FindContentMessage, srcId: NodeId
|
|
): seq[byte] =
|
|
const
|
|
contentOverhead = 1 + 1 # msg id + SSZ Union selector
|
|
maxPayloadSize = maxDiscv5PacketSize - talkRespOverhead - contentOverhead
|
|
enrOverhead = 4 # per added ENR, 4 bytes offset overhead
|
|
|
|
let contentId = p.toContentId(fc.contentKey).valueOr:
|
|
# Return empty response when content key validation fails
|
|
# TODO: Better would be to return no message at all? Needs changes on
|
|
# discv5 layer.
|
|
return @[]
|
|
|
|
let logDistance = p.logDistance(contentId, p.localNode.id)
|
|
portal_find_content_log_distance.observe(
|
|
int64(logDistance), labelValues = [$p.protocolId]
|
|
)
|
|
|
|
# Check first if content is in range, as this is a cheaper operation
|
|
if p.inRange(contentId):
|
|
let contentResult = p.dbGet(fc.contentKey, contentId)
|
|
if contentResult.isOk():
|
|
let content = contentResult.get()
|
|
if content.len <= maxPayloadSize:
|
|
return encodeMessage(
|
|
ContentMessage(
|
|
contentMessageType: contentType, content: ByteList[2048](content)
|
|
)
|
|
)
|
|
else:
|
|
let connectionId = p.stream.addContentRequest(srcId, content)
|
|
|
|
return encodeMessage(
|
|
ContentMessage(
|
|
contentMessageType: connectionIdType, connectionId: connectionId
|
|
)
|
|
)
|
|
|
|
# Node does not have the content, or content is not even in radius,
|
|
# send closest neighbours to the requested content id.
|
|
let
|
|
closestNodes = p.routingTable.neighbours(NodeId(contentId), seenOnly = true)
|
|
enrs = truncateEnrs(closestNodes, maxPayloadSize, enrOverhead)
|
|
portal_content_enrs_packed.observe(enrs.len().int64, labelValues = [$p.protocolId])
|
|
|
|
encodeMessage(ContentMessage(contentMessageType: enrsType, enrs: enrs))
|
|
|
|
proc handleOffer(p: PortalProtocol, o: OfferMessage, srcId: NodeId): seq[byte] =
|
|
# Early return when our contentQueue is full. This means there is a backlog
|
|
# of content to process and potentially gossip around. Don't accept more
|
|
# data in this case.
|
|
if p.stream.contentQueue.full():
|
|
return encodeMessage(
|
|
AcceptMessage(
|
|
connectionId: Bytes2([byte 0x00, 0x00]),
|
|
contentKeys: ContentKeysBitList.init(o.contentKeys.len),
|
|
)
|
|
)
|
|
|
|
var contentKeysBitList = ContentKeysBitList.init(o.contentKeys.len)
|
|
var contentKeys = ContentKeysList.init(@[])
|
|
# TODO: Do we need some protection against a peer offering lots (64x) of
|
|
# content that fits our Radius but is actually bogus?
|
|
# Additional TODO, but more of a specification clarification: What if we don't
|
|
# want any of the content? Reply with empty bitlist and a connectionId of
|
|
# all zeroes but don't actually allow an uTP connection?
|
|
for i, contentKey in o.contentKeys:
|
|
let contentIdResult = p.toContentId(contentKey)
|
|
if contentIdResult.isOk():
|
|
let contentId = contentIdResult.get()
|
|
|
|
let logDistance = p.logDistance(contentId, p.localNode.id)
|
|
portal_offer_log_distance.observe(
|
|
int64(logDistance), labelValues = [$p.protocolId]
|
|
)
|
|
|
|
if p.inRange(contentId):
|
|
if p.dbGet(contentKey, contentId).isErr:
|
|
contentKeysBitList.setBit(i)
|
|
discard contentKeys.add(contentKey)
|
|
else:
|
|
# Return empty response when content key validation fails
|
|
return @[]
|
|
|
|
let connectionId =
|
|
if contentKeysBitList.countOnes() != 0:
|
|
p.stream.addContentOffer(srcId, contentKeys)
|
|
else:
|
|
# When the node does not accept any of the content offered, reply with an
|
|
# all zeroes bitlist and connectionId.
|
|
# Note: What to do in this scenario is not defined in the Portal spec.
|
|
Bytes2([byte 0x00, 0x00])
|
|
|
|
encodeMessage(
|
|
AcceptMessage(connectionId: connectionId, contentKeys: contentKeysBitList)
|
|
)
|
|
|
|
proc messageHandler(
|
|
protocol: TalkProtocol,
|
|
request: seq[byte],
|
|
srcId: NodeId,
|
|
srcUdpAddress: Address,
|
|
nodeOpt: Opt[Node],
|
|
): seq[byte] =
|
|
doAssert(protocol of PortalProtocol)
|
|
|
|
logScope:
|
|
protocolId = p.protocolId
|
|
|
|
let p = PortalProtocol(protocol)
|
|
|
|
let decoded = decodeMessage(request)
|
|
if decoded.isOk():
|
|
let message = decoded.get()
|
|
trace "Received message request", srcId, srcUdpAddress, kind = message.kind
|
|
# Received a proper Portal message, check first if an ENR is provided by
|
|
# the discovery v5 layer and add it to the portal network routing table.
|
|
# If not provided through the handshake, try to get it from the discovery v5
|
|
# routing table.
|
|
# When the node would be eligable for the portal network routing table, it
|
|
# is possible that it exists in the base discv5 routing table as the same
|
|
# node ids are used. It is not certain at all however as more nodes might
|
|
# exists on the base layer, and it will also depend on the distance,
|
|
# order of lookups, etc.
|
|
# Note: As third measure, could run a findNodes request with distance 0.
|
|
if nodeOpt.isSome():
|
|
let node = nodeOpt.value()
|
|
let status = p.addNode(node)
|
|
trace "Adding new node to routing table after incoming request", status, node
|
|
else:
|
|
let nodeOpt = p.baseProtocol.getNode(srcId)
|
|
if nodeOpt.isSome():
|
|
let node = nodeOpt.value()
|
|
let status = p.addNode(node)
|
|
trace "Adding new node to routing table after incoming request", status, node
|
|
|
|
portal_message_requests_incoming.inc(labelValues = [$p.protocolId, $message.kind])
|
|
|
|
case message.kind
|
|
of MessageKind.ping:
|
|
p.handlePing(message.ping, srcId)
|
|
of MessageKind.findNodes:
|
|
p.handleFindNodes(message.findNodes)
|
|
of MessageKind.findContent:
|
|
p.handleFindContent(message.findContent, srcId)
|
|
of MessageKind.offer:
|
|
p.handleOffer(message.offer, srcId)
|
|
else:
|
|
# This would mean a that Portal wire response message is being send over a
|
|
# discv5 talkreq message.
|
|
debug "Invalid Portal wire message type over talkreq", kind = message.kind
|
|
@[]
|
|
else:
|
|
portal_message_decoding_failures.inc(labelValues = [$p.protocolId])
|
|
debug "Packet decoding error", error = decoded.error, srcId, srcUdpAddress
|
|
@[]
|
|
|
|
proc new*(
|
|
T: type PortalProtocol,
|
|
baseProtocol: protocol.Protocol,
|
|
protocolId: PortalProtocolId,
|
|
toContentId: ToContentIdHandler,
|
|
dbGet: DbGetHandler,
|
|
dbRadius: DbRadiusHandler,
|
|
stream: PortalStream,
|
|
bootstrapRecords: openArray[Record] = [],
|
|
distanceCalculator: DistanceCalculator = XorDistanceCalculator,
|
|
config: PortalProtocolConfig = defaultPortalProtocolConfig,
|
|
): T =
|
|
let proto = PortalProtocol(
|
|
protocolHandler: messageHandler,
|
|
protocolId: protocolId,
|
|
routingTable: RoutingTable.init(
|
|
baseProtocol.localNode, config.bitsPerHop, config.tableIpLimits, baseProtocol.rng,
|
|
distanceCalculator,
|
|
),
|
|
baseProtocol: baseProtocol,
|
|
toContentId: toContentId,
|
|
dbGet: dbGet,
|
|
dataRadius: dbRadius,
|
|
bootstrapRecords: @bootstrapRecords,
|
|
stream: stream,
|
|
radiusCache: RadiusCache.init(256),
|
|
offerQueue: newAsyncQueue[OfferRequest](concurrentOffers),
|
|
disablePoke: config.disablePoke,
|
|
pingTimings: Table[NodeId, chronos.Moment](),
|
|
)
|
|
|
|
proto.baseProtocol.registerTalkProtocol(@(proto.protocolId), proto).expect(
|
|
"Only one protocol should have this id"
|
|
)
|
|
|
|
proto
|
|
|
|
# Sends the discv5 talkreq message with provided Portal message, awaits and
|
|
# validates the proper response, and updates the Portal Network routing table.
|
|
proc reqResponse[Request: SomeMessage, Response: SomeMessage](
|
|
p: PortalProtocol, dst: Node, request: Request
|
|
): Future[PortalResult[Response]] {.async: (raises: [CancelledError]).} =
|
|
logScope:
|
|
protocolId = p.protocolId
|
|
|
|
trace "Send message request", dstId = dst.id, kind = messageKind(Request)
|
|
portal_message_requests_outgoing.inc(
|
|
labelValues = [$p.protocolId, $messageKind(Request)]
|
|
)
|
|
|
|
let talkresp =
|
|
await talkReq(p.baseProtocol, dst, @(p.protocolId), encodeMessage(request))
|
|
|
|
# Note: Failure of `decodeMessage` might also simply mean that the peer is
|
|
# not supporting the specific talk protocol, as according to specification
|
|
# an empty response needs to be send in that case.
|
|
# See: https://github.com/ethereum/devp2p/blob/master/discv5/discv5-wire.md#talkreq-request-0x05
|
|
|
|
let messageResponse = talkresp
|
|
.mapErr(
|
|
proc(x: cstring): string =
|
|
$x
|
|
)
|
|
.flatMap(
|
|
proc(x: seq[byte]): Result[Message, string] =
|
|
decodeMessage(x)
|
|
)
|
|
.flatMap(
|
|
proc(m: Message): Result[Response, string] =
|
|
getInnerMessage[Response](m)
|
|
)
|
|
|
|
if messageResponse.isOk():
|
|
trace "Received message response",
|
|
srcId = dst.id, srcAddress = dst.address, kind = messageKind(Response)
|
|
portal_message_response_incoming.inc(
|
|
labelValues = [$p.protocolId, $messageKind(Response)]
|
|
)
|
|
|
|
p.routingTable.setJustSeen(dst)
|
|
else:
|
|
debug "Error receiving message response",
|
|
error = messageResponse.error, srcId = dst.id, srcAddress = dst.address
|
|
p.pingTimings.del(dst.id)
|
|
p.routingTable.replaceNode(dst)
|
|
|
|
return messageResponse
|
|
|
|
proc pingImpl*(
|
|
p: PortalProtocol, dst: Node
|
|
): Future[PortalResult[PongMessage]] {.async: (raises: [CancelledError]).} =
|
|
let customPayload = CustomPayload(dataRadius: p.dataRadius())
|
|
let ping = PingMessage(
|
|
enrSeq: p.localNode.record.seqNum,
|
|
customPayload: ByteList[2048](SSZ.encode(customPayload)),
|
|
)
|
|
|
|
return await reqResponse[PingMessage, PongMessage](p, dst, ping)
|
|
|
|
proc findNodesImpl*(
|
|
p: PortalProtocol, dst: Node, distances: List[uint16, 256]
|
|
): Future[PortalResult[NodesMessage]] {.async: (raises: [CancelledError]).} =
|
|
let fn = FindNodesMessage(distances: distances)
|
|
|
|
# TODO Add nodes validation
|
|
return await reqResponse[FindNodesMessage, NodesMessage](p, dst, fn)
|
|
|
|
proc findContentImpl*(
|
|
p: PortalProtocol, dst: Node, contentKey: ContentKeyByteList
|
|
): Future[PortalResult[ContentMessage]] {.async: (raises: [CancelledError]).} =
|
|
let fc = FindContentMessage(contentKey: contentKey)
|
|
|
|
return await reqResponse[FindContentMessage, ContentMessage](p, dst, fc)
|
|
|
|
proc offerImpl*(
|
|
p: PortalProtocol, dst: Node, contentKeys: ContentKeysList
|
|
): Future[PortalResult[AcceptMessage]] {.async: (raises: [CancelledError]).} =
|
|
let offer = OfferMessage(contentKeys: contentKeys)
|
|
|
|
return await reqResponse[OfferMessage, AcceptMessage](p, dst, offer)
|
|
|
|
proc recordsFromBytes*(
|
|
rawRecords: List[ByteList[2048], 32]
|
|
): PortalResult[seq[Record]] =
|
|
var records: seq[Record]
|
|
for r in rawRecords.asSeq():
|
|
let record = enr.Record.fromBytes(r.asSeq()).valueOr:
|
|
# If any of the ENRs is invalid, fail immediatly. This is similar as what
|
|
# is done on the discovery v5 layer.
|
|
return err("Deserialization of an ENR failed")
|
|
|
|
records.add(record)
|
|
|
|
ok(records)
|
|
|
|
proc ping*(
|
|
p: PortalProtocol, dst: Node
|
|
): Future[PortalResult[PongMessage]] {.async: (raises: [CancelledError]).} =
|
|
let pongResponse = await p.pingImpl(dst)
|
|
|
|
if pongResponse.isOk():
|
|
# Update last time we pinged this node
|
|
p.pingTimings[dst.id] = now(chronos.Moment)
|
|
|
|
let pong = pongResponse.get()
|
|
# TODO: This should become custom per Portal Network
|
|
let customPayloadDecoded =
|
|
try:
|
|
SSZ.decode(pong.customPayload.asSeq(), CustomPayload)
|
|
except SerializationError:
|
|
# invalid custom payload
|
|
return err("Pong message contains invalid custom payload")
|
|
|
|
p.radiusCache.put(dst.id, customPayloadDecoded.dataRadius)
|
|
|
|
return pongResponse
|
|
|
|
proc findNodes*(
|
|
p: PortalProtocol, dst: Node, distances: seq[uint16]
|
|
): Future[PortalResult[seq[Node]]] {.async: (raises: [CancelledError]).} =
|
|
let nodesMessage = await p.findNodesImpl(dst, List[uint16, 256](distances))
|
|
if nodesMessage.isOk():
|
|
let records = recordsFromBytes(nodesMessage.get().enrs)
|
|
if records.isOk():
|
|
# TODO: distance function is wrong here for state, fix + tests
|
|
return ok(verifyNodesRecords(records.get(), dst, enrsResultLimit, distances))
|
|
else:
|
|
return err(records.error)
|
|
else:
|
|
return err(nodesMessage.error)
|
|
|
|
proc findContent*(
|
|
p: PortalProtocol, dst: Node, contentKey: ContentKeyByteList
|
|
): Future[PortalResult[FoundContent]] {.async: (raises: [CancelledError]).} =
|
|
logScope:
|
|
node = dst
|
|
contentKey
|
|
|
|
let contentMessageResponse = await p.findContentImpl(dst, contentKey)
|
|
|
|
if contentMessageResponse.isOk():
|
|
let m = contentMessageResponse.get()
|
|
case m.contentMessageType
|
|
of connectionIdType:
|
|
let nodeAddress = NodeAddress.init(dst)
|
|
if nodeAddress.isNone():
|
|
# It should not happen as we are already after the succesfull
|
|
# talkreq/talkresp cycle
|
|
error "Trying to connect to node with unknown address", id = dst.id
|
|
return err("Trying to connect to node with unknown address")
|
|
|
|
# uTP protocol uses BE for all values in the header, incl. connection id
|
|
let socket = (
|
|
await p.stream.connectTo(
|
|
nodeAddress.unsafeGet(), uint16.fromBytesBE(m.connectionId)
|
|
)
|
|
).valueOr:
|
|
debug "uTP connection error for find content", error
|
|
return err("Error connecting uTP socket")
|
|
|
|
try:
|
|
# Read all bytes from the socket
|
|
# This will either end with a FIN, or because the read action times out.
|
|
# A FIN does not necessarily mean that the data read is complete.
|
|
# Further validation is required, using a length prefix here might be
|
|
# beneficial for this.
|
|
let readFut = socket.read()
|
|
|
|
readFut.cancelCallback = proc(udate: pointer) {.gcsafe.} =
|
|
debug "Socket read cancelled", socketKey = socket.socketKey
|
|
# In case this `findContent` gets cancelled while reading the data,
|
|
# send a FIN and clean up the socket.
|
|
socket.close()
|
|
|
|
if await readFut.withTimeout(p.stream.contentReadTimeout):
|
|
let content = await readFut
|
|
# socket received remote FIN and drained whole buffer, it can be
|
|
# safely destroyed without notifing remote
|
|
debug "Socket read fully", socketKey = socket.socketKey
|
|
socket.destroy()
|
|
return ok(
|
|
FoundContent(src: dst, kind: Content, content: content, utpTransfer: true)
|
|
)
|
|
else:
|
|
debug "Socket read time-out", socketKey = socket.socketKey
|
|
# Note: This might look a bit strange, but not doing a socket.close()
|
|
# here as this is already done internally. utp_socket `checkTimeouts`
|
|
# already does a socket.destroy() on timeout. Might want to change the
|
|
# API on this later though.
|
|
return err("Reading data from socket timed out, content request failed")
|
|
except CancelledError as exc:
|
|
# even though we already installed cancelCallback on readFut, it is worth
|
|
# catching CancelledError in case that withTimeout throws CancelledError
|
|
# but readFut have already finished.
|
|
debug "Socket read cancelled", socketKey = socket.socketKey
|
|
|
|
socket.close()
|
|
raise exc
|
|
of contentType:
|
|
return ok(
|
|
FoundContent(
|
|
src: dst, kind: Content, content: m.content.asSeq(), utpTransfer: false
|
|
)
|
|
)
|
|
of enrsType:
|
|
let records = recordsFromBytes(m.enrs)
|
|
if records.isOk():
|
|
let verifiedNodes = verifyNodesRecords(records.get(), dst, enrsResultLimit)
|
|
|
|
return ok(FoundContent(src: dst, kind: Nodes, nodes: verifiedNodes))
|
|
else:
|
|
return err("Content message returned invalid ENRs")
|
|
else:
|
|
warn "FindContent failed due to find content request failure ",
|
|
error = contentMessageResponse.error
|
|
|
|
return err("No content response")
|
|
|
|
proc getContentKeys(o: OfferRequest): ContentKeysList =
|
|
case o.kind
|
|
of Direct:
|
|
var contentKeys: ContentKeysList
|
|
for info in o.contentList:
|
|
discard contentKeys.add(info.contentKey)
|
|
return contentKeys
|
|
of Database:
|
|
return o.contentKeys
|
|
|
|
func getMaxOfferedContentKeys*(protocolIdLen: uint32, maxKeySize: uint32): int =
|
|
## Calculates how many ContentKeys will fit in one offer message which
|
|
## will be small enouch to fit into discv5 limit.
|
|
## This is neccesarry as contentKeysLimit (64) is sometimes to big, and even
|
|
## half of this can be too much to fit into discv5 limits.
|
|
|
|
let maxTalkReqPayload = maxDiscv5PacketSize - getTalkReqOverhead(int(protocolIdLen))
|
|
# To calculate how much bytes, `n` content keys of size `maxKeySize` will take
|
|
# we can use following equation:
|
|
# bytes = (n * (maxKeySize + perContentKeyOverhead)) + offerMessageOverhead
|
|
# to calculate maximal number of keys which will given space this can be
|
|
# transformed to:
|
|
# n = trunc((bytes - offerMessageOverhead) / (maxKeySize + perContentKeyOverhead))
|
|
return ((maxTalkReqPayload - 5) div (int(maxKeySize) + 4))
|
|
|
|
proc offer(
|
|
p: PortalProtocol, o: OfferRequest
|
|
): Future[PortalResult[ContentKeysBitList]] {.async: (raises: [CancelledError]).} =
|
|
## Offer triggers offer-accept interaction with one peer
|
|
## Whole flow has two phases:
|
|
## 1. Come to an agreement on what content to transfer, by using offer and
|
|
## accept messages.
|
|
## 2. Open uTP stream from content provider to content receiver and transfer
|
|
## agreed content.
|
|
## There are two types of possible offer requests:
|
|
## Direct - when caller provides content to transfer. This way, content is
|
|
## guaranteed to be transferred as it stays in memory until whole transfer
|
|
## is completed.
|
|
## Database - when caller provides keys of content to be transferred. This
|
|
## way content is provided from database just before it is transferred through
|
|
## uTP socket. This is useful when there is a lot of content to be transferred
|
|
## to many peers, and keeping it all in memory could exhaust node resources.
|
|
## Main drawback is that content may be deleted from the node database
|
|
## by the cleanup process before it will be transferred, so this way does not
|
|
## guarantee content transfer.
|
|
let contentKeys = getContentKeys(o)
|
|
|
|
logScope:
|
|
node = o.dst
|
|
contentKeys
|
|
|
|
debug "Offering content"
|
|
|
|
portal_content_keys_offered.observe(
|
|
contentKeys.len().int64, labelValues = [$p.protocolId]
|
|
)
|
|
|
|
let acceptMessageResponse = await p.offerImpl(o.dst, contentKeys)
|
|
|
|
if acceptMessageResponse.isOk():
|
|
let m = acceptMessageResponse.get()
|
|
|
|
let contentKeysLen =
|
|
case o.kind
|
|
of Direct:
|
|
o.contentList.len()
|
|
of Database:
|
|
o.contentKeys.len()
|
|
|
|
if m.contentKeys.len() != contentKeysLen:
|
|
# TODO:
|
|
# When there is such system, the peer should get scored negatively here.
|
|
error "Accepted content key bitlist has invalid size"
|
|
return err("Accepted content key bitlist has invalid size")
|
|
|
|
let acceptedKeysAmount = m.contentKeys.countOnes()
|
|
portal_content_keys_accepted.observe(
|
|
acceptedKeysAmount.int64, labelValues = [$p.protocolId]
|
|
)
|
|
if acceptedKeysAmount == 0:
|
|
debug "No content accepted"
|
|
# Don't open an uTP stream if no content was requested
|
|
return ok(m.contentKeys)
|
|
|
|
let nodeAddress = NodeAddress.init(o.dst)
|
|
if nodeAddress.isNone():
|
|
# It should not happen as we are already after succesfull talkreq/talkresp
|
|
# cycle
|
|
error "Trying to connect to node with unknown address", id = o.dst.id
|
|
return err("Trying to connect to node with unknown address")
|
|
|
|
let socket = (
|
|
await p.stream.connectTo(
|
|
nodeAddress.unsafeGet(), uint16.fromBytesBE(m.connectionId)
|
|
)
|
|
).valueOr:
|
|
debug "uTP connection error for offer content", error
|
|
return err("Error connecting uTP socket")
|
|
|
|
template lenu32(x: untyped): untyped =
|
|
uint32(len(x))
|
|
|
|
case o.kind
|
|
of Direct:
|
|
for i, b in m.contentKeys:
|
|
if b:
|
|
let content = o.contentList[i].content
|
|
# TODO: stop using faststreams for this
|
|
var output = memoryOutput()
|
|
try:
|
|
output.write(toBytes(content.lenu32, Leb128).toOpenArray())
|
|
output.write(content)
|
|
except IOError as e:
|
|
# This should not happen in case of in-memory streams
|
|
raiseAssert e.msg
|
|
|
|
let dataWritten = (await socket.write(output.getOutput)).valueOr:
|
|
debug "Error writing requested data", error
|
|
# No point in trying to continue writing data
|
|
socket.close()
|
|
return err("Error writing requested data")
|
|
|
|
trace "Offered content item send", dataWritten = dataWritten
|
|
of Database:
|
|
for i, b in m.contentKeys:
|
|
if b:
|
|
let
|
|
contentKey = o.contentKeys[i]
|
|
contentIdResult = p.toContentId(contentKey)
|
|
if contentIdResult.isOk():
|
|
let
|
|
contentId = contentIdResult.get()
|
|
contentResult = p.dbGet(contentKey, contentId)
|
|
|
|
var output = memoryOutput()
|
|
if contentResult.isOk():
|
|
let content = contentResult.get()
|
|
try:
|
|
output.write(toBytes(content.lenu32, Leb128).toOpenArray())
|
|
output.write(content)
|
|
except IOError as e:
|
|
# This should not happen in case of in-memory streams
|
|
raiseAssert e.msg
|
|
else:
|
|
try:
|
|
# When data turns out missing, add a 0 size varint
|
|
output.write(toBytes(0'u8, Leb128).toOpenArray())
|
|
except IOError as e:
|
|
raiseAssert e.msg
|
|
|
|
let dataWritten = (await socket.write(output.getOutput)).valueOr:
|
|
debug "Error writing requested data", error
|
|
# No point in trying to continue writing data
|
|
socket.close()
|
|
return err("Error writing requested data")
|
|
|
|
trace "Offered content item send", dataWritten = dataWritten
|
|
await socket.closeWait()
|
|
debug "Content successfully offered"
|
|
|
|
return ok(m.contentKeys)
|
|
else:
|
|
warn "Offer failed due to accept request failure ",
|
|
error = acceptMessageResponse.error
|
|
return err("No accept response")
|
|
|
|
proc offer*(
|
|
p: PortalProtocol, dst: Node, contentKeys: ContentKeysList
|
|
): Future[PortalResult[ContentKeysBitList]] {.async: (raises: [CancelledError]).} =
|
|
let req = OfferRequest(dst: dst, kind: Database, contentKeys: contentKeys)
|
|
return await p.offer(req)
|
|
|
|
proc offer*(
|
|
p: PortalProtocol, dst: Node, content: seq[ContentKV]
|
|
): Future[PortalResult[ContentKeysBitList]] {.async: (raises: [CancelledError]).} =
|
|
if len(content) > contentKeysLimit:
|
|
return err("Cannot offer more than 64 content items")
|
|
|
|
let contentList = List[ContentKV, contentKeysLimit].init(content)
|
|
let req = OfferRequest(dst: dst, kind: Direct, contentList: contentList)
|
|
return await p.offer(req)
|
|
|
|
proc offerWorker(p: PortalProtocol) {.async: (raises: [CancelledError]).} =
|
|
while true:
|
|
let req = await p.offerQueue.popFirst()
|
|
|
|
let res = await p.offer(req)
|
|
if res.isOk():
|
|
portal_gossip_offers_successful.inc(labelValues = [$p.protocolId])
|
|
else:
|
|
portal_gossip_offers_failed.inc(labelValues = [$p.protocolId])
|
|
|
|
proc offerQueueEmpty*(p: PortalProtocol): bool =
|
|
p.offerQueue.empty()
|
|
|
|
proc lookupWorker(
|
|
p: PortalProtocol, dst: Node, target: NodeId
|
|
): Future[seq[Node]] {.async: (raises: [CancelledError]).} =
|
|
let distances = lookupDistances(target, dst.id)
|
|
let nodesMessage = await p.findNodes(dst, distances)
|
|
if nodesMessage.isOk():
|
|
let nodes = nodesMessage.get()
|
|
# Attempt to add all nodes discovered
|
|
for n in nodes:
|
|
discard p.addNode(n)
|
|
return nodes
|
|
else:
|
|
return @[]
|
|
|
|
proc lookup*(
|
|
p: PortalProtocol, target: NodeId
|
|
): Future[seq[Node]] {.async: (raises: [CancelledError]).} =
|
|
## Perform a lookup for the given target, return the closest n nodes to the
|
|
## target. Maximum value for n is `BUCKET_SIZE`.
|
|
# `closestNodes` holds the k closest nodes to target found, sorted by distance
|
|
# Unvalidated nodes are used for requests as a form of validation.
|
|
var closestNodes = p.routingTable.neighbours(target, BUCKET_SIZE, seenOnly = false)
|
|
|
|
var asked, seen = HashSet[NodeId]()
|
|
asked.incl(p.localNode.id) # No need to ask our own node
|
|
seen.incl(p.localNode.id) # No need to discover our own node
|
|
for node in closestNodes:
|
|
seen.incl(node.id)
|
|
|
|
var pendingQueries = newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](alpha)
|
|
var requestAmount = 0'i64
|
|
|
|
while true:
|
|
var i = 0
|
|
# Doing `alpha` amount of requests at once as long as closer non queried
|
|
# nodes are discovered.
|
|
while i < closestNodes.len and pendingQueries.len < alpha:
|
|
let n = closestNodes[i]
|
|
if not asked.containsOrIncl(n.id):
|
|
pendingQueries.add(p.lookupWorker(n, target))
|
|
requestAmount.inc()
|
|
inc i
|
|
|
|
trace "Pending lookup queries", total = pendingQueries.len
|
|
|
|
if pendingQueries.len == 0:
|
|
break
|
|
|
|
let query =
|
|
try:
|
|
await one(pendingQueries)
|
|
except ValueError:
|
|
raiseAssert("pendingQueries should not have been empty")
|
|
|
|
trace "Got lookup query response"
|
|
|
|
let index = pendingQueries.find(query)
|
|
if index != -1:
|
|
pendingQueries.del(index)
|
|
else:
|
|
error "Resulting query should have been in the pending queries"
|
|
|
|
let nodes = await query
|
|
# TODO: Remove node on timed-out query?
|
|
for n in nodes:
|
|
if not seen.containsOrIncl(n.id):
|
|
# If it wasn't seen before, insert node while remaining sorted
|
|
closestNodes.insert(
|
|
n,
|
|
closestNodes.lowerBound(
|
|
n,
|
|
proc(x: Node, n: Node): int =
|
|
cmp(p.distance(x.id, target), p.distance(n.id, target)),
|
|
),
|
|
)
|
|
|
|
if closestNodes.len > BUCKET_SIZE:
|
|
closestNodes.del(closestNodes.high())
|
|
|
|
portal_lookup_node_requests.observe(requestAmount, labelValues = [$p.protocolId])
|
|
p.lastLookup = now(chronos.Moment)
|
|
return closestNodes
|
|
|
|
proc triggerPoke*(
|
|
p: PortalProtocol,
|
|
nodes: seq[Node],
|
|
contentKey: ContentKeyByteList,
|
|
content: seq[byte],
|
|
) =
|
|
## In order to properly test gossip mechanisms (e.g. in Portal Hive),
|
|
## we need the option to turn off the POKE functionality as it influences
|
|
## how data moves around the network.
|
|
if p.disablePoke:
|
|
return
|
|
## Triggers asynchronous offer-accept interaction to provided nodes.
|
|
## Provided content should be in range of provided nodes.
|
|
for node in nodes:
|
|
if not p.offerQueue.full():
|
|
try:
|
|
let
|
|
contentKV = ContentKV(contentKey: contentKey, content: content)
|
|
list = List[ContentKV, contentKeysLimit].init(@[contentKV])
|
|
req = OfferRequest(dst: node, kind: Direct, contentList: list)
|
|
p.offerQueue.putNoWait(req)
|
|
except AsyncQueueFullError as e:
|
|
# Should not occur as full() check is done.
|
|
raiseAssert(e.msg)
|
|
else:
|
|
# Offer queue is full, do not start more offer-accept interactions
|
|
return
|
|
|
|
# TODO ContentLookup and Lookup look almost exactly the same, also lookups in other
|
|
# networks will probably be very similar. Extract lookup function to separate module
|
|
# and make it more generaic
|
|
proc contentLookup*(
|
|
p: PortalProtocol, target: ContentKeyByteList, targetId: UInt256
|
|
): Future[Opt[ContentLookupResult]] {.async: (raises: [CancelledError]).} =
|
|
## Perform a lookup for the given target, return the closest n nodes to the
|
|
## target. Maximum value for n is `BUCKET_SIZE`.
|
|
# `closestNodes` holds the k closest nodes to target found, sorted by distance
|
|
# Unvalidated nodes are used for requests as a form of validation.
|
|
var closestNodes = p.routingTable.neighbours(targetId, BUCKET_SIZE, seenOnly = false)
|
|
# Shuffling the order of the nodes in order to not always hit the same node
|
|
# first for the same request.
|
|
p.baseProtocol.rng[].shuffle(closestNodes)
|
|
|
|
var asked, seen = HashSet[NodeId]()
|
|
asked.incl(p.localNode.id) # No need to ask our own node
|
|
seen.incl(p.localNode.id) # No need to discover our own node
|
|
for node in closestNodes:
|
|
seen.incl(node.id)
|
|
|
|
var pendingQueries =
|
|
newSeqOfCap[Future[PortalResult[FoundContent]].Raising([CancelledError])](alpha)
|
|
var requestAmount = 0'i64
|
|
|
|
var nodesWithoutContent: seq[Node] = newSeq[Node]()
|
|
|
|
while true:
|
|
var i = 0
|
|
# Doing `alpha` amount of requests at once as long as closer non queried
|
|
# nodes are discovered.
|
|
while i < closestNodes.len and pendingQueries.len < alpha:
|
|
let n = closestNodes[i]
|
|
if not asked.containsOrIncl(n.id):
|
|
pendingQueries.add(p.findContent(n, target))
|
|
requestAmount.inc()
|
|
inc i
|
|
|
|
trace "Pending lookup queries", total = pendingQueries.len
|
|
|
|
if pendingQueries.len == 0:
|
|
break
|
|
|
|
let query =
|
|
try:
|
|
await one(pendingQueries)
|
|
except ValueError:
|
|
raiseAssert("pendingQueries should not have been empty")
|
|
|
|
trace "Got lookup query response"
|
|
|
|
let index = pendingQueries.find(query)
|
|
if index != -1:
|
|
pendingQueries.del(index)
|
|
else:
|
|
error "Resulting query should have been in the pending queries"
|
|
|
|
let contentResult = await query
|
|
if contentResult.isOk():
|
|
let content = contentResult.get()
|
|
|
|
case content.kind
|
|
of Nodes:
|
|
let maybeRadius = p.radiusCache.get(content.src.id)
|
|
if maybeRadius.isSome() and
|
|
p.inRange(content.src.id, maybeRadius.unsafeGet(), targetId):
|
|
# Only return nodes which may be interested in content.
|
|
# No need to check for duplicates in nodesWithoutContent
|
|
# as requests are never made two times to the same node.
|
|
nodesWithoutContent.add(content.src)
|
|
|
|
for n in content.nodes:
|
|
if not seen.containsOrIncl(n.id):
|
|
discard p.addNode(n)
|
|
# If it wasn't seen before, insert node while remaining sorted
|
|
closestNodes.insert(
|
|
n,
|
|
closestNodes.lowerBound(
|
|
n,
|
|
proc(x: Node, n: Node): int =
|
|
cmp(p.distance(x.id, targetId), p.distance(n.id, targetId)),
|
|
),
|
|
)
|
|
|
|
if closestNodes.len > BUCKET_SIZE:
|
|
closestNodes.del(closestNodes.high())
|
|
of Content:
|
|
# cancel any pending queries as the content has been found
|
|
for f in pendingQueries:
|
|
f.cancelSoon()
|
|
portal_lookup_content_requests.observe(
|
|
requestAmount, labelValues = [$p.protocolId]
|
|
)
|
|
return Opt.some(
|
|
ContentLookupResult.init(
|
|
content.content, content.utpTransfer, nodesWithoutContent
|
|
)
|
|
)
|
|
else:
|
|
# TODO: Should we do something with the node that failed responding our
|
|
# query?
|
|
discard
|
|
|
|
portal_lookup_content_failures.inc(labelValues = [$p.protocolId])
|
|
return Opt.none(ContentLookupResult)
|
|
|
|
proc traceContentLookup*(
|
|
p: PortalProtocol, target: ContentKeyByteList, targetId: UInt256
|
|
): Future[TraceContentLookupResult] {.async: (raises: [CancelledError]).} =
|
|
## Perform a lookup for the given target, return the closest n nodes to the
|
|
## target. Maximum value for n is `BUCKET_SIZE`.
|
|
# `closestNodes` holds the k closest nodes to target found, sorted by distance
|
|
# Unvalidated nodes are used for requests as a form of validation.
|
|
var closestNodes = p.routingTable.neighbours(targetId, BUCKET_SIZE, seenOnly = false)
|
|
# Shuffling the order of the nodes in order to not always hit the same node
|
|
# first for the same request.
|
|
p.baseProtocol.rng[].shuffle(closestNodes)
|
|
|
|
let ts = now(chronos.Moment)
|
|
var responses = Table[string, TraceResponse]()
|
|
var metadata = Table[string, NodeMetadata]()
|
|
|
|
var asked, seen = HashSet[NodeId]()
|
|
asked.incl(p.localNode.id) # No need to ask our own node
|
|
seen.incl(p.localNode.id) # No need to discover our own node
|
|
for node in closestNodes:
|
|
seen.incl(node.id)
|
|
|
|
# Local node should be part of the responses
|
|
responses["0x" & $p.localNode.id] =
|
|
TraceResponse(durationMs: 0, respondedWith: seen.toSeq())
|
|
|
|
metadata["0x" & $p.localNode.id] = NodeMetadata(
|
|
enr: p.localNode.record, distance: p.distance(p.localNode.id, targetId)
|
|
)
|
|
|
|
# We should also have metadata for all the closes nodes
|
|
# in order to be able to show cancelled requests
|
|
for cn in closestNodes:
|
|
metadata["0x" & $cn.id] =
|
|
NodeMetadata(enr: cn.record, distance: p.distance(cn.id, targetId))
|
|
|
|
var pendingQueries =
|
|
newSeqOfCap[Future[PortalResult[FoundContent]].Raising([CancelledError])](alpha)
|
|
var pendingNodes = newSeq[Node]()
|
|
var requestAmount = 0'i64
|
|
|
|
var nodesWithoutContent: seq[Node] = newSeq[Node]()
|
|
|
|
while true:
|
|
var i = 0
|
|
# Doing `alpha` amount of requests at once as long as closer non queried
|
|
# nodes are discovered.
|
|
while i < closestNodes.len and pendingQueries.len < alpha:
|
|
let n = closestNodes[i]
|
|
if not asked.containsOrIncl(n.id):
|
|
pendingQueries.add(p.findContent(n, target))
|
|
pendingNodes.add(n)
|
|
requestAmount.inc()
|
|
inc i
|
|
|
|
trace "Pending lookup queries", total = pendingQueries.len
|
|
|
|
if pendingQueries.len == 0:
|
|
break
|
|
|
|
let query =
|
|
try:
|
|
await one(pendingQueries)
|
|
except ValueError:
|
|
raiseAssert("pendingQueries should not have been empty")
|
|
trace "Got lookup query response"
|
|
|
|
let index = pendingQueries.find(query)
|
|
if index != -1:
|
|
pendingQueries.del(index)
|
|
pendingNodes.del(index)
|
|
else:
|
|
error "Resulting query should have been in the pending queries"
|
|
|
|
let contentResult = await query
|
|
|
|
if contentResult.isOk():
|
|
let content = contentResult.get()
|
|
|
|
case content.kind
|
|
of Nodes:
|
|
let duration = chronos.milliseconds(now(chronos.Moment) - ts)
|
|
|
|
let maybeRadius = p.radiusCache.get(content.src.id)
|
|
if maybeRadius.isSome() and
|
|
p.inRange(content.src.id, maybeRadius.unsafeGet(), targetId):
|
|
# Only return nodes which may be interested in content.
|
|
# No need to check for duplicates in nodesWithoutContent
|
|
# as requests are never made two times to the same node.
|
|
nodesWithoutContent.add(content.src)
|
|
|
|
var respondedWith = newSeq[NodeId]()
|
|
|
|
for n in content.nodes:
|
|
let dist = p.distance(n.id, targetId)
|
|
|
|
metadata["0x" & $n.id] = NodeMetadata(enr: n.record, distance: dist)
|
|
respondedWith.add(n.id)
|
|
|
|
if not seen.containsOrIncl(n.id):
|
|
discard p.addNode(n)
|
|
# If it wasn't seen before, insert node while remaining sorted
|
|
closestNodes.insert(
|
|
n,
|
|
closestNodes.lowerBound(
|
|
n,
|
|
proc(x: Node, n: Node): int =
|
|
cmp(p.distance(x.id, targetId), dist),
|
|
),
|
|
)
|
|
|
|
if closestNodes.len > BUCKET_SIZE:
|
|
closestNodes.del(closestNodes.high())
|
|
|
|
let distance = p.distance(content.src.id, targetId)
|
|
|
|
responses["0x" & $content.src.id] =
|
|
TraceResponse(durationMs: duration, respondedWith: respondedWith)
|
|
|
|
metadata["0x" & $content.src.id] =
|
|
NodeMetadata(enr: content.src.record, distance: distance)
|
|
of Content:
|
|
let duration = chronos.milliseconds(now(chronos.Moment) - ts)
|
|
|
|
# cancel any pending queries as the content has been found
|
|
for f in pendingQueries:
|
|
f.cancelSoon()
|
|
portal_lookup_content_requests.observe(
|
|
requestAmount, labelValues = [$p.protocolId]
|
|
)
|
|
|
|
let distance = p.distance(content.src.id, targetId)
|
|
|
|
responses["0x" & $content.src.id] =
|
|
TraceResponse(durationMs: duration, respondedWith: newSeq[NodeId]())
|
|
|
|
metadata["0x" & $content.src.id] =
|
|
NodeMetadata(enr: content.src.record, distance: distance)
|
|
|
|
var pendingNodeIds = newSeq[NodeId]()
|
|
|
|
for pn in pendingNodes:
|
|
pendingNodeIds.add(pn.id)
|
|
metadata["0x" & $pn.id] =
|
|
NodeMetadata(enr: pn.record, distance: p.distance(pn.id, targetId))
|
|
|
|
return TraceContentLookupResult(
|
|
content: Opt.some(content.content),
|
|
utpTransfer: content.utpTransfer,
|
|
trace: TraceObject(
|
|
origin: p.localNode.id,
|
|
targetId: targetId,
|
|
receivedFrom: Opt.some(content.src.id),
|
|
responses: responses,
|
|
metadata: metadata,
|
|
cancelled: pendingNodeIds,
|
|
startedAtMs: chronos.epochNanoSeconds(ts) div 1_000_000,
|
|
# nanoseconds to milliseconds
|
|
),
|
|
)
|
|
else:
|
|
# TODO: Should we do something with the node that failed responding our
|
|
# query?
|
|
discard
|
|
|
|
portal_lookup_content_failures.inc(labelValues = [$p.protocolId])
|
|
return TraceContentLookupResult(
|
|
content: Opt.none(seq[byte]),
|
|
utpTransfer: false,
|
|
trace: TraceObject(
|
|
origin: p.localNode.id,
|
|
targetId: targetId,
|
|
receivedFrom: Opt.none(NodeId),
|
|
responses: responses,
|
|
metadata: metadata,
|
|
cancelled: newSeq[NodeId](),
|
|
startedAtMs: chronos.epochNanoSeconds(ts) div 1_000_000,
|
|
# nanoseconds to milliseconds
|
|
),
|
|
)
|
|
|
|
proc query*(
|
|
p: PortalProtocol, target: NodeId, k = BUCKET_SIZE
|
|
): Future[seq[Node]] {.async: (raises: [CancelledError]).} =
|
|
## Query k nodes for the given target, returns all nodes found, including the
|
|
## nodes queried.
|
|
##
|
|
## This will take k nodes from the routing table closest to target and
|
|
## query them for nodes closest to target. If there are less than k nodes in
|
|
## the routing table, nodes returned by the first queries will be used.
|
|
var queryBuffer = p.routingTable.neighbours(target, k, seenOnly = false)
|
|
|
|
var asked, seen = HashSet[NodeId]()
|
|
asked.incl(p.localNode.id) # No need to ask our own node
|
|
seen.incl(p.localNode.id) # No need to discover our own node
|
|
for node in queryBuffer:
|
|
seen.incl(node.id)
|
|
|
|
var pendingQueries = newSeqOfCap[Future[seq[Node]].Raising([CancelledError])](alpha)
|
|
|
|
while true:
|
|
var i = 0
|
|
while i < min(queryBuffer.len, k) and pendingQueries.len < alpha:
|
|
let n = queryBuffer[i]
|
|
if not asked.containsOrIncl(n.id):
|
|
pendingQueries.add(p.lookupWorker(n, target))
|
|
inc i
|
|
|
|
trace "Pending lookup queries", total = pendingQueries.len
|
|
|
|
if pendingQueries.len == 0:
|
|
break
|
|
|
|
let query =
|
|
try:
|
|
await one(pendingQueries)
|
|
except ValueError:
|
|
raiseAssert("pendingQueries should not have been empty")
|
|
trace "Got lookup query response"
|
|
|
|
let index = pendingQueries.find(query)
|
|
if index != -1:
|
|
pendingQueries.del(index)
|
|
else:
|
|
error "Resulting query should have been in the pending queries"
|
|
|
|
let nodes = await query
|
|
# TODO: Remove node on timed-out query?
|
|
for n in nodes:
|
|
if not seen.containsOrIncl(n.id):
|
|
queryBuffer.add(n)
|
|
|
|
p.lastLookup = now(chronos.Moment)
|
|
return queryBuffer
|
|
|
|
proc queryRandom*(
|
|
p: PortalProtocol
|
|
): Future[seq[Node]] {.async: (raw: true, raises: [CancelledError]).} =
|
|
## Perform a query for a random target, return all nodes discovered.
|
|
p.query(NodeId.random(p.baseProtocol.rng[]))
|
|
|
|
proc getNClosestNodesWithRadius*(
|
|
p: PortalProtocol, targetId: NodeId, n: int, seenOnly: bool = false
|
|
): seq[(Node, UInt256)] =
|
|
let closestLocalNodes =
|
|
p.routingTable.neighbours(targetId, k = n, seenOnly = seenOnly)
|
|
|
|
var nodesWithRadiuses: seq[(Node, UInt256)]
|
|
for node in closestLocalNodes:
|
|
let radius = p.radiusCache.get(node.id)
|
|
if radius.isSome():
|
|
nodesWithRadiuses.add((node, radius.unsafeGet()))
|
|
return nodesWithRadiuses
|
|
|
|
proc neighborhoodGossip*(
|
|
p: PortalProtocol,
|
|
srcNodeId: Opt[NodeId],
|
|
contentKeys: ContentKeysList,
|
|
content: seq[seq[byte]],
|
|
): Future[int] {.async: (raises: [CancelledError]).} =
|
|
## Run neighborhood gossip for provided content.
|
|
## Returns the number of peers to which content was attempted to be gossiped.
|
|
if content.len() == 0:
|
|
return 0
|
|
|
|
var contentList = List[ContentKV, contentKeysLimit].init(@[])
|
|
for i, contentItem in content:
|
|
let contentKV = ContentKV(contentKey: contentKeys[i], content: contentItem)
|
|
discard contentList.add(contentKV)
|
|
|
|
# Just taking the first content item as target id.
|
|
# TODO: come up with something better?
|
|
let contentId = p.toContentId(contentList[0].contentKey).valueOr:
|
|
return 0
|
|
|
|
# For selecting the closest nodes to whom to gossip the content a mixed
|
|
# approach is taken:
|
|
# 1. Select the closest neighbours in the routing table
|
|
# 2. Check if the radius is known for these these nodes and whether they are
|
|
# in range of the content to be offered.
|
|
# 3. If more than n (= 8) nodes are in range, offer these nodes the content
|
|
# (max nodes set at 8).
|
|
# 4. If less than n nodes are in range, do a node lookup, and offer the nodes
|
|
# returned from the lookup the content (max nodes set at 8)
|
|
#
|
|
# This should give a bigger rate of success and avoid the data being stopped
|
|
# in its propagation than when looking only for nodes in the own routing
|
|
# table, but at the same time avoid unnecessary node lookups.
|
|
# It might still cause issues in data getting propagated in a wider id range.
|
|
const maxGossipNodes = 8
|
|
|
|
let closestLocalNodes =
|
|
p.routingTable.neighbours(NodeId(contentId), k = 16, seenOnly = true)
|
|
|
|
var gossipNodes: seq[Node]
|
|
for node in closestLocalNodes:
|
|
let radius = p.radiusCache.get(node.id)
|
|
if radius.isSome():
|
|
if p.inRange(node.id, radius.unsafeGet(), contentId):
|
|
if srcNodeId.isNone:
|
|
gossipNodes.add(node)
|
|
elif node.id != srcNodeId.get():
|
|
gossipNodes.add(node)
|
|
|
|
if gossipNodes.len >= 8: # use local nodes for gossip
|
|
portal_gossip_without_lookup.inc(labelValues = [$p.protocolId])
|
|
let numberOfGossipedNodes = min(gossipNodes.len, maxGossipNodes)
|
|
for node in gossipNodes[0 ..< numberOfGossipedNodes]:
|
|
let req = OfferRequest(dst: node, kind: Direct, contentList: contentList)
|
|
await p.offerQueue.addLast(req)
|
|
return numberOfGossipedNodes
|
|
else: # use looked up nodes for gossip
|
|
portal_gossip_with_lookup.inc(labelValues = [$p.protocolId])
|
|
let closestNodes = await p.lookup(NodeId(contentId))
|
|
let numberOfGossipedNodes = min(closestNodes.len, maxGossipNodes)
|
|
for node in closestNodes[0 ..< numberOfGossipedNodes]:
|
|
# Note: opportunistically not checking if the radius of the node is known
|
|
# and thus if the node is in radius with the content. Reason is, these
|
|
# should really be the closest nodes in the DHT, and thus are most likely
|
|
# going to be in range of the requested content.
|
|
let req = OfferRequest(dst: node, kind: Direct, contentList: contentList)
|
|
await p.offerQueue.addLast(req)
|
|
return numberOfGossipedNodes
|
|
|
|
proc neighborhoodGossipDiscardPeers*(
|
|
p: PortalProtocol,
|
|
srcNodeId: Opt[NodeId],
|
|
contentKeys: ContentKeysList,
|
|
content: seq[seq[byte]],
|
|
): Future[void] {.async: (raises: [CancelledError]).} =
|
|
discard await p.neighborhoodGossip(srcNodeId, contentKeys, content)
|
|
|
|
proc randomGossip*(
|
|
p: PortalProtocol,
|
|
srcNodeId: Opt[NodeId],
|
|
contentKeys: ContentKeysList,
|
|
content: seq[seq[byte]],
|
|
): Future[int] {.async: (raises: [CancelledError]).} =
|
|
## Run random gossip for provided content.
|
|
## Returns the number of peers to which content was attempted to be gossiped.
|
|
if content.len() == 0:
|
|
return 0
|
|
|
|
var contentList = List[ContentKV, contentKeysLimit].init(@[])
|
|
for i, contentItem in content:
|
|
let contentKV = ContentKV(contentKey: contentKeys[i], content: contentItem)
|
|
discard contentList.add(contentKV)
|
|
|
|
const maxGossipNodes = 4
|
|
let nodes = p.routingTable.randomNodes(maxGossipNodes)
|
|
|
|
for node in nodes[0 ..< nodes.len()]:
|
|
let req = OfferRequest(dst: node, kind: Direct, contentList: contentList)
|
|
await p.offerQueue.addLast(req)
|
|
return nodes.len()
|
|
|
|
proc randomGossipDiscardPeers*(
|
|
p: PortalProtocol,
|
|
srcNodeId: Opt[NodeId],
|
|
contentKeys: ContentKeysList,
|
|
content: seq[seq[byte]],
|
|
): Future[void] {.async: (raises: [CancelledError]).} =
|
|
discard await p.randomGossip(srcNodeId, contentKeys, content)
|
|
|
|
proc storeContent*(
|
|
p: PortalProtocol,
|
|
contentKey: ContentKeyByteList,
|
|
contentId: ContentId,
|
|
content: seq[byte],
|
|
) =
|
|
doAssert(p.dbPut != nil)
|
|
p.dbPut(contentKey, contentId, content)
|
|
|
|
proc seedTable*(p: PortalProtocol) =
|
|
## Seed the table with specifically provided Portal bootstrap nodes. These are
|
|
## nodes that must support the wire protocol for the specific content network.
|
|
# Note: We allow replacing the bootstrap nodes in the routing table as it is
|
|
# possible that some of these are not supporting the specific portal network.
|
|
# Other note: One could also pick nodes from the discv5 routing table to
|
|
# bootstrap the portal networks, however it would require a flag in the ENR to
|
|
# be added and there might be none in the routing table due to low amount of
|
|
# Portal nodes versus other nodes.
|
|
logScope:
|
|
protocolId = p.protocolId
|
|
|
|
for record in p.bootstrapRecords:
|
|
if p.addNode(record):
|
|
debug "Added bootstrap node", uri = toURI(record), protocolId = p.protocolId
|
|
else:
|
|
error "Bootstrap node could not be added",
|
|
uri = toURI(record), protocolId = p.protocolId
|
|
|
|
proc populateTable(p: PortalProtocol) {.async: (raises: [CancelledError]).} =
|
|
## Do a set of initial lookups to quickly populate the table.
|
|
# start with a self target query (neighbour nodes)
|
|
logScope:
|
|
protocolId = p.protocolId
|
|
|
|
let selfQuery = await p.query(p.localNode.id)
|
|
trace "Discovered nodes in self target query", nodes = selfQuery.len
|
|
|
|
for i in 0 ..< initialLookups:
|
|
let randomQuery = await p.queryRandom()
|
|
trace "Discovered nodes in random target query", nodes = randomQuery.len
|
|
|
|
debug "Total nodes in routing table after populate", total = p.routingTable.len()
|
|
|
|
proc revalidateNode*(p: PortalProtocol, n: Node) {.async: (raises: [CancelledError]).} =
|
|
let pong = await p.ping(n)
|
|
|
|
if pong.isOk():
|
|
let res = pong.get()
|
|
if res.enrSeq > n.record.seqNum:
|
|
# Request new ENR
|
|
let nodesMessage = await p.findNodes(n, @[0'u16])
|
|
if nodesMessage.isOk():
|
|
let nodes = nodesMessage.get()
|
|
if nodes.len > 0: # Normally a node should only return 1 record actually
|
|
discard p.addNode(nodes[0])
|
|
|
|
proc getNodeForRevalidation(p: PortalProtocol): Opt[Node] =
|
|
let node = p.routingTable.nodeToRevalidate()
|
|
if node.isNil:
|
|
return Opt.none(Node)
|
|
|
|
let now = now(chronos.Moment)
|
|
let timestamp = p.pingTimings.getOrDefault(node.id, now)
|
|
|
|
if (timestamp + revalidationTimeout) <= now:
|
|
Opt.some(node)
|
|
else:
|
|
Opt.none(Node)
|
|
|
|
proc revalidateLoop(p: PortalProtocol) {.async: (raises: []).} =
|
|
## Loop which revalidates the nodes in the routing table by sending the ping
|
|
## message.
|
|
try:
|
|
while true:
|
|
await sleepAsync(milliseconds(p.baseProtocol.rng[].rand(revalidateMax)))
|
|
let n = getNodeForRevalidation(p)
|
|
if n.isSome:
|
|
asyncSpawn p.revalidateNode(n.get())
|
|
except CancelledError:
|
|
trace "revalidateLoop canceled"
|
|
|
|
proc refreshLoop(p: PortalProtocol) {.async: (raises: []).} =
|
|
## Loop that refreshes the routing table by starting a random query in case
|
|
## no queries were done since `refreshInterval` or more.
|
|
## It also refreshes the majority address voted for via pong responses.
|
|
logScope:
|
|
protocolId = p.protocolId
|
|
|
|
try:
|
|
while true:
|
|
# TODO: It would be nicer and more secure if this was event based and/or
|
|
# steered from the routing table.
|
|
while p.routingTable.len() == 0:
|
|
p.seedTable()
|
|
await p.populateTable()
|
|
await sleepAsync(5.seconds)
|
|
|
|
let currentTime = now(chronos.Moment)
|
|
if currentTime > (p.lastLookup + refreshInterval):
|
|
let randomQuery = await p.queryRandom()
|
|
trace "Discovered nodes in random target query", nodes = randomQuery.len
|
|
debug "Total nodes in routing table", total = p.routingTable.len()
|
|
|
|
await sleepAsync(refreshInterval)
|
|
except CancelledError:
|
|
trace "refreshLoop canceled"
|
|
|
|
proc start*(p: PortalProtocol) =
|
|
p.refreshLoop = refreshLoop(p)
|
|
p.revalidateLoop = revalidateLoop(p)
|
|
|
|
for i in 0 ..< concurrentOffers:
|
|
p.offerWorkers.add(offerWorker(p))
|
|
|
|
proc stop*(p: PortalProtocol) =
|
|
if not p.revalidateLoop.isNil:
|
|
p.revalidateLoop.cancelSoon()
|
|
if not p.refreshLoop.isNil:
|
|
p.refreshLoop.cancelSoon()
|
|
|
|
for worker in p.offerWorkers:
|
|
worker.cancelSoon()
|
|
p.offerWorkers = @[]
|
|
|
|
proc resolve*(
|
|
p: PortalProtocol, id: NodeId
|
|
): Future[Opt[Node]] {.async: (raises: [CancelledError]).} =
|
|
## Resolve a `Node` based on provided `NodeId`.
|
|
##
|
|
## This will first look in the own routing table. If the node is known, it
|
|
## will try to contact if for newer information. If node is not known or it
|
|
## does not reply, a lookup is done to see if it can find a (newer) record of
|
|
## the node on the network.
|
|
if id == p.localNode.id:
|
|
return Opt.some(p.localNode)
|
|
|
|
let node = p.getNode(id)
|
|
if node.isSome():
|
|
let nodesMessage = await p.findNodes(node.get(), @[0'u16])
|
|
# TODO: Handle failures better. E.g. stop on different failures than timeout
|
|
if nodesMessage.isOk() and nodesMessage[].len > 0:
|
|
return Opt.some(nodesMessage[][0])
|
|
|
|
let discovered = await p.lookup(id)
|
|
for n in discovered:
|
|
if n.id == id:
|
|
if node.isSome() and node.get().record.seqNum >= n.record.seqNum:
|
|
return node
|
|
else:
|
|
return Opt.some(n)
|
|
|
|
return node
|
|
|
|
proc resolveWithRadius*(
|
|
p: PortalProtocol, id: NodeId
|
|
): Future[Opt[(Node, UInt256)]] {.async: (raises: [CancelledError]).} =
|
|
## Resolve a `Node` based on provided `NodeId`, also try to establish what
|
|
## is known radius of found node.
|
|
##
|
|
## This will first look in the own routing table. If the node is known, it
|
|
## will try to contact if for newer information. If node is not known or it
|
|
## does not reply, a lookup is done to see if it can find a (newer) record of
|
|
## the node on the network.
|
|
##
|
|
## If node is found, radius will be first checked in radius cache, it radius
|
|
## is not known node will be pinged to establish what is its current radius
|
|
##
|
|
|
|
let n = await p.resolve(id)
|
|
if n.isNone():
|
|
return Opt.none((Node, UInt256))
|
|
|
|
let node = n.unsafeGet()
|
|
|
|
let r = p.radiusCache.get(id)
|
|
if r.isSome():
|
|
return Opt.some((node, r.unsafeGet()))
|
|
|
|
let pongResult = await p.ping(node)
|
|
if pongResult.isOk():
|
|
let maybeRadius = p.radiusCache.get(id)
|
|
# After successful ping radius should already be in cache, but for the
|
|
# unlikely case that it is not, check it just to be sure.
|
|
# TODO: refactor ping to return node radius.
|
|
if maybeRadius.isNone():
|
|
return Opt.none((Node, UInt256))
|
|
else:
|
|
return Opt.some((node, maybeRadius.unsafeGet()))
|
|
else:
|
|
return Opt.none((Node, UInt256))
|