PM-2938: Leader selection (#9)

* PM-2938: Added Hash type and Keccak256.

* PM-2938: Add LeaderSelection and RoundRobin.

* PM-2938: LeaderSelection.Hashing

* PM-2938: Use hashing leader selection in HotStuff basic tests.

* PM-2938: Silence logs from networking in tests.

* PM-2966: Add some sleep in tests.

README.md

@@ -59,6 +59,12 @@ To run a single test class, use the `.single` method with the full path to the s
 mill __.storage.test.single io.iohk.metronome.storage.KVStoreStateSpec
 ```
 
+To experiment with the code, start an interactive session:
+
+```console
+mill -i metronome[2.13.4].hotstuff.consensus.console
+```
+
 ### Formatting the codebase
 
 Please configure your editor to use `scalafmt` on save. CI will be configured to check formatting.

build.sc

@@ -15,6 +15,7 @@ object VersionOf {
   val config           = "1.4.1"
   val `kind-projector` = "0.11.3"
   val logback          = "1.2.3"
+  val mantis           = "3.2.1-SNAPSHOT"
   val monix            = "3.3.0"
   val prometheus       = "0.10.0"
   val rocksdb          = "6.15.2"
@@ -25,7 +26,6 @@ object VersionOf {
   val shapeless        = "2.3.3"
   val `scodec-core`    = "1.11.7"
   val `scodec-bits`    = "1.1.12"
-  val `mantis-crypto`  = "3.2.1-SNAPSHOT"
 }
 
 // Using 2.12.13 instead of 2.12.10 to access @nowarn, to disable certain deperaction
@@ -150,20 +150,6 @@ class MetronomeModule(val crossScalaVersion: String) extends CrossScalaModule {
     )
   }
 
-  /** Generic Peer-to-Peer components that can multiplex protocols
-    * from different modules over a single authenticated TLS connection.
-    */
-  object networking extends SubModule {
-    override def moduleDeps: Seq[JavaModule] =
-      Seq(tracing, crypto)
-
-    override def ivyDeps = super.ivyDeps() ++ Agg(
-      ivy"io.iohk::scalanet:${VersionOf.scalanet}"
-    )
-
-    object test extends TestModule
-  }
-
   /** Storage abstractions, e.g. a generic key-value store. */
   object storage extends SubModule {
     override def ivyDeps = super.ivyDeps() ++ Agg(
@@ -193,14 +179,31 @@ class MetronomeModule(val crossScalaVersion: String) extends CrossScalaModule {
     override def description: String =
       "Cryptographic primitives to support HotStuff and BFT proof verification."
 
+    override def moduleDeps: Seq[PublishModule] =
+      Seq(core)
+
     override def ivyDeps = super.ivyDeps() ++ Agg(
-      ivy"io.iohk::mantis-crypto:${VersionOf.`mantis-crypto`}",
+      ivy"io.iohk::mantis-crypto:${VersionOf.mantis}",
       ivy"org.scodec::scodec-bits:${VersionOf.`scodec-bits`}"
     )
 
     object test extends TestModule
   }
 
+  /** Generic Peer-to-Peer components that can multiplex protocols
+    * from different modules over a single authenticated TLS connection.
+    */
+  object networking extends SubModule {
+    override def moduleDeps: Seq[JavaModule] =
+      Seq(tracing, crypto)
+
+    override def ivyDeps = super.ivyDeps() ++ Agg(
+      ivy"io.iohk::scalanet:${VersionOf.scalanet}"
+    )
+
+    object test extends TestModule
+  }
+
   /** Generic HotStuff BFT library. */
   object hotstuff extends SubModule {
 
@@ -221,7 +224,14 @@ class MetronomeModule(val crossScalaVersion: String) extends CrossScalaModule {
       */
     object service extends SubModule {
       override def moduleDeps: Seq[JavaModule] =
-        Seq(storage, tracing, crypto, hotstuff.consensus, hotstuff.forensics)
+        Seq(
+          storage,
+          tracing,
+          crypto,
+          networking,
+          hotstuff.consensus,
+          hotstuff.forensics
+        )
 
       override def ivyDeps = super.ivyDeps() ++ Agg(
         ivy"io.iohk::scalanet:${VersionOf.scalanet}"

metronome/core/src/metronome/core/Validated.scala

@@ -1,3 +1,14 @@
 package metronome.core
 
+/** Can be used to tag any particular type as validated, for example:
+  *
+  * ```
+  * def validateBlock(block: Block): Either[Error, Validated[Block]]
+  * def storeBlock(block: Validated[Block])
+  * ```
+  *
+  * It's a bit more lightweight than opting into the `ValidatedNel` from `cats`,
+  * mostly just serves as control that the right methods have been called in a
+  * pipeline.
+  */
 object Validated extends GenericTagger

metronome/crypto/src/metronome/crypto/hash/Hash.scala

@@ -0,0 +1,6 @@
+package metronome.crypto.hash
+
+import metronome.core.Tagger
+import scodec.bits.ByteVector
+
+object Hash extends Tagger[ByteVector]

metronome/crypto/src/metronome/crypto/hash/Keccak256.scala

@@ -0,0 +1,20 @@
+package metronome.crypto.hash
+
+import org.bouncycastle.crypto.digests.KeccakDigest
+import scodec.bits.{BitVector, ByteVector}
+
+object Keccak256 {
+  def apply(data: Array[Byte]): Hash = {
+    val output = new Array[Byte](32)
+    val digest = new KeccakDigest(256)
+    digest.update(data, 0, data.length)
+    digest.doFinal(output, 0)
+    Hash(ByteVector(output))
+  }
+
+  def apply(data: ByteVector): Hash =
+    apply(data.toArray)
+
+  def apply(data: BitVector): Hash =
+    apply(data.toByteArray)
+}

metronome/crypto/src/metronome/crypto/hash/package.scala

@@ -0,0 +1,5 @@
+package metronome.crypto
+
+package object hash {
+  type Hash = Hash.Tagged
+}

metronome/crypto/test/src/metronome/crypto/hash/Keccak256Spec.scala

@@ -0,0 +1,21 @@
+package metronome.crypto.hash
+
+import scodec.bits._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class Keccak256Spec extends AnyFlatSpec with Matchers {
+  behavior of "Keccak256"
+
+  it should "hash empty data" in {
+    Keccak256(
+      "".getBytes
+    ) shouldBe hex"c5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470"
+  }
+
+  it should "hash non-empty data" in {
+    Keccak256(
+      "abc".getBytes
+    ) shouldBe hex"4e03657aea45a94fc7d47ba826c8d667c0d1e6e33a64a036ec44f58fa12d6c45"
+  }
+}

metronome/hotstuff/consensus/src/metronome/hotstuff/consensus/Federation.scala

@@ -27,7 +27,7 @@ abstract case class Federation[PKey](
     publicKeys: IndexedSeq[PKey],
     // Maximum number of Byzantine nodes.
     maxFaulty: Int
-) {
+)(implicit ls: LeaderSelection) {
   private val publicKeySet = publicKeys.toSet
 
   /** Size of the federation. */
@@ -40,15 +40,15 @@ abstract case class Federation[PKey](
     publicKeySet.contains(publicKey)
 
   def leaderOf(viewNumber: ViewNumber): PKey =
-    publicKeys((viewNumber % size).toInt)
+    publicKeys(implicitly[LeaderSelection].leaderOf(viewNumber, size))
 }
 
 object Federation {
 
   /** Create a federation with the highest possible fault tolerance. */
   def apply[PKey](
       publicKeys: IndexedSeq[PKey]
-  ): Either[String, Federation[PKey]] =
+  )(implicit ls: LeaderSelection): Either[String, Federation[PKey]] =
     apply(publicKeys, maxByzantine(publicKeys.size))
 
   /** Create a federation with the fault tolerance possibly reduced from the theoretical
@@ -59,7 +59,7 @@ object Federation {
   def apply[PKey](
       publicKeys: IndexedSeq[PKey],
       maxFaulty: Int
-  ): Either[String, Federation[PKey]] = {
+  )(implicit ls: LeaderSelection): Either[String, Federation[PKey]] = {
     val f = maxByzantine(publicKeys.size)
     if (publicKeys.isEmpty) {
       Left("The federation cannot be empty!")

metronome/hotstuff/consensus/src/metronome/hotstuff/consensus/LeaderSelection.scala

@@ -0,0 +1,53 @@
+package metronome.hotstuff.consensus
+
+import metronome.crypto.hash.Keccak256
+import scodec.bits.ByteVector
+
+/** Strategy to pick the leader for a given view number from
+  * federation of with a fixed size.
+  */
+trait LeaderSelection {
+
+  /** Return the index of the federation member who should lead the view. */
+  def leaderOf(viewNumber: ViewNumber, size: Int): Int
+}
+
+object LeaderSelection {
+
+  /** Simple strategy cycling through leaders in a static order. */
+  object RoundRobin extends LeaderSelection {
+    override def leaderOf(viewNumber: ViewNumber, size: Int): Int =
+      (viewNumber % size).toInt
+  }
+
+  /** Leader assignment based on view-number has not been discussed in the Hotstuff
+    * paper and in general, it does not affect the safety and liveness.
+    * However, it does affect worst-case latency.
+    *
+    * Consider a static adversary under a round-robin leader change scheme.
+    * All the f nodes can set their public keys so that they are consecutive.
+    * In such a scenario those f consecutive leaders can create timeouts leading
+    * to an O(f) confirmation latency. (Recall that in a normal case, the latency is O(1)).
+    *
+    * A minor improvement to this is to assign leaders based on
+    * "publicKeys((H256(viewNumber).toInt % size).toInt)".
+    *
+    * This leader order randomization via a hash function will ensure that even
+    * if adversarial public keys are consecutive in PublicKey set, they are not
+    * necessarily consecutive in leader order.
+    *
+    * Note that the above policy will not ensure that adversarial leaders are never consecutive,
+    * but the probability of such occurrence will be lower under a static adversary.
+    */
+  object Hashing extends LeaderSelection {
+    override def leaderOf(viewNumber: ViewNumber, size: Int): Int = {
+      val bytes = ByteVector.fromLong(viewNumber) // big-endian
+      val hash  = Keccak256(bytes)
+      // If we prepend 0.toByte then it would treat it as unsigned, at the cost of an array copy.
+      // Instead of doing that I'll just make sure we deal with negative modulo.
+      val num = BigInt(hash.toArray)
+      val mod = (num % size).toInt
+      if (mod < 0) mod + size else mod
+    }
+  }
+}

metronome/hotstuff/consensus/test/src/metronome/hotstuff/consensus/FederationSpec.scala

@@ -7,6 +7,8 @@ import org.scalatest.prop.TableDrivenPropertyChecks._
 
 class FederationSpec extends AnyFlatSpec with Matchers with Inside {
 
+  implicit val ls = LeaderSelection.RoundRobin
+
   behavior of "Federation"
 
   it should "not create an empty federation" in {

metronome/hotstuff/consensus/test/src/metronome/hotstuff/consensus/LeaderSelectionProps.scala

@@ -0,0 +1,44 @@
+package metronome.hotstuff.consensus
+
+import metronome.core.Tagger
+import org.scalacheck._
+import org.scalacheck.Prop.forAll
+
+abstract class LeaderSelectionProps(name: String, val selector: LeaderSelection)
+    extends Properties(name) {
+
+  object Size extends Tagger[Int]
+  type Size = Size.Tagged
+
+  implicit val arbViewNumber: Arbitrary[ViewNumber] = Arbitrary {
+    Gen.posNum[Long].map(ViewNumber(_))
+  }
+
+  implicit val arbFederationSize: Arbitrary[Size] = Arbitrary {
+    Gen.posNum[Int].map(Size(_))
+  }
+
+  property("leaderOf") = forAll { (viewNumber: ViewNumber, size: Size) =>
+    val idx = selector.leaderOf(viewNumber, size)
+    0 <= idx && idx < size
+  }
+}
+
+object RoundRobinSelectionProps
+    extends LeaderSelectionProps(
+      "LeaderSelection.RoundRobin",
+      LeaderSelection.RoundRobin
+    ) {
+
+  property("round-robin") = forAll { (viewNumber: ViewNumber, size: Size) =>
+    val idx0 = selector.leaderOf(viewNumber, size)
+    val idx1 = selector.leaderOf(viewNumber.next, size)
+    idx1 == idx0 + 1 || idx0 == size - 1 && idx1 == 0
+  }
+}
+
+object HashingSelectionProps
+    extends LeaderSelectionProps(
+      "LeaderSelection.Hashing",
+      LeaderSelection.Hashing
+    )

metronome/hotstuff/consensus/test/src/metronome/hotstuff/consensus/basic/HotStuffProtocolProps.scala

@@ -1,7 +1,7 @@
 package metronome.hotstuff.consensus.basic
 
 import metronome.crypto.{GroupSignature, PartialSignature}
-import metronome.hotstuff.consensus.{ViewNumber, Federation}
+import metronome.hotstuff.consensus.{ViewNumber, Federation, LeaderSelection}
 import org.scalacheck.commands.Commands
 import org.scalacheck.{Properties, Gen, Prop}
 import org.scalacheck.Arbitrary.arbitrary
@@ -48,6 +48,8 @@ object HotStuffProtocolCommands extends Commands {
     override def parentBlockHash(b: TestBlock) = b.parentBlockHash
   }
 
+  implicit val leaderSelection = LeaderSelection.Hashing
+
   // Going to use publicKey == -1 * signingKey.
   def mockSigningKey(pk: TestAgreement.PKey): TestAgreement.SKey = -1 * pk
 
@@ -149,8 +151,9 @@ object HotStuffProtocolCommands extends Commands {
     // Using a signing key that works with the mock validation.
     def signingKey = mockSigningKey(publicKey)
 
-    def isLeader = viewNumber % n == ownIndex
-    def leader   = federation((viewNumber % n).toInt)
+    def leaderIndex = leaderSelection.leaderOf(viewNumber, n)
+    def isLeader    = leaderIndex == ownIndex
+    def leader      = federation(leaderIndex)
 
     def quorumSize = (n + f) / 2 + 1
   }

metronome/networking/test/resources/logback.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level %logger{36} %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <logger name="io.netty" level="OFF"/>
+
+    <logger name="io.iohk.scalanet.peergroup" level="OFF"/>
+
+    <root level="WARN">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+</configuration>

metronome/networking/test/src/io/iohk/metronome/networking/MockEncryptedConnectionProvider.scala

@@ -208,13 +208,18 @@ object MockEncryptedConnectionProvider {
       incomingEvents.poll
 
     override def sendMessage(m: TestMessage): Task[Unit] =
-      Task
-        .race(closeToken.get, sentMessages.update(current => m :: current))
-        .flatMap {
-          case Left(_) =>
-            Task.raiseError(ConnectionAlreadyClosed(remotePeerInfo._2))
-          case Right(_) => Task.now(())
-        }
+      closeToken.tryGet.flatMap {
+        case Some(_) =>
+          Task.raiseError(ConnectionAlreadyClosed(remotePeerInfo._2))
+        case None =>
+          Task
+            .race(closeToken.get, sentMessages.update(current => m :: current))
+            .flatMap {
+              case Left(_) =>
+                Task.raiseError(ConnectionAlreadyClosed(remotePeerInfo._2))
+              case Right(_) => Task.now(())
+            }
+      }
   }
 
   object MockEncryptedConnection {