This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-persistence-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 891b1fc0 Honor snapshot selection lower bounds #591 (#618)
891b1fc0 is described below
commit 891b1fc0e67b41c58072f359adca4a25fef8ab0d
Author: Goutam Adwant <[email protected]>
AuthorDate: Wed Sep 23 12:12:50 2026 -0700
Honor snapshot selection lower bounds #591 (#618)
* Honor snapshot selection lower bounds #591
* Apply inclusive sequence and timestamp bounds in current and legacy DAO
queries.
* Preserve existing custom DAO methods through concrete criteria defaults;
fail unsupported bounded calls explicitly rather than widening deletion.
* Add snapshot load/delete regressions and document the custom DAO
selection contract.
* Preserve custom snapshot DAO fallback behavior #591
Motivation:
Failing new criteria defaults on minimum bounds breaks existing custom
DAO recovery and bounded retention requests.
Modification:
Delegate default criteria calls to the existing upper-bound methods.
Document the legacy approximation and add custom DAO routing, deletion,
typed retention and bounded recovery regressions.
Result:
Existing custom DAOs retain their behavior while built-in DAOs continue
to enforce all four inclusive bounds.
Tests:
Custom DAO regressions and current/legacy H2 snapshot tests passed.
Custom DAO suites: 33 tests passed on Scala 2.13, 3.3 and 3.9 each.
Core: 387 passed, two pending, one previously observed events-by-tag
timeout, which also occurred on the isolated rerun.
Style, documentation and normal Scala 2.13 MiMa checks passed.
References:
Refs #591
---
.../jdbc/snapshot/JdbcSnapshotStore.scala | 29 +-----
.../jdbc/snapshot/dao/DefaultSnapshotDao.scala | 16 ++-
.../jdbc/snapshot/dao/SnapshotDao.scala | 69 ++++++++++++-
.../jdbc/snapshot/dao/SnapshotQueries.scala | 20 ++++
.../snapshot/dao/legacy/ByteArraySnapshotDao.scala | 22 ++++-
.../jdbc/snapshot/dao/legacy/SnapshotQueries.scala | 20 ++++
.../snapshot/CustomSnapshotDaoRecoverySpec.scala | 107 +++++++++++++++++++++
.../jdbc/snapshot/JdbcSnapshotStoreSpec.scala | 76 ++++++++++++++-
.../jdbc/snapshot/dao/SnapshotDaoSpec.scala | 93 ++++++++++++++++++
docs/src/main/paradox/custom-dao.md | 25 ++++-
10 files changed, 444 insertions(+), 33 deletions(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStore.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStore.scala
index fbe40cf3..0487da3d 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStore.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStore.scala
@@ -63,21 +63,8 @@ class JdbcSnapshotStore(config: Config) extends
SnapshotStore {
override def loadAsync(
persistenceId: String,
- criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] =
{
- val result = criteria match {
- case SnapshotSelectionCriteria(Long.MaxValue, Long.MaxValue, _, _) =>
- snapshotDao.latestSnapshot(persistenceId)
- case SnapshotSelectionCriteria(Long.MaxValue, maxTimestamp, _, _) =>
- snapshotDao.snapshotForMaxTimestamp(persistenceId, maxTimestamp)
- case SnapshotSelectionCriteria(maxSequenceNr, Long.MaxValue, _, _) =>
- snapshotDao.snapshotForMaxSequenceNr(persistenceId, maxSequenceNr)
- case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
- snapshotDao.snapshotForMaxSequenceNrAndMaxTimestamp(persistenceId,
maxSequenceNr, maxTimestamp)
- case _ => Future.successful(None)
- }
-
- result.map(_.map(toSelectedSnapshot))
- }
+ criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] =
+ snapshotDao.snapshotForCriteria(persistenceId,
criteria).map(_.map(toSelectedSnapshot))
override def saveAsync(metadata: SnapshotMetadata, snapshot: Any):
Future[Unit] =
snapshotDao.save(metadata, snapshot)
@@ -88,17 +75,7 @@ class JdbcSnapshotStore(config: Config) extends
SnapshotStore {
} yield ()
override def deleteAsync(persistenceId: String, criteria:
SnapshotSelectionCriteria): Future[Unit] =
- criteria match {
- case SnapshotSelectionCriteria(Long.MaxValue, Long.MaxValue, _, _) =>
- snapshotDao.deleteAllSnapshots(persistenceId)
- case SnapshotSelectionCriteria(Long.MaxValue, maxTimestamp, _, _) =>
- snapshotDao.deleteUpToMaxTimestamp(persistenceId, maxTimestamp)
- case SnapshotSelectionCriteria(maxSequenceNr, Long.MaxValue, _, _) =>
- snapshotDao.deleteUpToMaxSequenceNr(persistenceId, maxSequenceNr)
- case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
- snapshotDao.deleteUpToMaxSequenceNrAndMaxTimestamp(persistenceId,
maxSequenceNr, maxTimestamp)
- case _ => Future.successful(())
- }
+ snapshotDao.deleteByCriteria(persistenceId, criteria)
override def postStop(): Unit = {
if (slickDb.allowShutdown) {
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/DefaultSnapshotDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/DefaultSnapshotDao.scala
index 41f14272..bdc4ee24 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/DefaultSnapshotDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/DefaultSnapshotDao.scala
@@ -16,7 +16,7 @@ package org.apache.pekko.persistence.jdbc.snapshot.dao
import slick.jdbc.{ JdbcBackend, JdbcProfile }
import org.apache.pekko
-import pekko.persistence.SnapshotMetadata
+import pekko.persistence.{ SnapshotMetadata, SnapshotSelectionCriteria }
import pekko.persistence.jdbc.config.SnapshotConfig
import pekko.serialization.Serialization
import pekko.stream.Materializer
@@ -75,6 +75,20 @@ class DefaultSnapshotDao(
private def zeroOrOneSnapshot(rows: Seq[SnapshotRow]):
Option[(SnapshotMetadata, Any)] =
rows.headOption.map(row => toSnapshotData(row).get) // throw is from a
future map
+ override def snapshotForCriteria(
+ persistenceId: String,
+ criteria: SnapshotSelectionCriteria): Future[Option[(SnapshotMetadata,
Any)]] =
+ db.run(queries.selectOneByCriteria(
+ (persistenceId, criteria.maxSequenceNr, criteria.maxTimestamp,
criteria.minSequenceNr,
+ criteria.minTimestamp)).result)
+ .map(zeroOrOneSnapshot)
+
+ override def deleteByCriteria(persistenceId: String, criteria:
SnapshotSelectionCriteria): Future[Unit] =
+ db.run(queries.selectByCriteria(
+ (persistenceId, criteria.maxSequenceNr, criteria.maxTimestamp,
criteria.minSequenceNr,
+ criteria.minTimestamp)).delete)
+ .map(_ => ())(ExecutionContext.parasitic)
+
override def latestSnapshot(persistenceId: String):
Future[Option[(SnapshotMetadata, Any)]] =
db.run(queries.selectLatestByPersistenceId(persistenceId).result).flatMap
{ rows =>
rows.headOption match {
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDao.scala
index ab57813c..9ce5f764 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDao.scala
@@ -14,11 +14,76 @@
package org.apache.pekko.persistence.jdbc.snapshot.dao
-import org.apache.pekko.persistence.SnapshotMetadata
+import java.util.concurrent.ConcurrentHashMap
-import scala.concurrent.Future
+import org.apache.pekko.persistence.{ SnapshotMetadata,
SnapshotSelectionCriteria }
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.{ ExecutionContext, Future }
+
+private object SnapshotDaoWarnings {
+ private val warnedDaoClasses = ConcurrentHashMap.newKeySet[String]()
+
+ def ignoredMinimumBounds(dao: SnapshotDao, criteria:
SnapshotSelectionCriteria): Unit = {
+ val daoClass = dao.getClass.getName
+ if ((criteria.minSequenceNr != 0L || criteria.minTimestamp != 0L) &&
warnedDaoClasses.add(daoClass))
+ LoggerFactory.getLogger(dao.getClass).warn(
+ "Snapshot DAO [{}] uses the compatibility deleteByCriteria
implementation; minimum bounds are ignored. " +
+ "Override deleteByCriteria to apply all criteria bounds.",
+ daoClass)
+ }
+}
trait SnapshotDao {
+
+ /**
+ * Load the snapshot with the highest sequence number matching all inclusive
criteria bounds.
+ * The default delegates to the existing upper-bound methods for custom DAO
compatibility,
+ * then filters the returned snapshot against all bounds. Custom DAOs should
override this
+ * method to query all four bounds directly, as the built-in DAOs do;
otherwise a snapshot
+ * matching the criteria may be missed when timestamps are not ordered by
sequence number.
+ * Subclasses of a built-in DAO that override upper-bound loading methods
are bypassed by
+ * the built-in implementation and must also override this method to apply
their behavior.
+ */
+ def snapshotForCriteria(
+ persistenceId: String,
+ criteria: SnapshotSelectionCriteria): Future[Option[(SnapshotMetadata,
Any)]] = {
+ val snapshot = criteria match {
+ case SnapshotSelectionCriteria(Long.MaxValue, Long.MaxValue, _, _) =>
+ latestSnapshot(persistenceId)
+ case SnapshotSelectionCriteria(Long.MaxValue, maxTimestamp, _, _) =>
+ snapshotForMaxTimestamp(persistenceId, maxTimestamp)
+ case SnapshotSelectionCriteria(maxSequenceNr, Long.MaxValue, _, _) =>
+ snapshotForMaxSequenceNr(persistenceId, maxSequenceNr)
+ case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
+ snapshotForMaxSequenceNrAndMaxTimestamp(persistenceId, maxSequenceNr,
maxTimestamp)
+ }
+ snapshot.map(_.filter { case (metadata, _) => criteria.matches(metadata)
})(ExecutionContext.parasitic)
+ }
+
+ /**
+ * Delete only snapshots matching all inclusive criteria bounds for this
persistence ID.
+ * The default delegates to the existing upper-bound methods for custom DAO
compatibility,
+ * ignoring minimum bounds. Custom DAOs must override this method to apply
all four bounds,
+ * as the built-in DAOs do; otherwise snapshots below a minimum bound may
also be deleted.
+ * Subclasses of a built-in DAO that override upper-bound deletion methods
are bypassed by
+ * the built-in implementation and must also override this method to apply
their behavior.
+ */
+ def deleteByCriteria(persistenceId: String, criteria:
SnapshotSelectionCriteria): Future[Unit] = {
+ SnapshotDaoWarnings.ignoredMinimumBounds(this, criteria)
+
+ criteria match {
+ case SnapshotSelectionCriteria(Long.MaxValue, Long.MaxValue, _, _) =>
+ deleteAllSnapshots(persistenceId)
+ case SnapshotSelectionCriteria(Long.MaxValue, maxTimestamp, _, _) =>
+ deleteUpToMaxTimestamp(persistenceId, maxTimestamp)
+ case SnapshotSelectionCriteria(maxSequenceNr, Long.MaxValue, _, _) =>
+ deleteUpToMaxSequenceNr(persistenceId, maxSequenceNr)
+ case SnapshotSelectionCriteria(maxSequenceNr, maxTimestamp, _, _) =>
+ deleteUpToMaxSequenceNrAndMaxTimestamp(persistenceId, maxSequenceNr,
maxTimestamp)
+ }
+ }
+
def deleteAllSnapshots(persistenceId: String): Future[Unit]
def deleteUpToMaxSequenceNr(persistenceId: String, maxSequenceNr: Long):
Future[Unit]
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotQueries.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotQueries.scala
index f5399e1d..b43ed058 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotQueries.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotQueries.scala
@@ -36,6 +36,26 @@ class SnapshotQueries(val profile: JdbcProfile, override val
snapshotTableCfg: S
_selectAll(persistenceId).take(1)
val selectLatestByPersistenceId = Compiled(_selectLatestByPersistenceId _)
+ private def _selectByCriteria(
+ persistenceId: Rep[String],
+ maxSequenceNr: Rep[Long],
+ maxTimestamp: Rep[Long],
+ minSequenceNr: Rep[Long],
+ minTimestamp: Rep[Long]) =
+ _selectAll(persistenceId).filter(row =>
+ row.sequenceNumber <= maxSequenceNr && row.sequenceNumber >=
minSequenceNr &&
+ row.created <= maxTimestamp && row.created >= minTimestamp)
+ val selectByCriteria = Compiled(_selectByCriteria _)
+
+ private def _selectOneByCriteria(
+ persistenceId: Rep[String],
+ maxSequenceNr: Rep[Long],
+ maxTimestamp: Rep[Long],
+ minSequenceNr: Rep[Long],
+ minTimestamp: Rep[Long]) =
+ _selectByCriteria(persistenceId, maxSequenceNr, maxTimestamp,
minSequenceNr, minTimestamp).take(1)
+ val selectOneByCriteria = Compiled(_selectOneByCriteria _)
+
private def _selectByPersistenceIdAndSequenceNr(persistenceId: Rep[String],
sequenceNr: Rep[Long]) =
_selectAll(persistenceId).filter(_.sequenceNumber === sequenceNr)
val selectByPersistenceIdAndSequenceNr =
Compiled(_selectByPersistenceIdAndSequenceNr _)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/ByteArraySnapshotDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/ByteArraySnapshotDao.scala
index ba37bf6a..f570c4ac 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/ByteArraySnapshotDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/ByteArraySnapshotDao.scala
@@ -15,7 +15,7 @@
package org.apache.pekko.persistence.jdbc.snapshot.dao.legacy
import org.apache.pekko
-import pekko.persistence.SnapshotMetadata
+import pekko.persistence.{ SnapshotMetadata, SnapshotSelectionCriteria }
import pekko.persistence.jdbc.config.SnapshotConfig
import pekko.persistence.jdbc.snapshot.dao.legacy.SnapshotTables.SnapshotRow
import pekko.persistence.jdbc.snapshot.dao.SnapshotDao
@@ -44,6 +44,26 @@ class ByteArraySnapshotDao(
case Failure(cause) => throw cause
}
+ override def snapshotForCriteria(
+ persistenceId: String,
+ criteria: SnapshotSelectionCriteria): Future[Option[(SnapshotMetadata,
Any)]] =
+ if (criteria.minSequenceNr == 0L && criteria.minTimestamp == 0L)
+ super.snapshotForCriteria(persistenceId, criteria)
+ else
+ db.run(queries.selectOneByCriteria(
+ (persistenceId, criteria.maxSequenceNr, criteria.maxTimestamp,
criteria.minSequenceNr,
+ criteria.minTimestamp)).result)
+ .map(_.headOption.map(toSnapshotData))
+
+ override def deleteByCriteria(persistenceId: String, criteria:
SnapshotSelectionCriteria): Future[Unit] =
+ if (criteria.minSequenceNr == 0L && criteria.minTimestamp == 0L)
+ super.deleteByCriteria(persistenceId, criteria)
+ else
+ db.run(queries.selectByCriteria(
+ (persistenceId, criteria.maxSequenceNr, criteria.maxTimestamp,
criteria.minSequenceNr,
+ criteria.minTimestamp)).delete)
+ .map(_ => ())(ExecutionContext.parasitic)
+
override def latestSnapshot(persistenceId: String):
Future[Option[(SnapshotMetadata, Any)]] =
for {
rows <- db.run(queries.selectLatestByPersistenceId(persistenceId).result)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotQueries.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotQueries.scala
index d6c0ed2e..a8c189ce 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotQueries.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/legacy/SnapshotQueries.scala
@@ -36,6 +36,26 @@ class SnapshotQueries(val profile: JdbcProfile, override val
snapshotTableCfg: L
_selectAll(persistenceId).take(1)
val selectLatestByPersistenceId = Compiled(_selectLatestByPersistenceId _)
+ private def _selectByCriteria(
+ persistenceId: Rep[String],
+ maxSequenceNr: Rep[Long],
+ maxTimestamp: Rep[Long],
+ minSequenceNr: Rep[Long],
+ minTimestamp: Rep[Long]) =
+ _selectAll(persistenceId).filter(row =>
+ row.sequenceNumber <= maxSequenceNr && row.sequenceNumber >=
minSequenceNr &&
+ row.created <= maxTimestamp && row.created >= minTimestamp)
+ val selectByCriteria = Compiled(_selectByCriteria _)
+
+ private def _selectOneByCriteria(
+ persistenceId: Rep[String],
+ maxSequenceNr: Rep[Long],
+ maxTimestamp: Rep[Long],
+ minSequenceNr: Rep[Long],
+ minTimestamp: Rep[Long]) =
+ _selectByCriteria(persistenceId, maxSequenceNr, maxTimestamp,
minSequenceNr, minTimestamp).take(1)
+ val selectOneByCriteria = Compiled(_selectOneByCriteria _)
+
private def _selectByPersistenceIdAndSequenceNr(persistenceId: Rep[String],
sequenceNr: Rep[Long]) =
_selectAll(persistenceId).filter(_.sequenceNumber === sequenceNr)
val selectByPersistenceIdAndSequenceNr =
Compiled(_selectByPersistenceIdAndSequenceNr _)
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/CustomSnapshotDaoRecoverySpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/CustomSnapshotDaoRecoverySpec.scala
new file mode 100644
index 00000000..8a7701d1
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/CustomSnapshotDaoRecoverySpec.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.snapshot
+
+import scala.concurrent.{ ExecutionContext, Future }
+
+import org.apache.pekko
+import pekko.persistence.{
+ DeleteSnapshotsSuccess,
+ Persistence,
+ SelectedSnapshot,
+ SnapshotMetadata,
+ SnapshotSelectionCriteria
+}
+import pekko.persistence.SnapshotProtocol.{ DeleteSnapshots, LoadSnapshot,
LoadSnapshotResult }
+import pekko.persistence.jdbc.SharedActorSystemTestSpec
+import pekko.persistence.jdbc.config.SnapshotConfig
+import pekko.persistence.jdbc.db.SlickDatabase
+import pekko.persistence.jdbc.snapshot.dao.{ DefaultSnapshotDao, SnapshotDao }
+import pekko.persistence.jdbc.testkit.internal.H2
+import pekko.serialization.Serialization
+import pekko.stream.Materializer
+import pekko.testkit.TestProbe
+import com.typesafe.config.ConfigValueFactory
+import slick.jdbc.{ JdbcBackend, JdbcProfile }
+
+// Models an existing custom DAO that implements only the pre-criteria API.
+class UpperBoundSnapshotDao(
+ db: JdbcBackend#Database,
+ profile: JdbcProfile,
+ config: SnapshotConfig,
+ serialization: Serialization)(implicit ec: ExecutionContext, mat:
Materializer)
+ extends DefaultSnapshotDao(db, profile, config, serialization)
+ with SnapshotDao {
+ override def snapshotForCriteria(
+ persistenceId: String,
+ criteria: SnapshotSelectionCriteria): Future[Option[(SnapshotMetadata,
Any)]] =
+ super[SnapshotDao].snapshotForCriteria(persistenceId, criteria)
+
+ override def deleteByCriteria(persistenceId: String, criteria:
SnapshotSelectionCriteria): Future[Unit] =
+ super[SnapshotDao].deleteByCriteria(persistenceId, criteria)
+}
+
+class CustomSnapshotDaoRecoverySpec
+ extends SharedActorSystemTestSpec(
+ "h2-application.conf",
+ Map("jdbc-snapshot-store.dao" ->
ConfigValueFactory.fromAnyRef(classOf[UpperBoundSnapshotDao].getName))) {
+
+ private lazy val snapshots = new DefaultSnapshotDao(
+ db,
+ SlickDatabase.profile(config, "slick"),
+ new
SnapshotConfig(system.settings.config.getConfig("jdbc-snapshot-store")),
+ serialization)
+ private lazy val snapshotStore =
Persistence(system).snapshotStoreFor("jdbc-snapshot-store")
+
+ override def beforeAll(): Unit = {
+ dropAndCreate(H2)
+ super.beforeAll()
+ }
+
+ it should "filter an upper-bound custom DAO result against lower bounds" in {
+ val persistenceId = "custom-bounded-load"
+ val older = SnapshotMetadata(persistenceId, 2L, 200L)
+ snapshots.save(older, "older").futureValue
+
+ val criteria = SnapshotSelectionCriteria(minSequenceNr = 3L, minTimestamp
= 201L)
+ val probe = TestProbe()
+ snapshotStore.tell(LoadSnapshot(persistenceId, criteria, Long.MaxValue),
probe.ref)
+ probe.expectMsg(LoadSnapshotResult(None, Long.MaxValue))
+
+ val matching = SnapshotMetadata(persistenceId, 4L, 400L)
+ snapshots.save(matching, "matching").futureValue
+ snapshotStore.tell(LoadSnapshot(persistenceId, criteria, Long.MaxValue),
probe.ref)
+ probe.expectMsg(LoadSnapshotResult(Some(SelectedSnapshot(matching,
"matching")), Long.MaxValue))
+ }
+
+ it should "preserve upper-bound deletion for a custom DAO" in {
+ val persistenceId = "custom-bounded-delete"
+ Seq(1L, 2L, 4L, 6L).foldLeft(Future.successful(())) { (saved, sequenceNr)
=>
+ saved.flatMap(_ =>
+ snapshots.save(SnapshotMetadata(persistenceId, sequenceNr, sequenceNr
* 100), sequenceNr.toInt))
+ }.futureValue
+
+ val criteria = SnapshotSelectionCriteria(maxSequenceNr = 4L, minSequenceNr
= 2L)
+ val probe = TestProbe()
+ snapshotStore.tell(DeleteSnapshots(persistenceId, criteria), probe.ref)
+ probe.expectMsg(DeleteSnapshotsSuccess(criteria))
+
+ snapshots.snapshotForMaxSequenceNr(persistenceId, 4L).futureValue shouldBe
None
+ snapshots.latestSnapshot(persistenceId).futureValue.map(_._1.sequenceNr)
shouldBe Some(6L)
+ }
+}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStoreSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStoreSpec.scala
index 56b91bf4..80fe2f26 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStoreSpec.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/JdbcSnapshotStoreSpec.scala
@@ -14,14 +14,22 @@
package org.apache.pekko.persistence.jdbc.snapshot
+import scala.concurrent.Future
+
import org.apache.pekko
-import pekko.persistence.CapabilityFlag
+import pekko.persistence.{ CapabilityFlag, SelectedSnapshot, SnapshotMetadata,
SnapshotSelectionCriteria }
+import pekko.persistence.SnapshotProtocol.{ DeleteSnapshots, LoadSnapshot,
LoadSnapshotResult }
+import pekko.persistence.DeleteSnapshotsSuccess
import pekko.persistence.jdbc.config._
import pekko.persistence.jdbc.util.{ ClasspathResources, DropCreate }
import pekko.persistence.jdbc.db.SlickDatabase
import pekko.persistence.jdbc.testkit.internal.{ H2, SchemaType,
SchemaUtilsImpl }
import pekko.persistence.snapshot.SnapshotStoreSpec
-import pekko.testkit.TestKit
+import pekko.persistence.jdbc.snapshot.dao.{ DefaultSnapshotDao, SnapshotDao }
+import pekko.persistence.jdbc.snapshot.dao.legacy.ByteArraySnapshotDao
+import pekko.serialization.SerializationExtension
+import pekko.stream.SystemMaterializer
+import pekko.testkit.{ TestKit, TestProbe }
import com.typesafe.config.{ Config, ConfigFactory }
import org.scalatest.BeforeAndAfterAll
@@ -46,6 +54,70 @@ abstract class JdbcSnapshotStoreSpec(config: Config,
schemaType: SchemaType)
protected override def supportsSerialization: CapabilityFlag = newDao
protected override def supportsMetadata: CapabilityFlag = newDao
+ private lazy val boundsDao: SnapshotDao = {
+ val profile = SlickDatabase.profile(config, "slick")
+ val snapshotConfig = new
SnapshotConfig(system.settings.config.getConfig("jdbc-snapshot-store"))
+ val serialization = SerializationExtension(system)
+ implicit val mat = SystemMaterializer(system).materializer
+ if (newDao) new DefaultSnapshotDao(db, profile, snapshotConfig,
serialization)
+ else new ByteArraySnapshotDao(db, profile, snapshotConfig, serialization)
+ }
+
+ private def seedBoundsSnapshots(persistenceId: String):
Seq[SnapshotMetadata] = {
+ // Deliberately do not order timestamps by sequence number.
+ val metadata = Seq(100L, 400L, 200L, 300L, 500L).zipWithIndex.map { case
(timestamp, index) =>
+ SnapshotMetadata(persistenceId, index + 1L, timestamp)
+ }
+ Future.traverse(metadata)(md => boundsDao.save(md,
s"bounds-${md.sequenceNr}")).futureValue
+ metadata
+ }
+
+ private def allSequenceNumbers(persistenceId: String): Future[Seq[Long]] =
boundsDao match {
+ case dao: DefaultSnapshotDao =>
+ import dao.queries.profile.api._
+
db.run(dao.queries.selectAll(persistenceId).result).map(_.map(_.sequenceNumber))
+ case dao: ByteArraySnapshotDao =>
+ import dao.queries.profile.api._
+
db.run(dao.queries.selectAll(persistenceId).result).map(_.map(_.sequenceNumber))
+ case dao => Future.failed(new IllegalStateException(s"Unexpected snapshot
DAO: ${dao.getClass.getName}"))
+ }
+
+ private val boundsCases = Seq(
+ ("default bounds", SnapshotSelectionCriteria.Latest, Seq(1L, 2L, 3L, 4L,
5L)),
+ ("minimum sequence number", SnapshotSelectionCriteria(minSequenceNr = 3),
Seq(3L, 4L, 5L)),
+ ("unordered timestamps", SnapshotSelectionCriteria(maxSequenceNr = 4,
minTimestamp = 350), Seq(2L)),
+ ("all bounds", SnapshotSelectionCriteria(4, 400, 2, 300), Seq(2L, 4L)),
+ ("equal bounds", SnapshotSelectionCriteria(3, 200, 3, 200), Seq(3L)),
+ ("empty interval", SnapshotSelectionCriteria(maxSequenceNr = 2,
minSequenceNr = 4), Seq.empty[Long]))
+
+ "Snapshot selection bounds" must {
+ boundsCases.foreach { case (label, criteria, matching) =>
+ s"load with $label" in {
+ val persistenceId = s"$pid-bounds"
+ val metadata = seedBoundsSnapshots(persistenceId)
+ val probe = TestProbe()
+ snapshotStore.tell(LoadSnapshot(persistenceId, criteria,
Long.MaxValue), probe.ref)
+ val expected = matching.lastOption.map { sequenceNr =>
+ SelectedSnapshot(metadata((sequenceNr - 1).toInt),
s"bounds-$sequenceNr")
+ }
+ probe.expectMsg(LoadSnapshotResult(expected, Long.MaxValue))
+ }
+
+ s"delete with $label without deleting other snapshots" in {
+ val persistenceId = s"$pid-bounds"
+ val metadata = seedBoundsSnapshots(persistenceId)
+ val otherMetadata = seedBoundsSnapshots(s"$persistenceId-other")
+ val probe = TestProbe()
+ snapshotStore.tell(DeleteSnapshots(persistenceId, criteria), probe.ref)
+ probe.expectMsg(DeleteSnapshotsSuccess(criteria))
+ val survivors = metadata.filterNot(md =>
matching.contains(md.sequenceNr))
+ allSequenceNumbers(persistenceId).futureValue shouldBe
survivors.map(_.sequenceNr).reverse
+ allSequenceNumbers(otherMetadata.head.persistenceId).futureValue
shouldBe
+ otherMetadata.map(_.sequenceNr).reverse
+ }
+ }
+ }
+
override def beforeAll(): Unit = {
dropAndCreate(schemaType)
super.beforeAll()
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDaoSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDaoSpec.scala
new file mode 100644
index 00000000..e5a3b6d0
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/snapshot/dao/SnapshotDaoSpec.scala
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.snapshot.dao
+
+import org.apache.pekko.persistence.{ SnapshotMetadata,
SnapshotSelectionCriteria }
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.Future
+
+class SnapshotDaoSpec extends AnyWordSpec with Matchers with ScalaFutures {
+ private class CustomSnapshotDao extends SnapshotDao {
+ var calls = Vector.empty[(String, String, Seq[Long])]
+ val snapshot = Some((SnapshotMetadata("custom", 3, 200), "snapshot"))
+
+ private def loaded(method: String, persistenceId: String, bounds: Long*) =
{
+ calls :+= ((method, persistenceId, bounds.toList))
+ Future.successful(snapshot)
+ }
+
+ private def deleted(method: String, persistenceId: String, bounds: Long*)
= {
+ calls :+= ((method, persistenceId, bounds.toList))
+ Future.successful(())
+ }
+
+ override def latestSnapshot(persistenceId: String) = loaded("latest",
persistenceId)
+ override def snapshotForMaxTimestamp(persistenceId: String, timestamp:
Long) =
+ loaded("timestamp", persistenceId, timestamp)
+ override def snapshotForMaxSequenceNr(persistenceId: String, sequenceNr:
Long) =
+ loaded("sequence", persistenceId, sequenceNr)
+ override def snapshotForMaxSequenceNrAndMaxTimestamp(persistenceId:
String, sequenceNr: Long, timestamp: Long) =
+ loaded("both", persistenceId, sequenceNr, timestamp)
+
+ override def deleteAllSnapshots(persistenceId: String) = deleted("latest",
persistenceId)
+ override def deleteUpToMaxTimestamp(persistenceId: String, timestamp:
Long) =
+ deleted("timestamp", persistenceId, timestamp)
+ override def deleteUpToMaxSequenceNr(persistenceId: String, sequenceNr:
Long) =
+ deleted("sequence", persistenceId, sequenceNr)
+ override def deleteUpToMaxSequenceNrAndMaxTimestamp(persistenceId: String,
sequenceNr: Long, timestamp: Long) =
+ deleted("both", persistenceId, sequenceNr, timestamp)
+
+ override def delete(persistenceId: String, sequenceNr: Long) =
deleted("single", persistenceId, sequenceNr)
+ override def save(metadata: SnapshotMetadata, snapshot: Any) =
+ Future.failed(new UnsupportedOperationException("save"))
+ }
+
+ "SnapshotDao criteria defaults" should {
+ val upperBounds = Seq(
+ (SnapshotSelectionCriteria.Latest, "latest", Seq.empty[Long]),
+ (SnapshotSelectionCriteria(maxTimestamp = 200), "timestamp", Seq(200L)),
+ (SnapshotSelectionCriteria(maxSequenceNr = 3), "sequence", Seq(3L)),
+ (SnapshotSelectionCriteria(3, 200), "both", Seq(3L, 200L)))
+
+ upperBounds.foreach { case (upper, method, arguments) =>
+ s"preserve the existing $method load and delete methods" in {
+ val dao = new CustomSnapshotDao
+ Seq(upper, upper.copy(minSequenceNr = 1L, minTimestamp = 1L)).foreach
{ criteria =>
+ dao.snapshotForCriteria("custom", criteria).futureValue shouldBe
dao.snapshot
+ dao.deleteByCriteria("custom", criteria).futureValue shouldBe (())
+ }
+ dao.calls shouldBe Vector.fill(4)((method, "custom", arguments))
+ }
+ }
+
+ "filter an upper-bound result that does not satisfy a lower bound" in {
+ val dao = new CustomSnapshotDao
+ val criteria = SnapshotSelectionCriteria(minSequenceNr = 4L,
minTimestamp = 201L)
+ dao.snapshotForCriteria("custom", criteria).futureValue shouldBe None
+ dao.calls shouldBe Vector(("latest", "custom", Seq.empty))
+
+ dao.deleteByCriteria("custom", criteria).futureValue shouldBe (())
+ dao.calls shouldBe Vector(
+ ("latest", "custom", Seq.empty),
+ ("latest", "custom", Seq.empty))
+ }
+ }
+}
diff --git a/docs/src/main/paradox/custom-dao.md
b/docs/src/main/paradox/custom-dao.md
index f9861233..c388aadc 100644
--- a/docs/src/main/paradox/custom-dao.md
+++ b/docs/src/main/paradox/custom-dao.md
@@ -40,6 +40,30 @@ class MyCustomSnapshotDao(db: JdbcBackend#Database, val
profile: JdbcProfile, sn
As you can see, the custom DAOs get a _Slick database_, a _Slick profile_, the
journal or snapshot _configuration_, an
_org.apache.pekko.serialization.Serialization_, an _ExecutionContext_ and
_Materializer_ injected after constructed.
You should register the Fully Qualified Class Name in `application.conf` so
that the custom DAOs will be used.
+## Snapshot selection criteria
+
+Snapshot loading and criteria-based deletion use
`SnapshotDao.snapshotForCriteria` and
+`SnapshotDao.deleteByCriteria`. Both built-in snapshot DAOs apply all four
bounds in
+`SnapshotSelectionCriteria`: `minSequenceNr`, `maxSequenceNr`, `minTimestamp`,
and
+`maxTimestamp`. Bounds are inclusive, and both the sequence number and
timestamp must
+match. Loading returns the matching snapshot with the highest sequence number;
deletion
+removes only matching snapshots for the requested persistence ID. An empty
interval
+matches no snapshots.
+
+The existing upper-bound methods remain available. The default
`snapshotForCriteria`
+implementation delegates to those methods and post-filters their result
against all bounds,
+preserving safe recovery for existing custom DAOs. It may return no snapshot
when timestamps
+are not ordered by sequence number even if an older snapshot matches. The
default
+`deleteByCriteria` implementation preserves the previous upper-bound deletion
behavior and
+may delete snapshots below a minimum bound, including when the interval is
empty. Custom DAOs
+should override both criteria methods to query all four bounds directly, as
the built-in DAOs
+do.
+
+Subclasses of a built-in snapshot DAO that override its upper-bound methods
must also override
+`snapshotForCriteria` and `deleteByCriteria` if they need to apply custom
loading or deletion
+behavior. Criteria-based operations in the built-in DAOs execute their
four-bound queries
+directly.
+
For more information please review the two default implementations
`org.apache.pekko.persistence.jdbc.dao.bytea.journal.ByteArrayJournalDao` and
`org.apache.pekko.persistence.jdbc.dao.bytea.snapshot.ByteArraySnapshotDao` or
the demo custom DAO example from the
[demo-akka-persistence](https://github.com/dnvriend/demo-akka-persistence-jdbc)
site.
@@@warning { title="Binary compatibility" }
@@ -49,4 +73,3 @@ For example 4.0.0 is not binary backwards compatible with
3.5.x. There may also
the APIs for customer DAOs if new capabilities must be added to to the traits.
@@@
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
