Merge pull request #104 from lerna-stack/improve-multi-jvm-test-stability

Improve test stability

Author: xirc

.github/workflows/ci.yml

@@ -7,7 +7,13 @@ on:
     branches: [ master, feature/** ]
 
 env:
-  SBT_OPTS: -Dlerna.enable.discipline
+  # classic timefactor doesn't affect classic default-timeout
+  SBT_OPTS: >-
+    -Dlerna.enable.discipline
+    -Dakka.test.timefactor=3.0
+    -Dakka.actor.testkit.typed.timefactor=3.0
+    -Dakka.test.default-timeout=15s
+    -Dakka.testconductor.barrier-timeout=90s
 
 jobs:
   test:
@@ -51,7 +57,7 @@ jobs:
 
     - name: Run integration tests
       continue-on-error: true # results are reported by action-junit-report
-      run: sh ./scripts/run-multijvm-test.sh 6
+      run: sh ./scripts/run-multijvm-test.sh 1
 
     - name: Publish test report
       uses: mikepenz/action-junit-report@v2

build.sbt

@@ -32,6 +32,8 @@ lazy val lerna = (project in file("."))
     name := "akka-entity-replication",
     fork in Test := true,
     parallelExecution in Test := false,
+    javaOptions in Test ++= sbtJavaOptions,
+    jvmOptions in MultiJvm ++= sbtJavaOptions,
     libraryDependencies ++= Seq(
         "com.typesafe.akka" %% "akka-cluster-typed"    % akkaVersion,
         "com.typesafe.akka" %% "akka-stream"           % akkaVersion,
@@ -83,6 +85,20 @@ lazy val lerna = (project in file("."))
     mimaReportSignatureProblems := true, // check also generic parameters
   )
 
+/**
+  * This is used to pass specific system properties (mostly from CI environment variables)
+  * to the forked process by sbt
+  */
+val sbtJavaOptions: Seq[String] = {
+  // selects properties starting with the following
+  val includes = Set(
+    "akka.",
+  )
+  sys.props.collect {
+    case (k, v) if includes.exists(k.startsWith) => s"-D$k=$v"
+  }.toSeq
+}
+
 addCommandAlias(
   "testCoverage",
   Seq(

src/multi-jvm/resources/multi-jvm-testing.conf

@@ -28,6 +28,11 @@ akka.actor {
   }
 }
 
+akka.cluster {
+  // Not rquired for tesintg
+  jmx.enabled = off
+}
+
 akka.test {
   single-expect-default = 15s
 }

src/multi-jvm/scala/lerna/akka/entityreplication/ConsistencyTestNormal.scala

@@ -4,6 +4,7 @@ import java.util.concurrent.atomic.AtomicInteger
 import akka.actor.{ ActorRef, Props }
 import akka.remote.testkit.MultiNodeSpec
 import lerna.akka.entityreplication.ConsistencyTestBase.{ ConsistencyTestBaseConfig, ConsistencyTestReplicationActor }
+import lerna.akka.entityreplication.util.AtLeastOnceComplete
 import org.scalatest.Inside
 
 import scala.annotation.nowarn
@@ -45,16 +46,16 @@ class ConsistencyTestNormal extends MultiNodeSpec(ConsistencyTestBaseConfig) wit
       extractShardId = ConsistencyTestReplicationActor.extractShardId,
     )
 
+    enterBarrier("ClusterReplication started")
+
     // check the ClusterReplication healthiness
     val requestId = generateUniqueId()
-    awaitAssert {
-      clusterReplication ! GetStatus(id = "check-healthiness", requestId)
-      expectMsgType[Status](max = 1.seconds)
-    }
-    ignoreMsg {
-      // ignore Status messages that were sent for checking healthy
-      case Status(_, `requestId`) => true
-    }
+    AtLeastOnceComplete
+      .askTo(
+        clusterReplication,
+        GetStatus(id = "check-healthiness", requestId),
+        retryInterval = 1.seconds,
+      ).await
   }
 
   "正常系（直列に処理した場合）" should {

src/multi-jvm/scala/lerna/akka/entityreplication/RaftActorCompactionSpec.scala

@@ -153,7 +153,7 @@ class RaftActorCompactionSpec extends MultiNodeSpec(RaftActorCompactionSpecConfi
       runOn(node1, node2, node3) {
         awaitAssert {
           replicationActor(myself) ! DummyReplicationActor.GetState(entityId)
-          expectMsg(DummyReplicationActor.State(1))
+          expectMsg(max = 1.second, DummyReplicationActor.State(1))
         }
       }
       enterBarrier("all ReplicationActor applied an event")
@@ -178,14 +178,14 @@ class RaftActorCompactionSpec extends MultiNodeSpec(RaftActorCompactionSpecConfi
         // ReplicationActor which has not been isolated applied all events
         awaitAssert {
           replicationActor(myself) ! DummyReplicationActor.GetState(entityId)
-          expectMsg(DummyReplicationActor.State(4))
+          expectMsg(max = 1.second, DummyReplicationActor.State(4))
         }
       }
       runOn(node3) {
         // ReplicationActor which has been isolated did not apply any events
         awaitAssert {
           replicationActor(myself) ! DummyReplicationActor.GetState(entityId)
-          expectMsg(DummyReplicationActor.State(1))
+          expectMsg(max = 1.second, DummyReplicationActor.State(1))
         }
       }
       enterBarrier("ReplicationActor which has not been isolated applied all events")

src/multi-jvm/scala/lerna/akka/entityreplication/RaftEventSourcedSpec.scala

@@ -199,14 +199,14 @@ class RaftEventSourcedSpec extends MultiNodeSpec(RaftEventSourcedSpecConfig) wit
         awaitAssert(
           {
             clusterReplication ! DummyReplicationActor.Increment(entityId, requestId1, amount = 3)
-            expectMsgType[DummyReplicationActor.State].knownRequestId should contain(requestId1)
+            expectMsgType[DummyReplicationActor.State](max = 1.second).knownRequestId should contain(requestId1)
           },
           initializationTimeout,
         )
 
         awaitAssert {
           clusterReplication ! DummyReplicationActor.Increment(entityId, requestId2, amount = 10)
-          expectMsgType[DummyReplicationActor.State].knownRequestId should contain(requestId2)
+          expectMsgType[DummyReplicationActor.State](max = 1.second).knownRequestId should contain(requestId2)
         }
 
         val readJournal =

src/multi-jvm/scala/lerna/akka/entityreplication/ReplicationRegionSpec.scala

@@ -285,7 +285,7 @@ class ReplicationRegionSpec extends MultiNodeSpec(ReplicationRegionSpecConfig) w
         val replicationActorPath = s"/system/sharding/raft-shard-$typeName-$role/$clusterShard/$raftShardId/$entityId"
         awaitAssert {
           system.actorSelection(replicationActorPath) ! GetStatus(entityId)
-          expectMsg(Status(3))
+          expectMsg(max = 1.second, Status(3))
         }
       }
     }

src/multi-jvm/scala/lerna/akka/entityreplication/raft/RaftActorMultiNodeSpec.scala

@@ -654,6 +654,6 @@ class RaftActorMultiNodeSpec extends MultiNodeSpec(RaftActorSpecConfig) with STM
 
   protected def getState(raftActor: ActorRef): RaftState = {
     raftActor ! GetState
-    expectMsgType[RaftState]
+    expectMsgType[RaftState](max = 1.second)
   }
 }

src/multi-jvm/scala/lerna/akka/entityreplication/typed/ReplicatedEntityMultiNodeSpec.scala

@@ -1,16 +1,18 @@
 package lerna.akka.entityreplication.typed
 
 import akka.actor.testkit.typed.scaladsl.ActorTestKit
-import akka.actor.typed.ActorRef
+import akka.actor.typed.{ ActorRef, ActorSystem }
 import akka.actor.typed.receptionist.{ Receptionist, ServiceKey }
 import akka.actor.typed.scaladsl.Behaviors
 import akka.actor.typed.scaladsl.adapter._
 import akka.remote.testconductor.RoleName
 import akka.remote.testkit.{ MultiNodeConfig, MultiNodeSpec }
 import com.typesafe.config.{ ConfigFactory, ConfigValueFactory }
 import lerna.akka.entityreplication.raft.routing.MemberIndex
+import lerna.akka.entityreplication.util.AtLeastOnceComplete
 import lerna.akka.entityreplication.{ STMultiNodeSerializable, STMultiNodeSpec }
 
+import java.util.UUID
 import java.util.concurrent.atomic.AtomicInteger
 import scala.jdk.CollectionConverters._
 
@@ -53,7 +55,7 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
   import ReplicatedEntityMultiNodeSpec._
   import ReplicatedEntityMultiNodeSpecConfig._
 
-  private[this] val typedSystem = system.toTyped
+  private implicit val typedSystem: ActorSystem[_] = system.toTyped
 
   private[this] val clusterReplication = ClusterReplication(typedSystem)
 
@@ -90,28 +92,30 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
       val probe = actorTestKit.createTestProbe[Receptionist.Listing]()
 
       runOn(node1) {
-        val entity  = clusterReplication.entityRefFor(PingPongEntity.typeKey, createSeqEntityId())
-        val replyTo = actorTestKit.createTestProbe[PingPongEntity.Pong]()
+        val entity = clusterReplication.entityRefFor(PingPongEntity.typeKey, createSeqEntityId())
 
-        entity ! PingPongEntity.Ping(replyTo.ref)
-        replyTo.receiveMessage().count should be(1)
-        entity ! PingPongEntity.Ping(replyTo.ref)
-        replyTo.receiveMessage().count should be(2)
+        def ping(requestId: String): PingPongEntity.Pong = {
+          AtLeastOnceComplete
+            .askTo(entity, PingPongEntity.Ping(_, requestId), retryInterval = remainingOrDefault / 5).await
+        }
+
+        ping(createUniqueRequestId()).count should be(1)
+        ping(createUniqueRequestId()).count should be(2)
       }
       runOn(node1, node2, node3) {
         // find entity
         var actor: ActorRef[PingPongEntity.Command] = null
         awaitAssert {
           typedSystem.receptionist ! Receptionist.Subscribe(PingPongEntity.serviceKey, probe.ref)
-          val listing = probe.receiveMessage()
+          val listing = probe.receiveMessage(max = remainingOrDefault / 5)
           actor = listing.serviceInstances(PingPongEntity.serviceKey).head
           typedSystem.receptionist ! Receptionist.Deregister(PingPongEntity.serviceKey, actor)
         }
         // check state
         val stateReceiver = actorTestKit.createTestProbe[PingPongEntity.State]()
         awaitAssert {
           actor ! PingPongEntity.UnsafeGetState(stateReceiver.ref)
-          stateReceiver.receiveMessage().count should be(2)
+          stateReceiver.receiveMessage(max = remainingOrDefault / 5).count should be(2)
         }
       }
     }
@@ -120,18 +124,17 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
       clusterReplication.init(PingPongEntity())
 
       runOn(node1) {
-        val entity  = clusterReplication.entityRefFor(PingPongEntity.typeKey, createSeqEntityId())
-        val replyTo = actorTestKit.createTestProbe[PingPongEntity.Pong]()
+        val entity = clusterReplication.entityRefFor(PingPongEntity.typeKey, createSeqEntityId())
 
-        entity ! PingPongEntity.Ping(replyTo.ref)
-        replyTo.receiveMessage().count should be(1)
-        entity ! PingPongEntity.Ping(replyTo.ref)
-        replyTo.receiveMessage().count should be(2)
-        entity ! PingPongEntity.Break()
-        awaitAssert {
-          entity ! PingPongEntity.Ping(replyTo.ref)
-          replyTo.receiveMessage(max = remainingOrDefault / 5).count should be(3)
+        def ping(requestId: String): PingPongEntity.Pong = {
+          AtLeastOnceComplete
+            .askTo(entity, PingPongEntity.Ping(_, requestId), retryInterval = remainingOrDefault / 5).await
         }
+
+        ping(createUniqueRequestId()).count should be(1)
+        ping(createUniqueRequestId()).count should be(2)
+        entity ! PingPongEntity.Break()
+        ping(createUniqueRequestId()).count should be(3)
       }
     }
 
@@ -147,7 +150,7 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
       runOn(node1, node2, node3) {
         awaitAssert {
           typedSystem.receptionist ! Receptionist.Subscribe(EphemeralEntity.serviceKey, probe.ref)
-          val listing = probe.receiveMessage()
+          val listing = probe.receiveMessage(max = remainingOrDefault / 5)
           actor = listing.serviceInstances(EphemeralEntity.serviceKey).head
           typedSystem.receptionist ! Receptionist.Deregister(EphemeralEntity.serviceKey, actor)
         }
@@ -182,7 +185,7 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
         val subscriber = actorTestKit.createTestProbe[Receptionist.Listing]()
         awaitAssert {
           typedSystem.receptionist ! Receptionist.Subscribe(EphemeralEntity.serviceKey, subscriber.ref)
-          val listing = subscriber.receiveMessage()
+          val listing = subscriber.receiveMessage(max = remainingOrDefault / 5)
           actor = listing.serviceInstances(EphemeralEntity.serviceKey).head
           typedSystem.receptionist ! Receptionist.Deregister(EphemeralEntity.serviceKey, actor)
         }
@@ -204,6 +207,8 @@ class ReplicatedEntityMultiNodeSpec extends MultiNodeSpec(ReplicatedEntityMultiN
 
   private[this] val idGenerator                 = new AtomicInteger(0)
   private[this] def createSeqEntityId(): String = s"replication-${idGenerator.incrementAndGet()}"
+
+  private def createUniqueRequestId(): String = UUID.randomUUID().toString
 }
 
 object ReplicatedEntityMultiNodeSpec {
@@ -212,23 +217,27 @@ object ReplicatedEntityMultiNodeSpec {
     val typeKey: ReplicatedEntityTypeKey[Command] = ReplicatedEntityTypeKey("PingPong")
     val serviceKey: ServiceKey[Command]           = ServiceKey[Command]("PingPongService")
 
-    sealed trait Command                                      extends STMultiNodeSerializable
-    final case class Ping(replyTo: ActorRef[Pong])            extends Command
-    final case class Pong(count: Int)                         extends STMultiNodeSerializable
-    final case class Break()                                  extends Command
-    final case class UnsafeGetState(replyTo: ActorRef[State]) extends Command
+    sealed trait Command                                              extends STMultiNodeSerializable
+    final case class Ping(replyTo: ActorRef[Pong], requestId: String) extends Command
+    final case class Pong(count: Int)                                 extends STMultiNodeSerializable
+    final case class Break()                                          extends Command
+    final case class UnsafeGetState(replyTo: ActorRef[State])         extends Command
 
-    sealed trait Event         extends STMultiNodeSerializable
-    final case class CountUp() extends Event
+    sealed trait Event                          extends STMultiNodeSerializable
+    final case class CountUp(requestId: String) extends Event
 
-    final case class State(count: Int) extends STMultiNodeSerializable {
+    final case class State(count: Int, processedRequests: Set[String]) extends STMultiNodeSerializable {
 
       def onMessage(message: Command): Effect[Event, State] =
         message match {
-          case Ping(replyTo) =>
-            Effect
-              .replicate(CountUp())
-              .thenReply(replyTo)(s => Pong(s.count))
+          case Ping(replyTo, requestId) =>
+            if (processedRequests.contains(requestId)) {
+              Effect.reply(replyTo)(Pong(count))
+            } else {
+              Effect
+                .replicate(CountUp(requestId))
+                .thenReply(replyTo)(s => Pong(s.count))
+            }
           case Break() =>
             Effect
               .none[Event, State]
@@ -243,7 +252,7 @@ object ReplicatedEntityMultiNodeSpec {
 
       def applyEvent(event: Event): State =
         event match {
-          case _ => copy(count = count + 1)
+          case CountUp(requestId) => copy(count = count + 1, processedRequests = processedRequests + requestId)
         }
     }
 
@@ -253,7 +262,7 @@ object ReplicatedEntityMultiNodeSpec {
           context.system.receptionist ! Receptionist.Register(serviceKey, context.self)
           ReplicatedEntityBehavior[Command, Event, State](
             entityContext = entityContext,
-            emptyState = State(count = 0),
+            emptyState = State(count = 0, processedRequests = Set.empty),
             commandHandler = _ onMessage _,
             eventHandler = _ applyEvent _,
           )

src/multi-jvm/scala/lerna/akka/entityreplication/typed/ReplicatedEntitySnapshotMultiNodeSpec.scala

@@ -96,7 +96,7 @@ class ReplicatedEntitySnapshotMultiNodeSpec
         awaitAssert {
           // update state
           entityRef ! DummyEntity.Increment(transactionSeq1, amount = 1, replyTo.ref)
-          replyTo.receiveMessage(1.seconds).count should be(1)
+          replyTo.receiveMessage(500.millis).count should be(1)
         }
       }
       enterBarrier("sent a command")
@@ -106,7 +106,7 @@ class ReplicatedEntitySnapshotMultiNodeSpec
         awaitAssert {
           actorRef = findEntityActorRef()
           actorRef ! DummyEntity.GetState(replyTo.ref)
-          replyTo.receiveMessage(1.seconds).count should be(1)
+          replyTo.receiveMessage(500.millis).count should be(1)
         }
       }
       enterBarrier("all entities applied an event")
@@ -120,7 +120,7 @@ class ReplicatedEntitySnapshotMultiNodeSpec
           var latestCount = 1
           awaitAssert {
             entityRef ! DummyEntity.Increment(transactionSeq, amount = 1, replyTo.ref)
-            val reply = replyTo.receiveMessage(1.seconds)
+            val reply = replyTo.receiveMessage(500.millis)
             reply.count should be > latestCount
             latestCount = reply.count
           }
@@ -132,14 +132,14 @@ class ReplicatedEntitySnapshotMultiNodeSpec
         // entity which has not been isolated applied all events
         awaitAssert {
           actorRef ! DummyEntity.GetState(replyTo.ref)
-          replyTo.receiveMessage(1.seconds).count should be(5)
+          replyTo.receiveMessage(500.millis).count should be(5)
         }
       }
       runOn(node3) {
         // entity which has been isolated did not apply any events
         awaitAssert {
           actorRef ! DummyEntity.GetState(replyTo.ref)
-          replyTo.receiveMessage(1.seconds).count should be(1)
+          replyTo.receiveMessage(500.millis).count should be(1)
         }
       }
       enterBarrier("entity which has not been isolated applied all events")
@@ -151,11 +151,11 @@ class ReplicatedEntitySnapshotMultiNodeSpec
         awaitAssert {
           // attempt to create entities on all nodes
           entityRef ! DummyEntity.Increment(transactionSeq2, amount = 0, replyTo.ref)
-          replyTo.receiveMessage(1.seconds)
+          replyTo.receiveMessage(500.millis)
           // All ReplicationActor states will eventually be the same as any other after the isolation is resolved
           actorRef = findEntityActorRef()
           actorRef ! DummyEntity.GetState(replyTo.ref)
-          replyTo.receiveMessage(1.seconds).count should be(5)
+          replyTo.receiveMessage(500.millis).count should be(5)
         }
       }
     }