Adding persistence test module

fjtirado · fjtirado · commit 30f0d36d22f6 · 2026-01-29T21:03:28.000+01:00
Signed-off-by: fjtirado &lt;ftirados@redhat.com&gt;
diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/DefaultPersistenceInstanceReader.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/DefaultPersistenceInstanceReader.java
@@ -17,7 +17,6 @@
 
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowInstance;
-import java.util.Collection;
 import java.util.Optional;
 import java.util.stream.Stream;
 
@@ -29,16 +28,6 @@ protected DefaultPersistenceInstanceReader(PersistenceInstanceStore store) {
     this.store = store;
   }
 
-  @Override
-  public Stream<WorkflowInstance> scan(
-      WorkflowDefinition definition, Collection<String> instanceIds) {
-    PersistenceInstanceTransaction transaction = store.begin();
-    return instanceIds.stream()
-        .map(id -> read(transaction, definition, id))
-        .flatMap(Optional::stream)
-        .onClose(() -> transaction.commit());
-  }
-
   @Override
   public Optional<WorkflowInstance> find(WorkflowDefinition definition, String instanceId) {
     PersistenceInstanceTransaction transaction = store.begin();
@@ -57,10 +46,10 @@ private Optional<WorkflowInstance> read(
   }
 
   @Override
-  public Stream<WorkflowInstance> scanAll(WorkflowDefinition definition) {
+  public Stream<WorkflowInstance> scanAll(WorkflowDefinition definition, String applicationId) {
     PersistenceInstanceTransaction transaction = store.begin();
     return transaction
-        .scanAll(definition)
+        .scanAll(applicationId, definition)
         .onClose(() -> transaction.commit())
         .map(v -> new WorkflowPersistenceInstance(definition, v));
   }
diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceReader.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceReader.java
@@ -17,15 +17,16 @@
 
 import io.serverlessworkflow.impl.WorkflowDefinition;
 import io.serverlessworkflow.impl.WorkflowInstance;
-import java.util.Collection;
 import java.util.Optional;
 import java.util.stream.Stream;
 
 public interface PersistenceInstanceReader {
 
-  Stream<WorkflowInstance> scanAll(WorkflowDefinition definition);
+  default Stream<WorkflowInstance> scanAll(WorkflowDefinition definition) {
+    return scanAll(definition, definition.application().id());
+  }
 
-  Stream<WorkflowInstance> scan(WorkflowDefinition definition, Collection<String> instanceIds);
+  Stream<WorkflowInstance> scanAll(WorkflowDefinition definition, String applicationId);
 
   Optional<WorkflowInstance> find(WorkflowDefinition definition, String instanceId);
 }
diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceTransaction.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceTransaction.java
@@ -40,7 +40,7 @@ public interface PersistenceInstanceTransaction {
 
   void clearStatus(WorkflowContextData workflowContext);
 
-  Stream<PersistenceWorkflowInfo> scanAll(WorkflowDefinition definition);
+  Stream<PersistenceWorkflowInfo> scanAll(String applicationId, WorkflowDefinition definition);
 
   Optional<PersistenceWorkflowInfo> readWorkflowInfo(
       WorkflowDefinition definition, String instanceId);
diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java
@@ -31,13 +31,13 @@ public class WorkflowPersistenceInstance extends WorkflowMutableInstance {
   public WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWorkflowInfo info) {
     super(definition, info.id(), info.input());
     this.info = info;
+    this.startedAt = info.startedAt();
   }
 
   @Override
   public CompletableFuture<WorkflowModel> start() {
     return startExecution(
         () -> {
-          startedAt = info.startedAt();
           if (info.status() == WorkflowStatus.SUSPENDED) {
             internalSuspend();
           }
diff --git a/impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceTransaction.java b/impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceTransaction.java
@@ -32,12 +32,16 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-public abstract class BigMapInstanceTransaction<V, T, S> implements PersistenceInstanceTransaction {
+public abstract class BigMapInstanceTransaction<V, T, S, A>
+    implements PersistenceInstanceTransaction {
 
   @Override
   public void writeInstanceData(WorkflowContextData workflowContext) {
+    String key = key(workflowContext);
     instanceData(workflowContext.definition())
-        .put(key(workflowContext), marshallInstance(workflowContext.instanceData()));
+        .put(key, marshallInstance(workflowContext.instanceData()));
+    applicationData()
+        .put(key, marshallApplicationId(workflowContext.definition().application().id()));
   }
 
   @Override
@@ -53,28 +57,36 @@ public void writeCompletedTask(WorkflowContextData workflowContext, TaskContextD
     tasks(key(workflowContext))
         .put(
             taskContext.position().jsonPointer(),
-            marshallTaskRetried(workflowContext, (TaskContext) taskContext));
+            marshallTaskCompleted(workflowContext, (TaskContext) taskContext));
   }
 
   @Override
-  public Stream<PersistenceWorkflowInfo> scanAll(WorkflowDefinition definition) {
+  public Stream<PersistenceWorkflowInfo> scanAll(
+      String applicationId, WorkflowDefinition definition) {
     Map<String, V> instances = instanceData(definition);
+    Map<String, A> applicationData = applicationData();
     Map<String, S> status = status(definition);
     return instances.entrySet().stream()
+        .filter(e -> testAppl(applicationData, e.getKey(), applicationId))
         .map(
             e ->
                 readPersistenceInfo(
                     e.getKey(), e.getValue(), tasks(e.getKey()), status.get(e.getKey())));
   }
 
+  private boolean testAppl(Map<String, A> applicationData, String key, String applicationId) {
+    A item = applicationData.get(key);
+    return item == null || unmarshallApplicationId(item).equals(applicationId);
+  }
+
   @Override
   public Optional<PersistenceWorkflowInfo> readWorkflowInfo(
       WorkflowDefinition definition, String key) {
     Map<String, V> instances = instanceData(definition);
     return instances.containsKey(key)
-        ? Optional.empty()
-        : Optional.of(
-            readPersistenceInfo(key, instances.get(key), tasks(key), status(definition).get(key)));
+        ? Optional.of(
+            readPersistenceInfo(key, instances.get(key), tasks(key), status(definition).get(key)))
+        : Optional.empty();
   }
 
   @Override
@@ -117,6 +129,8 @@ private String key(WorkflowContextData workflowContext) {
     return workflowContext.instanceData().id();
   }
 
+  protected abstract Map<String, A> applicationData();
+
   protected abstract Map<String, V> instanceData(WorkflowDefinitionData definition);
 
   protected abstract Map<String, S> status(WorkflowDefinitionData workflowContext);
@@ -131,6 +145,8 @@ protected abstract T marshallTaskCompleted(
   protected abstract T marshallTaskRetried(
       WorkflowContextData workflowContext, TaskContext taskContext);
 
+  protected abstract A marshallApplicationId(String id);
+
   protected abstract S marshallStatus(WorkflowStatus status);
 
   protected abstract PersistenceTaskInfo unmarshallTaskInfo(T taskData);
@@ -139,5 +155,7 @@ protected abstract T marshallTaskRetried(
 
   protected abstract WorkflowStatus unmarshallStatus(S statusData);
 
+  protected abstract String unmarshallApplicationId(A a);
+
   protected abstract void removeTasks(String key);
 }
diff --git a/impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BytesMapInstanceTransaction.java b/impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BytesMapInstanceTransaction.java
@@ -22,6 +22,7 @@
 import io.serverlessworkflow.impl.WorkflowStatus;
 import io.serverlessworkflow.impl.executors.AbstractTaskExecutor;
 import io.serverlessworkflow.impl.executors.TransitionInfo;
+import io.serverlessworkflow.impl.marshaller.MarshallingUtils;
 import io.serverlessworkflow.impl.marshaller.TaskStatus;
 import io.serverlessworkflow.impl.marshaller.WorkflowBufferFactory;
 import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
@@ -34,7 +35,7 @@
 import java.io.ByteArrayOutputStream;
 
 public abstract class BytesMapInstanceTransaction
-    extends BigMapInstanceTransaction<byte[], byte[], byte[]> {
+    extends BigMapInstanceTransaction<byte[], byte[], byte[], byte[]> {
 
   private static final byte VERSION_0 = 0;
   private static final byte VERSION_1 = 1;
@@ -92,6 +93,14 @@ protected void writeModel(WorkflowOutputBuffer writer, WorkflowModel model) {
     writer.writeObject(model);
   }
 
+  protected byte[] marshallApplicationId(String id) {
+    return MarshallingUtils.writeString(factory, id);
+  }
+
+  protected String unmarshallApplicationId(byte[] value) {
+    return MarshallingUtils.readString(factory, value);
+  }
+
   @Override
   protected byte[] marshallTaskRetried(
       WorkflowContextData workflowContext, TaskContext taskContext) {
diff --git a/impl/persistence/jackson-marshaller/src/main/java/io/serverlessworkflow/impl/marshaller/jackson/JacksonModelMarshaller.java b/impl/persistence/jackson-marshaller/src/main/java/io/serverlessworkflow/impl/marshaller/jackson/JacksonModelMarshaller.java
@@ -38,7 +38,8 @@ public void write(WorkflowOutputBuffer buffer, JacksonModel object) {
   @Override
   public JacksonModel read(WorkflowInputBuffer buffer) {
     try {
-      return JsonUtils.mapper().readValue(buffer.readBytes(), JacksonModel.class);
+      JacksonModel model = JsonUtils.mapper().readValue(buffer.readBytes(), JacksonModel.class);
+      return model == null ? JacksonModel.NULL : model;
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
diff --git a/impl/persistence/mvstore/pom.xml b/impl/persistence/mvstore/pom.xml
@@ -16,5 +16,9 @@
           <groupId>io.serverlessworkflow</groupId>
           <artifactId>serverlessworkflow-persistence-big-map</artifactId>
         </dependency>
+        <dependency>
+          <groupId>io.serverlessworkflow</groupId>
+          <artifactId>serverlessworkflow-persistence-tests</artifactId>
+        </dependency>
     </dependencies>
 </project>
diff --git a/impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStore.java b/impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStore.java
@@ -41,7 +41,7 @@ public void close() {
   }
 
   @Override
-  public BigMapInstanceTransaction<byte[], byte[], byte[]> begin() {
+  public BigMapInstanceTransaction<byte[], byte[], byte[], byte[]> begin() {
     return new MVStoreTransaction(mvStore.begin(), factory);
   }
 }
diff --git a/impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStoreTransaction.java b/impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStoreTransaction.java
@@ -82,4 +82,9 @@ public void commit() {
   public void rollback() {
     transaction.rollback();
   }
+
+  @Override
+  protected Map<String, byte[]> applicationData() {
+    return transaction.openMap("APPLICATION");
+  }
 }
diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2020-Present The Serverless Workflow Specification Authors
+ *
+ * Licensed 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 io.serverlessworkflow.impl.persistence.mvstore;
+
+import io.serverlessworkflow.impl.persistence.PersistenceInstanceStore;
+import io.serverlessworkflow.impl.persistence.test.AbstractPersistenceTest;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.junit.jupiter.api.AfterEach;
+
+class MVStorePersistenceStoreTest extends AbstractPersistenceTest {
+
+  private static final String DB_NAME = "dbtest.db";
+
+  @Override
+  protected PersistenceInstanceStore persistenceStore() {
+    return new MVStorePersistenceStore(DB_NAME);
+  }
+
+  @AfterEach
+  void destroy() throws IOException {
+    Files.delete(Path.of(DB_NAME));
+  }
+}
diff --git a/impl/persistence/pom.xml b/impl/persistence/pom.xml
@@ -13,5 +13,6 @@
       <module>mvstore</module>
       <module>bigmap</module>
       <module>api</module>
+      <module>tests</module>
     </modules>
 </project>
diff --git a/impl/persistence/tests/pom.xml b/impl/persistence/tests/pom.xml
@@ -0,0 +1,58 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.serverlessworkflow</groupId>
+    <artifactId>serverlessworkflow-persistence</artifactId>
+    <version>8.0.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>serverlessworkflow-persistence-tests</artifactId>
+   <name>Serverless Workflow :: Impl :: Persistence:: Tests</name>
+  <dependencies>
+    <dependency>
+        <groupId>org.junit.jupiter</groupId>
+        <artifactId>junit-jupiter-engine</artifactId>
+        <scope>compile</scope>
+    </dependency>
+    <dependency>
+        <groupId>org.mockito</groupId>
+        <artifactId>mockito-core</artifactId>
+        <scope>compile</scope>
+     </dependency>
+     <dependency>
+        <groupId>org.junit.jupiter</groupId>
+        <artifactId>junit-jupiter-api</artifactId>
+        <scope>compile</scope>
+    </dependency>
+    <dependency>
+        <groupId>org.junit.jupiter</groupId>
+        <artifactId>junit-jupiter-params</artifactId>
+        <scope>compile</scope>
+    </dependency>
+    <dependency>
+        <groupId>org.assertj</groupId>
+        <artifactId>assertj-core</artifactId>
+        <scope>compile</scope>
+    </dependency>
+    <dependency>
+        <groupId>ch.qos.logback</groupId>
+        <artifactId>logback-classic</artifactId>
+        <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.serverlessworkflow</groupId>
+      <artifactId>serverlessworkflow-persistence-api</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>io.serverlessworkflow</groupId>
+      <artifactId>serverlessworkflow-impl-jackson</artifactId>
+    </dependency>
+     <dependency>
+        <groupId>io.serverlessworkflow</groupId>
+        <artifactId>serverlessworkflow-api</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>io.serverlessworkflow</groupId>
+        <artifactId>serverlessworkflow-persistence-jackson-marshaller</artifactId>
+    </dependency>    
+  </dependencies>
+</project>
diff --git a/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractPersistenceTest.java b/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractPersistenceTest.java
diff --git a/impl/persistence/tests/src/main/resources/simple-expression.yaml b/impl/persistence/tests/src/main/resources/simple-expression.yaml
diff --git a/impl/pom.xml b/impl/pom.xml

Original file line number	Diff line number	Diff line change
`@@ -31,13 +31,13 @@ public class WorkflowPersistenceInstance extends WorkflowMutableInstance {`
`31`	`31`	`public WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWorkflowInfo info) {`
`32`	`32`	`super(definition, info.id(), info.input());`
`33`	`33`	`this.info = info;`
	`34`	`+ this.startedAt = info.startedAt();`
`34`	`35`	`}`
`35`	`36`
`36`	`37`	`@Override`
`37`	`38`	`public CompletableFuture<WorkflowModel> start() {`
`38`	`39`	`return startExecution(`
`39`	`40`	`() -> {`
`40`		`- startedAt = info.startedAt();`
`41`	`41`	`if (info.status() == WorkflowStatus.SUSPENDED) {`
`42`	`42`	`internalSuspend();`
`43`	`43`	`}`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ public void close() {`
`41`	`41`	`}`
`42`	`42`
`43`	`43`	`@Override`
`44`		`- public BigMapInstanceTransaction<byte[], byte[], byte[]> begin() {`
	`44`	`+ public BigMapInstanceTransaction<byte[], byte[], byte[], byte[]> begin() {`
`45`	`45`	`return new MVStoreTransaction(mvStore.begin(), factory);`
`46`	`46`	`}`
`47`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -82,4 +82,9 @@ public void commit() {`
`82`	`82`	`public void rollback() {`
`83`	`83`	`transaction.rollback();`
`84`	`84`	`}`
	`85`	`+`
	`86`	`+ @Override`
	`87`	`+ protected Map<String, byte[]> applicationData() {`
	`88`	`+ return transaction.openMap("APPLICATION");`
	`89`	`+ }`
`85`	`90`	`}`