diff --git a/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala b/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala index 5388c353519..64b683204d6 100644 --- a/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala +++ b/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala @@ -247,6 +247,56 @@ abstract class JournalSpec(config: Config) journal ! ReplayMessages(0, Long.MaxValue, Long.MaxValue, pid, receiverProbe.ref) receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) } + + "replay all surviving messages when the lower bound equals the last deleted sequence number" in { + val deleteProbe = TestProbe() + journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref) + deleteProbe.expectMsg(DeleteMessagesSuccess(3L)) + + journal ! ReplayMessages(3L, Long.MaxValue, Long.MaxValue, pid, receiverProbe.ref) + receiverProbe.expectMsg(replayedMessage(4)) + receiverProbe.expectMsg(replayedMessage(5)) + receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) + } + + "replay surviving messages within bounds when the replay window spans a deleted prefix" in { + val deleteProbe = TestProbe() + journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref) + deleteProbe.expectMsg(DeleteMessagesSuccess(3L)) + + journal ! ReplayMessages(1, 4, Long.MaxValue, pid, receiverProbe.ref) + receiverProbe.expectMsg(replayedMessage(4)) + receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) + } + + "return only recovery success when the upper bound falls within a deleted prefix" in { + val deleteProbe = TestProbe() + journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref) + deleteProbe.expectMsg(DeleteMessagesSuccess(3L)) + + journal ! ReplayMessages(1, 2, Long.MaxValue, pid, receiverProbe.ref) + receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) + } + + "return only recovery success when the upper bound equals the last deleted sequence number" in { + val deleteProbe = TestProbe() + journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref) + deleteProbe.expectMsg(DeleteMessagesSuccess(3L)) + + journal ! ReplayMessages(1, 3, Long.MaxValue, pid, receiverProbe.ref) + receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) + } + + "replay from the first surviving message when the lower bound equals the first surviving sequence number" in { + val deleteProbe = TestProbe() + journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref) + deleteProbe.expectMsg(DeleteMessagesSuccess(3L)) + + journal ! ReplayMessages(4, Long.MaxValue, Long.MaxValue, pid, receiverProbe.ref) + receiverProbe.expectMsg(replayedMessage(4)) + receiverProbe.expectMsg(replayedMessage(5)) + receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L)) + } } "A Journal optionally".may {