multi-master replication

- update to Akka 2.4-M2 for DistributedData
- add Status CRDT and tests

Author: rkuhn

chapter12/build.sbt

@@ -1,3 +1,3 @@
 import Build._
 
-libraryDependencies ++= Seq(akkaContrib, playJson, sbtIO, akkaTestkit, junit, scalatest)
+libraryDependencies ++= Seq(akkaContrib, akkaDData, playJson, sbtIO, akkaTestkit, junit, scalatest)

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/ActivePassive.scala

@@ -14,8 +14,10 @@ import scala.collection.immutable.TreeMap
 import scala.concurrent.forkjoin.ThreadLocalRandom
 import scala.util.Random
 import scala.annotation.tailrec
-import akka.contrib.pattern.ClusterSingletonManager
-import akka.contrib.pattern.ClusterSingletonProxy
+import akka.cluster.singleton.ClusterSingletonManager
+import akka.cluster.singleton.ClusterSingletonProxy
+import akka.cluster.singleton.ClusterSingletonManagerSettings
+import akka.cluster.singleton.ClusterSingletonProxySettings
 import com.typesafe.config.ConfigFactory
 import scala.io.StdIn
 
@@ -33,8 +35,8 @@ object ActivePassive {
   class Active(localReplica: ActorRef, replicationFactor: Int, maxQueueSize: Int) extends Actor with Stash with ActorLogging {
     private val MaxOutstanding = maxQueueSize / 2
 
-    private var theStore = Map.empty[String, JsValue]
-    private var seqNr = Iterator from 0
+    private var theStore: Map[String, JsValue] = _
+    private var seqNr: Iterator[Int] = _
     private val toReplicate = Queue.empty[Replicate]
     private var replicating = TreeMap.empty[Int, (Replicate, Int)]
 
@@ -146,9 +148,11 @@ object ActivePassive {
       case Replicate(s, _, _, replyTo) if s - expectedSeq < 0 =>
         replyTo ! Replicated(s)
       case r: Replicate if r.seq == expectedSeq =>
+        val nextStore = theStore + (r.key -> r.value)
+        persist(name, expectedSeq, nextStore)
         r.replyTo ! Replicated(r.seq)
         applied.enqueue(r)
-        context.become(upToDate(theStore + (r.key -> r.value), r.seq + 1))
+        context.become(upToDate(nextStore, expectedSeq + 1))
       case r: Replicate =>
         if (r.seq - expectedSeq > maxLag)
           fallBehind(expectedSeq, TreeMap(r.seq -> r))
@@ -283,9 +287,12 @@ object ActivePassive {
   def start(port: Option[Int]): ActorSystem = {
     val system = ActorSystem("ActivePassive", roleConfig("backend", port) withFallback commonConfig)
     val localReplica = system.actorOf(Props(new Passive(3, 3.seconds, 100)), "passive")
+    val settings = ClusterSingletonManagerSettings(system)
+      .withSingletonName("active")
+      .withRole("backend")
+      .withHandOverRetryInterval(150.millis)
     val managerProps =
-      ClusterSingletonManager.props(Props(new Active(localReplica, 2, 120)), "active", PoisonPill,
-        role = Some("backend"), retryInterval = 150.millis)
+      ClusterSingletonManager.props(Props(new Active(localReplica, 2, 120)), PoisonPill, settings)
     val manager = system.actorOf(managerProps, "activeManager")
     system
   }
@@ -300,7 +307,10 @@ object ActivePassive {
 
     awaitMembers(sys, systems.length + 1)
 
-    val proxy = sys.actorOf(ClusterSingletonProxy.props("/user/activeManager/active", Some("backend")), "proxy")
+    val proxySettings = ClusterSingletonProxySettings(sys)
+      .withRole("backend")
+      .withSingletonName("active")
+    val proxy = sys.actorOf(ClusterSingletonProxy.props("/user/activeManager", proxySettings), "proxy")
 
     val useStorage = sys.actorOf(Props(new UseStorage(proxy)), "useStorage")
     useStorage ! Run(0)

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/MultiMasterCRDT.scala

@@ -1,28 +1,65 @@
 package com.reactivedesignpatterns.chapter12
 
-class MultiMasterCRDT {
+import akka.cluster.ddata.ReplicatedData
 
-  import akka.actor._
-  
-  case class Greet(whom: String)
-  
-  class Greeter extends Actor {
-    def receive = {
-      case Greet(whom) =>
-        sender() ! s"Hello $whom!"
-        val delegate = context.actorOf(grumpyProps)
-        context.become(grumpy(delegate))
-    }
-    def grumpy(delegate: ActorRef): Receive = {
-      case g: Greet => delegate forward g
-    }
+object MultiMasterCRDT {
+
+  sealed trait Status extends ReplicatedData {
+    type T = Status
+    def merge(that: Status): Status = mergeStatus(this, that)
+    
+    def predecessors: Set[Status]
+    def successors: Set[Status]
+  }
+
+  case object New extends Status {
+    lazy val predecessors: Set[Status] = Set.empty
+    lazy val successors: Set[Status] = Set(Scheduled)
+  }
+  case object Scheduled extends Status {
+    lazy val predecessors: Set[Status] = Set(New)
+    lazy val successors: Set[Status] = Set(Executing, Cancelled)
+  }
+  case object Executing extends Status {
+    lazy val predecessors: Set[Status] = Set(Scheduled)
+    lazy val successors: Set[Status] = Set(Aborted)
+  }
+  case object Finished extends Status {
+    lazy val predecessors: Set[Status] = Set(Executing, Aborted)
+    lazy val successors: Set[Status] = Set.empty
+  }
+  case object Cancelled extends Status {
+    lazy val predecessors: Set[Status] = Set(New, Scheduled)
+    lazy val successors: Set[Status] = Set(Aborted)
+  }
+  case object Aborted extends Status {
+    lazy val predecessors: Set[Status] = Set(Cancelled, Executing)
+    lazy val successors: Set[Status] = Set(Finished)
+  }
+
+  def mergeStatus(left: Status, right: Status): Status = {
+    /*
+     * Keep the left Status in hand and determine whether it is a predecessor of
+     * the candidate, moving on to the candidate’s successor if not successful.
+     * The list of exclusions is used to avoid performing already determined
+     * unsuccessful comparisons again.
+     */
+    def innerLoop(candidate: Status, exclude: Set[Status]): Status =
+      if (isSuccessor(candidate, left, exclude)) {
+        candidate
+      } else {
+        val nextExclude = exclude + candidate
+        val branches = candidate.successors.map(succ => innerLoop(succ, nextExclude))
+        branches.reduce((l, r) => if (isSuccessor(l, r, nextExclude)) r else l)
+      }
+    def isSuccessor(candidate: Status, fixed: Status, exclude: Set[Status]): Boolean =
+      if (candidate == fixed) true
+      else {
+        val toSearch = candidate.predecessors -- exclude
+        toSearch.exists(pred => isSuccessor(pred, fixed, exclude))
+      }
+
+    innerLoop(right, Set.empty)
   }
-  
-  val grumpyProps = Props(new Actor {
-    def receive = {
-      case Greet(whom) =>
-        sender() ! s"Go away, $whom!"
-    }
-  })
 
 }

chapter12/src/main/scala/com/reactivedesignpatterns/chapter12/MultiMasterRAFT.scala

@@ -0,0 +1,71 @@
+package com.reactivedesignpatterns.chapter12
+
+import org.scalatest.WordSpec
+import org.scalatest.Matchers
+import org.scalactic.ConversionCheckedTripleEquals
+
+class MultiMasterCRDTSpec extends WordSpec with Matchers with ConversionCheckedTripleEquals {
+  import MultiMasterCRDT._
+
+  "A CRDT" must {
+    val allStatus = List(New, Scheduled, Executing, Finished, Cancelled, Aborted)
+
+    "have a correct mergeStatus function" in {
+      mergeStatus(New, New) should ===(New)
+      mergeStatus(New, Scheduled) should ===(Scheduled)
+      mergeStatus(New, Executing) should ===(Executing)
+      mergeStatus(New, Finished) should ===(Finished)
+      mergeStatus(New, Cancelled) should ===(Cancelled)
+      mergeStatus(New, Aborted) should ===(Aborted)
+      mergeStatus(Scheduled, New) should ===(Scheduled)
+      mergeStatus(Scheduled, Scheduled) should ===(Scheduled)
+      mergeStatus(Scheduled, Executing) should ===(Executing)
+      mergeStatus(Scheduled, Finished) should ===(Finished)
+      mergeStatus(Scheduled, Cancelled) should ===(Cancelled)
+      mergeStatus(Scheduled, Aborted) should ===(Aborted)
+      mergeStatus(Executing, New) should ===(Executing)
+      mergeStatus(Executing, Scheduled) should ===(Executing)
+      mergeStatus(Executing, Executing) should ===(Executing)
+      mergeStatus(Executing, Finished) should ===(Finished)
+      mergeStatus(Executing, Cancelled) should ===(Aborted)
+      mergeStatus(Executing, Aborted) should ===(Aborted)
+      mergeStatus(Finished, New) should ===(Finished)
+      mergeStatus(Finished, Scheduled) should ===(Finished)
+      mergeStatus(Finished, Executing) should ===(Finished)
+      mergeStatus(Finished, Finished) should ===(Finished)
+      mergeStatus(Finished, Cancelled) should ===(Finished)
+      mergeStatus(Finished, Aborted) should ===(Finished)
+      mergeStatus(Cancelled, New) should ===(Cancelled)
+      mergeStatus(Cancelled, Scheduled) should ===(Cancelled)
+      mergeStatus(Cancelled, Executing) should ===(Aborted)
+      mergeStatus(Cancelled, Finished) should ===(Finished)
+      mergeStatus(Cancelled, Cancelled) should ===(Cancelled)
+      mergeStatus(Cancelled, Aborted) should ===(Aborted)
+      mergeStatus(Aborted, New) should ===(Aborted)
+      mergeStatus(Aborted, Scheduled) should ===(Aborted)
+      mergeStatus(Aborted, Executing) should ===(Aborted)
+      mergeStatus(Aborted, Finished) should ===(Finished)
+      mergeStatus(Aborted, Cancelled) should ===(Aborted)
+      mergeStatus(Aborted, Aborted) should ===(Aborted)
+    }
+
+    "have a symmetrical mergeStatus function" in {
+      for {
+        left <- allStatus
+        right <- allStatus
+      } withClue(s"mergeStatus($left, $right): ") {
+        mergeStatus(left, right) should ===(mergeStatus(right, left))
+      }
+    }
+    
+    "merge Finished always to Finished" in {
+      for {
+        other <- allStatus
+      } withClue(s"mergeStatus(Finished, $other): ") {
+        mergeStatus(Finished, other) should ===(Finished)
+      }
+    }
+
+  }
+
+}

chapter12/src/test/scala/com/reactivedesignpatterns/chapter12/MultiMasterCRDTSpec.scala

@@ -1,11 +1,12 @@
 import sbt._
 
 object Build {
-  val akkaVersion = "2.3.11"
+  val akkaVersion = "2.4-M2"
 
   val akkaActor = "com.typesafe.akka" %% "akka-actor" % akkaVersion
   val akkaTestkit = "com.typesafe.akka" %% "akka-testkit" % akkaVersion % "test"
   val akkaContrib = "com.typesafe.akka" %% "akka-contrib" % akkaVersion
+  val akkaDData = "com.typesafe.akka" %% "akka-distributed-data-experimental" % akkaVersion
 
   val sbtIO = "org.scala-sbt" %% "io" % "0.13.8"