fix: backtracking when fallback is enabled

Author: leviramsey

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

@@ -28,6 +28,7 @@ import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue
 import software.amazon.awssdk.services.dynamodb.model.QueryRequest
 import software.amazon.awssdk.services.dynamodb.model.QueryResponse
+import scala.concurrent.Promise
 
 /**
  * INTERNAL API
@@ -42,6 +43,38 @@ import software.amazon.awssdk.services.dynamodb.model.QueryResponse
   private val serialization = SerializationExtension(system)
   private val fallbackStoreProvider = FallbackStoreProvider(system)
 
+  private val _backtrackingBreadcrumbSerId = Promise[Int]()
+  def backtrackingBreadcrumbSerId(): Int = {
+    _backtrackingBreadcrumbSerId.future.value match {
+      case Some(v) => v.get
+      case None =>
+        val serializerIds = serialization.bindings.iterator.map(_._2.identifier).toSet
+        // Use an unfolding iterator to find a serializer ID that's not bound.  There shouldn't ever be more
+        // than a few thousand serializer IDs bound (and the docs suggest real serializers should only have
+        // positive serializer IDs ("couple of billion")), so this shouldn't have to iterate far.
+        //
+        // Ranges that aren't indexable by an Int (viz. have more than Int.MaxValue elements) are surprisingly
+        // restrictive, thus the unfold
+        Iterator
+          .unfold(Int.MinValue) { s =>
+            if (s != Int.MaxValue) Some(s -> (s + 1))
+            else None
+          }
+          // stop fast if some other thread found this before we do
+          .find { i => !serializerIds(i) || _backtrackingBreadcrumbSerId.isCompleted } match {
+          case None =>
+            // Over 4 billion serializers... really?
+            _backtrackingBreadcrumbSerId.tryFailure(new NoSuchElementException("All serializer IDs used?"))
+
+          case Some(id) =>
+            _backtrackingBreadcrumbSerId.trySuccess(id)
+        }
+        // first get is safe, we just ensured completion
+        // second get will throw if we exhausted serializers, but that's a danger we're prepared to face...
+        _backtrackingBreadcrumbSerId.future.value.get.get
+    }
+  }
+
   private val bySliceProjectionExpression = {
     import JournalAttributes._
     s"$Pid, $SeqNr, $Timestamp, $EventSerId, $EventSerManifest, $Tags, $BreadcrumbSerId, $BreadcrumbSerManifest"
@@ -182,6 +215,13 @@ import software.amazon.awssdk.services.dynamodb.model.QueryResponse
   }
 
   // implements BySliceQuery.Dao
+  // NB: if backtracking and an event that was saved to the fallback store is encountered,
+  // the payload will be None (as with any backtracking event) and the serId of the returned
+  // item will be one not used by any bound serializer.
+  //
+  // Without a payload, the serializer ID is kind of meaningless (and the events-by-slice
+  // queries in the read journal will ignore the serializer ID unless it is the filtered
+  // payload serializer).
   override def itemsBySlice(
       entityType: String,
       slice: Int,
@@ -277,7 +317,8 @@ import software.amazon.awssdk.services.dynamodb.model.QueryResponse
               writeTimestamp = getTimestamp(item),
               readTimestamp = InstantFactory.now(),
               payload = None, // lazy loaded for backtracking
-              serId = item.get(EventSerId).n().toInt,
+              serId =
+                if (item.containsKey(EventSerId)) item.get(EventSerId).n().toInt else backtrackingBreadcrumbSerId(),
               serManifest = "",
               writerUuid = "", // not need in this query
               tags = if (item.containsKey(Tags)) item.get(Tags).ss().asScala.toSet else Set.empty,

s3-fallback-store/src/test/resources/application-test.conf

@@ -20,3 +20,7 @@ akka.persistence.dynamodb.journal.fallback-store {
 }
 
 akka.persistence.dynamodb.snapshot.fallback-store = ${akka.persistence.dynamodb.journal.fallback-store}
+
+akka.persistence.dynamodb.query.refresh-interval = 500ms
+akka.persistence.dynamodb.query.backtracking.window = 500ms
+akka.persistence.dynamodb.query.backtracking.behind-current-time = 1s

s3-fallback-store/src/test/scala/akka/persistence/s3fallback/LargeSnapshotAndEventSpec.scala

@@ -10,6 +10,7 @@ import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
 import akka.actor.typed.ActorSystem
 import akka.persistence.dynamodb.TestActors
 import akka.persistence.dynamodb.TestData
+import akka.persistence.dynamodb.internal.EnvelopeOrigin
 import akka.persistence.dynamodb.internal.JournalAttributes
 import akka.persistence.dynamodb.internal.SnapshotAttributes
 import akka.persistence.dynamodb.query.scaladsl.DynamoDBReadJournal
@@ -199,6 +200,27 @@ class LargeSnapshotAndEventSpec
       cebsEvent.getEvent() shouldBe padding
 
       cebsProbe.cancel()
+
+      // We want to force backtracking to see the first event, so persist another
+      //reincarnation ! PersistWithAck("ignore-me", probe.ref)
+      //probe.expectMessage(Done)
+
+      val liveEventsBySlicesSubscriber = query
+        .eventsBySlices[String](entityType, slice, slice, TimestampOffset.Zero)
+        .filter(_.persistenceId == persistenceId.id)
+        .runWith(sinkProbe)
+
+      val lebsProbe = liveEventsBySlicesSubscriber.request(2)
+      val forwardLebsEvent = lebsProbe.expectNext()
+
+      forwardLebsEvent.event shouldBe padding
+
+      val backtrackLebsEvent = lebsProbe.expectNext()
+
+      backtrackLebsEvent.eventOption shouldBe empty
+      backtrackLebsEvent.source shouldBe EnvelopeOrigin.SourceBacktracking
+
+      lebsProbe.cancel()
     }
   }