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

* feat: Recovery based on only the last event

* bump: Akka core 2.10.1

* EventEnvelope constructor

* sql server limit 1

Author: patriknw

core/src/main/scala/akka/persistence/r2dbc/internal/QueryDao.scala

@@ -26,6 +26,8 @@ private[r2dbc] trait QueryDao extends BySliceQuery.Dao[SerializedJournalRow] {
   def timestampOfEvent(persistenceId: String, seqNr: Long): Future[Option[Instant]]
   def loadEvent(persistenceId: String, seqNr: Long, includePayload: Boolean): Future[Option[SerializedJournalRow]]
 
+  def loadLastEvent(persistenceId: String, toSeqNr: Long, includeDeleted: Boolean): Future[Option[SerializedJournalRow]]
+
   def eventsByPersistenceId(
       persistenceId: String,
       fromSequenceNr: Long,

core/src/main/scala/akka/persistence/r2dbc/internal/postgres/PostgresQueryDao.scala

@@ -12,6 +12,7 @@ import scala.concurrent.Future
 import scala.concurrent.duration.Duration
 import scala.concurrent.duration.FiniteDuration
 
+import io.r2dbc.spi.Row
 import io.r2dbc.spi.Statement
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -137,23 +138,44 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
   protected def selectOneEventSql(slice: Int): String =
     sqlCache.get(slice, "selectOneEventSql") {
       sql"""
-      SELECT slice, entity_type, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, meta_ser_id, meta_ser_manifest, meta_payload, tags
+      SELECT entity_type, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, meta_ser_id, meta_ser_manifest, meta_payload, tags
       FROM ${journalTable(slice)}
       WHERE persistence_id = ? AND seq_nr = ? AND deleted = $sqlFalse"""
     }
 
   protected def selectOneEventWithoutPayloadSql(slice: Int): String =
     sqlCache.get(slice, "selectOneEventWithoutPayloadSql") {
       sql"""
-      SELECT slice, entity_type, db_timestamp, CURRENT_TIMESTAMP AS read_db_timestamp, event_ser_id, event_ser_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
+      SELECT entity_type, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
       FROM ${journalTable(slice)}
       WHERE persistence_id = ? AND seq_nr = ? AND deleted = $sqlFalse"""
     }
 
+  protected def selectLastEventSql(slice: Int): String =
+    sqlCache.get(slice, "selectLastEventSql") {
+      sql"""
+      SELECT entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
+      FROM ${journalTable(slice)}
+      WHERE persistence_id = ? AND seq_nr <= ? AND deleted = $sqlFalse
+      ORDER BY seq_nr DESC
+      LIMIT 1"""
+    }
+
+  protected def selectLastEventIncludeDeletedSql(slice: Int): String =
+    sqlCache.get(slice, "selectLastEventIncludeDeletedSql") {
+      sql"""
+      SELECT entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags, deleted
+      FROM ${journalTable(slice)}
+      WHERE persistence_id = ? AND seq_nr <= ?
+      ORDER BY seq_nr DESC
+      LIMIT 1
+      """
+    }
+
   protected def selectEventsSql(slice: Int): String =
     sqlCache.get(slice, "selectEventsSql") {
       sql"""
-      SELECT slice, entity_type, seq_nr, db_timestamp, CURRENT_TIMESTAMP AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
+      SELECT entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
       from ${journalTable(slice)}
       WHERE persistence_id = ? AND seq_nr >= ? AND seq_nr <= ?
       AND deleted = false
@@ -164,7 +186,7 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
   protected def selectEventsIncludeDeletedSql(slice: Int): String =
     sqlCache.get(slice, "selectEventsIncludeDeletedSql") {
       sql"""
-      SELECT slice, entity_type, seq_nr, db_timestamp, CURRENT_TIMESTAMP AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags, deleted
+      SELECT entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags, deleted
       from ${journalTable(slice)}
       WHERE persistence_id = ? AND seq_nr >= ? AND seq_nr <= ?
       ORDER BY seq_nr
@@ -393,7 +415,7 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
             Some(row.getPayload("event_payload"))
           else None
         SerializedJournalRow(
-          slice = row.get[Integer]("slice", classOf[Integer]),
+          slice = slice,
           entityType = row.get("entity_type", classOf[String]),
           persistenceId,
           seqNr,
@@ -408,6 +430,42 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
       })
   }
 
+  override def loadLastEvent(
+      persistenceId: String,
+      toSeqNr: Long,
+      includeDeleted: Boolean): Future[Option[SerializedJournalRow]] = {
+    val slice = persistenceExt.sliceForPersistenceId(persistenceId)
+    val executor = executorProvider.executorFor(slice)
+    val selectSql = if (includeDeleted) selectLastEventIncludeDeletedSql(slice) else selectLastEventSql(slice)
+    executor.selectOne("select last event")(
+      connection => {
+        connection
+          .createStatement(selectSql)
+          .bind(0, persistenceId)
+          .bind(1, toSeqNr)
+      },
+      row => {
+        if (includeDeleted && row.get[java.lang.Boolean]("deleted", classOf[java.lang.Boolean])) {
+          // deleted row
+          deletedJournalRow(slice, persistenceId, row)
+        } else {
+          SerializedJournalRow(
+            slice = slice,
+            entityType = row.get("entity_type", classOf[String]),
+            persistenceId,
+            seqNr = row.get[java.lang.Long]("seq_nr", classOf[java.lang.Long]),
+            dbTimestamp = row.getTimestamp("db_timestamp"),
+            readDbTimestamp = row.getTimestamp("read_db_timestamp"),
+            payload = Some(row.getPayload("event_payload")),
+            serId = row.get[Integer]("event_ser_id", classOf[Integer]),
+            serManifest = row.get("event_ser_manifest", classOf[String]),
+            writerUuid = row.get("writer", classOf[String]),
+            tags = row.getTags("tags"),
+            metadata = readMetadata(row))
+        }
+      })
+  }
+
   override def eventsByPersistenceId(
       persistenceId: String,
       fromSequenceNr: Long,
@@ -424,22 +482,10 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
       row =>
         if (includeDeleted && row.get[java.lang.Boolean]("deleted", classOf[java.lang.Boolean])) {
           // deleted row
-          SerializedJournalRow(
-            slice = row.get[Integer]("slice", classOf[Integer]),
-            entityType = row.get("entity_type", classOf[String]),
-            persistenceId = persistenceId,
-            seqNr = row.get[java.lang.Long]("seq_nr", classOf[java.lang.Long]),
-            dbTimestamp = row.getTimestamp("db_timestamp"),
-            readDbTimestamp = row.getTimestamp("read_db_timestamp"),
-            payload = None,
-            serId = 0,
-            serManifest = "",
-            writerUuid = "",
-            tags = Set.empty,
-            metadata = None)
+          deletedJournalRow(slice, persistenceId, row)
         } else {
           SerializedJournalRow(
-            slice = row.get[Integer]("slice", classOf[Integer]),
+            slice = slice,
             entityType = row.get("entity_type", classOf[String]),
             persistenceId = persistenceId,
             seqNr = row.get[java.lang.Long]("seq_nr", classOf[java.lang.Long]),
@@ -459,6 +505,22 @@ private[r2dbc] class PostgresQueryDao(executorProvider: R2dbcExecutorProvider) e
     Source.futureSource(result.map(Source(_))).mapMaterializedValue(_ => NotUsed)
   }
 
+  private def deletedJournalRow(slice: Int, persistenceId: String, row: Row): SerializedJournalRow = {
+    SerializedJournalRow(
+      slice = slice,
+      entityType = row.get("entity_type", classOf[String]),
+      persistenceId = persistenceId,
+      seqNr = row.get[java.lang.Long]("seq_nr", classOf[java.lang.Long]),
+      dbTimestamp = row.getTimestamp("db_timestamp"),
+      readDbTimestamp = row.getTimestamp("read_db_timestamp"),
+      payload = None,
+      serId = 0,
+      serManifest = "",
+      writerUuid = "",
+      tags = Set.empty,
+      metadata = None)
+  }
+
   protected def bindPersistenceIdsForEntityTypeAfterSql(
       stmt: Statement,
       entityType: String,

core/src/main/scala/akka/persistence/r2dbc/internal/sqlserver/SqlServerQueryDao.scala

@@ -163,6 +163,26 @@ private[r2dbc] class SqlServerQueryDao(executorProvider: R2dbcExecutorProvider)
     stmt
   }
 
+  override protected def selectLastEventSql(slice: Int): String =
+    sqlCache.get(slice, "selectLastEventSql") {
+      sql"""
+      SELECT TOP(1) entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags
+      FROM ${journalTable(slice)}
+      WHERE persistence_id = ? AND seq_nr <= ? AND deleted = $sqlFalse
+      ORDER BY seq_nr DESC
+      """
+    }
+
+  override protected def selectLastEventIncludeDeletedSql(slice: Int): String =
+    sqlCache.get(slice, "selectLastEventIncludeDeletedSql") {
+      sql"""
+      SELECT TOP(1) entity_type, seq_nr, db_timestamp, $sqlDbTimestamp AS read_db_timestamp, event_ser_id, event_ser_manifest, event_payload, writer, adapter_manifest, meta_ser_id, meta_ser_manifest, meta_payload, tags, deleted
+      FROM ${journalTable(slice)}
+      WHERE persistence_id = ? AND seq_nr <= ?
+      ORDER BY seq_nr DESC
+      """
+    }
+
   override protected def persistenceIdsForEntityTypeAfterSql(minSlice: Int): String =
     sqlCache.get(minSlice, "persistenceIdsForEntityTypeAfterSql") {
       sql"""

core/src/main/scala/akka/persistence/r2dbc/journal/R2dbcJournal.scala

@@ -220,7 +220,11 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
 
   override def replayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(
       recoveryCallback: PersistentRepr => Unit): Future[Long] = {
-    log.debug("replayMessages [{}] [{}]", persistenceId, fromSequenceNr)
+    if (fromSequenceNr == -1)
+      log.debug("replayMessages [{}] [{}]", persistenceId, "last")
+    else
+      log.debug("replayMessages [{}] [{}]", persistenceId, fromSequenceNr)
+
     val pendingWrite = Option(writesInProgress.get(persistenceId)) match {
       case Some(f) =>
         log.debug("Write in progress for [{}], deferring replayMessages until write completed", persistenceId)
@@ -229,7 +233,23 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
       case None => FutureDone
     }
     pendingWrite.flatMap { _ =>
-      if (toSequenceNr == Long.MaxValue && max == Long.MaxValue) {
+      if (toSequenceNr <= 0 || max == 0) {
+        // no replay
+        journalDao.readHighestSequenceNr(persistenceId, fromSequenceNr)
+      } else if (fromSequenceNr == -1) {
+        // recover from last event only
+        query.internalLastEventByPersistenceId(persistenceId, toSequenceNr, includeDeleted = true).map {
+          case Some(item) =>
+            // payload is empty for deleted item
+            if (item.payload.isDefined) {
+              val repr = deserializeRow(serialization, item)
+              recoveryCallback(repr)
+            }
+            item.seqNr
+          case None =>
+            0L
+        }
+      } else if (toSequenceNr == Long.MaxValue && max == Long.MaxValue) {
         // this is the normal case, highest sequence number from last event
         query
           .internalCurrentEventsByPersistenceId(
@@ -246,9 +266,6 @@ private[r2dbc] final class R2dbcJournal(config: Config, cfgPath: String) extends
             }
             item.seqNr
           })
-      } else if (toSequenceNr <= 0) {
-        // no replay
-        journalDao.readHighestSequenceNr(persistenceId, fromSequenceNr)
       } else {
         // replay to custom sequence number
 

core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala

@@ -194,7 +194,7 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
       1L,
       eventOption = None,
       timestamp.toEpochMilli,
-      eventMetadata = None,
+      _eventMetadata = None,
       entityType,
       slice,
       filtered = true,
@@ -725,6 +725,16 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
       .mapMaterializedValue(_ => NotUsed)
   }
 
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[r2dbc] def internalLastEventByPersistenceId(
+      persistenceId: String,
+      toSequenceNr: Long,
+      includeDeleted: Boolean): Future[Option[SerializedJournalRow]] = {
+    queryDao.loadLastEvent(persistenceId, toSequenceNr, includeDeleted)
+  }
+
   // EventTimestampQuery
   override def timestampOf(persistenceId: String, sequenceNr: Long): Future[Option[Instant]] = {
     val result = queryDao.timestampOfEvent(persistenceId, sequenceNr)

core/src/main/scala/akka/persistence/r2dbc/snapshot/R2dbcSnapshotStore.scala

@@ -38,6 +38,8 @@ object R2dbcSnapshotStore {
             .deserialize(serializedMeta.payload, serializedMeta.serializerId, serializedMeta.serializerManifest)
             .get)),
       serialization.deserialize(snap.snapshot, snap.serializerId, snap.serializerManifest).get)
+
+  private val FutureNone = Future.successful(None)
 }
 
 /**
@@ -49,6 +51,7 @@ object R2dbcSnapshotStore {
 @InternalApi
 private[r2dbc] final class R2dbcSnapshotStore(cfg: Config, cfgPath: String) extends SnapshotStore {
   import R2dbcSnapshotStore.deserializeSnapshotRow
+  import R2dbcSnapshotStore.FutureNone
 
   private implicit val ec: ExecutionContext = context.dispatcher
   private val serialization: Serialization = SerializationExtension(context.system)
@@ -69,10 +72,14 @@ private[r2dbc] final class R2dbcSnapshotStore(cfg: Config, cfgPath: String) exte
   private val dao = settings.connectionFactorySettings.dialect.createSnapshotDao(executorProvider)
   private val queryDao = settings.connectionFactorySettings.dialect.createQueryDao(executorProvider)
 
-  def loadAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] =
-    dao
-      .load(persistenceId, criteria)
-      .map(_.map(row => deserializeSnapshotRow(row, serialization)))
+  def loadAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] = {
+    if (criteria.maxSequenceNr <= 0)
+      FutureNone
+    else
+      dao
+        .load(persistenceId, criteria)
+        .map(_.map(row => deserializeSnapshotRow(row, serialization)))
+  }
 
   def saveAsync(metadata: SnapshotMetadata, snapshot: Any): Future[Unit] = {
     val entityType = PersistenceId.extractEntityType(metadata.persistenceId)

core/src/test/scala/akka/persistence/r2dbc/TestActors.scala

@@ -14,6 +14,7 @@ import akka.persistence.typed.ReplicaId
 import akka.persistence.typed.ReplicationId
 import akka.persistence.typed.SnapshotCompleted
 import akka.persistence.typed.scaladsl.EventSourcedBehavior
+import akka.persistence.typed.scaladsl.Recovery
 import akka.persistence.typed.scaladsl.ReplicatedEventSourcing
 import akka.persistence.typed.state.scaladsl.DurableStateBehavior
 
@@ -82,6 +83,16 @@ object TestActors {
       }
     }
 
+    def withRecovery(pid: PersistenceId, recovery: Recovery): Behavior[Command] = {
+      Behaviors.setup { context =>
+        eventSourcedBehavior(pid, context)
+          .snapshotWhen { case (_, event, _) =>
+            event.toString.contains("snap")
+          }
+          .withRecovery(recovery)
+      }
+    }
+
     def eventSourcedBehavior(
         pid: PersistenceId,
         context: ActorContext[Command]): EventSourcedBehavior[Command, Any, String] = {
@@ -198,8 +209,11 @@ object TestActors {
         ReplicationId(entityType, entityId, ReplicaId("dc-1")),
         Set(ReplicaId("dc-1")),
         R2dbcReadJournal.Identifier) { replicationContext =>
-        Persister.eventSourcedBehavior(PersistenceId(entityType, entityId), context)
+        Persister.eventSourcedBehavior(replicationContext.persistenceId, context)
       }
     }
   }
+
+  def replicatedEventSourcedPersistenceId(entityType: String, entityId: String): PersistenceId =
+    ReplicationId(entityType, entityId, ReplicaId("dc-1")).persistenceId
 }

core/src/test/scala/akka/persistence/r2dbc/journal/R2dbcJournalSpec.scala

@@ -33,12 +33,14 @@ object R2dbcJournalSpec {
 }
 
 class R2dbcJournalSpec extends JournalSpec(R2dbcJournalSpec.config) with TestDbLifecycle {
-  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.off()
+  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = false
+  override protected def supportsReplayOnlyLast: CapabilityFlag = true
   override def typedSystem: ActorSystem[_] = system.toTyped
 }
 
 class R2dbcJournalWithMetaSpec extends JournalSpec(R2dbcJournalSpec.configWithMeta) with TestDbLifecycle {
-  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = CapabilityFlag.off()
-  protected override def supportsMetadata: CapabilityFlag = CapabilityFlag.on()
+  override protected def supportsRejectingNonSerializableObjects: CapabilityFlag = false
+  protected override def supportsMetadata: CapabilityFlag = true
+  override protected def supportsReplayOnlyLast: CapabilityFlag = true
   override def typedSystem: ActorSystem[_] = system.toTyped
 }

core/src/test/scala/akka/persistence/r2dbc/query/EventsByPersistenceIdSpec.scala

@@ -155,14 +155,15 @@ class EventsByPersistenceIdSpec
         val probe = testKit.createTestProbe[Done]()
         val entityType = nextEntityType()
         val entityId = "entity-1"
+        val pid = TestActors.replicatedEventSourcedPersistenceId(entityType, entityId).id
 
         val persister = testKit.spawn(TestActors.replicatedEventSourcedPersister(entityType, entityId))
         persister ! Persister.PersistWithAck("e-1", probe.ref)
         probe.expectMessage(Done)
         persister ! Persister.PersistWithAck("e-2", probe.ref)
         probe.expectMessage(Done)
 
-        val sub = doQuery(PersistenceId(entityType, entityId).id, 0, Long.MaxValue)
+        val sub = doQuery(pid, 0, Long.MaxValue)
           .runWith(TestSink())
           .request(10)
 

core/src/test/scala/akka/persistence/r2dbc/query/EventsBySliceSpec.scala

@@ -242,7 +242,7 @@ class EventsBySliceSpec
         val probe = testKit.createTestProbe[Done]()
         val entityType = nextEntityType()
         val entityId = "entity-1"
-        val persistenceId = PersistenceId(entityType, entityId)
+        val persistenceId = TestActors.replicatedEventSourcedPersistenceId(entityType, entityId)
         val slice = query.sliceForPersistenceId(persistenceId.id)
 
         val persister = testKit.spawn(TestActors.replicatedEventSourcedPersister(entityType, entityId))