🔀Merge branch 'master' into fix-inconsistency-after-installsnapshot

negokaz · negokaz · commit 86a13dccb357 · 2022-02-28T19:39:42.000+09:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Starting a follower member later than leader completes a compaction may break ReplicatedLog of the follower [#105](https://github.com/lerna-stack/akka-entity-replication/issues/105)
 - The Raft leader uses the same previous `LogEntryIndex` and `Term` to all batched `AppendEntries` messages [#123](https://github.com/lerna-stack/akka-entity-replication/issues/123)
 - Raft Actors doesn't accept a `RequestVote(lastLogIndex < log.lastLogIndex, lastLogTerm > log.lastLogTerm)` message [#125](https://github.com/lerna-stack/akka-entity-replication/issues/125)
+- A new event is created even though all past events have not been applied [#130](https://github.com/lerna-stack/akka-entity-replication/issues/130)
 - `InstallSnapshot` can miss snapshots to copy [PR#128](https://github.com/lerna-stack/akka-entity-replication/pull/128)
 
   ⚠️ This change adds a new persistence event. This might don't allow downgrading after upgrading.
diff --git a/src/main/scala/lerna/akka/entityreplication/ReplicationActor.scala b/src/main/scala/lerna/akka/entityreplication/ReplicationActor.scala
@@ -144,6 +144,9 @@ trait ReplicationActor[StateData] extends Actor with Stash with akka.lerna.Stash
             innerApplyEvent(logEntry.event.event, logEntry.index)
             changeState(ready)
             internalStash.unstashAll()
+          case ReplicationFailed =>
+            changeState(ready)
+            internalStash.unstashAll()
           case ReplicationSucceeded(_, logEntryIndex, responseInstanceId) if responseInstanceId.contains(instanceId) =>
             changeState(ready)
             internalStash.unstashAll()
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/Leader.scala b/src/main/scala/lerna/akka/entityreplication/raft/Leader.scala
@@ -186,17 +186,29 @@ private[raft] trait Leader { this: RaftActor =>
     }
 
   private[this] def replicate(replicate: Replicate): Unit = {
-    cancelHeartbeatTimeoutTimer()
-    applyDomainEvent(AppendedEvent(EntityEvent(replicate.entityId, replicate.event))) { _ =>
-      applyDomainEvent(
-        StartedReplication(
-          ClientContext(replicate.replyTo, replicate.instanceId, replicate.originSender),
-          currentData.replicatedLog.lastLogIndex,
-        ),
-      ) { _ =>
-        publishAppendEntries()
-      }
+    replicate.entityId match {
+      case Some(normalizedEntityId) // from entity(ReplicationActor)
+          if currentData.hasUncommittedLogEntryOf(normalizedEntityId) =>
+        if (log.isWarningEnabled)
+          log.warning(
+            s"Failed to replicate the event (${replicate.event.getClass.getName}) since an uncommitted event exists for the entity (entityId: ${normalizedEntityId.raw}). Replicating new events is allowed after the event is committed",
+          )
+        replicate.replyTo ! ReplicationFailed
+
+      case _ =>
+        cancelHeartbeatTimeoutTimer()
+        applyDomainEvent(AppendedEvent(EntityEvent(replicate.entityId, replicate.event))) { _ =>
+          applyDomainEvent(
+            StartedReplication(
+              ClientContext(replicate.replyTo, replicate.instanceId, replicate.originSender),
+              currentData.replicatedLog.lastLogIndex,
+            ),
+          ) { _ =>
+            publishAppendEntries()
+          }
+        }
     }
+
   }
 
   private[this] def receiveReplicationResponse(event: ReplicationResponse): Unit =
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala b/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala
@@ -323,6 +323,12 @@ private[entityreplication] trait RaftMemberData
     replicatedLog.sliceEntries(from, to).filter(_.event.entityId.contains(entityId))
   }
 
+  def hasUncommittedLogEntryOf(entityId: NormalizedEntityId): Boolean = {
+    replicatedLog
+      .entriesAfter(index = commitIndex) // uncommitted entries
+      .exists(_.event.entityId.contains(entityId))
+  }
+
   def alreadyVotedOthers(candidate: MemberIndex): Boolean = votedFor.exists(candidate != _)
 
   def hasMatchLogEntry(prevLogIndex: LogEntryIndex, prevLogTerm: Term): Boolean = {
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/RaftProtocol.scala b/src/main/scala/lerna/akka/entityreplication/raft/RaftProtocol.scala
@@ -52,6 +52,7 @@ private[entityreplication] object RaftProtocol {
   final case class Replica(logEntry: LogEntry)                                            extends EntityCommand
   final case class TakeSnapshot(metadata: EntitySnapshotMetadata, replyTo: ActorRef)      extends EntityCommand
   final case object RecoveryTimeout                                                       extends EntityCommand
+  final case object ReplicationFailed                                                     extends EntityCommand
 
   sealed trait ReplicationResponse
 
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/model/ReplicatedLog.scala b/src/main/scala/lerna/akka/entityreplication/raft/model/ReplicatedLog.scala
@@ -36,6 +36,9 @@ private[entityreplication] final case class ReplicatedLog private[model] (
     entries.slice(toSeqIndex(from), until = toSeqIndex(to.next()))
   }
 
+  def entriesAfter(index: LogEntryIndex): Iterator[LogEntry] =
+    entries.iterator.drop(n = toSeqIndex(index) + 1)
+
   def nonEmpty: Boolean = entries.nonEmpty
 
   def append(event: EntityEvent, term: Term): ReplicatedLog = {
diff --git a/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Inactive.scala b/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Inactive.scala
@@ -37,6 +37,7 @@ private[entityreplication] class Inactive[Command, Event, State](
         case _: RaftProtocol.RecoveryState        => Behaviors.unhandled
         case _: RaftProtocol.ReplicationSucceeded => Behaviors.unhandled
         case RaftProtocol.RecoveryTimeout         => Behaviors.unhandled
+        case RaftProtocol.ReplicationFailed       => Behaviors.unhandled
       }.receiveSignal(setup.onSignal(setup.emptyState))
 
   def receiveActivate(command: RaftProtocol.Activate): Behavior[EntityCommand] = {
diff --git a/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Ready.scala b/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Ready.scala
@@ -76,6 +76,7 @@ private[entityreplication] class Ready[Command, Event, State](
         case _: RaftProtocol.RecoveryState        => Behaviors.unhandled
         case _: RaftProtocol.ReplicationSucceeded => Behaviors.unhandled
         case RaftProtocol.RecoveryTimeout         => Behaviors.unhandled
+        case RaftProtocol.ReplicationFailed       => Behaviors.unhandled
       }.receiveSignal(setup.onSignal(readyState.entityState))
 
   def receiveProcessCommand(command: RaftProtocol.ProcessCommand, state: BehaviorState): Behavior[EntityCommand] = {
diff --git a/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Recovering.scala b/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/Recovering.scala
@@ -92,6 +92,7 @@ private[entityreplication] class Recovering[Command, Event, State](
               Behaviors.same
             case _: RaftProtocol.Activate             => Behaviors.unhandled
             case _: RaftProtocol.ReplicationSucceeded => Behaviors.unhandled
+            case RaftProtocol.ReplicationFailed       => Behaviors.unhandled
           }.receiveSignal(setup.onSignal(setup.emptyState))
       }
     }
diff --git a/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/WaitForReplication.scala b/src/main/scala/lerna/akka/entityreplication/typed/internal/behavior/WaitForReplication.scala
@@ -39,6 +39,7 @@ private[entityreplication] class WaitForReplication[Command, Event, State](
       .receiveMessage[EntityCommand] {
         case command: RaftProtocol.Replica              => receiveReplica(command, state)
         case command: RaftProtocol.ReplicationSucceeded => receiveReplicationSucceeded(command, state)
+        case RaftProtocol.ReplicationFailed             => Ready.behavior(setup, transformReadyState(state)) // Discard side effects
         case command: RaftProtocol.TakeSnapshot         => receiveTakeSnapshot(command, state.entityState)
         case command: RaftProtocol.ProcessCommand =>
           setup.stashBuffer.stash(command)
diff --git a/src/multi-jvm/scala/lerna/akka/entityreplication/raft/RaftActorMultiNodeSpec.scala b/src/multi-jvm/scala/lerna/akka/entityreplication/raft/RaftActorMultiNodeSpec.scala
@@ -315,6 +315,7 @@ class RaftActorMultiNodeSpec extends MultiNodeSpec(RaftActorSpecConfig) with STM
 
       var leaderMember: RaftTestFSMRef   = null
       var followerMember: RaftTestFSMRef = null
+
       runOn(node1) {
         leaderMember = createRaftActor(
           shardId,
@@ -328,9 +329,9 @@ class RaftActorMultiNodeSpec extends MultiNodeSpec(RaftActorSpecConfig) with STM
         val leaderData = getState(leaderMember).stateData
         val log = leaderData.replicatedLog.merge(
           Seq(
-            LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "correct1"), Term(2)),
-            LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "correct2"), Term(2)),
-            LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "correct3"), Term(2)),
+            LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "correct1"), Term(1)),
+            LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "correct2"), Term(1)),
+            LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "correct3"), Term(1)),
           ),
           LogEntryIndex.initial(),
         )
@@ -362,13 +363,14 @@ class RaftActorMultiNodeSpec extends MultiNodeSpec(RaftActorSpecConfig) with STM
       val dummyEvent = "dummyEvent"
 
       val expectedLog = Seq(
-        LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "correct1"), Term(2)),
-        LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "correct2"), Term(2)),
-        LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "correct3"), Term(2)),
+        LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "correct1"), Term(1)),
+        LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "correct2"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "correct3"), Term(1)),
         LogEntry(LogEntryIndex(4), EntityEvent(Option(entityId), dummyEvent), Term(1)),
       )
 
       runOn(node1) {
+        awaitCond(getState(leaderMember).stateData.commitIndex == LogEntryIndex(3)) // event = "correct3"
         val instanceId = EntityInstanceId(1)
         leaderMember ! Replicate(dummyEvent, testActor, entityId, instanceId, ActorRef.noSender)
         inside(expectMsgType[ReplicationSucceeded]) {
diff --git a/src/test/scala/lerna/akka/entityreplication/ReplicationActorSpec.scala b/src/test/scala/lerna/akka/entityreplication/ReplicationActorSpec.scala
@@ -111,6 +111,22 @@ class ReplicationActorSpec extends TestKit(ActorSystem("ReplicationActorSpec", c
       raftActorProbe.expectMsgType[Replicate]
     }
 
+    "transition to `ready` state and processes the next command without executing any side effect if ReplicationActor receives ReplicationFailed in `waitForReplicationResponse` state" in {
+      val client           = TestProbe()
+      val raftActorProbe   = TestProbe()
+      val replicationActor = createReplicationActor(raftActorProbe)
+
+      client.send(replicationActor, Count())
+      raftActorProbe.expectMsgType[Replicate]
+
+      client.send(replicationActor, Count())
+      raftActorProbe.expectNoMessage() // the command is stashed
+
+      replicationActor ! ReplicationFailed // from raftActor
+      client.expectNoMessage()
+      raftActorProbe.expectMsgType[Replicate]
+    }
+
     "replace instanceId when it restarted" in {
       val watchProbe       = TestProbe()
       val raftActorProbe   = TestProbe()
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/RaftActorLeaderSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/RaftActorLeaderSpec.scala
@@ -1,12 +1,14 @@
 package lerna.akka.entityreplication.raft
 
 import akka.Done
+import akka.actor.testkit.typed.scaladsl.LoggingTestKit
+import akka.actor.typed.scaladsl.adapter._
 import akka.actor.{ ActorRef, ActorSystem }
 import akka.testkit.{ TestKit, TestProbe }
 import com.typesafe.config.ConfigFactory
 import lerna.akka.entityreplication.{ ClusterReplicationSettings, ReplicationRegion }
 import lerna.akka.entityreplication.model.{ EntityInstanceId, NormalizedEntityId, NormalizedShardId, TypeName }
-import lerna.akka.entityreplication.raft.RaftProtocol.Replicate
+import lerna.akka.entityreplication.raft.RaftProtocol.{ Replicate, ReplicationFailed }
 import lerna.akka.entityreplication.raft.model._
 import lerna.akka.entityreplication.raft.protocol.RaftCommands._
 import lerna.akka.entityreplication.testkit.CustomTestProbe._
@@ -24,6 +26,7 @@ import scala.concurrent.duration.DurationInt
 class RaftActorLeaderSpec extends TestKit(ActorSystem()) with RaftActorSpecBase with Inside {
 
   import RaftActor._
+  private implicit val typedSystem: akka.actor.typed.ActorSystem[Nothing] = system.toTyped
 
   private[this] val entityId = NormalizedEntityId.from("test-entity")
   private[this] val shardId  = NormalizedShardId.from("test-shard")
@@ -629,6 +632,52 @@ class RaftActorLeaderSpec extends TestKit(ActorSystem()) with RaftActorSpecBase
           msg.index
       } should contain theSameElementsAs (Set(follower1Index, follower2Index))
     }
+
+    "reply ReplicationFailed to replicationActor and log warn message if replication is in progress" in {
+      val replicationActor1 = TestProbe()
+      val replicationActor2 = TestProbe()
+      val entityId1         = NormalizedEntityId("test-1")
+      val entityId2         = NormalizedEntityId("test-2")
+      val entityInstanceId  = EntityInstanceId(1)
+
+      val leader     = createRaftActor()
+      val term       = Term(1)
+      val leaderData = createLeaderData(term)
+      setState(leader, Candidate, leaderData)
+      setState(leader, Leader, leaderData)
+
+      leader ! Replicate(
+        event = "a",
+        replyTo = replicationActor1.ref,
+        entityId1,
+        entityInstanceId,
+        originSender = system.deadLetters,
+      )
+      replicationActor1.expectNoMessage()
+
+      leader ! Replicate(
+        event = "b",
+        replicationActor2.ref,
+        entityId2,
+        entityInstanceId,
+        originSender = system.deadLetters,
+      )
+      replicationActor2.expectNoMessage()
+
+      LoggingTestKit
+        .warn(
+          "Failed to replicate the event (java.lang.String) since an uncommitted event exists for the entity (entityId: test-1). Replicating new events is allowed after the event is committed",
+        ).expect {
+          leader ! Replicate(
+            event = "c", // java.lang.String
+            replicationActor1.ref,
+            entityId1,
+            entityInstanceId,
+            originSender = system.deadLetters,
+          )
+          replicationActor1.expectMsg(ReplicationFailed)
+        }
+    }
   }
 
   private[this] val settings = ClusterReplicationSettings.create(system)
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/model/ReplicatedLogSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/model/ReplicatedLogSpec.scala
@@ -428,4 +428,46 @@ class ReplicatedLogSpec extends WordSpecLike with Matchers {
 
   }
 
+  "ReplicatedLog.entriesAfter" should {
+
+    "returns entries with deleted up to the specified LogEntryIndex" in {
+      val logEntries = Seq(
+        LogEntry(LogEntryIndex(1), EntityEvent(None, "a"), Term(1)),
+        LogEntry(LogEntryIndex(2), EntityEvent(None, "b"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(None, "c"), Term(1)),
+      )
+
+      val log = new ReplicatedLog(logEntries)
+
+      log.entriesAfter(index = LogEntryIndex.initial()).map(_.event.event).toList shouldBe List("a", "b", "c")
+      log.entriesAfter(index = LogEntryIndex(1)).map(_.event.event).toList shouldBe List("b", "c")
+      log.entriesAfter(index = LogEntryIndex(2)).map(_.event.event).toList shouldBe List("c")
+      log.entriesAfter(index = LogEntryIndex(3)).map(_.event.event).toList shouldBe List()
+      log.entriesAfter(index = LogEntryIndex(4)).map(_.event.event).toList shouldBe List()
+    }
+
+    "returns entries with deleted up to the specified LogEntryIndex when the log is compressed" in {
+      val logEntries = Seq(
+        LogEntry(LogEntryIndex(1), EntityEvent(None, "a"), Term(1)),
+        LogEntry(LogEntryIndex(2), EntityEvent(None, "b"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(None, "c"), Term(1)),
+        LogEntry(LogEntryIndex(4), EntityEvent(None, "d"), Term(1)),
+        LogEntry(LogEntryIndex(5), EntityEvent(None, "e"), Term(1)),
+      )
+
+      val log = new ReplicatedLog(logEntries).deleteOldEntries(to = LogEntryIndex(2), preserveLogSize = 3)
+      require(log.entries.map(_.index.underlying) == List(3, 4, 5))
+      require(log.entries.map(_.event.event) == List("c", "d", "e"))
+
+      log.entriesAfter(index = LogEntryIndex.initial()).map(_.event.event).toList shouldBe List("c", "d", "e")
+      log.entriesAfter(index = LogEntryIndex(1)).map(_.event.event).toList shouldBe List("c", "d", "e")
+      log.entriesAfter(index = LogEntryIndex(2)).map(_.event.event).toList shouldBe List("c", "d", "e")
+      log.entriesAfter(index = LogEntryIndex(3)).map(_.event.event).toList shouldBe List("d", "e")
+      log.entriesAfter(index = LogEntryIndex(4)).map(_.event.event).toList shouldBe List("e")
+      log.entriesAfter(index = LogEntryIndex(5)).map(_.event.event).toList shouldBe List()
+      log.entriesAfter(index = LogEntryIndex(6)).map(_.event.event).toList shouldBe List()
+    }
+
+  }
+
 }
diff --git a/src/test/scala/lerna/akka/entityreplication/typed/ReplicatedEntityBehaviorSpec.scala b/src/test/scala/lerna/akka/entityreplication/typed/ReplicatedEntityBehaviorSpec.scala
@@ -406,6 +406,38 @@ class ReplicatedEntityBehaviorSpec extends WordSpec with BeforeAndAfterAll with
       testkit.stop(bankAccount)
     }
 
+    "transitions to `Ready` state and processes the next command without executing any side effect if ReplicationActor receives ReplicationFailed in `WaitForReplication` state" in {
+      val bankAccount = spawnEntity(BankAccountBehavior(entityContext))
+
+      {
+        // recover the entity
+        import SnapshotProtocol._
+        val metadata = EntitySnapshotMetadata(normalizedEntityId, nextLogEntryIndex())
+        val state    = EntityState(BankAccountBehavior.Account(100))
+        inside(snapshotStoreProbe.receiveMessage()) {
+          case SnapshotProtocol.FetchSnapshot(_, replyTo) =>
+            replyTo ! SnapshotProtocol.SnapshotFound(EntitySnapshot(metadata, state))
+        }
+        inside(shardProbe.receiveMessage()) {
+          case FetchEntityEvents(_, _, _, replyTo) =>
+            replyTo ! FetchEntityEventsResponse(Seq())
+        }
+      }
+
+      val clientProbe = bankAccount.askWithTestProbe(BankAccountBehavior.GetBalance)
+      shardProbe.expectMessageType[RaftProtocol.Replicate]
+      clientProbe.expectNoMessage()
+
+      bankAccount.askWithTestProbe(BankAccountBehavior.GetBalance)
+      shardProbe.expectNoMessage() // the command is stashed
+
+      bankAccount.asEntity ! RaftProtocol.ReplicationFailed // from raftActor
+      clientProbe.expectNoMessage()
+      shardProbe.expectMessageType[RaftProtocol.Replicate]
+
+      testkit.stop(bankAccount)
+    }
+
     "can process commands after receiving Replica even if replication is in progress" in {
       val bankAccount = spawnEntity(BankAccountBehavior(entityContext))
 

Original file line number	Diff line number	Diff line change
`@@ -36,6 +36,9 @@ private[entityreplication] final case class ReplicatedLog private[model] (`
`36`	`36`	`entries.slice(toSeqIndex(from), until = toSeqIndex(to.next()))`
`37`	`37`	`}`
`38`	`38`
	`39`	`+ def entriesAfter(index: LogEntryIndex): Iterator[LogEntry] =`
	`40`	`+ entries.iterator.drop(n = toSeqIndex(index) + 1)`
	`41`	`+`
`39`	`42`	`def nonEmpty: Boolean = entries.nonEmpty`
`40`	`43`
`41`	`44`	`def append(event: EntityEvent, term: Term): ReplicatedLog = {`
Original file line number	Diff line number	Diff line change
`@@ -92,6 +92,7 @@ private[entityreplication] class Recovering[Command, Event, State](`
`92`	`92`	`Behaviors.same`
`93`	`93`	`case _: RaftProtocol.Activate => Behaviors.unhandled`
`94`	`94`	`case _: RaftProtocol.ReplicationSucceeded => Behaviors.unhandled`
	`95`	`+ case RaftProtocol.ReplicationFailed => Behaviors.unhandled`
`95`	`96`	`}.receiveSignal(setup.onSignal(setup.emptyState))`
`96`	`97`	`}`
`97`	`98`	`}`