[Fix #1691] Add metadata to mvstore

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java

@@ -53,7 +53,7 @@ public class WorkflowMutableInstance implements WorkflowInstance {
   protected AtomicReference<CompletableFuture<WorkflowModel>> futureRef = new AtomicReference<>();
   protected Instant completedAt;
 
-  protected final Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
+  protected Map<String, Object> additionalObjects = new ConcurrentHashMap<>();
 
   protected final Map<String, Integer> iterationsMap = new ConcurrentHashMap<>();
 
@@ -448,6 +448,10 @@ public void addCancelable(CompletableFuture<?> cancelable) {
     }
   }
 
+  protected void setMetadata(Map<String, Object> metadata) {
+    this.additionalObjects = metadata;
+  }
+
   @Override
   public <T> T addMetadataIfAbsent(String key, Supplier<T> supplier) {
     return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get());

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java

@@ -122,6 +122,16 @@ public WorkflowOutputBuffer writeBytes(byte[] bytes) {
     return this;
   }
 
+  @Override
+  public WorkflowOutputBuffer writeRawBytes(byte[] bytes) {
+    try {
+      output.write(bytes);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+    return this;
+  }
+
   @Override
   public void close() {
     try {

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java

@@ -173,6 +173,10 @@ public static WorkflowModel readModel(WorkflowBufferFactory factory, byte[] valu
     return readValue(factory, value, b -> (WorkflowModel) b.readObject());
   }
 
+  public static Object readObject(WorkflowBufferFactory factory, byte[] value) {
+    return readValue(factory, value, b -> b.readObject());
+  }
+
   public static Instant readInstant(WorkflowBufferFactory factory, byte[] value) {
     return readValue(factory, value, WorkflowInputBuffer::readInstant);
   }

impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java

@@ -41,6 +41,8 @@ public interface WorkflowOutputBuffer extends AutoCloseable {
 
   WorkflowOutputBuffer writeBytes(byte[] bytes);
 
+  WorkflowOutputBuffer writeRawBytes(byte[] bytes);
+
   WorkflowOutputBuffer writeInstant(Instant instant);
 
   default WorkflowOutputBuffer writeOffsetDateTime(OffsetDateTime time) {

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CompletedTaskInfo.java

@@ -17,14 +17,16 @@
 
 import io.serverlessworkflow.impl.WorkflowModel;
 import java.time.Instant;
+import java.util.Map;
 
 public record CompletedTaskInfo(
     Instant instant,
     WorkflowModel model,
     WorkflowModel context,
     Boolean isEndNode,
     String nextPosition,
-    int iteration)
+    int iteration,
+    Map<String, Object> additionalObjects)
     implements PersistenceTaskInfo {
 
   public CompletedTaskInfo(
@@ -35,4 +37,14 @@ public CompletedTaskInfo(
       String nextPosition) {
     this(instant, model, context, isEndNode, nextPosition, 1);
   }
+
+  public CompletedTaskInfo(
+      Instant instant,
+      WorkflowModel model,
+      WorkflowModel context,
+      Boolean isEndNode,
+      String nextPosition,
+      int iteration) {
+    this(instant, model, context, isEndNode, nextPosition, iteration, Map.of());
+  }
 }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/WorkflowPersistenceInstance.java

@@ -83,6 +83,7 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) {
                   : workflow.definition().taskExecutor(completedTaskInfo.nextPosition()),
               completedTaskInfo.isEndNode()));
       workflow.context(completedTaskInfo.context());
+      setMetadata(completedTaskInfo.additionalObjects());
     } else if (taskInfo instanceof RetriedTaskInfo retriedTaskInfo) {
       if (context.retryAttempt() == 0) {
         context.retryAttempt(retriedTaskInfo.retryAttempt());

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/AbstractHashFactory.java

@@ -0,0 +1,32 @@
+/*
+ * 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.hashing;
+
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import java.util.Optional;
+
+public abstract class AbstractHashFactory implements HashFactory {
+
+  @Override
+  public Optional<HashItem> fromBuffer(byte id, WorkflowInputBuffer buffer) {
+    return Optional.ofNullable(
+        switch (id) {
+          case MD5HashItem.ID -> new MD5HashItem(buffer);
+          case IntegerHashItem.ID -> new IntegerHashItem(buffer);
+          default -> null;
+        });
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DisabledHashFactory.java

@@ -0,0 +1,32 @@
+/*
+ * 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.hashing;
+
+import java.util.Optional;
+
+public class DisabledHashFactory extends AbstractHashFactory {
+
+  private static final HashFactory instance = new DisabledHashFactory();
+
+  public static HashFactory instance() {
+    return instance;
+  }
+
+  @Override
+  public Optional<HashItem> fromData(byte[] data) {
+    return Optional.empty();
+  }
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java

@@ -0,0 +1,26 @@
+/*
+ * 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.hashing;
+
+import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer;
+import java.util.Optional;
+
+public interface HashFactory {
+
+  Optional<HashItem> fromData(byte[] data);
+
+  Optional<HashItem> fromBuffer(byte id, WorkflowInputBuffer buffer);
+}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashItem.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.hashing;
+
+import io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer;
+
+public interface HashItem {
+
+  final byte HASHING_DISABLED = 0;
+
+  /**
+   * byte identifying this hash item type
+   *
+   * @return any byte except 0, which is reserved value
+   */
+  byte id();
+
+  void writeKey(WorkflowOutputBuffer buffer);
+
+  String key();
+}