Enhance DurableStateStore TCK with OCC, serialization, and soft-delete tests

docs/src/main/paradox/persistence-journals.md

@@ -134,6 +134,14 @@ Scala
 Java
 :  @@snip [LambdaPersistencePluginDocTest.java](/docs/src/test/java/jdocs/persistence/LambdaPersistencePluginDocTest.java) { #snapshot-store-tck-java }
 
+To include the `DurableStateStore` TCK tests in your test suite simply extend the provided @scala[`DurableStateStoreSpec`]@java[`JavaDurableStateStoreSpec`]:
+
+Scala
+:  @@snip [PersistencePluginDocSpec.scala](/docs/src/test/scala/docs/persistence/PersistencePluginDocSpec.scala) { #durable-state-tck-scala }
+
+Java
+:  @@snip [LambdaPersistencePluginDocTest.java](/docs/src/test/java/jdocs/persistence/LambdaPersistencePluginDocTest.java) { #durable-state-tck-java }
+
 In case your plugin requires some setting up (starting a mock database, removing temporary files etc.) you can override the
 `beforeAll` and `afterAll` methods to hook into the tests lifecycle:
 

docs/src/test/java/jdocs/persistence/LambdaPersistencePluginDocTest.java

@@ -27,6 +27,7 @@
 import org.apache.pekko.persistence.*;
 import org.apache.pekko.persistence.japi.journal.JavaJournalSpec;
 import org.apache.pekko.persistence.japi.snapshot.JavaSnapshotStoreSpec;
+import org.apache.pekko.persistence.japi.state.JavaDurableStateStoreSpec;
 import org.apache.pekko.persistence.journal.japi.*;
 import org.apache.pekko.persistence.journal.leveldb.SharedLeveldbJournal;
 import org.apache.pekko.persistence.journal.leveldb.SharedLeveldbStore;
@@ -172,6 +173,30 @@ public MySnapshotStoreTest() {
         // #snapshot-store-tck-java
       };
 
+  static Object o3b =
+      new Object() {
+        // #durable-state-tck-java
+        class MyDurableStateStoreTest extends JavaDurableStateStoreSpec {
+
+          public MyDurableStateStoreTest() {
+            super(
+                ConfigFactory.parseString(
+                    "pekko.persistence.state.plugin = \"my.durable-state.plugin\""));
+          }
+
+          @Override
+          public CapabilityFlag supportsDeleteWithRevisionCheck() {
+            return CapabilityFlag.on();
+          }
+
+          @Override
+          public CapabilityFlag supportsUpsertWithRevisionCheck() {
+            return CapabilityFlag.on();
+          }
+        }
+        // #durable-state-tck-java
+      };
+
   static Object o4 =
       new Object() {
         // https://github.com/pekko/pekko/issues/26826

docs/src/test/scala/docs/persistence/PersistencePluginDocSpec.scala

@@ -222,6 +222,27 @@ object PersistenceTCKDoc {
     }
     // #snapshot-store-tck-scala
   }
+  object example2b {
+    import org.apache.pekko.persistence.state.DurableStateStoreSpec
+
+    // #durable-state-tck-scala
+    class MyDurableStateStoreSpec
+        extends DurableStateStoreSpec(
+          config = ConfigFactory.parseString("""
+        pekko.persistence.state.plugin = "my.durable-state.plugin"
+        """)) {
+
+      override def supportsDeleteWithRevisionCheck: CapabilityFlag =
+        true // or CapabilityFlag.on
+
+      override def supportsUpsertWithRevisionCheck: CapabilityFlag =
+        true // or CapabilityFlag.on
+
+      override def supportsSerialization: CapabilityFlag =
+        true // or CapabilityFlag.on
+    }
+    // #durable-state-tck-scala
+  }
   object example3 {
     import java.io.File
 

persistence-tck/src/main/scala/org/apache/pekko/persistence/CapabilityFlags.scala

@@ -86,3 +86,33 @@ trait SnapshotStoreCapabilityFlags extends CapabilityFlags {
   protected def supportsMetadata: CapabilityFlag
 }
 //#snapshot-store-flags
+
+//#durable-state-store-flags
+trait DurableStateStoreCapabilityFlags extends CapabilityFlags {
+
+  /**
+   * When `true` enables tests which check if the durable state store properly rejects
+   * a `deleteObject` call when the revision does not match the stored revision.
+   */
+  protected def supportsDeleteWithRevisionCheck: CapabilityFlag
+
+  /**
+   * When `true` enables tests which check if the durable state store properly rejects
+   * an `upsertObject` call when the revision is stale (does not match the expected next revision).
+   * This is the optimistic concurrency check on writes.
+   */
+  protected def supportsUpsertWithRevisionCheck: CapabilityFlag
+
+  /**
+   * When `true` enables tests which check if the durable state store properly serializes
+   * and deserializes values via the configured Pekko serializer infrastructure.
+   */
+  protected def supportsSerialization: CapabilityFlag
+
+  /**
+   * When `true` enables tests which check the deprecated single-argument `deleteObject(persistenceId)`
+   * overload, which deletes regardless of the current revision.
+   */
+  protected def supportsSoftDelete: CapabilityFlag
+}
+//#durable-state-store-flags

persistence-tck/src/main/scala/org/apache/pekko/persistence/japi/state/JavaDurableStateStoreSpec.scala

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.japi.state
+
+import scala.collection.immutable
+
+import org.apache.pekko
+import pekko.persistence.CapabilityFlag
+import pekko.persistence.state.DurableStateStoreSpec
+
+import org.scalatest.{ Args, ConfigMap, Filter, Status, Suite, TestData }
+
+import com.typesafe.config.Config
+
+/**
+ * JAVA API
+ *
+ * This spec aims to verify custom pekko-persistence [[pekko.persistence.state.DurableStateStore]]
+ * implementations.
+ * Plugin authors are highly encouraged to include it in their plugin's test suites.
+ *
+ * In case your durable state store plugin needs some kind of setup or teardown, override the
+ * `beforeAll` or `afterAll` methods (don't forget to call `super` in your overridden methods).
+ *
+ * @see [[pekko.persistence.state.DurableStateStoreSpec]]
+ */
+class JavaDurableStateStoreSpec(config: Config) extends DurableStateStoreSpec(config) {
+  override protected def supportsDeleteWithRevisionCheck: CapabilityFlag = CapabilityFlag.off()
+
+  override protected def supportsUpsertWithRevisionCheck: CapabilityFlag = CapabilityFlag.off()
+
+  override protected def supportsSoftDelete: CapabilityFlag = CapabilityFlag.off()
+
+  override def runTests(testName: Option[String], args: Args): Status =
+    super.runTests(testName, args)
+
+  override def runTest(testName: String, args: Args): Status =
+    super.runTest(testName, args)
+
+  override def run(testName: Option[String], args: Args): Status =
+    super.run(testName, args)
+
+  override def testDataFor(testName: String, theConfigMap: ConfigMap): TestData =
+    super.testDataFor(testName, theConfigMap)
+
+  override def testNames: Set[String] =
+    super.testNames
+
+  override def tags: Map[String, Set[String]] =
+    super.tags
+
+  override def rerunner: Option[String] =
+    super.rerunner
+
+  override def expectedTestCount(filter: Filter): Int =
+    super.expectedTestCount(filter)
+
+  override def suiteId: String =
+    super.suiteId
+
+  override def suiteName: String =
+    super.suiteName
+
+  override def runNestedSuites(args: Args): Status =
+    super.runNestedSuites(args)
+
+  override def nestedSuites: immutable.IndexedSeq[Suite] =
+    super.nestedSuites
+}