Persit EntitySnapshot as an event

CHANGELOG.md

@@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add function of Disabling raft actor [PR#173](https://github.com/lerna-stack/akka-entity-replication/pull/173)
 - Add diagnostic info to logs of sending replication results
   [PR#179](https://github.com/lerna-stack/akka-entity-replication/pull/179)
+- Persist EntitySnapshot as an event
+  [PR#184](https://github.com/lerna-stack/akka-entity-replication/pull/184)
 
 ### Fixed
 - RaftActor might delete committed entries

src/main/resources/reference.conf

@@ -70,6 +70,16 @@ lerna.akka.entityreplication {
       max-snapshot-batch-size = 1000
     }
 
+    // Entity snapshot store settings
+    // All entity snapshot histries are persisted for data recovery.
+    entity-snapshot-store {
+
+      // Frequency of taking snapshot of entity snapshot.
+      // By taking snapshot, optimizing recovery times of an Entity.
+      // Increasing this value increases snapshot writing, decreasing it increases recovery times.
+      snapshot-every = 1
+    }
+
     sharding = ${akka.cluster.sharding} {
       // Maximum number of messages that are buffered by a ShardRegion actor.
       // Make it smaller than the default value to discard messages that are too old.

src/main/scala/lerna/akka/entityreplication/raft/RaftSettings.scala

@@ -50,6 +50,8 @@ trait RaftSettings {
 
   def snapshotSyncMaxSnapshotBatchSize: Int
 
+  def snapshotStoreSnapshotEvery: Int
+
   def clusterShardingConfig: Config
 
   def raftActorAutoStartFrequency: FiniteDuration

src/main/scala/lerna/akka/entityreplication/raft/RaftSettingsImpl.scala

@@ -27,6 +27,7 @@ private[entityreplication] final case class RaftSettingsImpl(
     snapshotSyncCopyingParallelism: Int,
     snapshotSyncPersistenceOperationTimeout: FiniteDuration,
     snapshotSyncMaxSnapshotBatchSize: Int,
+    snapshotStoreSnapshotEvery: Int,
     clusterShardingConfig: Config,
     raftActorAutoStartFrequency: FiniteDuration,
     raftActorAutoStartNumberOfActors: Int,
@@ -156,6 +157,12 @@ private[entityreplication] object RaftSettingsImpl {
       s"snapshot-sync.max-snapshot-batch-size (${snapshotSyncMaxSnapshotBatchSize}) should be larger than 0",
     )
 
+    val snapshotStoreSnapshotEvery: Int = config.getInt("entity-snapshot-store.snapshot-every")
+    require(
+      snapshotStoreSnapshotEvery > 0,
+      s"entity-snapshot-store.snapshot-every ($snapshotStoreSnapshotEvery) should be larger than 0",
+    )
+
     val clusterShardingConfig: Config = config.getConfig("sharding")
 
     val raftActorAutoStartFrequency: FiniteDuration =
@@ -237,6 +244,7 @@ private[entityreplication] object RaftSettingsImpl {
       snapshotSyncCopyingParallelism = snapshotSyncCopyingParallelism,
       snapshotSyncPersistenceOperationTimeout = snapshotSyncPersistenceOperationTimeout,
       snapshotSyncMaxSnapshotBatchSize = snapshotSyncMaxSnapshotBatchSize,
+      snapshotStoreSnapshotEvery = snapshotStoreSnapshotEvery,
       clusterShardingConfig = clusterShardingConfig,
       raftActorAutoStartFrequency = raftActorAutoStartFrequency,
       raftActorAutoStartNumberOfActors = raftActorAutoStartNumberOfActors,

src/main/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStore.scala

@@ -2,7 +2,7 @@ package lerna.akka.entityreplication.raft.snapshot
 
 import akka.actor.{ ActorLogging, ActorRef, Props, ReceiveTimeout }
 import akka.persistence
-import akka.persistence.{ PersistentActor, Recovery, SnapshotSelectionCriteria }
+import akka.persistence.PersistentActor
 import lerna.akka.entityreplication.model.{ NormalizedEntityId, TypeName }
 import lerna.akka.entityreplication.raft.RaftSettings
 import lerna.akka.entityreplication.raft.routing.MemberIndex
@@ -33,6 +33,8 @@ private[entityreplication] class SnapshotStore(
     with ActorLogging {
   import SnapshotProtocol._
 
+  private var replyRefCache: Option[ActorRef] = None
+
   override def persistenceId: String =
     SnapshotStore.persistenceId(typeName, entityId, selfMemberIndex)
 
@@ -42,20 +44,30 @@ private[entityreplication] class SnapshotStore(
 
   context.setReceiveTimeout(settings.compactionSnapshotCacheTimeToLive)
 
-  // SequenceNr is always 0 because SnapshotStore doesn't persist events (only persist snapshots).
-  override def recovery: Recovery =
-    Recovery(
-      toSequenceNr = 0L,
-      fromSnapshot = SnapshotSelectionCriteria(maxSequenceNr = 0L, maxTimestamp = Long.MaxValue),
-    )
-
   override def receiveRecover: Receive = {
+    case snapshot: EntitySnapshot => context.become(hasSnapshot(snapshot))
     case akka.persistence.SnapshotOffer(_, s: EntitySnapshot) =>
       context.become(hasSnapshot(s))
   }
 
   override def receiveCommand: Receive = hasNoSnapshot
 
+  override protected def onPersistFailure(cause: Throwable, event: Any, seqNr: Long): Unit = {
+    if (log.isWarningEnabled) {
+      log.warning(
+        "Saving snapshot failed - {}: {}",
+        cause.getClass.getCanonicalName,
+        cause.getMessage,
+      )
+    }
+    super.onPersistFailure(cause, event, seqNr)
+    replyRefCache.fold(
+      if (log.isWarningEnabled) log.warning("missing reply reference - {}", cause.getClass.getCanonicalName),
+    )(
+      _ ! SaveSnapshotFailure(event.asInstanceOf[EntitySnapshot].metadata),
+    )
+  }
+
   def hasNoSnapshot: Receive = {
     case command: Command =>
       command match {
@@ -81,12 +93,19 @@ private[entityreplication] class SnapshotStore(
       // reduce IO: don't save if same as cached snapshot
       command.replyTo ! SaveSnapshotSuccess(command.snapshot.metadata)
     } else {
-      saveSnapshot(command.snapshot)
-      context.become(savingSnapshot(command.replyTo, command.snapshot, prevSnapshot))
+      replyRefCache = Option(command.replyTo)
+      persistAsync(command.snapshot) { _ =>
+        command.replyTo ! SaveSnapshotSuccess(command.snapshot.metadata)
+        if (lastSequenceNr % settings.snapshotStoreSnapshotEvery == 0 && lastSequenceNr != 0) {
+          saveSnapshot(command.snapshot)
+        }
+        context.become(hasSnapshot(command.snapshot))
+      }
+      context.become(savingSnapshot(prevSnapshot))
     }
   }
 
-  def savingSnapshot(replyTo: ActorRef, snapshot: EntitySnapshot, prevSnapshot: Option[EntitySnapshot]): Receive = {
+  def savingSnapshot(prevSnapshot: Option[EntitySnapshot]): Receive = {
     case command: Command =>
       command match {
         case cmd: SaveSnapshot =>
@@ -103,23 +122,24 @@ private[entityreplication] class SnapshotStore(
               replyTo ! SnapshotProtocol.SnapshotNotFound(entityId)
           }
       }
-    case _: persistence.SaveSnapshotSuccess =>
-      replyTo ! SaveSnapshotSuccess(snapshot.metadata)
-      context.become(hasSnapshot(snapshot))
-    case failure: persistence.SaveSnapshotFailure =>
-      if (log.isWarningEnabled)
-        log.warning(
-          "Saving snapshot failed - {}: {}",
-          failure.cause.getClass.getCanonicalName,
-          failure.cause.getMessage,
-        )
-      replyTo ! SaveSnapshotFailure(snapshot.metadata)
   }
 
   override def unhandled(message: Any): Unit =
     message match {
       case ReceiveTimeout =>
         context.stop(self)
+      case _: persistence.SaveSnapshotSuccess =>
+        if (log.isDebugEnabled) {
+          log.debug("Saving EntitySnapshot as a snapshot succeeded.")
+        }
+      case failure: persistence.SaveSnapshotFailure =>
+        if (log.isWarningEnabled) {
+          log.warning(
+            "Saving EntitySnapshot as a snapshot failed. - {}: {}",
+            failure.cause.getClass.getCanonicalName,
+            failure.cause.getMessage,
+          )
+        }
       case _ =>
         super.unhandled(message)
     }

src/test/scala/lerna/akka/entityreplication/raft/RaftSettingsSpec.scala

@@ -35,6 +35,7 @@ final class RaftSettingsSpec extends TestKit(ActorSystem("RaftSettingsSpec")) wi
       settings.snapshotSyncCopyingParallelism shouldBe 10
       settings.snapshotSyncPersistenceOperationTimeout shouldBe 10.seconds
       settings.snapshotSyncMaxSnapshotBatchSize shouldBe 1_000
+      settings.snapshotStoreSnapshotEvery shouldBe 1
       settings.clusterShardingConfig shouldBe defaultConfig.getConfig("lerna.akka.entityreplication.raft.sharding")
       settings.raftActorAutoStartFrequency shouldBe 3.seconds
       settings.raftActorAutoStartNumberOfActors shouldBe 5
@@ -179,6 +180,29 @@ final class RaftSettingsSpec extends TestKit(ActorSystem("RaftSettingsSpec")) wi
       }
     }
 
+    "throw an IllegalArgumentException if the given entity-snapshot-store.snapshot-every is less than or equal to 0" in {
+      {
+        val config = ConfigFactory
+          .parseString("""
+              |lerna.akka.entityreplication.raft.entity-snapshot-store.snapshot-every = -1
+              |""".stripMargin)
+          .withFallback(defaultConfig)
+        a[IllegalArgumentException] shouldBe thrownBy {
+          RaftSettings(config)
+        }
+      }
+      {
+        val config = ConfigFactory
+          .parseString("""
+              |lerna.akka.entityreplication.raft.entity-snapshot-store.snapshot-every = 0
+              |""".stripMargin)
+          .withFallback(defaultConfig)
+        a[IllegalArgumentException] shouldBe thrownBy {
+          RaftSettings(config)
+        }
+      }
+    }
+
     "throw an IllegalArgumentException if the given raft-actor-auto-start.frequency is 0 milli" in {
       val config = ConfigFactory
         .parseString("""