Merge pull request #108 from input-output-hk/ETCM-414-not-bonding-with-incoming-until-enrolled

ETCM-414: Incoming pings don't trigger bonds until enroll is finished.

Author: aakoshh

README.md

@@ -82,7 +82,7 @@ mill __.test
 
 A single test suite can be executed with the `single` helper command, for example:
 ```bash
-mill scalanet.test.single io.iohk.scalanet.crypto.SignatureVerificationSpec
+mill scalanet.ut.single io.iohk.scalanet.crypto.SignatureVerificationSpec
 ```
 
 ### Publishing

build.sc

@@ -68,7 +68,7 @@ trait ScalanetModule extends ScalaModule {
 trait ScalanetPublishModule extends PublishModule {
   def description: String
 
-  override def publishVersion = "0.4.3-SNAPSHOT"
+  override def publishVersion = "0.4.4-SNAPSHOT"
 
   override def pomSettings = PomSettings(
     description = description,

scalanet/discovery/it/src/io/iohk/scalanet/kademlia/KademliaIntegrationSpec.scala

@@ -188,11 +188,16 @@ abstract class KademliaIntegrationSpec(name: String)
     val lowRefConfig = defaultConfig.copy(refreshRate = 1.seconds)
     val randomNode = generatePeerRecordWithKey
     (for {
+      // Start a node which would bootstrap from a peer that's not running yet.
       node1 <- startNode(initialNodes = Set(randomNode._1), testConfig = lowRefConfig)
+      // Start a standalone node.
       node2 <- startNode(initialNodes = Set(), testConfig = lowRefConfig)
+      // A chain of nodes bootstrapping from each other
       node3 <- startNode(initialNodes = Set(node2.self), testConfig = lowRefConfig)
       node4 <- startNode(initialNodes = Set(node3.self), testConfig = lowRefConfig)
       _ <- Resource.liftF(Task.sleep(10.seconds))
+      // Now start a node that node1 wanted to bootstrap from, and join all the other nodes.
+      // When node1 refreshes it will try to connect to node5 again and discover the others.
       node5 <- startNode(
         selfRecordWithKey = randomNode,
         initialNodes = Set(node2.self, node3.self, node4.self),

scalanet/discovery/src/io/iohk/scalanet/discovery/ethereum/v4/DiscoveryService.scala

@@ -95,7 +95,8 @@ object DiscoveryService {
           // Start handling requests, we need them during enrolling so the peers can ping and bond with us.
           cancelToken <- network.startHandling(service)
           // Contact the bootstrap nodes.
-          enroll = service.enroll()
+          // Setting the enrolled status here because we could potentially repeat enrollment until it succeeds.
+          enroll = service.enroll().guarantee(stateRef.update(_.setEnrolled))
           // Periodically discover new nodes.
           discover = service.lookupRandom.delayExecution(config.discoveryPeriod).loopForever
           // Enrollment can be run in the background if it takes very long.
@@ -152,7 +153,9 @@ object DiscoveryService {
       // Deferred results so we can ensure there's only one concurrent Ping to a given peer.
       bondingResultsMap: Map[Peer[A], BondingResults],
       // Deferred ENR fetches so we only do one at a time to a given peer.
-      fetchEnrMap: Map[Peer[A], FetchEnrResult]
+      fetchEnrMap: Map[Peer[A], FetchEnrResult],
+      // Indicate whether enrollment hash finished.
+      hasEnrolled: Boolean
   ) {
     def isSelf(peer: Peer[A]): Boolean =
       peer.id == node.id
@@ -191,6 +194,9 @@ object DiscoveryService {
     def clearBondingResults(peer: Peer[A]): State[A] =
       copy(bondingResultsMap = bondingResultsMap - peer)
 
+    def clearLastPongTimestamp(peer: Peer[A]): State[A] =
+      copy(lastPongTimestampMap = lastPongTimestampMap - peer)
+
     def withEnrFetch(peer: Peer[A], result: FetchEnrResult): State[A] =
       copy(fetchEnrMap = fetchEnrMap.updated(peer, result))
 
@@ -230,6 +236,9 @@ object DiscoveryService {
         kBuckets = kBuckets.remove(Node.kademliaId(peerId)),
         kademliaIdToNodeId = kademliaIdToNodeId - Node.kademliaId(peerId)
       )
+
+    def setEnrolled: State[A] =
+      copy(hasEnrolled = true)
   }
   protected[v4] object State {
     def apply[A](
@@ -245,7 +254,8 @@ object DiscoveryService {
       enrMap = Map(node.id -> enr),
       lastPongTimestampMap = Map.empty[Peer[A], Timestamp],
       bondingResultsMap = Map.empty[Peer[A], BondingResults],
-      fetchEnrMap = Map.empty[Peer[A], FetchEnrResult]
+      fetchEnrMap = Map.empty[Peer[A], FetchEnrResult],
+      hasEnrolled = false
     )
   }
 
@@ -328,6 +338,10 @@ object DiscoveryService {
           for {
             // Complete any deferred waiting for a ping from this peer, if we initiated the bonding.
             _ <- completePing(caller)
+            // To protect against an eclipse attack filling up the k-table after a reboot,
+            // only try to bond with an incoming Ping's peer after the initial enrollment
+            // hash finished.
+            hasEnrolled <- stateRef.get.map(_.hasEnrolled)
             _ <- isBonded(caller)
               .ifM(
                 // We may already be bonded but the remote node could have changed its address.
@@ -338,6 +352,7 @@ object DiscoveryService {
                 bond(caller)
               )
               .startAndForget
+              .whenA(hasEnrolled)
             // Return the latet local ENR sequence.
             enrSeq <- localEnrSeq
           } yield Some(Some(enrSeq))
@@ -745,7 +760,11 @@ object DiscoveryService {
               .findNode(peer)(target)
               .flatMap {
                 case None =>
-                  Task(logger.debug(s"Received no response for neighbors for $target from ${peer.address}")).as(Nil)
+                  for {
+                    _ <- Task(logger.debug(s"Received no response for neighbors for $target from ${peer.address}"))
+                    // The other node has possibly unbonded from us, or it was still enrolling when we bonded. Try bonding next time.
+                    _ <- stateRef.update(_.clearLastPongTimestamp(peer))
+                  } yield Nil
                 case Some(neighbors) =>
                   Task(logger.debug(s"Received ${neighbors.size} neighbors for $target from ${peer.address}"))
                     .as(neighbors.toList)

scalanet/discovery/ut/src/io/iohk/scalanet/discovery/ethereum/v4/DiscoveryServiceSpec.scala

@@ -512,14 +512,34 @@ class DiscoveryServiceSpec extends AsyncFlatSpec with Matchers {
 
   behavior of "ping"
 
-  it should "respond with the ENR sequence and bond with the caller" in test {
+  it should "respond with the ENR sequence but not bond with the caller before enrollment" in test {
+    new Fixture {
+      override val test = for {
+        hasEnrolled <- stateRef.get.map(_.hasEnrolled)
+        maybeEnrSeq <- service.ping(remotePeer)(None)
+        // Shouldn't start bonding, but in case it does, wait until it finishes.
+        _ <- stateRef.get.flatMap(_.bondingResultsMap.get(remotePeer).fold(Task.unit)(_.pongReceived.get.void))
+        state <- stateRef.get
+      } yield {
+        hasEnrolled shouldBe false
+        maybeEnrSeq shouldBe Some(Some(localENR.content.seq))
+        state.nodeMap should not contain key(remotePeer.id)
+        state.enrMap should not contain key(remotePeer.id)
+        state.lastPongTimestampMap should not contain key(remotePeer)
+      }
+    }
+  }
+
+  it should "respond with the ENR sequence and bond with the caller after enrollment" in test {
     new Fixture {
       override lazy val rpc = unimplementedRPC.copy(
         ping = _ => _ => Task(Some(None)),
         enrRequest = _ => _ => Task(Some(remoteENR))
       )
       override val test = for {
+        _ <- stateRef.update(_.setEnrolled)
         maybeEnrSeq <- service.ping(remotePeer)(None)
+        // Wait for any ongoing bonding and ENR fetching to finish.
         _ <- stateRef.get.flatMap(_.bondingResultsMap.get(remotePeer).fold(Task.unit)(_.pongReceived.get.void))
         _ <- stateRef.get.flatMap(_.fetchEnrMap.get(remotePeer).fold(Task.unit)(_.get.void))
         state <- stateRef.get