Add tests for deleting old data

CHANGELOG.md

@@ -10,7 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Support deleting old events and snapshots
   [#201](https://github.com/lerna-stack/akka-entity-replication/issues/201),
-  [PR#204](https://github.com/lerna-stack/akka-entity-replication/pull/204)
+  [PR#204](https://github.com/lerna-stack/akka-entity-replication/pull/204),
+  [PR#212](https://github.com/lerna-stack/akka-entity-replication/pull/212)
 
 ### Changed
 - Rollback deletes only tagged events after the given sequence number

core/src/test/scala/lerna/akka/entityreplication/raft/RaftActorPersistenceDeletionSpec.scala

@@ -16,12 +16,14 @@ import lerna.akka.entityreplication.raft.protocol.RaftCommands._
 import lerna.akka.entityreplication.raft.routing.MemberIndex
 import lerna.akka.entityreplication.raft.snapshot.SnapshotProtocol._
 import org.scalactic.TypeCheckedTripleEquals
+import org.scalatest.BeforeAndAfterAll
 
 final class RaftActorPersistenceDeletionSpec
     extends TestKit(
       ActorSystem("RaftActorPersistenceDeletionSpec", RaftActorSpecBase.configWithPersistenceTestKits),
     )
     with RaftActorSpecBase
+    with BeforeAndAfterAll
     with TypeCheckedTripleEquals {
 
   private implicit val typedSystem: typed.ActorSystem[Nothing] = system.toTyped
@@ -36,6 +38,11 @@ final class RaftActorPersistenceDeletionSpec
     snapshotTestKit.resetPolicy()
   }
 
+  override def afterAll(): Unit = {
+    try shutdown(system)
+    finally super.afterAll()
+  }
+
   private def configFor(
       deleteBeforeRelativeSequenceNr: Long,
       deleteOldEvents: Boolean = false,
@@ -265,6 +272,47 @@ final class RaftActorPersistenceDeletionSpec
       )
     }
 
+    "not delete events if it fails a snapshot save" in new Fixture {
+      val raftActor = createFollower(
+        configFor(
+          deleteBeforeRelativeSequenceNr = 0,
+          deleteOldEvents = true,
+        ),
+      )
+
+      // Arrange: append new entries, which will trigger compaction.
+      appendNewEntriesTo(
+        raftActor,
+        currentTerm = Term(1),
+        prevLogTerm = Term(1),
+        LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "event-1"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "event-2"), Term(1)),
+        LogEntry(LogEntryIndex(4), EntityEvent(Option(entityId), "event-3"), Term(1)),
+      )
+
+      promoteBeforeDelete(this, raftActor)
+
+      // Arrange: store events to verify no event deletion.
+      val eventsBeforeCompaction = persistenceTestKit.persistedInStorage(persistenceId)
+      assert(eventsBeforeCompaction.nonEmpty)
+
+      LoggingTestKit.warn("Failed to saveSnapshot").expect {
+        // Arrange: the next snapshot save will fail.
+        snapshotTestKit.failNextPersisted(persistenceId)
+        // Act: complete compaction, which will trigger a snapshot save.
+        advanceCompaction(raftActor, newEventSourcingIndex = LogEntryIndex(4))
+      }
+
+      // Assert:
+      assertForDuration(
+        {
+          val eventsAfterCompaction = persistenceTestKit.persistedInStorage(persistenceId)
+          assert(eventsAfterCompaction.take(eventsBeforeCompaction.size) === eventsBeforeCompaction)
+        },
+        max = remainingOrDefault,
+      )
+    }
+
     "delete events matching the criteria if it successfully saves a snapshot" in new Fixture {
       val raftActor = createFollower(
         configFor(
@@ -416,6 +464,54 @@ final class RaftActorPersistenceDeletionSpec
       )
     }
 
+    "not delete snapshots if it fails a snapshot save" in new Fixture {
+      val raftActor = createFollower(
+        configFor(
+          deleteBeforeRelativeSequenceNr = 0,
+          deleteOldSnapshots = true,
+        ),
+      )
+
+      // Arrange: complete the first compaction, which will trigger a snapshot save.
+      appendNewEntriesTo(
+        raftActor,
+        currentTerm = Term(1),
+        prevLogTerm = Term(1),
+        LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "event-1"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "event-2"), Term(1)),
+      )
+      advanceCompaction(raftActor, newEventSourcingIndex = LogEntryIndex(3))
+      val snapshot1 = snapshotTestKit.expectNextPersistedType[PersistentStateData.PersistentState](persistenceId)
+
+      // Arrange: append new entries, which will trigger the second compaction.
+      appendNewEntriesTo(
+        raftActor,
+        currentTerm = Term(1),
+        prevLogTerm = Term(1),
+        LogEntry(LogEntryIndex(4), EntityEvent(Option(entityId), "event-3"), Term(1)),
+        LogEntry(LogEntryIndex(5), EntityEvent(Option(entityId), "event-4"), Term(1)),
+        LogEntry(LogEntryIndex(6), EntityEvent(Option(entityId), "event-5"), Term(1)),
+      )
+
+      promoteBeforeDelete(this, raftActor)
+
+      LoggingTestKit.warn("Failed to saveSnapshot").expect {
+        // Arrange: the next snapshot save will fail.
+        snapshotTestKit.failNextPersisted(persistenceId)
+        // Act: complete the second compaction, which will trigger a snapshot save and deletion.
+        advanceCompaction(raftActor, newEventSourcingIndex = LogEntryIndex(6))
+      }
+
+      // Assert:
+      assertForDuration(
+        {
+          val snapshotsAfterCompaction = snapshotTestKit.persistedInStorage(persistenceId)
+          assert(snapshotsAfterCompaction.map(_._2) === Seq(snapshot1))
+        },
+        max = remainingOrDefault,
+      )
+    }
+
     "delete snapshots matching the criteria if it successfully saves a snapshot" in new Fixture {
       val raftActor = createFollower(
         configFor(

core/src/test/scala/lerna/akka/entityreplication/raft/eventsourced/CommitLogStoreActorSpec.scala

@@ -663,6 +663,36 @@ final class CommitLogStoreActorSpec
       )
     }
 
+    "not delete events if it fails a snapshot save" in new Fixture {
+      val commitLogStoreActor = spawnCommitLogStoreActor(
+        configFor(
+          snapshotEvery = 3,
+          deleteOldEvents = true,
+          deleteBeforeRelativeSequenceNr = 1,
+        ),
+      )
+
+      LoggingTestKit.warn("Failed to saveSnapshot").expect {
+        // Arrange: the next snapshot save will fail.
+        snapshotTestKit.failNextPersisted(persistenceId)
+        // Act: save events, which will trigger a snapshot save.
+        appendCommittedEntriesTo(
+          commitLogStoreActor,
+          LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "event-1"), Term(1)),
+          LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "event-2"), Term(1)),
+          LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "event-3"), Term(1)),
+        )
+      }
+
+      // Assert:
+      assertForDuration(
+        {
+          assert(persistenceTestKit.persistedInStorage(persistenceId) === Seq("event-1", "event-2", "event-3"))
+        },
+        max = remainingOrDefault,
+      )
+    }
+
     "delete events matching the criteria if it successfully saves a snapshot" in new Fixture {
       val commitLogStoreActor = spawnCommitLogStoreActor(
         configFor(
@@ -779,6 +809,50 @@ final class CommitLogStoreActorSpec
       )
     }
 
+    "not delete snapshots if it fails a snapshot save" in new Fixture {
+      val commitLogStoreActor = spawnCommitLogStoreActor(
+        configFor(
+          snapshotEvery = 2,
+          deleteOldSnapshots = true,
+          deleteBeforeRelativeSequenceNr = 2,
+        ),
+      )
+
+      // Prepare: save snapshots.
+      appendCommittedEntriesTo(
+        commitLogStoreActor,
+        LogEntry(LogEntryIndex(1), EntityEvent(Option(entityId), "event-1"), Term(1)),
+        LogEntry(LogEntryIndex(2), EntityEvent(Option(entityId), "event-2"), Term(1)),
+        LogEntry(LogEntryIndex(3), EntityEvent(Option(entityId), "event-3"), Term(1)),
+        LogEntry(LogEntryIndex(4), EntityEvent(Option(entityId), "event-4"), Term(1)),
+      )
+      snapshotTestKit.expectNextPersisted(persistenceId, CommitLogStoreActor.State(LogEntryIndex(2)))
+      snapshotTestKit.expectNextPersisted(persistenceId, CommitLogStoreActor.State(LogEntryIndex(4)))
+
+      LoggingTestKit.warn("Failed to saveSnapshot").expect {
+        // Arrange: the next snapshot save will fail.
+        snapshotTestKit.failNextPersisted(persistenceId)
+        // Act: save events, which will trigger a snapshot save.
+        appendCommittedEntriesTo(
+          commitLogStoreActor,
+          LogEntry(LogEntryIndex(5), EntityEvent(Option(entityId), "event-5"), Term(1)),
+          LogEntry(LogEntryIndex(6), EntityEvent(Option(entityId), "event-6"), Term(1)),
+        )
+      }
+
+      // Assert:
+      assertForDuration(
+        {
+          val allSnapshots = Seq(
+            CommitLogStoreActor.State(LogEntryIndex(2)),
+            CommitLogStoreActor.State(LogEntryIndex(4)),
+          )
+          assert(snapshotTestKit.persistedInStorage(persistenceId).map(_._2) === allSnapshots)
+        },
+        max = remainingOrDefault,
+      )
+    }
+
     "delete snapshots matching the criteria if it successfully saves a snapshot" in new Fixture {
       val commitLogStoreActor = spawnCommitLogStoreActor(
         configFor(

core/src/test/scala/lerna/akka/entityreplication/raft/snapshot/SnapshotStorePersistenceDeletionSpec.scala

@@ -11,6 +11,7 @@ import lerna.akka.entityreplication.raft.model.LogEntryIndex
 import lerna.akka.entityreplication.raft.{ ActorSpec, RaftSettings }
 import lerna.akka.entityreplication.raft.routing.MemberIndex
 import org.scalactic.TypeCheckedTripleEquals
+import org.scalatest.BeforeAndAfterAll
 
 import java.util.UUID
 
@@ -19,6 +20,7 @@ final class SnapshotStorePersistenceDeletionSpec
       ActorSystem("SnapshotStorePersistenceDeletionSpec", ShardSnapshotStoreSpecBase.configWithPersistenceTestKits),
     )
     with ActorSpec
+    with BeforeAndAfterAll
     with TypeCheckedTripleEquals {
 
   import lerna.akka.entityreplication.raft.snapshot.SnapshotProtocol._
@@ -35,6 +37,11 @@ final class SnapshotStorePersistenceDeletionSpec
     snapshotTestKit.resetPolicy()
   }
 
+  override def afterAll(): Unit = {
+    try shutdown(system)
+    finally super.afterAll()
+  }
+
   private def configFor(
       snapshotEvery: Int,
       deleteBeforeRelativeSequenceNr: Long,
@@ -113,6 +120,38 @@ final class SnapshotStorePersistenceDeletionSpec
     )
   }
 
+  "not delete events if it fails a snapshot save" in new Fixture {
+    val snapshotStore = spawnSnapshotStore(
+      configFor(
+        snapshotEvery = 2,
+        deleteBeforeRelativeSequenceNr = 1,
+        deleteOldEvents = true,
+      ),
+    )
+
+    // Arrange: save the first event.
+    val entityState1 = createEntitySnapshotData(LogEntryIndex(1))
+    saveEntitySnapshot(snapshotStore, entityState1)
+    persistenceTestKit.expectNextPersisted(persistenceId, entityState1)
+
+    val entityState2 = createEntitySnapshotData(LogEntryIndex(2))
+    LoggingTestKit.warn("Failed to saveSnapshot").expect {
+      // Arrange: the next snapshot save will fail.
+      snapshotTestKit.failNextPersisted(persistenceId)
+      // Act: save the second event, which will trigger a snapshot save.
+      saveEntitySnapshot(snapshotStore, entityState2)
+      persistenceTestKit.expectNextPersisted(persistenceId, entityState2)
+    }
+
+    // Assert:
+    assertForDuration(
+      {
+        assert(persistenceTestKit.persistedInStorage(persistenceId) === Seq(entityState1, entityState2))
+      },
+      max = remainingOrDefault,
+    )
+  }
+
   "delete events matching the criteria if it successfully saves a snapshot" in new Fixture {
     val snapshotStore = spawnSnapshotStore(
       configFor(
@@ -227,6 +266,43 @@ final class SnapshotStorePersistenceDeletionSpec
     )
   }
 
+  "not delete snapshots if it fails a snapshot save" in new Fixture {
+    val snapshotStore = spawnSnapshotStore(
+      configFor(
+        snapshotEvery = 1,
+        deleteBeforeRelativeSequenceNr = 1,
+        deleteOldSnapshots = true,
+      ),
+    )
+
+    // Arrange: save the first snapshot.
+    val entityState1 = createEntitySnapshotData(LogEntryIndex(1))
+    saveEntitySnapshot(snapshotStore, entityState1)
+    snapshotTestKit.expectNextPersisted(persistenceId, entityState1)
+
+    // Arrange: save the second snapshot.
+    val entityState2 = createEntitySnapshotData(LogEntryIndex(2))
+    saveEntitySnapshot(snapshotStore, entityState2)
+    snapshotTestKit.expectNextPersisted(persistenceId, entityState2)
+
+    LoggingTestKit.warn("Failed to saveSnapshot").expect {
+      // Arrange: the next snapshot save will fail.
+      snapshotTestKit.failNextPersisted(persistenceId)
+      // Act: save the third snapshot, which will trigger a snapshot save.
+      val entityState3 = createEntitySnapshotData(LogEntryIndex(3))
+      saveEntitySnapshot(snapshotStore, entityState3)
+    }
+
+    // Assert:
+    assertForDuration(
+      {
+        val allSnapshots = Seq(entityState1, entityState2)
+        assert(snapshotTestKit.persistedInStorage(persistenceId).map(_._2) === allSnapshots)
+      },
+      max = remainingOrDefault,
+    )
+  }
+
   "delete snapshots matching the criteria if it successfully saves a snapshot" in new Fixture {
     val snapshotStore = spawnSnapshotStore(
       configFor(