[Fix #782] Adding transaction support

Author: fjtirado

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceReader.java

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public abstract class BigMapInstanceReader<V, T, S> implements PersistenceInstanceReader {
@@ -36,43 +37,75 @@ protected BigMapInstanceReader(BigMapInstanceStore<String, V, T, S> store) {
     this.store = store;
   }
 
+  private <Result> Result doTransaction(
+      Function<BigMapInstanceTransaction<String, V, T, S>, Result> operations) {
+    BigMapInstanceTransaction<String, V, T, S> transaction = store.begin();
+    try {
+      Result result = operations.apply(transaction);
+      transaction.commit();
+      return result;
+    } catch (Exception ex) {
+      transaction.rollback();
+      throw ex;
+    }
+  }
+
   @Override
   public Map<String, WorkflowInstance> readAll(WorkflowDefinition definition) {
-    Map<String, V> instances = store.instanceData(definition);
-    Map<String, S> status = store.status(definition);
-    return instances.entrySet().stream()
-        .map(
-            e ->
-                restore(
-                    definition,
-                    e.getKey(),
-                    e.getValue(),
-                    store.tasks(e.getKey()),
-                    status.get(e.getKey())))
-        .collect(Collectors.toMap(WorkflowInstance::id, i -> i));
+    return doTransaction(
+        t -> {
+          Map<String, V> instances = t.instanceData(definition);
+          Map<String, S> status = t.status(definition);
+          return instances.entrySet().stream()
+              .map(
+                  e ->
+                      restore(
+                          definition,
+                          e.getKey(),
+                          e.getValue(),
+                          t.tasks(e.getKey()),
+                          status.get(e.getKey())))
+              .collect(Collectors.toMap(WorkflowInstance::id, i -> i));
+        });
   }
 
   @Override
   public Map<String, WorkflowInstance> read(
       WorkflowDefinition definition, Collection<String> instanceIds) {
-    return instanceIds.stream()
-        .map(id -> read(definition, id))
-        .flatMap(Optional::stream)
-        .collect(Collectors.toMap(WorkflowInstance::id, id -> id));
+    return doTransaction(
+        t -> {
+          Map<String, V> instances = t.instanceData(definition);
+          Map<String, S> status = t.status(definition);
+          return instanceIds.stream()
+              .map(id -> read(instances, status, t.tasks(id), definition, id))
+              .flatMap(Optional::stream)
+              .collect(Collectors.toMap(WorkflowInstance::id, id -> id));
+        });
   }
 
   @Override
   public Optional<WorkflowInstance> read(WorkflowDefinition definition, String instanceId) {
-    Map<String, V> instances = store.instanceData(definition);
+    return doTransaction(
+        t ->
+            read(
+                t.instanceData(definition),
+                t.status(definition),
+                t.tasks(instanceId),
+                definition,
+                instanceId));
+  }
+
+  private Optional<WorkflowInstance> read(
+      Map<String, V> instances,
+      Map<String, S> status,
+      Map<String, T> tasks,
+      WorkflowDefinition definition,
+      String instanceId) {
     return instances.containsKey(instanceId)
         ? Optional.empty()
         : Optional.of(
             restore(
-                definition,
-                instanceId,
-                instances.get(instanceId),
-                store.tasks(instanceId),
-                store.status(definition).get(instanceId)));
+                definition, instanceId, instances.get(instanceId), tasks, status.get(instanceId)));
   }
 
   public void close() {}

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceStore.java

@@ -15,16 +15,6 @@
  */
 package io.serverlessworkflow.impl.persistence.bigmap;
 
-import io.serverlessworkflow.impl.WorkflowDefinitionData;
-import java.util.Map;
-
 public interface BigMapInstanceStore<K, V, T, S> extends AutoCloseable {
-
-  Map<K, V> instanceData(WorkflowDefinitionData definition);
-
-  Map<K, S> status(WorkflowDefinitionData workflowContext);
-
-  Map<String, T> tasks(K instanceId);
-
-  void cleanupTasks(K instanceId);
+  BigMapInstanceTransaction<K, V, T, S> begin();
 }

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceTransaction.java

@@ -0,0 +1,34 @@
+/*
+ * 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.bigmap;
+
+import io.serverlessworkflow.impl.WorkflowDefinitionData;
+import java.util.Map;
+
+public interface BigMapInstanceTransaction<K, V, T, S> {
+
+  Map<K, V> instanceData(WorkflowDefinitionData definition);
+
+  Map<K, S> status(WorkflowDefinitionData workflowContext);
+
+  Map<String, T> tasks(K instanceId);
+
+  void cleanupTasks(K instanceId);
+
+  void commit();
+
+  void rollback();
+}

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BigMapInstanceWriter.java

@@ -21,6 +21,7 @@
 import io.serverlessworkflow.impl.WorkflowInstanceData;
 import io.serverlessworkflow.impl.WorkflowStatus;
 import io.serverlessworkflow.impl.persistence.PersistenceInstanceWriter;
+import java.util.function.Consumer;
 
 public abstract class BigMapInstanceWriter<K, V, T, S> implements PersistenceInstanceWriter {
 
@@ -30,11 +31,23 @@ protected BigMapInstanceWriter(BigMapInstanceStore<K, V, T, S> store) {
     this.store = store;
   }
 
+  private void doTransaction(Consumer<BigMapInstanceTransaction<K, V, T, S>> operations) {
+    BigMapInstanceTransaction<K, V, T, S> transaction = store.begin();
+    try {
+      operations.accept(transaction);
+      transaction.commit();
+    } catch (Exception ex) {
+      transaction.rollback();
+      throw ex;
+    }
+  }
+
   @Override
   public void started(WorkflowContextData workflowContext) {
-    store
-        .instanceData(workflowContext.definition())
-        .put(key(workflowContext), marshallInstance(workflowContext.instanceData()));
+    doTransaction(
+        t ->
+            t.instanceData(workflowContext.definition())
+                .put(key(workflowContext), marshallInstance(workflowContext.instanceData())));
   }
 
   @Override
@@ -57,33 +70,35 @@ public void taskStarted(WorkflowContextData workflowContext, TaskContextData tas
 
   @Override
   public void taskCompleted(WorkflowContextData workflowContext, TaskContextData taskContext) {
-    K key = key(workflowContext);
-    store
-        .tasks(key)
-        .put(
-            taskContext.position().jsonPointer(),
-            marshallTaskCompleted(workflowContext, (TaskContext) taskContext));
+    doTransaction(
+        t ->
+            t.tasks(key(workflowContext))
+                .put(
+                    taskContext.position().jsonPointer(),
+                    marshallTaskCompleted(workflowContext, (TaskContext) taskContext)));
   }
 
   @Override
   public void suspended(WorkflowContextData workflowContext) {
-    store
-        .status(workflowContext.definition())
-        .put(key(workflowContext), marshallStatus(WorkflowStatus.SUSPENDED));
+    doTransaction(
+        t ->
+            t.status(workflowContext.definition())
+                .put(key(workflowContext), marshallStatus(WorkflowStatus.SUSPENDED)));
   }
 
   @Override
   public void resumed(WorkflowContextData workflowContext) {
-    store
-        .status(workflowContext.definition())
-        .put(key(workflowContext), marshallStatus(WorkflowStatus.RUNNING));
+    doTransaction(t -> t.status(workflowContext.definition()).remove(key(workflowContext)));
   }
 
   protected void removeProcessInstance(WorkflowContextData workflowContext) {
-    K key = key(workflowContext);
-    store.instanceData(workflowContext.definition()).remove(key);
-    store.status(workflowContext.definition()).remove(key);
-    store.cleanupTasks(key);
+    doTransaction(
+        t -> {
+          K key = key(workflowContext);
+          t.instanceData(workflowContext.definition()).remove(key);
+          t.status(workflowContext.definition()).remove(key);
+          t.cleanupTasks(key);
+        });
   }
 
   protected abstract K key(WorkflowContextData workflowContext);

impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStore.java

@@ -15,61 +15,26 @@
  */
 package io.serverlessworkflow.impl.persistence.mvstore;
 
-import io.serverlessworkflow.api.types.Document;
-import io.serverlessworkflow.api.types.Workflow;
-import io.serverlessworkflow.impl.WorkflowDefinitionData;
 import io.serverlessworkflow.impl.persistence.bigmap.BigMapInstanceStore;
-import java.util.Map;
+import io.serverlessworkflow.impl.persistence.bigmap.BigMapInstanceTransaction;
 import org.h2.mvstore.MVStore;
+import org.h2.mvstore.tx.TransactionStore;
 
 public class MVStorePersistenceStore
     implements BigMapInstanceStore<String, byte[], byte[], byte[]> {
-  private final MVStore mvStore;
-
-  protected static final String ID_SEPARATOR = "-";
+  private final TransactionStore mvStore;
 
   public MVStorePersistenceStore(String dbName) {
-    this.mvStore = MVStore.open(dbName);
-  }
-
-  protected static String identifier(Workflow workflow, String sep) {
-    Document document = workflow.getDocument();
-    return document.getNamespace() + sep + document.getName() + sep + document.getVersion();
+    this.mvStore = new TransactionStore(MVStore.open(dbName));
   }
 
   @Override
   public void close() {
-    if (!mvStore.isClosed()) {
-      mvStore.close();
-    }
-  }
-
-  @Override
-  public Map<String, byte[]> instanceData(WorkflowDefinitionData workflowContext) {
-    return openMap(workflowContext, "instances");
-  }
-
-  @Override
-  public Map<String, byte[]> tasks(String instanceId) {
-    return mvStore.openMap(mapTaskName(instanceId));
-  }
-
-  @Override
-  public Map<String, byte[]> status(WorkflowDefinitionData workflowContext) {
-    return openMap(workflowContext, "status");
-  }
-
-  private Map<String, byte[]> openMap(WorkflowDefinitionData workflowDefinition, String suffix) {
-    return mvStore.openMap(
-        identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix);
-  }
-
-  private String mapTaskName(String instanceId) {
-    return instanceId + ID_SEPARATOR + "tasks";
+    mvStore.close();
   }
 
   @Override
-  public void cleanupTasks(String instanceId) {
-    mvStore.removeMap(mapTaskName(instanceId));
+  public BigMapInstanceTransaction<String, byte[], byte[], byte[]> begin() {
+    return new MVStoreTransaction(mvStore.begin());
   }
 }

impl/persistence/mvstore/src/main/java/io/serverlessworkflow/impl/persistence/mvstore/MVStoreTransaction.java

@@ -0,0 +1,84 @@
+/*
+ * 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.api.types.Document;
+import io.serverlessworkflow.api.types.Workflow;
+import io.serverlessworkflow.impl.WorkflowDefinitionData;
+import io.serverlessworkflow.impl.persistence.bigmap.BigMapInstanceTransaction;
+import java.util.Map;
+import org.h2.mvstore.tx.Transaction;
+import org.h2.mvstore.tx.TransactionMap;
+
+public class MVStoreTransaction
+    implements BigMapInstanceTransaction<String, byte[], byte[], byte[]> {
+
+  protected static final String ID_SEPARATOR = "-";
+
+  private final Transaction transaction;
+
+  public MVStoreTransaction(Transaction transaction) {
+    this.transaction = transaction;
+  }
+
+  protected static String identifier(Workflow workflow, String sep) {
+    Document document = workflow.getDocument();
+    return document.getNamespace() + sep + document.getName() + sep + document.getVersion();
+  }
+
+  @Override
+  public Map<String, byte[]> instanceData(WorkflowDefinitionData workflowContext) {
+    return openMap(workflowContext, "instances");
+  }
+
+  @Override
+  public Map<String, byte[]> tasks(String instanceId) {
+    return taskMap(instanceId);
+  }
+
+  @Override
+  public Map<String, byte[]> status(WorkflowDefinitionData workflowContext) {
+    return openMap(workflowContext, "status");
+  }
+
+  @Override
+  public void cleanupTasks(String instanceId) {
+    transaction.removeMap(taskMap(instanceId));
+  }
+
+  private TransactionMap<String, byte[]> taskMap(String instanceId) {
+    return transaction.openMap(mapTaskName(instanceId));
+  }
+
+  private Map<String, byte[]> openMap(WorkflowDefinitionData workflowDefinition, String suffix) {
+    return transaction.openMap(
+        identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix);
+  }
+
+  private String mapTaskName(String instanceId) {
+    return instanceId + ID_SEPARATOR + "tasks";
+  }
+
+  @Override
+  public void commit() {
+    transaction.commit();
+  }
+
+  @Override
+  public void rollback() {
+    transaction.rollback();
+  }
+}