feat: fallback to S3 for large snapshots and events (#155)

Author: leviramsey

.editorconfig

@@ -0,0 +1,10 @@
+root = true
+
+[*]
+charset = utf-8
+insert_final_newline = true
+
+# No need to enforce an editor config for Java/Scala files, as the build automatically styles
+[*.conf]
+indent_style = space
+indent_size = 2

.github/workflows/build-test.yml

@@ -79,7 +79,7 @@ jobs:
 
       - name: Start DB
         run: |-
-          docker compose -f docker/docker-compose.yml up --wait
+          docker compose -f docker/docker-compose-with-s3.yml up --wait
 
       - name: sbt test
         run: |-

CONTRIBUTING.md

@@ -10,6 +10,12 @@ It can be started with the docker-compose file in the docker folder:
 docker-compose -f docker/docker-compose.yml up --wait
 ```
 
+Or if running tests of the S3 fallback functionality (`s3Fallback / test` in sbt)
+
+```
+docker-compose -f docker/docker-compose-with-s3.yml up --wait
+```
+
 ## General Workflow
 
 This is the process for committing code into main.

build.sbt

@@ -115,7 +115,7 @@ lazy val root = (project in file("."))
     publishTo := Some(Resolver.file("Unused transient repository", file("target/unusedrepo"))))
   .enablePlugins(ScalaUnidocPlugin)
   .disablePlugins(SitePlugin, MimaPlugin, CiReleasePlugin)
-  .aggregate(core, docs)
+  .aggregate(core, s3Fallback, docs)
 
 def suffixFileFilter(suffix: String): FileFilter = new SimpleFileFilter(f => f.getAbsolutePath.endsWith(suffix))
 
@@ -125,6 +125,14 @@ lazy val core = (project in file("core"))
   .enablePlugins(AutomateHeaderPlugin)
   .disablePlugins(CiReleasePlugin)
 
+lazy val s3Fallback = (project in file("s3-fallback-store"))
+  .settings(common)
+  .settings(name := "akka-persistence-dynamodb-s3-fallback-store")
+  .settings(libraryDependencies ++= Dependencies.s3Fallback)
+  .dependsOn(core % "compile->compile;test->test")
+  .enablePlugins(AutomateHeaderPlugin)
+  .disablePlugins(CiReleasePlugin)
+
 lazy val docs = project
   .in(file("docs"))
   .enablePlugins(AkkaParadoxPlugin, ParadoxSitePlugin, PreprocessPlugin, PublishRsyncPlugin)

core/src/main/resources/reference.conf

@@ -52,6 +52,34 @@ akka.persistence.dynamodb {
     # replay filter not needed for this plugin
     replay-filter.mode = off
 
+    # Configure the fallback store which will be used to store a batch of events when
+    # the write's size exceeds the threshold herein
+    fallback-store {
+      # If enabling the fallback store, set to the config path of the fallback store plugin to use
+      # 
+      # The referred-to config path must at least contain a value named 'class' set to the FQCN of
+      # the fallback store plugin.  For example if `plugin = "fallback-store"`, your config will need
+      #
+      #  fallback-store {
+      #    class = "package.containing.my.FallbackStore"
+      #
+      #    # Additional fallback store specific configuration would go here
+      #  }
+      #
+      # If not set, no fallback store will be used
+      plugin = ""
+
+      # The journal will estimate the size of the write (including attributes): if the estimate exceeds this
+      # value and the fallback store is enabled (by setting `plugin`, above), then the events
+      # will be written to the fallback store and a "breadcrumb" describing how to
+      # retrieve the event from the fallback store will be written.
+      threshold = 300 KiB
+
+      # Write at most this many events from one batch (e.g. `persistAll`) concurrently to the fallback store.
+      # Additionally, when reading events, at most this many events (for one query) will have in-flight
+      # retrievals from the fallback store
+      batch-size = 16
+    }
   }
 }
 // #journal-settings
@@ -73,7 +101,35 @@ akka.persistence.dynamodb {
     by-slice-idx = ""
 
     # Enables an optimization in Akka for avoiding snapshot deletes in retention.
+    #
+    # Note that the data model for this plugin can only store one snapshot per
+    # persistence ID: disabling this will not allow multiple snapshots per
+    # persistence ID
     only-one-snapshot = true
+
+    # Configure the fallback store which will be used to store a snapshot when the
+    # write's size exceeds the threshold herein
+    fallback-store {
+      # If enabling the fallback store, set to the config path of the fallback store plugin to use
+      #
+      # The referred-to config path must at least contain a value named 'class' set to the FQCN of
+      # the fallback store plugin.  For example if `plugin = "fallback-store"`, your config will need
+      #
+      #  fallback-store {
+      #    class = "package.containing.my.FallbackStore"
+      #
+      #    # Additional fallback store specific configuration would go here
+      #  }
+      #
+      # If not set, no fallback store will be used
+      plugin = ""
+
+      # The snapshot store will estimate the size of the write (including attributes): if
+      # the estimate exceeds this value and the fallback store is enabled (by setting `plugin`
+      # 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
+    }
   }
 }
 // #snapshot-settings

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

@@ -10,6 +10,8 @@ import scala.jdk.CollectionConverters._
 import scala.jdk.DurationConverters._
 
 import akka.actor.typed.ActorSystem
+import akka.annotation.ApiMayChange
+import akka.annotation.DoNotInherit
 import akka.annotation.InternalStableApi
 import akka.util.Helpers
 import com.typesafe.config.Config
@@ -61,6 +63,10 @@ object DynamoDBSettings {
 
     val clockSkewSettings = new ClockSkewSettings(config)
 
+    val journalFallbackSettings = JournalFallbackSettings(config.getConfig("journal.fallback-store"))
+
+    val snapshotFallbackSettings = SnapshotFallbackSettings(config.getConfig("snapshot.fallback-store"))
+
     new DynamoDBSettings(
       journalTable,
       journalPublishEvents,
@@ -70,7 +76,9 @@ object DynamoDBSettings {
       timeToLiveSettings,
       journalBySliceGsi,
       snapshotBySliceGsi,
-      clockSkewSettings)
+      clockSkewSettings,
+      journalFallbackSettings,
+      snapshotFallbackSettings)
   }
 
   /**
@@ -93,7 +101,16 @@ final class DynamoDBSettings private (
     val timeToLiveSettings: TimeToLiveSettings,
     val journalBySliceGsi: String,
     val snapshotBySliceGsi: String,
-    val clockSkewSettings: ClockSkewSettings)
+    val clockSkewSettings: ClockSkewSettings,
+    val journalFallbackSettings: JournalFallbackSettings,
+    val snapshotFallbackSettings: SnapshotFallbackSettings) {
+  override def toString: String =
+    s"DynamoDBSettings(journalTable=$journalTable, journalPublishEvents=$journalPublishEvents, " +
+    s"snapshotTable=$snapshotTable, querySettings=$querySettings, cleanupSettings=$cleanupSettings, " +
+    s"timeToLiveSettings=$timeToLiveSettings, journalBySliceGsi=$journalBySliceGsi, " +
+    s"snapshotBySliceGsi=$snapshotBySliceGsi, clockSkewSettings=$clockSkewSettings, " +
+    s"journalFallbackSettings=$journalFallbackSettings, snapshotFallbackSettings=$snapshotFallbackSettings)"
+}
 
 final class QuerySettings(config: Config) {
   val refreshInterval: FiniteDuration = config.getDuration("refresh-interval").toScala
@@ -320,6 +337,55 @@ final class ClockSkewSettings(config: Config) {
   override def toString: String = s"ClockSkewSettings($warningTolerance)"
 }
 
+/** Not for user extension */
+@DoNotInherit
+sealed abstract class FallbackSettings(val plugin: String, val threshold: Int) {
+  require(threshold > 0, "threshold must be positive")
+  require(threshold <= (400 * 1000), "threshold must be at most 400 KB")
+
+  // Must be overridden in subclasses
+  override def toString: String = throw new scala.NotImplementedError
+
+  def isEnabled: Boolean = plugin.nonEmpty
+}
+
+@ApiMayChange
+final class SnapshotFallbackSettings(plugin: String, threshold: Int) extends FallbackSettings(plugin, threshold) {
+  override def toString: String =
+    if (isEnabled)
+      s"SnapshotFallbackSettings(plugin=$plugin, threshold=${threshold}B)"
+    else "disabled"
+}
+
+object SnapshotFallbackSettings {
+  def apply(config: Config): SnapshotFallbackSettings = {
+    val plugin = config.getString("plugin")
+    val threshold = java.lang.Long.min(config.getBytes("threshold"), 400 * 1000).toInt
+
+    new SnapshotFallbackSettings(plugin, threshold)
+  }
+}
+
+@ApiMayChange
+final class JournalFallbackSettings(commonSettings: SnapshotFallbackSettings, val batchSize: Int)
+    extends FallbackSettings(commonSettings.plugin, commonSettings.threshold) {
+  require(!commonSettings.isEnabled || batchSize > 0, "batch size must be positive")
+
+  override def toString: String =
+    if (isEnabled)
+      s"JournalFallbackSettings(plugin=$plugin, threshold=${threshold}B, batchSize=$batchSize)"
+    else "disabled"
+}
+
+object JournalFallbackSettings {
+  def apply(config: Config): JournalFallbackSettings = {
+    val commonSettings = SnapshotFallbackSettings(config)
+    val batchSize = config.getInt("batch-size")
+
+    new JournalFallbackSettings(commonSettings, batchSize)
+  }
+}
+
 private[akka] object ConfigHelpers {
   def optString(config: Config, path: String): Option[String] = {
     if (config.hasPath(path)) {