Merge pull request #181 from lerna-stack/add-unit-tests-of-snapshot-store

negokaz · web-flow · commit da0458a5411b · 2022-10-28T09:58:16.000+09:00
Add unit tests of SnapshotStore
diff --git a/src/main/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStore.scala b/src/main/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStore.scala
@@ -17,6 +17,11 @@ private[entityreplication] object SnapshotStore {
       selfMemberIndex: MemberIndex,
   ): Props =
     Props(new SnapshotStore(typeName, entityId, settings, selfMemberIndex))
+
+  /** Returns a persistence ID of SnapshotStore */
+  def persistenceId(typeName: TypeName, entityId: NormalizedEntityId, selfMemberIndex: MemberIndex): String =
+    ActorIds.persistenceId("SnapshotStore", typeName.underlying, entityId.underlying, selfMemberIndex.role)
+
 }
 
 private[entityreplication] class SnapshotStore(
@@ -29,7 +34,7 @@ private[entityreplication] class SnapshotStore(
   import SnapshotProtocol._
 
   override def persistenceId: String =
-    ActorIds.persistenceId("SnapshotStore", typeName.underlying, entityId.underlying, selfMemberIndex.role)
+    SnapshotStore.persistenceId(typeName, entityId, selfMemberIndex)
 
   override def journalPluginId: String = settings.journalPluginId
 
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreFailureSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreFailureSpec.scala
@@ -1,49 +1,43 @@
 package lerna.akka.entityreplication.raft.snapshot
 
+import akka.actor.testkit.typed.scaladsl.LoggingTestKit
+import akka.actor.typed.scaladsl.adapter.ClassicActorSystemOps
+
 import java.util.concurrent.atomic.AtomicInteger
 import akka.actor.{ ActorRef, ActorSystem }
 import akka.persistence.testkit.scaladsl.SnapshotTestKit
-import akka.persistence.testkit.{ PersistenceTestKitPlugin, PersistenceTestKitSnapshotPlugin }
+import akka.persistence.testkit.{
+  ProcessingResult,
+  ProcessingSuccess,
+  SnapshotOperation,
+  SnapshotStorage,
+  WriteSnapshot,
+}
 import akka.testkit.TestKit
-import com.typesafe.config.{ Config, ConfigFactory }
 import lerna.akka.entityreplication.model.{ NormalizedEntityId, TypeName }
 import lerna.akka.entityreplication.raft.model.LogEntryIndex
 import lerna.akka.entityreplication.raft.routing.MemberIndex
-import lerna.akka.entityreplication.raft.snapshot.ShardSnapshotStoreFailureSpec._
 import lerna.akka.entityreplication.raft.snapshot.SnapshotProtocol._
 import lerna.akka.entityreplication.raft.{ ActorSpec, RaftSettings }
 import lerna.akka.entityreplication.testkit.KryoSerializable
 
+import scala.concurrent.Promise
+import scala.util.Using
+
 object ShardSnapshotStoreFailureSpec {
   final case object DummyState extends KryoSerializable
-
-  def configWithPersistenceTestKits: Config = {
-    PersistenceTestKitPlugin.config
-      .withFallback(PersistenceTestKitSnapshotPlugin.config)
-      .withFallback(raftPersistenceConfigWithPersistenceTestKits)
-      .withFallback(ConfigFactory.load())
-  }
-
-  private val raftPersistenceConfigWithPersistenceTestKits: Config = ConfigFactory.parseString(
-    s"""
-       |lerna.akka.entityreplication.raft.persistence {
-       |  journal.plugin = ${PersistenceTestKitPlugin.PluginId}
-       |  snapshot-store.plugin = ${PersistenceTestKitSnapshotPlugin.PluginId}
-       |  # Might be possible to use PersistenceTestKitReadJournal
-       |  // query.plugin = ""
-       |}
-       |""".stripMargin,
-  )
-
 }
 
 class ShardSnapshotStoreFailureSpec
     extends TestKit(
-      ActorSystem("ShardSnapshotStoreFailureSpec", ShardSnapshotStoreFailureSpec.configWithPersistenceTestKits),
+      ActorSystem("ShardSnapshotStoreFailureSpec", ShardSnapshotStoreSpecBase.configWithPersistenceTestKits),
     )
     with ActorSpec {
+  import ShardSnapshotStoreFailureSpec._
 
   private val snapshotTestKit = SnapshotTestKit(system)
+  private val typeName        = TypeName.from("test")
+  private val memberIndex     = MemberIndex("test-role")
 
   override def beforeEach(): Unit = {
     super.beforeEach()
@@ -55,9 +49,9 @@ class ShardSnapshotStoreFailureSpec
     planAutoKill {
       childActorOf(
         ShardSnapshotStore.props(
-          TypeName.from("test"),
+          typeName,
           RaftSettings(system.settings.config),
-          MemberIndex("test-role"),
+          memberIndex,
         ),
       )
     }
@@ -70,10 +64,11 @@ class ShardSnapshotStoreFailureSpec
   "ShardSnapshotStore（読み込みの異常）" should {
 
     "FetchSnapshot に失敗した場合は応答無し（クライアント側でタイムアウトの実装が必要）" in {
-      val entityId           = generateUniqueEntityId()
-      val shardSnapshotStore = createShardSnapshotStore()
+      val entityId                   = generateUniqueEntityId()
+      val shardSnapshotStore         = createShardSnapshotStore()
+      val snapshotStorePersistenceId = SnapshotStore.persistenceId(typeName, entityId, memberIndex)
 
-      snapshotTestKit.failNextRead()
+      snapshotTestKit.failNextRead(snapshotStorePersistenceId)
       shardSnapshotStore ! FetchSnapshot(entityId, replyTo = testActor)
       expectNoMessage()
     }
@@ -82,16 +77,100 @@ class ShardSnapshotStoreFailureSpec
   "ShardSnapshotStore（書き込みの異常）" should {
 
     "SaveSnapshot に失敗した場合は SaveSnapshotFailure が返信される" in {
+      val entityId                   = generateUniqueEntityId()
+      val shardSnapshotStore         = createShardSnapshotStore()
+      val snapshotStorePersistenceId = SnapshotStore.persistenceId(typeName, entityId, memberIndex)
+      val metadata                   = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
+      val dummyEntityState           = EntityState(DummyState)
+      val snapshot                   = EntitySnapshot(metadata, dummyEntityState)
+
+      snapshotTestKit.failNextPersisted(snapshotStorePersistenceId)
+      shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+      expectMsg(SaveSnapshotFailure(metadata))
+    }
+  }
+
+  "ShardSnapshotStore (with time-consuming writes)" should {
+
+    // Emulates a time-consuming write
+    class TimeConsumingWriteSnapshotPolicy extends SnapshotStorage.SnapshotPolicies.PolicyType with AutoCloseable {
+      val processingResultPromise = Promise[ProcessingResult]()
+      override def tryProcess(persistenceId: String, processingUnit: SnapshotOperation): ProcessingResult = {
+        processingUnit match {
+          case _: WriteSnapshot => processingResultPromise.future.await
+          case _                => ProcessingSuccess
+        }
+      }
+      override def close(): Unit = {
+        processingResultPromise.trySuccess(ProcessingSuccess)
+      }
+    }
+
+    "reply with `SnapshotNotFound` to `FetchSnapshot` if it has no EntitySnapshot and is saving an EntitySnapshot" ignore {
+      // TODO Change SnapshotStore.savingSnapshot such that this test passes.
       val entityId           = generateUniqueEntityId()
       val shardSnapshotStore = createShardSnapshotStore()
-      val metadata           = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
-      val dummyEntityState   = EntityState(DummyState)
-      val snapshot           = EntitySnapshot(metadata, dummyEntityState)
+      val metadata           = EntitySnapshotMetadata(entityId, LogEntryIndex(1))
+      val snapshot           = EntitySnapshot(metadata, EntityState(DummyState))
+
+      Using(new TimeConsumingWriteSnapshotPolicy()) { timeConsumingWriteSnapshotPolicy =>
+        // Prepare: SnapshotStore is saving the snapshot
+        snapshotTestKit.withPolicy(timeConsumingWriteSnapshotPolicy)
+        shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+
+        // Test:
+        shardSnapshotStore ! FetchSnapshot(entityId, replyTo = testActor)
+        expectMsg(SnapshotNotFound)
+      }
+    }
 
-      snapshotTestKit.failNextPersisted()
-      shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
-      expectMsg(SaveSnapshotFailure(metadata))
+    "reply with `SnapshotFound` to `FetchSnapshot` if it has an EntitySnapshot and is saving a new EntitySnapshot" in {
+      val entityId           = generateUniqueEntityId()
+      val shardSnapshotStore = createShardSnapshotStore()
+
+      val firstSnapshotMetadata = EntitySnapshotMetadata(entityId, LogEntryIndex(1))
+      val firstSnapshot =
+        EntitySnapshot(firstSnapshotMetadata, EntityState(DummyState))
+      shardSnapshotStore ! SaveSnapshot(firstSnapshot, replyTo = testActor)
+      expectMsg(SaveSnapshotSuccess(firstSnapshotMetadata))
+
+      Using(new TimeConsumingWriteSnapshotPolicy()) { timeConsumingWriteSnapshotPolicy =>
+        // Prepare: SnapshotStore is saving the second snapshot
+        snapshotTestKit.withPolicy(timeConsumingWriteSnapshotPolicy)
+        val secondSnapshot =
+          EntitySnapshot(EntitySnapshotMetadata(entityId, LogEntryIndex(5)), EntityState(DummyState))
+        shardSnapshotStore ! SaveSnapshot(secondSnapshot, replyTo = testActor)
+
+        // Test:
+        shardSnapshotStore ! FetchSnapshot(entityId, replyTo = testActor)
+        expectMsg(SnapshotFound(firstSnapshot))
+      }
     }
+
+    "reply with nothing to `SaveSnapshot` and log a warning if it is saving an EntitySnapshot" in {
+      implicit val typedSystem: akka.actor.typed.ActorSystem[Nothing] = system.toTyped
+
+      val entityId           = generateUniqueEntityId()
+      val shardSnapshotStore = createShardSnapshotStore()
+      val metadata           = EntitySnapshotMetadata(entityId, LogEntryIndex(1))
+      val snapshot           = EntitySnapshot(metadata, EntityState(DummyState))
+
+      Using(new TimeConsumingWriteSnapshotPolicy()) { timeConsumingWriteSnapshotPolicy =>
+        // Prepare: SnapshotStore is saving the snapshot
+        snapshotTestKit.withPolicy(timeConsumingWriteSnapshotPolicy)
+        shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+
+        // Test:
+        LoggingTestKit
+          .warn(
+            s"Saving snapshot for an entity ($entityId) currently. Consider to increase log-size-threshold or log-size-check-interval.",
+          ).expect {
+            shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+          }
+        expectNoMessage()
+      }
+    }
+
   }
 
 }
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreSpecBase.scala b/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreSpecBase.scala
@@ -0,0 +1,26 @@
+package lerna.akka.entityreplication.raft.snapshot
+
+import akka.persistence.testkit.{ PersistenceTestKitPlugin, PersistenceTestKitSnapshotPlugin }
+import com.typesafe.config.{ Config, ConfigFactory }
+
+object ShardSnapshotStoreSpecBase {
+
+  def configWithPersistenceTestKits: Config = {
+    PersistenceTestKitPlugin.config
+      .withFallback(PersistenceTestKitSnapshotPlugin.config)
+      .withFallback(raftPersistenceConfigWithPersistenceTestKits)
+      .withFallback(ConfigFactory.load())
+  }
+
+  private val raftPersistenceConfigWithPersistenceTestKits: Config = ConfigFactory.parseString(
+    s"""
+       |lerna.akka.entityreplication.raft.persistence {
+       |  journal.plugin = ${PersistenceTestKitPlugin.PluginId}
+       |  snapshot-store.plugin = ${PersistenceTestKitSnapshotPlugin.PluginId}
+       |  # Might be possible to use PersistenceTestKitReadJournal
+       |  // query.plugin = ""
+       |}
+       |""".stripMargin,
+  )
+
+}
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreSuccessSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/snapshot/ShardSnapshotStoreSuccessSpec.scala
@@ -2,6 +2,7 @@ package lerna.akka.entityreplication.raft.snapshot
 
 import java.util.concurrent.atomic.AtomicInteger
 import akka.actor.{ ActorRef, ActorSystem, PoisonPill }
+import akka.persistence.testkit.scaladsl.SnapshotTestKit
 import akka.testkit.TestKit
 import com.typesafe.config.{ Config, ConfigFactory }
 import lerna.akka.entityreplication.model.{ NormalizedEntityId, TypeName }
@@ -14,31 +15,66 @@ object ShardSnapshotStoreSuccessSpec {
   final case object DummyState extends KryoSerializable
 }
 
-class ShardSnapshotStoreSuccessSpec extends TestKit(ActorSystem()) with ActorSpec {
+class ShardSnapshotStoreSuccessSpec
+    extends TestKit(
+      ActorSystem("ShardSnapshotStoreSuccessSpec", ShardSnapshotStoreSpecBase.configWithPersistenceTestKits),
+    )
+    with ActorSpec {
   import ShardSnapshotStoreSuccessSpec._
   import lerna.akka.entityreplication.raft.snapshot.SnapshotProtocol._
 
+  private val snapshotTestKit        = SnapshotTestKit(system)
+  private val typeName               = TypeName.from("test")
+  private val memberIndex            = MemberIndex("test-role")
   private[this] val dummyEntityState = EntityState(DummyState)
 
+  override def beforeEach(): Unit = {
+    super.beforeEach()
+    snapshotTestKit.clearAll()
+    snapshotTestKit.resetPolicy()
+  }
+
   "ShardSnapshotStore（正常系）" should {
 
     "SaveSnapshot に成功した場合は SaveSnapshotSuccess が返信される" in {
-      val entityId           = generateUniqueEntityId()
-      val shardSnapshotStore = createShardSnapshotStore()
-      val metadata           = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
-      val snapshot           = EntitySnapshot(metadata, dummyEntityState)
+      val entityId                   = generateUniqueEntityId()
+      val shardSnapshotStore         = createShardSnapshotStore()
+      val snapshotStorePersistenceId = SnapshotStore.persistenceId(typeName, entityId, memberIndex)
+      val metadata                   = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
+      val snapshot                   = EntitySnapshot(metadata, dummyEntityState)
+
+      shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+      snapshotTestKit.expectNextPersisted(snapshotStorePersistenceId, snapshot)
+      expectMsg(SaveSnapshotSuccess(metadata))
+    }
+
+    "persist nothing and reply with SaveSnapshotSuccess to SaveSnapshot if it has the same EntitySnapshot" in {
+      val entityId                   = generateUniqueEntityId()
+      val shardSnapshotStore         = createShardSnapshotStore()
+      val snapshotStorePersistenceId = SnapshotStore.persistenceId(typeName, entityId, memberIndex)
+      val metadata                   = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
+      val snapshot                   = EntitySnapshot(metadata, dummyEntityState)
+
+      // Prepare:
+      shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+      snapshotTestKit.expectNextPersisted(snapshotStorePersistenceId, snapshot)
+      expectMsg(SaveSnapshotSuccess(metadata))
 
+      // Test:
       shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+      snapshotTestKit.expectNothingPersisted(snapshotStorePersistenceId)
       expectMsg(SaveSnapshotSuccess(metadata))
     }
 
     "FetchSnapshot に成功した場合は一度停止しても SnapshotFound でスナップショットが返信される" in {
-      val entityId           = generateUniqueEntityId()
-      val shardSnapshotStore = createShardSnapshotStore()
-      val metadata           = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
-      val snapshot           = EntitySnapshot(metadata, dummyEntityState)
+      val entityId                   = generateUniqueEntityId()
+      val shardSnapshotStore         = createShardSnapshotStore()
+      val snapshotStorePersistenceId = SnapshotStore.persistenceId(typeName, entityId, memberIndex)
+      val metadata                   = EntitySnapshotMetadata(entityId, LogEntryIndex.initial())
+      val snapshot                   = EntitySnapshot(metadata, dummyEntityState)
 
       shardSnapshotStore ! SaveSnapshot(snapshot, replyTo = testActor)
+      snapshotTestKit.expectNextPersisted(snapshotStorePersistenceId, snapshot)
       expectMsg(SaveSnapshotSuccess(metadata))
 
       // terminate SnapshotStore
@@ -84,9 +120,9 @@ class ShardSnapshotStoreSuccessSpec extends TestKit(ActorSystem()) with ActorSpe
     planAutoKill {
       childActorOf(
         ShardSnapshotStore.props(
-          TypeName.from("test"),
+          typeName,
           RaftSettings(additionalConfig.withFallback(system.settings.config)),
-          MemberIndex("test-role"),
+          memberIndex,
         ),
       )
     }
diff --git a/src/test/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStoreSpec.scala b/src/test/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStoreSpec.scala
@@ -0,0 +1,24 @@
+package lerna.akka.entityreplication.raft.snapshot
+
+import akka.actor.ActorSystem
+import akka.testkit.TestKit
+import lerna.akka.entityreplication.model.{ NormalizedEntityId, TypeName }
+import lerna.akka.entityreplication.raft.ActorSpec
+import lerna.akka.entityreplication.raft.routing.MemberIndex
+
+final class SnapshotStoreSpec extends TestKit(ActorSystem("SnapshotStoreSpec")) with ActorSpec {
+
+  "SnapshotStore.persistenceId" should {
+
+    "return a persistence ID for the given type name, entity ID, and member index" in {
+      val persistenceId = SnapshotStore.persistenceId(
+        TypeName.from("test-type-name"),
+        NormalizedEntityId.from("test-entity-id"),
+        MemberIndex("test-member-index"),
+      )
+      assert(persistenceId === "SnapshotStore:test-type-name:test-entity-id:test-member-index")
+    }
+
+  }
+
+}
