feat: Recovery based on only the last event (#32629)

* Introduced in a compatible way so that it still works if used with
  a plugin that doesn't implement this. In that case the snapshot will
  still not be loaded but all events will be replayed instead of only
  the last.

Author: patriknw

akka-docs/src/main/paradox/persistence-plugins.md

@@ -20,6 +20,8 @@ AWS DynamoDB can be used as backend for Akka Persistence with the [Akka Persiste
 
 @ref:[Durable State](./typed/index-persistence-durable-state.md) is not supported by the DynamoDB plugin.
 
+Recovery from only last event is not supported by the DynamoDB plugin.
+
 ## JDBC plugin
 
 Relational databases with JDBC-drivers are supported through [Akka Persistence JDBC](https://doc.akka.io/libraries/akka-persistence-jdbc/current/). For new projects, the @ref:[R2DBC plugin](#r2dbc-plugin) is recommended.
@@ -38,6 +40,7 @@ Example of concrete features _not_ supported by the Cassandra and JDBC plugins:
 * Projections starting from snapshots
 * Scalability of many Projections
 * Durable State entities (partly supported by JDBC plugin)
+* Recovery from only last event
 
 ## Enabling a plugin
 

akka-docs/src/main/paradox/typed/persistence.md

@@ -546,6 +546,35 @@ Please refer to @ref[snapshots](persistence-snapshot.md#snapshots) if you need t
 
 In any case, the highest sequence number will always be recovered so you can keep persisting new events without corrupting your event log.
 
+@@@ warning
+
+Disable of recovery is not normal behavior of an event sourced actor, since events and snapshots are not used for
+the recovery of the actor.
+
+@@@
+
+### Recovery from only last event
+
+For some use cases it is enough to recover the actor from the last event, as an optimization to not replay all events.
+You can enable this recovery mode with:
+
+Scala
+:  @@snip [BasicPersistentBehaviorCompileOnly.scala](/akka-persistence-typed/src/test/scala/docs/akka/persistence/typed/BasicPersistentBehaviorCompileOnly.scala) { #replay-last }
+
+Java
+:  @@snip [BasicPersistentBehaviorTest.java](/akka-persistence-typed/src/test/java/jdocs/akka/persistence/typed/BasicPersistentBehaviorTest.java) { #replay-last }
+
+Snapshots are not loaded when recovery from last event is selected.
+
+@@@ warning
+
+Recovery from only last event is not normal behavior of an event sourced actor, since it typically would need
+all events, or a snapshot and events after the snapshot, to recover its state.
+
+This feature is currently only supported by the R2DBC plugin.
+
+@@@
+
 ## Tagging
 
 Persistence allows you to use event tags without using an @ref[`EventAdapter`](../persistence.md#event-adapters):

akka-persistence-tck/src/main/scala/akka/persistence/CapabilityFlags.scala

@@ -58,6 +58,11 @@ trait JournalCapabilityFlags extends CapabilityFlags {
    */
   protected def supportsMetadata: CapabilityFlag
 
+  /**
+   * When `true` enables tests which check if the Journal can replay only the last event.
+   */
+  protected def supportsReplayOnlyLast: CapabilityFlag
+
 }
 //#journal-flags
 

akka-persistence-tck/src/main/scala/akka/persistence/journal/JournalSpec.scala

@@ -55,6 +55,8 @@ abstract class JournalSpec(config: Config)
 
   override protected def supportsMetadata: CapabilityFlag = false
 
+  override protected def supportsReplayOnlyLast: CapabilityFlag = false
+
   override protected def beforeEach(): Unit = {
     super.beforeEach()
     senderProbe = TestProbe()
@@ -188,6 +190,12 @@ abstract class JournalSpec(config: Config)
       journal ! ReplayMessages(0, Long.MaxValue, Long.MaxValue, "non-existing-pid", receiverProbe.ref)
       receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 0L))
     }
+    "retrieve highest sequenceNr without replaying events" in {
+      journal ! ReplayMessages(0, 0, 1, pid, receiverProbe.ref)
+      // no replayedMessage
+      receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+    }
+
     "not replay permanently deleted messages (range deletion)" in {
       val receiverProbe2 = TestProbe()
       val cmd = DeleteMessagesTo(pid, 3, receiverProbe2.ref)
@@ -236,6 +244,7 @@ abstract class JournalSpec(config: Config)
       journal ! ReplayMessages(0, Long.MaxValue, Long.MaxValue, pid, receiverProbe.ref)
       receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
     }
+
   }
 
   "A Journal optionally".may {
@@ -348,5 +357,14 @@ abstract class JournalSpec(config: Config)
 
       }
     }
+
+    optional(flag = supportsReplayOnlyLast) {
+      "replay only last when fromSequenceNr is -1" in {
+        journal ! ReplayMessages(-1, Long.MaxValue, Long.MaxValue, pid, receiverProbe.ref)
+        receiverProbe.expectMsg(replayedMessage(snr = 5))
+        receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+      }
+    }
+
   }
 }

akka-persistence-tck/src/main/scala/akka/persistence/snapshot/SnapshotStoreSpec.scala

@@ -4,8 +4,6 @@
 
 package akka.persistence.snapshot
 
-import scala.collection.immutable.Seq
-
 import com.typesafe.config.Config
 import com.typesafe.config.ConfigFactory
 
@@ -222,5 +220,6 @@ abstract class SnapshotStoreSpec(config: Config)
         }
       }
     }
+
   }
 }

akka-persistence-tck/src/test/scala/akka/persistence/journal/inmem/InmemJournalSpec.scala

@@ -9,5 +9,7 @@ import akka.persistence.PersistenceSpec
 import akka.persistence.journal.JournalSpec
 
 class InmemJournalSpec extends JournalSpec(config = PersistenceSpec.config("inmem", "InmemJournalSpec")) {
-  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.off()
+  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = false
+
+  override protected def supportsReplayOnlyLast: CapabilityFlag = true
 }

akka-persistence-testkit/src/main/scala/akka/persistence/testkit/PersistenceTestKitPlugin.scala

@@ -59,10 +59,15 @@ class PersistenceTestKitPlugin(@nowarn("msg=never used") cfg: Config, cfgPath: S
 
   override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(
       recoveryCallback: PersistentRepr => Unit): Future[Unit] =
-    Future.fromTry(
-      Try(
+    Future.fromTry(Try {
+      val highest = storage.tryReadSeqNumber(persistenceId)
+      if (highest != 0L && max != 0L) {
+        val to = math.min(toSequenceNr, highest)
+        // read only last when fromSequenceNr is -1
+        val from = if (fromSequenceNr == -1) to else fromSequenceNr
+
         storage
-          .tryRead(persistenceId, fromSequenceNr, toSequenceNr, max)
+          .tryRead(persistenceId, from, to, max)
           .map { repr =>
             // we keep the tags in the repr, so remove those here
             repr.payload match {
@@ -71,7 +76,11 @@ class PersistenceTestKitPlugin(@nowarn("msg=never used") cfg: Config, cfgPath: S
             }
 
           }
-          .foreach(recoveryCallback)))
+          .foreach(recoveryCallback)
+      } else {
+        Future.successful(())
+      }
+    })
 
   override def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] =
     Future.fromTry(Try {

akka-persistence-testkit/src/test/scala/akka/persistence/testkit/scaladsl/PersistenceTestKitJournalCompatSpec.scala

@@ -38,6 +38,8 @@ class PersistenceTestKitJournalCompatSpec extends JournalSpec(config = Persisten
 
   override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = true
   override protected def supportsMetadata: CapabilityFlag = true
+  override protected def supportsReplayOnlyLast: CapabilityFlag = true
+
 }
 
 class PersistenceTestKitSnapshotStoreCompatSpec

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/scaladsl/EventSourcedBehaviorReplayLastSpec.scala

@@ -0,0 +1,101 @@
+/*
+ * Copyright (C) 2017-2024 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.persistence.typed.scaladsl
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.annotation.nowarn
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+import akka.actor.testkit.typed.scaladsl._
+import akka.actor.typed.ActorRef
+import akka.actor.typed.scaladsl.ActorContext
+import akka.actor.typed.scaladsl.Behaviors
+import akka.persistence.testkit.PersistenceTestKitPlugin
+import akka.persistence.testkit.PersistenceTestKitSnapshotPlugin
+import akka.persistence.typed.PersistenceId
+import akka.serialization.jackson.CborSerializable
+
+object EventSourcedBehaviorReplayLastSpec extends Matchers {
+
+  sealed trait Command extends CborSerializable
+  final case class Increment(id: String) extends Command
+  final case class GetValue(replyTo: ActorRef[State]) extends Command
+
+  sealed trait Event extends CborSerializable
+  final case class Incremented(id: String) extends Event
+
+  final case class State(value: Int, history: Vector[String]) extends CborSerializable
+
+  def counter(
+      @nowarn("msg=never used") ctx: ActorContext[Command],
+      persistenceId: PersistenceId,
+      probe: Option[ActorRef[(State, Event)]] = None): EventSourcedBehavior[Command, Event, State] = {
+    EventSourcedBehavior[Command, Event, State](
+      persistenceId,
+      emptyState = State(0, Vector.empty),
+      commandHandler = (state, cmd) =>
+        cmd match {
+          case Increment(id) =>
+            Effect.persist(Incremented(id))
+
+          case GetValue(replyTo) =>
+            replyTo ! state
+            Effect.none
+
+        },
+      eventHandler = (state, evt) =>
+        evt match {
+          case Incremented(id) =>
+            probe.foreach(_ ! ((state, evt)))
+            State(state.value + 1, state.history :+ id)
+        })
+  }.withRecovery(Recovery.replayOnlyLast)
+
+}
+
+class EventSourcedBehaviorReplayLastSpec
+    extends ScalaTestWithActorTestKit(
+      PersistenceTestKitPlugin.config.withFallback(PersistenceTestKitSnapshotPlugin.config))
+    with AnyWordSpecLike
+    with LogCapturing {
+
+  import EventSourcedBehaviorReplayLastSpec._
+
+  val pidCounter = new AtomicInteger(0)
+  private def nextPid(): PersistenceId = PersistenceId.ofUniqueId(s"c${pidCounter.incrementAndGet()}")
+
+  "EventSourcedBehavior with Recovery.replayOnlyLast" must {
+    "recover from last event only" in {
+      val pid = nextPid()
+      val c = spawn(Behaviors.setup[Command](ctx => counter(ctx, pid)))
+      val replyProbe = TestProbe[State]()
+
+      c ! Increment("a")
+      c ! Increment("b")
+      c ! Increment("c")
+      c ! GetValue(replyProbe.ref)
+      replyProbe.expectMessage(State(3, Vector("a", "b", "c")))
+      testKit.stop(c)
+      replyProbe.expectTerminated(c)
+
+      val probe = TestProbe[(State, Event)]()
+      val c2 = spawn(Behaviors.setup[Command](ctx => counter(ctx, pid, Some(probe.ref))))
+      // replayed the last event
+      probe.expectMessage((State(0, Vector.empty), Incremented("c")))
+      probe.expectNoMessage()
+
+      c2 ! GetValue(replyProbe.ref)
+      replyProbe.expectMessage(State(1, Vector("c")))
+
+      c2 ! Increment("d")
+      c2 ! GetValue(replyProbe.ref)
+      replyProbe.expectMessage(State(2, Vector("c", "d")))
+    }
+
+  }
+}

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/scaladsl/EventSourcedBehaviorWatchSpec.scala

@@ -17,7 +17,6 @@ import akka.actor.testkit.typed.scaladsl.TestProbe
 import akka.actor.typed._
 import akka.actor.typed.scaladsl.ActorContext
 import akka.actor.typed.scaladsl.Behaviors
-import akka.persistence.{ Recovery => ClassicRecovery }
 import akka.persistence.typed.NoOpEventAdapter
 import akka.persistence.typed.PersistenceId
 import akka.persistence.typed.RecoveryCompleted
@@ -69,7 +68,7 @@ class EventSourcedBehaviorWatchSpec
       NoOpEventAdapter.instance[String],
       NoOpSnapshotAdapter.instance[String],
       snapshotWhen = SnapshotWhenPredicate.noSnapshot,
-      ClassicRecovery(),
+      Recovery.default,
       RetentionCriteria.disabled,
       holdingRecoveryPermit = false,
       settings = settings,