Add routing table and required lookup code to Portal protocol (#773)

* Add routing table and required lookup code to Portal protocol

* Enable Portal lookup and validate loops + minor adjustments

fluffy/fluffy.nim

@@ -77,6 +77,7 @@ proc run(config: PortalConf) {.raises: [CatchableError, Defect].} =
   let bridgeClient = initializeBridgeClient(config.bridgeUri)
 
   d.start()
+  portal.start()
 
   runForever()
 

fluffy/network/portal_protocol.nim

@@ -8,9 +8,9 @@
 {.push raises: [Defect].}
 
 import
-  std/sequtils,
+  std/[sequtils, sets, algorithm],
-  stew/[results, byteutils], chronicles,
+  stew/[results, byteutils], chronicles, chronos,
-  eth/rlp, eth/p2p/discoveryv5/[protocol, node, enr],
+  eth/rlp, eth/p2p/discoveryv5/[protocol, node, enr, routing_table, random2],
   ../content,
   ./messages
 
@@ -21,12 +21,32 @@ logScope:
 
 const
   PortalProtocolId* = "portal".toBytes()
+  Alpha = 3 ## Kademlia concurrency factor
+  LookupRequestLimit = 3 ## Amount of distances requested in a single Findnode
+  ## message for a lookup or query
+  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
 
 type
   PortalProtocol* = ref object of TalkProtocol
+    routingTable: RoutingTable
     baseProtocol*: protocol.Protocol
     dataRadius*: UInt256
     contentStorage*: ContentStorage
+    lastLookup: chronos.Moment
+    refreshLoop: Future[void]
+    revalidateLoop: Future[void]
+
+  PortalResult*[T] = Result[T, cstring]
+
+# TODO:
+# - setJustSeen and replaceNode on (all) message replies
+# - On incoming portal ping of unknown node: add node to routing table by
+# grabbing ENR from discv5 routing table (might not have it)?
+# - ENRs with portal protocol capabilities as field?
 
 proc handlePing(p: PortalProtocol, ping: PingMessage):
     seq[byte] =
@@ -47,7 +67,7 @@ proc handleFindNode(p: PortalProtocol, fn: FindNodeMessage): seq[byte] =
     let distances = fn.distances.asSeq()
     if distances.all(proc (x: uint16): bool = return x <= 256):
       let
-        nodes = p.baseProtocol.neighboursAtDistances(distances, seenOnly = true)
+        nodes = p.routingTable.neighboursAtDistances(distances, seenOnly = true)
         enrs = nodes.map(proc(x: Node): ByteList = ByteList(x.record.raw))
 
       # TODO: Fixed here to total message of 1 for now, as else we would need to
@@ -73,7 +93,7 @@ proc handleFindContent(p: PortalProtocol, fc: FindContentMessage): seq[byte] =
       enrs: enrs, payload: ByteList(content.get())))
   else:
     let
-      closestNodes = p.baseProtocol.neighbours(
+      closestNodes = p.routingTable.neighbours(
         NodeId(readUintBE[256](contentId.data)), seenOnly = true)
       payload = ByteList(@[]) # Empty payload when enrs are send
       enrs =
@@ -120,13 +140,16 @@ proc new*(T: type PortalProtocol, baseProtocol: protocol.Protocol,
     baseProtocol: baseProtocol,
     dataRadius: dataRadius)
 
+  proto.routingTable.init(baseProtocol.localNode, DefaultBitsPerHop,
+    DefaultTableIpLimits, baseProtocol.rng)
+
   proto.baseProtocol.registerTalkProtocol(PortalProtocolId, proto).expect(
     "Only one protocol should have this id")
 
   return proto
 
 proc ping*(p: PortalProtocol, dst: Node):
-    Future[DiscResult[PongMessage]] {.async.} =
+    Future[PortalResult[PongMessage]] {.async.} =
   let ping = PingMessage(enrSeq: p.baseProtocol.localNode.record.seqNum,
     dataRadius: p.dataRadius)
 
@@ -137,7 +160,7 @@ proc ping*(p: PortalProtocol, dst: Node):
     encodeMessage(ping))
 
   if talkresp.isOk():
-    let decoded = decodeMessage(talkresp.get().response)
+    let decoded = decodeMessage(talkresp.get())
     if decoded.isOk():
       let message = decoded.get()
       if message.kind == pong:
@@ -150,7 +173,7 @@ proc ping*(p: PortalProtocol, dst: Node):
     return err(talkresp.error)
 
 proc findNode*(p: PortalProtocol, dst: Node, distances: List[uint16, 256]):
-    Future[DiscResult[NodesMessage]] {.async.} =
+    Future[PortalResult[NodesMessage]] {.async.} =
   let fn = FindNodeMessage(distances: distances)
 
   trace "Send message request", dstId = dst.id, kind = MessageKind.findnode
@@ -158,11 +181,11 @@ proc findNode*(p: PortalProtocol, dst: Node, distances: List[uint16, 256]):
     encodeMessage(fn))
 
   if talkresp.isOk():
-    let decoded = decodeMessage(talkresp.get().response)
+    let decoded = decodeMessage(talkresp.get())
     if decoded.isOk():
       let message = decoded.get()
       if message.kind == nodes:
-        # TODO: Verify nodes here
+        # TODO: Verify nodes here?
         return ok(message.nodes)
       else:
         return err("Invalid message response received")
@@ -172,7 +195,7 @@ proc findNode*(p: PortalProtocol, dst: Node, distances: List[uint16, 256]):
     return err(talkresp.error)
 
 proc findContent*(p: PortalProtocol, dst: Node, contentKey: ContentKey):
-    Future[DiscResult[FoundContentMessage]] {.async.} =
+    Future[PortalResult[FoundContentMessage]] {.async.} =
   let fc = FindContentMessage(contentKey: contentKey)
 
   trace "Send message request", dstId = dst.id, kind = MessageKind.findcontent
@@ -180,7 +203,7 @@ proc findContent*(p: PortalProtocol, dst: Node, contentKey: ContentKey):
     encodeMessage(fc))
 
   if talkresp.isOk():
-    let decoded = decodeMessage(talkresp.get().response)
+    let decoded = decodeMessage(talkresp.get())
     if decoded.isOk():
       let message = decoded.get()
       if message.kind == foundcontent:
@@ -191,3 +214,231 @@ proc findContent*(p: PortalProtocol, dst: Node, contentKey: ContentKey):
       return err(decoded.error)
   else:
     return err(talkresp.error)
+
+proc recordsFromBytes(rawRecords: List[ByteList, 32]): seq[Record] =
+  var records: seq[Record]
+  for r in rawRecords.asSeq():
+    var record: Record
+    if record.fromBytes(r.asSeq()):
+      records.add(record)
+
+  records
+
+proc lookupDistances(target, dest: NodeId): seq[uint16] =
+  var distances: seq[uint16]
+  let td = logDist(target, dest)
+  distances.add(td)
+  var i = 1'u16
+  while distances.len < LookupRequestLimit:
+    if td + i < 256:
+      distances.add(td + i)
+    if td - i > 0'u16:
+      distances.add(td - i)
+    inc i
+
+proc lookupWorker(p: PortalProtocol, destNode: Node, target: NodeId):
+    Future[seq[Node]] {.async.} =
+  var nodes: seq[Node]
+  let distances = lookupDistances(target, destNode.id)
+
+  let nodesMessage = await p.findNode(destNode,  List[uint16, 256](distances))
+  if nodesMessage.isOk():
+    let records = recordsFromBytes(nodesMessage.get().enrs)
+    let verifiedNodes = verifyNodesRecords(records, destNode, @[0'u16])
+    nodes.add(verifiedNodes)
+
+    # Attempt to add all nodes discovered
+    for n in nodes:
+      discard p.routingTable.addNode(n)
+
+  return nodes
+
+proc lookup*(p: PortalProtocol, target: NodeId): Future[seq[Node]] {.async.} =
+  ## 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 = initHashSet[NodeId]()
+  asked.incl(p.baseProtocol.localNode.id) # No need to ask our own node
+  seen.incl(p.baseProtocol.localNode.id) # No need to discover our own node
+  for node in closestNodes:
+    seen.incl(node.id)
+
+  var pendingQueries = newSeqOfCap[Future[seq[Node]]](Alpha)
+
+  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))
+      inc i
+
+    trace "Pending lookup queries", total = pendingQueries.len
+
+    if pendingQueries.len == 0:
+      break
+
+    let query = await one(pendingQueries)
+    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 = query.read
+    # 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(distanceTo(x, target), distanceTo(n, target))
+        ))
+
+        if closestNodes.len > BUCKET_SIZE:
+          closestNodes.del(closestNodes.high())
+
+  p.lastLookup = now(chronos.Moment)
+  return closestNodes
+
+proc query*(p: PortalProtocol, target: NodeId, k = BUCKET_SIZE): Future[seq[Node]]
+    {.async.} =
+  ## 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 = initHashSet[NodeId]()
+  asked.incl(p.baseProtocol.localNode.id) # No need to ask our own node
+  seen.incl(p.baseProtocol.localNode.id) # No need to discover our own node
+  for node in queryBuffer:
+    seen.incl(node.id)
+
+  var pendingQueries = newSeqOfCap[Future[seq[Node]]](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 = await one(pendingQueries)
+    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 = query.read
+    # 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]] =
+  ## Perform a query for a random target, return all nodes discovered.
+  p.query(NodeId.random(p.baseProtocol.rng[]))
+
+proc seedTable(p: PortalProtocol) =
+  # TODO: Just picking something here for now. Should definitely add portal
+  # protocol info k:v pair in the ENRs and filter on that.
+  let closestNodes = p.baseProtocol.neighbours(
+    NodeId.random(p.baseProtocol.rng[]), seenOnly = true)
+
+  for node in closestNodes:
+    if p.routingTable.addNode(node) == Added:
+      debug "Added node from discv5 routing table", uri = toURI(node.record)
+    else:
+      debug "Node from discv5 routing table could not be added", uri = toURI(node.record)
+
+proc populateTable(p: PortalProtocol) {.async.} =
+  ## Do a set of initial lookups to quickly populate the table.
+  # start with a self target query (neighbour nodes)
+  let selfQuery = await p.query(p.baseProtocol.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.} =
+  let pong = await p.ping(n)
+
+  if pong.isOK():
+    let res = pong.get()
+    if res.enrSeq > n.record.seqNum:
+      # Request new ENR
+      let nodes = await p.findNode(n, List[uint16, 256](@[0'u16]))
+      if nodes.isOk():
+        let records = recordsFromBytes(nodes.get().enrs)
+        let verifiedNodes = verifyNodesRecords(records, n, @[0'u16])
+        if verifiedNodes.len > 0:
+          discard p.routingTable.addNode(verifiedNodes[0])
+
+proc revalidateLoop(p: PortalProtocol) {.async.} =
+  ## 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 = p.routingTable.nodeToRevalidate()
+      if not n.isNil:
+        asyncSpawn p.revalidateNode(n)
+  except CancelledError:
+    trace "revalidateLoop canceled"
+
+proc refreshLoop(p: PortalProtocol) {.async.} =
+  ## 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.
+  try:
+    await p.populateTable()
+
+    while true:
+      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.seedTable()
+
+  p.refreshLoop = refreshLoop(p)
+  p.revalidateLoop = revalidateLoop(p)
+
+proc stop*(p: PortalProtocol) =
+  if not p.revalidateLoop.isNil:
+    p.revalidateLoop.cancel()
+  if not p.refreshLoop.isNil:
+    p.refreshLoop.cancel()

fluffy/network/portalcli.nim

@@ -215,6 +215,7 @@ proc run(config: DiscoveryConf) =
 
   of noCommand:
     d.start()
+    portal.start()
     waitfor(discover(d))
 
 when isMainModule:

fluffy/tests/test_portal.nim

@@ -62,7 +62,8 @@ procSuite "Portal Tests":
         nodes.get().total == 1'u8
         nodes.get().enrs.len() == 1
 
-    block: # Find nothing
+    block: # Find nothing: this should result in nothing as we haven't started
+      # the seeding of the portal protocol routing table yet.
       let nodes = await proto1.findNode(proto2.baseProtocol.localNode,
         List[uint16, 256](@[]))
 
@@ -72,10 +73,16 @@ procSuite "Portal Tests":
         nodes.get().enrs.len() == 0
 
     block: # Find for distance
-      # ping in one direction to add, ping in the other to update as seen.
+      # ping in one direction to add, ping in the other to update as seen,
+      # adding the node in the discovery v5 routing table. Could also launch
+      # with bootstrap node instead.
       check (await node1.ping(node2.localNode)).isOk()
       check (await node2.ping(node1.localNode)).isOk()
 
+      # Start the portal protocol to seed nodes from the discoveryv5 routing
+      # table.
+      proto2.start()
+
       let distance = logDist(node1.localNode.id, node2.localNode.id)
       let nodes = await proto1.findNode(proto2.baseProtocol.localNode,
         List[uint16, 256](@[distance]))
@@ -85,6 +92,7 @@ procSuite "Portal Tests":
         nodes.get().total == 1'u8
         nodes.get().enrs.len() == 1
 
+    proto2.stop()
     await node1.closeWait()
     await node2.closeWait()
 
@@ -102,6 +110,10 @@ procSuite "Portal Tests":
     check (await node1.ping(node2.localNode)).isOk()
     check (await node2.ping(node1.localNode)).isOk()
 
+    # Start the portal protocol to seed nodes from the discoveryv5 routing
+    # table.
+    proto2.start()
+
     var nodeHash: NodeHash
 
     let contentKey = ContentKey(networkId: 0'u16,
@@ -118,5 +130,6 @@ procSuite "Portal Tests":
       foundContent.get().enrs.len() == 1
       foundContent.get().payload.len() == 0
 
+    proto2.stop()
     await node1.closeWait()
     await node2.closeWait()

vendor/nim-eth

@@ -1 +1 @@
-Subproject commit dd02d1be23d0d9d4ff929db30ff68bc859e35204
+Subproject commit 9bc4fa366af2f75dd2c8608964dc6dc3614b3f81