feat: allow fallback store to be eagerly initialized (#170)

Author: leviramsey

core/src/main/resources/reference.conf

@@ -79,6 +79,9 @@ akka.persistence.dynamodb {
       # Additionally, when reading events, at most this many events (for one query) will have in-flight
       # retrievals from the fallback store
       batch-size = 16
+
+      # Whether to start the fallback store at startup (eager = on) or defer until needed (eager = off)
+      eager = off
     }
   }
 }
@@ -129,6 +132,9 @@ akka.persistence.dynamodb {
       # above), then the snapshot will be written to the fallback store and a "breadcrumb"
       # describing how to retrieve the snapshot from the fallback store will be written.
       threshold = 300 KiB
+
+      # Whether to start the fallback store at startup (eager = on) or defer until needed (eager = off)
+      eager = off
     }
   }
 }

core/src/main/scala/akka/persistence/dynamodb/DynamoDBSettings.scala

@@ -339,7 +339,7 @@ final class ClockSkewSettings(config: Config) {
 
 /** Not for user extension */
 @DoNotInherit
-sealed abstract class FallbackSettings(val plugin: String, val threshold: Int) {
+sealed abstract class FallbackSettings(val plugin: String, val threshold: Int, val eager: Boolean) {
   require(threshold > 0, "threshold must be positive")
   require(threshold <= (400 * 1000), "threshold must be at most 400 KB")
 
@@ -350,7 +350,8 @@ sealed abstract class FallbackSettings(val plugin: String, val threshold: Int) {
 }
 
 @ApiMayChange
-final class SnapshotFallbackSettings(plugin: String, threshold: Int) extends FallbackSettings(plugin, threshold) {
+final class SnapshotFallbackSettings(plugin: String, threshold: Int, eager: Boolean)
+    extends FallbackSettings(plugin, threshold, eager) {
   override def toString: String =
     if (isEnabled)
       s"SnapshotFallbackSettings(plugin=$plugin, threshold=${threshold}B)"
@@ -361,14 +362,15 @@ object SnapshotFallbackSettings {
   def apply(config: Config): SnapshotFallbackSettings = {
     val plugin = config.getString("plugin")
     val threshold = java.lang.Long.min(config.getBytes("threshold"), 400 * 1000).toInt
+    val eager = config.getBoolean("eager")
 
-    new SnapshotFallbackSettings(plugin, threshold)
+    new SnapshotFallbackSettings(plugin, threshold, eager)
   }
 }
 
 @ApiMayChange
 final class JournalFallbackSettings(commonSettings: SnapshotFallbackSettings, val batchSize: Int)
-    extends FallbackSettings(commonSettings.plugin, commonSettings.threshold) {
+    extends FallbackSettings(commonSettings.plugin, commonSettings.threshold, commonSettings.eager) {
   require(!commonSettings.isEnabled || batchSize > 0, "batch size must be positive")
 
   override def toString: String =

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

@@ -159,6 +159,7 @@ class FallbackStoreProvider(system: ActorSystem[_]) extends Extension {
             .toList)
         .map { _ => Done }(ExecutionContext.parasitic)
     }
+
   def eventFallbackStoreFor(configLocation: String): EventFallbackStore[AnyRef] = {
     val storePair =
       fallbackStores.computeIfAbsent(configLocation, configLocation => pair(constructFallbackStore(configLocation)))
@@ -227,6 +228,15 @@ class FallbackStoreProvider(system: ActorSystem[_]) extends Extension {
     }
   }
 
+  /** INTERNAL API */
+  @InternalApi
+  final private[akka] def getFallbackStore(id: String): Option[CommonFallbackStore[AnyRef]] =
+    fallbackStores.get(id) match {
+      case null | (null, null) => None
+      case (null, snap)        => Some(snap)
+      case (evt, _)            => Some(evt)
+    }
+
   private val fallbackStores =
     new ConcurrentHashMap[String, (EventFallbackStore[AnyRef], SnapshotFallbackStore[AnyRef])]
 }

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

@@ -550,4 +550,8 @@ import software.amazon.awssdk.services.dynamodb.model.Update
         Future.failed(c.getCause)
       }(ExecutionContext.parasitic)
   }
+
+  if (settings.journalFallbackSettings.isEnabled && settings.journalFallbackSettings.eager) {
+    fallbackStoreProvider.eventFallbackStoreFor(settings.journalFallbackSettings.plugin)
+  }
 }

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

@@ -563,4 +563,8 @@ import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest
     (conditions.result().mkString(" AND "), attributes)
   }
 
+  if (settings.snapshotFallbackSettings.isEnabled && settings.snapshotFallbackSettings.eager) {
+    // constructs and saves in a concurrent hashmap for later use
+    fallbackStoreProvider.snapshotFallbackStoreFor(settings.snapshotFallbackSettings.plugin)
+  }
 }

core/src/test/scala/akka/persistence/dynamodb/cleanup/EventSourcedCleanupSpec.scala

@@ -102,7 +102,7 @@ class EventSourcedCleanupSpec
           val from = (iteration * batchSize) + 1
           iteration = iteration + 1
           val to = Math.min(maxSeqNumber, from + batchSize - 1)
-          val expectedMsg = s"Deleted events from [$from] to [$to] for persistenceId [$pid], consumed [8.0] WCU"
+          val expectedMsg = s"Deleted events from [$from] to [$to] for persistenceId [$pid], consumed [4.0] WCU"
           event.message == expectedMsg
         }
         .expect {
@@ -168,7 +168,7 @@ class EventSourcedCleanupSpec
           val from = (iteration * batchSize) + 1
           iteration = iteration + 1
           val to = Math.min(maxSeqNumber, from + batchSize - 1)
-          val expectedMsg = s"Deleted events from [$from] to [$to] for persistenceId [$pid], consumed [8.0] WCU"
+          val expectedMsg = s"Deleted events from [$from] to [$to] for persistenceId [$pid], consumed [4.0] WCU"
           event.message == expectedMsg
         }
         .expect {
@@ -547,7 +547,7 @@ class EventSourcedCleanupSpec
           val to = Math.min(x, from + batchSize - 1)
           val expectedMessage =
             s"Updated expiry of events for persistenceId [$pid], for sequence numbers [$from] to [$to]," +
-            s" expiring at [$expiryTimestamp], consumed [8.0] WCU"
+            s" expiring at [$expiryTimestamp], consumed [4.0] WCU"
           event.message == expectedMessage
         }
         .expect {
@@ -686,7 +686,7 @@ class EventSourcedCleanupSpec
           val to = Math.min(n, from + batchSize - 1)
           val expectedMessage =
             s"Updated expiry of events for persistenceId [$pid], for sequence numbers [$from] to [$to]," +
-            s" expiring at [$expiryTimestamp], consumed [8.0] WCU"
+            s" expiring at [$expiryTimestamp], consumed [4.0] WCU"
           event.message == expectedMessage
         }
         .expect {
@@ -828,7 +828,7 @@ class EventSourcedCleanupSpec
           val to = Math.min(n, from + batchSize - 1)
           val expectedMessage =
             s"Updated expiry of events for persistenceId [$pid], for sequence numbers [$from] to [$to]," +
-            s" expiring at [$expiryTimestamp], consumed [8.0] WCU"
+            s" expiring at [$expiryTimestamp], consumed [4.0] WCU"
           event.message == expectedMessage
         }
         .expect {

core/src/test/scala/akka/persistence/dynamodb/journal/DynamoDBJournalSpec.scala

@@ -109,6 +109,11 @@ object DynamoDBJournalSpec {
       }
       """)
       .withFallback(TestConfig.config)
+
+  def configWithEagerFallback: Config =
+    ConfigFactory
+      .parseString("akka.persistence.dynamodb.journal.fallback-store.eager = on")
+      .withFallback(configWithSmallFallback)
 }
 
 abstract class DynamoDBJournalBaseSpec(config: Config)
@@ -298,6 +303,11 @@ class DynamoDBJournalWithFallbackSpec
 
   "A DynamoDB journal with fallback enabled" should {
 
+    "not eagerly initialize the fallback store when eager = off" in new Setup {
+      FallbackStoreProvider(system).getFallbackStore(
+        system.settings.config.getString("akka.persistence.dynamodb.journal.fallback-store.plugin")) shouldBe empty
+    }
+
     "not interact with the fallback store for small events" in withFallbackStoreProbe { (fallbackStore, invocations) =>
       new Setup {
         val probe = TypedTestProbe[Any]()
@@ -412,3 +422,29 @@ class DynamoDBJournalWithFallbackSpec
     invocationsProbe.cancel()
   }
 }
+
+class DynamoDBJournalWithEagerFallbackSpec
+    extends ScalaTestWithActorTestKit(DynamoDBJournalSpec.configWithEagerFallback)
+    with AnyWordSpecLike
+    with TestData
+    with TestDbLifecycle
+    with LogCapturing {
+
+  def typedSystem: ActorSystem[_] = testKit.system
+
+  class Setup {
+    val journal = persistenceExt.journalFor("akka.persistence.dynamodb.journal")
+  }
+
+  "A DynamoDB journal with fallback enabled" should {
+
+    "eagerly initialize the fallback store when eager = on" in new Setup {
+      val cfg = system.settings.config.getConfig("akka.persistence.dynamodb.journal.fallback-store")
+      val plugin = cfg.getString("plugin")
+
+      eventually {
+        FallbackStoreProvider(system).getFallbackStore(plugin) shouldNot be(empty)
+      }
+    }
+  }
+}

core/src/test/scala/akka/persistence/dynamodb/snapshot/DynamoDBSnapshotStoreSpec.scala

@@ -91,6 +91,11 @@ object DynamoDBSnapshotStoreSpec {
       }
       """)
       .withFallback(config)
+
+  def configWithEagerFallback: Config =
+    ConfigFactory
+      .parseString("akka.persistence.dynamodb.snapshot.fallback-store.eager = on")
+      .withFallback(configWithSmallFallback)
 }
 
 abstract class DynamoDBSnapshotStoreBaseSpec(config: Config)
@@ -212,6 +217,11 @@ class DynamoDBSnapshotFallbackSpec
   }
 
   "A DynamoDB snapshot store with fallback enabled" should {
+    "not eagerly initialize the fallback store when eager = off" in new Setup {
+      FallbackStoreProvider(system).getFallbackStore(
+        system.settings.config.getString("akka.persistence.dynamodb.journal.fallback-store.plugin")) shouldBe empty
+    }
+
     "not interact with the fallback store for small snapshots" in withFallbackStoreProbe {
       (fallbackStore, invocations) =>
         new Setup {
@@ -319,6 +329,31 @@ class DynamoDBSnapshotFallbackSpec
     }
   }
 
+  class DynamoDBSnapshotWithEagerFallbackSpec
+      extends ScalaTestWithActorTestKit(DynamoDBSnapshotStoreSpec.configWithEagerFallback)
+      with AnyWordSpecLike
+      with TestData
+      with TestDbLifecycle
+      with LogCapturing {
+
+    def typedSystem: ActorSystem[_] = testKit.system
+
+    class Setup {
+      val snapshotStore = persistenceExt.snapshotStoreFor("akka.persistence.dynamodb.snapshot")
+    }
+
+    "A DynamoDB snapshot store with fallback enabled" should {
+      "eagerly initialize the fallback store when eager = on" in new Setup {
+        val cfg = system.settings.config.getConfig("akka.persistence.dynamodb.snapshot.fallback-store")
+        val plugin = cfg.getString("plugin")
+
+        eventually {
+          FallbackStoreProvider(system).getFallbackStore(plugin) shouldNot be(empty)
+        }
+      }
+    }
+  }
+
   private def withFallbackStoreProbe(
       test: (InMemFallbackStore, TestSubscriber.Probe[InMemFallbackStore.Invocation]) => Unit): Unit = {