From 037b79a6834c2ca767ad4228855cb593540d9c5c Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Wed, 23 Sep 2026 17:13:36 +0200 Subject: [PATCH 1/4] [Fix #1691] Add metadata to mvstore Signed-off-by: Francisco Javier Tirado Sarti --- .../impl/WorkflowMutableInstance.java | 6 +- .../impl/marshaller/DefaultOutputBuffer.java | 10 + .../impl/marshaller/MarshallingUtils.java | 4 + .../impl/marshaller/WorkflowOutputBuffer.java | 2 + .../impl/persistence/CompletedTaskInfo.java | 14 +- .../persistence/PersistenceInstanceInfo.java | 9 +- .../persistence/PersistenceWorkflowInfo.java | 13 +- .../impl/persistence/RetriedTaskInfo.java | 9 +- .../WorkflowPersistenceInstance.java | 3 + .../hashing/DefaultHashFactory.java | 52 +++++ .../impl/persistence/hashing/HashFactory.java | 26 +++ .../impl/persistence/hashing/HashItem.java | 33 ++++ .../persistence/hashing/IntegerHashItem.java | 51 +++++ .../impl/persistence/hashing/MD5HashItem.java | 63 ++++++ .../bigmap/BigMapInstanceTransaction.java | 11 +- .../bigmap/BytesMapInstanceTransaction.java | 187 +++++++++++++++--- .../mvstore/MVStorePersistenceStore.java | 20 +- .../mvstore/MVStoreTransaction.java | 70 +++++-- .../HashingMVStorePersistenceTest.java | 37 ++++ .../mvstore/MD5MVStorePersistenceTest.java | 37 ++++ ...onHashingMVStorePersistenceStoreTest.java} | 21 +- .../test/AbstractHandlerPersistenceTest.java | 2 + 22 files changed, 621 insertions(+), 59 deletions(-) create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashItem.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java create mode 100644 impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java create mode 100644 impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java rename impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/{MVStorePersistenceStoreTest.java => NonHashingMVStorePersistenceStoreTest.java} (63%) diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java index 0783b7565..16562dbbe 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java +++ b/impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowMutableInstance.java @@ -53,7 +53,7 @@ public class WorkflowMutableInstance implements WorkflowInstance { protected AtomicReference> futureRef = new AtomicReference<>(); protected Instant completedAt; - protected final Map additionalObjects = new ConcurrentHashMap<>(); + private Map additionalObjects = new ConcurrentHashMap<>(); protected final Map iterationsMap = new ConcurrentHashMap<>(); @@ -448,6 +448,10 @@ public void addCancelable(CompletableFuture cancelable) { } } + protected void setMetadata(Map metadata) { + this.additionalObjects = new ConcurrentHashMap<>(metadata); + } + @Override public T addMetadataIfAbsent(String key, Supplier supplier) { return (T) additionalObjects.computeIfAbsent(key, k -> supplier.get()); diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java index 1d5192790..27b19f885 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/DefaultOutputBuffer.java +++ b/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 { diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java b/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java index 760ce91c6..c765798fe 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/MarshallingUtils.java +++ b/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); } diff --git a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java b/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java index 59d22ef4f..63d6b2b6b 100644 --- a/impl/core/src/main/java/io/serverlessworkflow/impl/marshaller/WorkflowOutputBuffer.java +++ b/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) { diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CompletedTaskInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CompletedTaskInfo.java index b4eb9760a..3a03e66a5 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CompletedTaskInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/CompletedTaskInfo.java @@ -17,6 +17,7 @@ import io.serverlessworkflow.impl.WorkflowModel; import java.time.Instant; +import java.util.Map; public record CompletedTaskInfo( Instant instant, @@ -24,7 +25,8 @@ public record CompletedTaskInfo( WorkflowModel context, Boolean isEndNode, String nextPosition, - int iteration) + int iteration, + Map 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()); + } } diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceInfo.java index a4a3a8350..1d0bdcdab 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceInstanceInfo.java @@ -17,5 +17,12 @@ import io.serverlessworkflow.impl.WorkflowModel; import java.time.Instant; +import java.util.Map; -public record PersistenceInstanceInfo(Instant startedAt, WorkflowModel input) {} +public record PersistenceInstanceInfo( + Instant startedAt, WorkflowModel input, Map metadata) { + + public PersistenceInstanceInfo(Instant startedAt, WorkflowModel input) { + this(startedAt, input, Map.of()); + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceWorkflowInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceWorkflowInfo.java index 0c5d27bf2..bb21fad2f 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceWorkflowInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceWorkflowInfo.java @@ -25,4 +25,15 @@ public record PersistenceWorkflowInfo( Instant startedAt, WorkflowModel input, WorkflowStatus status, - Map tasks) {} + Map tasks, + Map metadata) { + + public PersistenceWorkflowInfo( + String id, + Instant startedAt, + WorkflowModel input, + WorkflowStatus status, + Map tasks) { + this(id, startedAt, input, status, tasks, Map.of()); + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/RetriedTaskInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/RetriedTaskInfo.java index e01130768..2b32c63ec 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/RetriedTaskInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/RetriedTaskInfo.java @@ -15,4 +15,11 @@ */ package io.serverlessworkflow.impl.persistence; -public record RetriedTaskInfo(int retryAttempt) implements PersistenceTaskInfo {} +import java.util.Map; + +public record RetriedTaskInfo(int retryAttempt, Map metadata) + implements PersistenceTaskInfo { + public RetriedTaskInfo(int retryAttempt) { + this(retryAttempt, Map.of()); + } +} 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 index b132c937d..a64fede81 100644 --- 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 @@ -48,6 +48,7 @@ private WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWo } }); this.startedAt = info.startedAt(); + setMetadata(info.metadata()); } @Override @@ -83,6 +84,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()); @@ -98,6 +100,7 @@ public void restoreContext(WorkflowContext workflow, TaskContext context) { } searchContext = tryContext.parent(); } + setMetadata(retriedTaskInfo.metadata()); } } } diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java new file mode 100644 index 000000000..5498200c7 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java @@ -0,0 +1,52 @@ +/* + * 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 class DefaultHashFactory implements HashFactory { + + @Override + public Optional fromBuffer(byte id, WorkflowInputBuffer buffer) { + return Optional.ofNullable( + switch (id) { + case MD5HashItem.ID -> new MD5HashItem(buffer); + case IntegerHashItem.ID -> new IntegerHashItem(buffer); + case HashItem.HASHING_DISABLED -> null; + default -> throw new UnsupportedOperationException("Unsupported id " + id); + }); + } + + @Override + public Optional fromData(byte[] data) { + if (md5Condition(data)) { + return Optional.of(new MD5HashItem(data)); + } else if (intCondition(data)) { + return Optional.of(new IntegerHashItem(data)); + } else { + return Optional.empty(); + } + } + + protected boolean intCondition(byte[] data) { + return data.length > IntegerHashItem.SIZE_THRESHOLD; + } + + protected boolean md5Condition(byte[] data) { + return data.length > MD5HashItem.SIZE_THRESHOLD; + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java new file mode 100644 index 000000000..b580b311f --- /dev/null +++ b/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 fromData(byte[] data); + + Optional fromBuffer(byte id, WorkflowInputBuffer buffer); +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashItem.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashItem.java new file mode 100644 index 000000000..cedea8a96 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashItem.java @@ -0,0 +1,33 @@ +/* + * 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(); +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java new file mode 100644 index 000000000..ccd4ddf86 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java @@ -0,0 +1,51 @@ +/* + * 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 io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer; +import java.util.Arrays; + +public class IntegerHashItem implements HashItem { + + public static final byte ID = 1; + public static final int SIZE_THRESHOLD = 4 * 4; + + private final int hashCode; + + public IntegerHashItem(WorkflowInputBuffer input) { + this.hashCode = input.readInt(); + } + + public IntegerHashItem(byte[] data) { + this.hashCode = Arrays.hashCode(data); + } + + @Override + public byte id() { + return ID; + } + + @Override + public void writeKey(WorkflowOutputBuffer buffer) { + buffer.writeInt(hashCode); + } + + @Override + public String key() { + return Integer.toString(hashCode); + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java new file mode 100644 index 000000000..1a37103e1 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java @@ -0,0 +1,63 @@ +/* + * 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 io.serverlessworkflow.impl.marshaller.WorkflowOutputBuffer; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; + +public class MD5HashItem implements HashItem { + + public static final byte ID = 2; + public static final int SIZE_THRESHOLD = 128 * 2; + + @Override + public byte id() { + return ID; + } + + private final byte[] hashCode; + + public MD5HashItem(WorkflowInputBuffer input) { + this.hashCode = input.readBytes(); + } + + public MD5HashItem(byte[] data) { + try { + this.hashCode = MessageDigest.getInstance("MD5").digest(data); + } catch (NoSuchAlgorithmException ex) { + throw new IllegalStateException(ex); + } + } + + @Override + public void writeKey(WorkflowOutputBuffer buffer) { + buffer.writeBytes(hashCode); + } + + private static final char[] hexCode = "0123456789ABCDEF".toCharArray(); + + @Override + public String key() { + StringBuilder r = new StringBuilder(hashCode.length * 2); + for (byte b : hashCode) { + r.append(hexCode[(b >> 4) & 0xF]); + r.append(hexCode[(b & 0xF)]); + } + return r.toString(); + } +} 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 index 92a2d808f..635d392e0 100644 --- 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 @@ -155,7 +155,7 @@ private void clearStatus(WorkflowDefinitionData definition, String key) { protected PersistenceWorkflowInfo readPersistenceInfo( String instanceId, V instanceData, Map tasksData, S status) { - PersistenceInstanceInfo instanceInfo = unmarshallInstanceInfo(instanceData); + PersistenceInstanceInfo instanceInfo = unmarshallInstanceInfo(instanceId, instanceData); return new PersistenceWorkflowInfo( instanceId, instanceInfo.startedAt(), @@ -163,7 +163,9 @@ protected PersistenceWorkflowInfo readPersistenceInfo( status == null ? null : unmarshallStatus(status), tasksData.entrySet().stream() .collect( - Collectors.toMap(Entry::getKey, entry -> unmarshallTaskInfo(entry.getValue())))); + Collectors.toMap( + Entry::getKey, entry -> unmarshallTaskInfo(instanceId, entry.getValue()))), + instanceInfo.metadata()); } private String key(WorkflowContextData workflowContext) { @@ -200,9 +202,10 @@ protected abstract T marshallTaskRetried( protected abstract S marshallStatus(WorkflowStatus status); - protected abstract PersistenceTaskInfo unmarshallTaskInfo(T taskData); + protected abstract PersistenceTaskInfo unmarshallTaskInfo(String instanceId, T taskData); - protected abstract PersistenceInstanceInfo unmarshallInstanceInfo(V instanceData); + protected abstract PersistenceInstanceInfo unmarshallInstanceInfo( + String instanceId, V instanceData); protected abstract WorkflowStatus unmarshallStatus(S statusData); 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 index 33ce047da..70508c74a 100644 --- 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 @@ -34,12 +34,19 @@ import io.serverlessworkflow.impl.persistence.PersistenceInstanceInfo; import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo; import io.serverlessworkflow.impl.persistence.RetriedTaskInfo; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashItem; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.UncheckedIOException; import java.net.URI; import java.time.OffsetDateTime; +import java.util.Arrays; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.locks.Lock; public abstract class BytesMapInstanceTransaction extends BigMapInstanceTransaction { @@ -47,24 +54,40 @@ public abstract class BytesMapInstanceTransaction private static final byte VERSION_0 = 0; private static final byte VERSION_1 = 1; private static final byte VERSION_2 = 2; - private static final byte[] PROCESSED_VALUE = new byte[] {1}; + private static final byte VERSION_3 = 3; + private static final byte[] PROCESSED_VALUE = {1}; - private final WorkflowBufferFactory factory; + private final WorkflowBufferFactory bufferFactory; + private final HashFactory hashFactory; + private final Lock hashLock; - protected BytesMapInstanceTransaction(WorkflowBufferFactory factory) { - this.factory = factory; + protected BytesMapInstanceTransaction( + WorkflowBufferFactory factory, HashFactory hashFactory, Lock hashLock) { + this.bufferFactory = factory; + this.hashFactory = hashFactory; + this.hashLock = hashLock; + } + + protected abstract Map blobData(String instanceId); + + protected abstract void removeBlobData(String instanceId); + + @Override + public void removeProcessInstance(WorkflowContextData workflowContext) { + super.removeProcessInstance(workflowContext); + removeBlobData(workflowContext.instanceData().id()); } @Override protected byte[] marshallTaskCompleted(WorkflowContextData contextData, TaskContext taskContext) { try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - WorkflowOutputBuffer writer = factory.output(bytes)) { - writer.writeByte(VERSION_2); + WorkflowOutputBuffer writer = bufferFactory.output(bytes)) { + writer.writeByte(VERSION_3); writer.writeEnum(TaskStatus.COMPLETED); writer.writeInstant(taskContext.completedAt()); - writeModel(writer, taskContext.output()); - writeModel(writer, contextData.context()); + writeLargeObject(contextData.instanceData(), writer, taskContext.output()); + writeLargeObject(contextData.instanceData(), writer, contextData.context()); TransitionInfo transition = taskContext.transition(); writer.writeBoolean(transition.isEndNode()); AbstractTaskExecutor next = (AbstractTaskExecutor) transition.next(); @@ -75,16 +98,37 @@ protected byte[] marshallTaskCompleted(WorkflowContextData contextData, TaskCont writer.writeString(next.position().jsonPointer()); } writer.writeInt(taskContext.iteration()); + writeMetadata(contextData.instanceData(), writer); return bytes.toByteArray(); } catch (IOException e) { throw new UncheckedIOException(e); } } + private void writeMetadata(WorkflowInstanceData instanceData, WorkflowOutputBuffer writer) { + Map additionalObjects = new HashMap<>(instanceData.metadata()); + writer.writeInt(additionalObjects.size()); + additionalObjects.forEach( + (k, v) -> { + writer.writeString(k); + writeLargeObject(instanceData, writer, v); + }); + } + + private void writeLargeObject( + WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, Object obj) { + final byte[] bytes = MarshallingUtils.writeObject(bufferFactory, obj); + hashFactory + .fromData(bytes) + .ifPresentOrElse( + item -> writeLargeObject(item, instanceData, writer, bytes), + () -> legacyWriteObject(writer, bytes)); + } + @Override protected byte[] marshallStatus(WorkflowStatus status) { try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - WorkflowOutputBuffer writer = factory.output(bytes)) { + WorkflowOutputBuffer writer = bufferFactory.output(bytes)) { writer.writeByte(VERSION_0); writer.writeEnum(status); return bytes.toByteArray(); @@ -95,45 +139,42 @@ protected byte[] marshallStatus(WorkflowStatus status) { @Override protected byte[] marshallInstance(WorkflowInstanceData instance) { - try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - WorkflowOutputBuffer writer = factory.output(bytes)) { - writer.writeByte(VERSION_0); + WorkflowOutputBuffer writer = bufferFactory.output(bytes)) { + writer.writeByte(VERSION_1); writer.writeInstant(instance.startedAt()); - writeModel(writer, instance.input()); + writer.writeObject(instance.input()); + writeMetadata(instance, writer); return bytes.toByteArray(); } catch (IOException e) { throw new UncheckedIOException(e); } } - protected void writeModel(WorkflowOutputBuffer writer, WorkflowModel model) { - writer.writeObject(model); - } - protected byte[] marshallApplicationId(String id) { - return MarshallingUtils.writeString(factory, id); + return MarshallingUtils.writeString(bufferFactory, id); } protected String unmarshallApplicationId(byte[] value) { - return MarshallingUtils.readString(factory, value); + return MarshallingUtils.readString(bufferFactory, value); } @Override protected byte[] marshallTaskRetried( WorkflowContextData workflowContext, TaskContext taskContext) { ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - try (WorkflowOutputBuffer writer = factory.output(bytes)) { - writer.writeByte(VERSION_2); + try (WorkflowOutputBuffer writer = bufferFactory.output(bytes)) { + writer.writeByte(VERSION_3); writer.writeEnum(TaskStatus.RETRIED); writer.writeInt(taskContext.retryAttempt()); + writeMetadata(workflowContext.instanceData(), writer); } return bytes.toByteArray(); } @Override - protected PersistenceTaskInfo unmarshallTaskInfo(byte[] taskData) { - try (WorkflowInputBuffer buffer = factory.input(new ByteArrayInputStream(taskData))) { + protected PersistenceTaskInfo unmarshallTaskInfo(String instanceId, byte[] taskData) { + try (WorkflowInputBuffer buffer = bufferFactory.input(new ByteArrayInputStream(taskData))) { byte version = buffer.readByte(); switch (version) { case VERSION_0: @@ -142,11 +183,40 @@ protected PersistenceTaskInfo unmarshallTaskInfo(byte[] taskData) { return readVersion1(buffer); case VERSION_2: return readVersion2(buffer); + case VERSION_3: + return readVersion3(instanceId, buffer); } throw new UnsupportedOperationException("Unknown version " + version); } } + private PersistenceTaskInfo readVersion3(String instanceId, WorkflowInputBuffer buffer) { + TaskStatus taskStatus = buffer.readEnum(TaskStatus.class); + switch (taskStatus) { + case COMPLETED: + return new CompletedTaskInfo( + buffer.readInstant(), + (WorkflowModel) readLargeObject(instanceId, buffer), + (WorkflowModel) readLargeObject(instanceId, buffer), + buffer.readBoolean(), + buffer.readBoolean() ? buffer.readString() : null, + buffer.readInt(), + readMetadata(instanceId, buffer)); + case RETRIED: + return new RetriedTaskInfo(buffer.readInt(), readMetadata(instanceId, buffer)); + } + throw new UnsupportedOperationException("Unknown status " + taskStatus); + } + + private Map readMetadata(String instanceId, WorkflowInputBuffer buffer) { + int size = buffer.readInt(); + Map map = new HashMap<>(size); + while (size-- > 0) { + map.put(buffer.readString(), readLargeObject(instanceId, buffer)); + } + return map; + } + private PersistenceTaskInfo readVersion2(WorkflowInputBuffer buffer) { TaskStatus taskStatus = buffer.readEnum(TaskStatus.class); switch (taskStatus) { @@ -184,17 +254,74 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) { buffer.readBoolean() ? buffer.readString() : null); } + protected abstract T executeNewTransaction(Callable runnable); + + private void writeLargeObject( + HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) { + + boolean duplicated; + String key = item.key(); + hashLock.lock(); + try { + duplicated = + executeNewTransaction( + () -> { + Map map = blobData(instanceData.id()); + byte[] value = map.get(key); + if (value == null) { + map.put(key, bytes); + return false; + } else { + return !Arrays.equals(value, bytes); + } + }); + } finally { + hashLock.unlock(); + } + + if (duplicated) { + legacyWriteObject(writer, bytes); + } else { + writer.writeByte(item.id()); + item.writeKey(writer); + } + } + + private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { + return hashFactory + .fromBuffer(buffer.readByte(), buffer) + .map( + item -> { + Map map = blobData(instanceId); + ByteArrayInputStream bytes = new ByteArrayInputStream(map.get(item.key())); + try (WorkflowInputBuffer input = bufferFactory.input(bytes)) { + return input.readObject(); + } + }) + .orElseGet(() -> buffer.readObject()); + } + + private void legacyWriteObject(WorkflowOutputBuffer writer, byte[] bytes) { + writer.writeByte(HashItem.HASHING_DISABLED).writeRawBytes(bytes); + } + @Override - protected PersistenceInstanceInfo unmarshallInstanceInfo(byte[] instanceData) { - try (WorkflowInputBuffer buffer = factory.input(new ByteArrayInputStream(instanceData))) { - buffer.readByte(); // version byte not used at the moment - return new PersistenceInstanceInfo(buffer.readInstant(), (WorkflowModel) buffer.readObject()); + protected PersistenceInstanceInfo unmarshallInstanceInfo(String instanceId, byte[] instanceData) { + try (WorkflowInputBuffer buffer = bufferFactory.input(new ByteArrayInputStream(instanceData))) { + byte version = buffer.readByte(); // version byte not used at the moment + + return version == VERSION_1 + ? new PersistenceInstanceInfo( + buffer.readInstant(), + (WorkflowModel) buffer.readObject(), + readMetadata(instanceId, buffer)) + : new PersistenceInstanceInfo(buffer.readInstant(), (WorkflowModel) buffer.readObject()); } } @Override protected WorkflowStatus unmarshallStatus(byte[] statusData) { - try (WorkflowInputBuffer buffer = factory.input(new ByteArrayInputStream(statusData))) { + try (WorkflowInputBuffer buffer = bufferFactory.input(new ByteArrayInputStream(statusData))) { buffer.readByte(); // version byte not used at the moment return buffer.readEnum(WorkflowStatus.class); } @@ -202,7 +329,7 @@ protected WorkflowStatus unmarshallStatus(byte[] statusData) { protected byte[] marshallCloudEvent(CloudEvent event) { try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(); - WorkflowOutputBuffer writer = factory.output(bytes)) { + WorkflowOutputBuffer writer = bufferFactory.output(bytes)) { writer.writeEnum(event.getSpecVersion()); writer.writeString(event.getId()); writer.writeString(event.getType()); @@ -221,7 +348,7 @@ protected byte[] marshallCloudEvent(CloudEvent event) { protected CloudEvent unmarshallCloudEvent(byte[] eventData) { try (ByteArrayInputStream bytes = new ByteArrayInputStream(eventData); - WorkflowInputBuffer reader = factory.input(bytes)) { + WorkflowInputBuffer reader = bufferFactory.input(bytes)) { CloudEventBuilder builder = CloudEventBuilder.fromSpecVersion(reader.readEnum(SpecVersion.class)); builder.withId(reader.readString()); 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 index 298115a7c..deb5cbc3b 100644 --- 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 @@ -19,22 +19,34 @@ import io.serverlessworkflow.impl.marshaller.WorkflowBufferFactory; import io.serverlessworkflow.impl.persistence.PersistenceInstanceStore; import io.serverlessworkflow.impl.persistence.bigmap.BigMapInstanceTransaction; +import io.serverlessworkflow.impl.persistence.hashing.DefaultHashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.h2.mvstore.MVStore; import org.h2.mvstore.tx.TransactionStore; public class MVStorePersistenceStore implements PersistenceInstanceStore { private final TransactionStore transactionStore; private final MVStore mvStore; - private WorkflowBufferFactory factory; + private final Lock hashLock = new ReentrantLock(); + private final WorkflowBufferFactory bufferFactory; + private final HashFactory hashFactory; public MVStorePersistenceStore(String dbName) { this(dbName, DefaultBufferFactory.factory()); } - public MVStorePersistenceStore(String dbName, WorkflowBufferFactory factory) { + public MVStorePersistenceStore(String dbName, WorkflowBufferFactory bufferFactory) { + this(dbName, bufferFactory, new DefaultHashFactory()); + } + + public MVStorePersistenceStore( + String dbName, WorkflowBufferFactory bufferFactory, HashFactory hashFactory) { this.mvStore = MVStore.open(dbName); this.transactionStore = new TransactionStore(mvStore); - this.factory = factory; + this.bufferFactory = bufferFactory; + this.hashFactory = hashFactory; } @Override @@ -44,6 +56,6 @@ public void close() { @Override public BigMapInstanceTransaction begin() { - return new MVStoreTransaction(mvStore, transactionStore.begin(), factory); + return new MVStoreTransaction(mvStore, transactionStore, bufferFactory, hashFactory, hashLock); } } 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 index aa2c0c83b..ea839c0b8 100644 --- 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 @@ -20,22 +20,35 @@ import io.serverlessworkflow.impl.WorkflowDefinitionData; import io.serverlessworkflow.impl.marshaller.WorkflowBufferFactory; import io.serverlessworkflow.impl.persistence.bigmap.BytesMapInstanceTransaction; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; +import java.util.ArrayDeque; +import java.util.Deque; import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.locks.Lock; import org.h2.mvstore.MVStore; import org.h2.mvstore.tx.Transaction; import org.h2.mvstore.tx.TransactionMap; +import org.h2.mvstore.tx.TransactionStore; public class MVStoreTransaction extends BytesMapInstanceTransaction { protected static final String ID_SEPARATOR = "-"; private static final String PROCESSED_PREFIX = "PROCESSED" + ID_SEPARATOR; - private final Transaction transaction; + private final TransactionStore transactionStore; + private final Deque transactionStack = new ArrayDeque<>(); private final MVStore store; - public MVStoreTransaction(MVStore store, Transaction transaction, WorkflowBufferFactory factory) { - super(factory); - this.transaction = transaction; + public MVStoreTransaction( + MVStore store, + TransactionStore transactionStore, + WorkflowBufferFactory bufferFactory, + HashFactory hashFactory, + Lock hashLock) { + super(bufferFactory, hashFactory, hashLock); + this.transactionStore = transactionStore; + transactionStack.add(transactionStore.begin()); this.store = store; } @@ -59,18 +72,22 @@ public Map status(WorkflowDefinitionData workflowContext) { return openMap(workflowContext, "status"); } + private Transaction currentTransaction() { + return transactionStack.peekLast(); + } + @Override public void removeTasks(String instanceId) { - transaction.removeMap(taskMap(instanceId)); + currentTransaction().removeMap(taskMap(instanceId)); } private TransactionMap taskMap(String instanceId) { - return transaction.openMap(mapTaskName(instanceId)); + return currentTransaction().openMap(mapTaskName(instanceId)); } private Map openMap(WorkflowDefinitionData workflowDefinition, String suffix) { - return transaction.openMap( - identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix); + return currentTransaction() + .openMap(identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix); } private String mapTaskName(String instanceId) { @@ -79,33 +96,58 @@ private String mapTaskName(String instanceId) { @Override public void commit(WorkflowDefinitionData definition) { - transaction.commit(); + currentTransaction().commit(); } @Override public void rollback(WorkflowDefinitionData definition) { - transaction.rollback(); + currentTransaction().rollback(); } @Override protected Map applicationData() { - return transaction.openMap("APPLICATION"); + return currentTransaction().openMap("APPLICATION"); } @Override protected Map cloudEvents(String regId) { - return transaction.openMap("CLOUDEVENTS" + ID_SEPARATOR + regId); + return currentTransaction().openMap("CLOUDEVENTS" + ID_SEPARATOR + regId); } @Override protected Map processedCloudEvents(String regId) { - return transaction.openMap(PROCESSED_PREFIX + regId); + return currentTransaction().openMap(PROCESSED_PREFIX + regId); } @Override protected void deleteAllProcessedMaps() { store.getMapNames().stream() .filter(s -> s.startsWith(PROCESSED_PREFIX)) - .forEach(s -> transaction.removeMap(transaction.openMap(s))); + .forEach(s -> currentTransaction().removeMap(currentTransaction().openMap(s))); + } + + @Override + protected Map blobData(String instanceId) { + return currentTransaction().openMap(instanceId + ID_SEPARATOR + "blobs"); + } + + @Override + protected void removeBlobData(String instanceId) { + currentTransaction().removeMap((TransactionMap) blobData(instanceId)); + } + + @Override + protected T executeNewTransaction(Callable runnable) { + transactionStack.add(transactionStore.begin()); + try { + T result = runnable.call(); + commit(null); + return result; + } catch (Exception ex) { + rollback(null); + throw new IllegalStateException(ex); + } finally { + transactionStack.pollLast(); + } } } diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java new file mode 100644 index 000000000..61ec4b2e6 --- /dev/null +++ b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java @@ -0,0 +1,37 @@ +/* + * 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.hashing.DefaultHashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; + +public class HashingMVStorePersistenceTest extends NonHashingMVStorePersistenceStoreTest { + + @Override + protected HashFactory hashFactory() { + return new DefaultHashFactory() { + @Override + protected boolean intCondition(byte[] data) { + return true; + } + + @Override + protected boolean md5Condition(byte[] data) { + return false; + } + }; + } +} diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java new file mode 100644 index 000000000..7c764bd43 --- /dev/null +++ b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java @@ -0,0 +1,37 @@ +/* + * 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.hashing.DefaultHashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; + +public class MD5MVStorePersistenceTest extends NonHashingMVStorePersistenceStoreTest { + + @Override + protected HashFactory hashFactory() { + return new DefaultHashFactory() { + @Override + protected boolean intCondition(byte[] data) { + return false; + } + + @Override + protected boolean md5Condition(byte[] data) { + return true; + } + }; + } +} 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/NonHashingMVStorePersistenceStoreTest.java similarity index 63% rename from impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java rename to impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/NonHashingMVStorePersistenceStoreTest.java index 1412d1767..5858977c7 100644 --- 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/NonHashingMVStorePersistenceStoreTest.java @@ -15,20 +15,37 @@ */ package io.serverlessworkflow.impl.persistence.mvstore; +import io.serverlessworkflow.impl.marshaller.DefaultBufferFactory; import io.serverlessworkflow.impl.persistence.PersistenceInstanceStore; +import io.serverlessworkflow.impl.persistence.hashing.DefaultHashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashFactory; 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 { +class NonHashingMVStorePersistenceStoreTest extends AbstractPersistenceTest { private static final String DB_NAME = "dbtest.db"; @Override protected PersistenceInstanceStore persistenceStore() { - return new MVStorePersistenceStore(DB_NAME); + return new MVStorePersistenceStore(DB_NAME, DefaultBufferFactory.factory(), hashFactory()); + } + + protected HashFactory hashFactory() { + return new DefaultHashFactory() { + @Override + protected boolean intCondition(byte[] data) { + return false; + } + + @Override + protected boolean md5Condition(byte[] data) { + return false; + } + }; } @AfterEach diff --git a/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java b/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java index 0cfc77dfa..95711eaee 100644 --- a/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java +++ b/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java @@ -76,6 +76,7 @@ void setup() { when(workflowContext.context()).thenReturn(context); when(workflowContext.definition()).thenReturn(definition); when(workflowContext.instanceData()).thenReturn(workflowInstance); + when(workflowInstance.metadata()).thenReturn(Map.of("Javierito", "rules")); when(workflowInstance.startedAt()).thenReturn(beforeStart.plus(Duration.ofMillis(1))); when(workflowInstance.context()).thenReturn(context); when(workflowInstance.id()).thenReturn(app.idFactory().get()); @@ -182,6 +183,7 @@ void testWorkflowInstance() throws InterruptedException { verify(updateTContext).transition(transition.capture()); assertThat(transition.getValue().isEndNode()).isTrue(); assertThat(instance.incIteration(position2)).isEqualTo(3); + assertThat(instance.metadata()).isEqualTo(Map.of("Javierito", "rules")); // workflow completed handlers.writer().completed(workflowContext).join(); From f3bc27dc4e7a94f809de488a26c5027a8f09867e Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Thu, 24 Sep 2026 13:23:26 +0200 Subject: [PATCH 2/4] [Fix #1691] MapInMemory approach Signed-off-by: Francisco Javier Tirado Sarti --- .../impl/persistence/PersistenceUtils.java | 40 ++++ .../hashing/HashMappingCoordinator.java | 182 ++++++++++++++++++ .../persistence/hashing/HashMappingInfo.java | 18 ++ .../bigmap/BytesMapInstanceTransaction.java | 82 ++++---- .../mvstore/MVStorePersistenceStore.java | 5 +- .../mvstore/MVStoreTransaction.java | 64 ++---- 6 files changed, 303 insertions(+), 88 deletions(-) create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java new file mode 100644 index 000000000..2ddf71cf9 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java @@ -0,0 +1,40 @@ +/* + * 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; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.stream.Collectors; + +public class PersistenceUtils { + + private PersistenceUtils() {} + + public static Map> mapMapToMapList(Map> map) { + return map.entrySet().stream() + .collect(Collectors.toMap(Entry::getKey, e -> PersistenceUtils.mapToList(e.getValue()))); + } + + public static List mapToList(Map map) { + List list = new ArrayList<>(map.size()); + for (int i = 0; i < map.size(); i++) { + list.add(map.get(i)); + } + return list; + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java new file mode 100644 index 000000000..c1c28af61 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java @@ -0,0 +1,182 @@ +/* + * 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.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.stream.Collectors; + +public class HashMappingCoordinator { + + public static final HashMappingCoordinator build( + Function>> retriever, + Consumer>> writer) { + return new HashMappingCoordinator(retriever, writer); + } + + private static class BytesWithFlag { + private final byte[] bytes; + private boolean persisted; + + public BytesWithFlag(byte[] bytes) { + this.bytes = bytes; + this.persisted = false; + } + + public BytesWithFlag(byte[] bytes, boolean persisted) { + this.bytes = bytes; + this.persisted = persisted; + } + + public void persist() { + persisted = true; + } + + public boolean isTransient() { + return !persisted; + } + } + + private record PendingWrite(String key, int index, BytesWithFlag bytes) {} + + private final Function>> retriever; + private final Consumer>> writer; + + private HashMappingCoordinator( + Function>> retriever, + Consumer>> writer) { + this.retriever = retriever; + this.writer = writer; + } + + private static Map>> mappingInfo = + new ConcurrentHashMap<>(); + + private Map> pending = new HashMap<>(); + + private HashMappingInfo from(PendingWrite pending) { + return new HashMappingInfo(pending.key, pending.index, pending.bytes.bytes); + } + + public int calculateIndex(String instanceId, HashItem item, byte[] bytes) { + Map> instanceMap = getInstanceMap(instanceId); + String key = item.key(); + synchronized (instanceMap) { + List list = instanceMap.computeIfAbsent(key, __ -> new ArrayList<>()); + if (list.isEmpty()) { + int index = 0; + BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes); + list.add(bytesWithFlag); + addWrite(instanceId, key, index, bytesWithFlag); + return index; + } + for (int i = 0; i < list.size(); i++) { + BytesWithFlag bytesWithFlag = list.get(i); + if (Arrays.equals(bytesWithFlag.bytes, bytes)) { + if (bytesWithFlag.isTransient()) { + addWrite(instanceId, key, i, bytesWithFlag); + } + return i; + } + } + BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes); + list.add(bytesWithFlag); + int index = list.size() - 1; + addWrite(instanceId, key, index, bytesWithFlag); + return index; + } + } + + private void addWrite(String instanceId, String key, int index, BytesWithFlag bytes) { + pending + .computeIfAbsent(instanceId, __ -> new ArrayList<>()) + .add(new PendingWrite(key, index, bytes)); + } + + public Optional readBytes(String instanceId, HashItem item, int index) { + Map> instanceMap = getInstanceMap(instanceId); + synchronized (instanceMap) { + return Optional.ofNullable(instanceMap.get(item.key())) + .filter(list -> list.size() > index) + .map(list -> list.get(index)) + .map(bytes -> bytes.bytes); + } + } + + private Map> getInstanceMap(String instanceId) { + return mappingInfo.computeIfAbsent( + instanceId, + k -> + retriever.apply(instanceId).entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + e -> + e.getValue().stream() + .map(b -> new BytesWithFlag(b, true)) + .collect(Collectors.toCollection(ArrayList::new))))); + } + + public void persist() { + Map> result = new HashMap<>(); + for (Map.Entry> item : pending.entrySet()) { + String instanceId = item.getKey(); + List list = new ArrayList<>(); + result.put(instanceId, list); + Map> instanceMap = mappingInfo.get(instanceId); + if (instanceMap != null) { + synchronized (instanceMap) { + item.getValue() + .forEach( + v -> { + if (v.bytes.isTransient()) { + list.add(from(v)); + } + }); + } + } + } + writer.accept(result); + } + + public void afterCommit() { + for (Map.Entry> item : pending.entrySet()) { + String instanceId = item.getKey(); + Map> instanceMap = mappingInfo.get(instanceId); + if (instanceMap != null) { + synchronized (instanceMap) { + item.getValue().forEach(v -> v.bytes.persist()); + } + } + } + pending.clear(); + } + + public void afterRollback() { + pending.clear(); + } + + public void afterRemove(String instanceId) { + mappingInfo.remove(instanceId); + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java new file mode 100644 index 000000000..70ba19acb --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java @@ -0,0 +1,18 @@ +/* + * 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; + +public record HashMappingInfo(String key, int index, byte[] bytes) {} 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 index 70508c74a..193af3308 100644 --- 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 @@ -33,20 +33,21 @@ import io.serverlessworkflow.impl.persistence.CompletedTaskInfo; import io.serverlessworkflow.impl.persistence.PersistenceInstanceInfo; import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo; +import io.serverlessworkflow.impl.persistence.PersistenceUtils; import io.serverlessworkflow.impl.persistence.RetriedTaskInfo; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; import io.serverlessworkflow.impl.persistence.hashing.HashItem; +import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator; +import io.serverlessworkflow.impl.persistence.hashing.HashMappingInfo; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.UncheckedIOException; import java.net.URI; import java.time.OffsetDateTime; -import java.util.Arrays; import java.util.HashMap; +import java.util.List; import java.util.Map; -import java.util.concurrent.Callable; -import java.util.concurrent.locks.Lock; public abstract class BytesMapInstanceTransaction extends BigMapInstanceTransaction { @@ -56,16 +57,37 @@ public abstract class BytesMapInstanceTransaction private static final byte VERSION_2 = 2; private static final byte VERSION_3 = 3; private static final byte[] PROCESSED_VALUE = {1}; + private static final String SEPARATOR = ":"; private final WorkflowBufferFactory bufferFactory; private final HashFactory hashFactory; - private final Lock hashLock; + protected final HashMappingCoordinator hashCoordinator; - protected BytesMapInstanceTransaction( - WorkflowBufferFactory factory, HashFactory hashFactory, Lock hashLock) { + protected BytesMapInstanceTransaction(WorkflowBufferFactory factory, HashFactory hashFactory) { this.bufferFactory = factory; this.hashFactory = hashFactory; - this.hashLock = hashLock; + this.hashCoordinator = + HashMappingCoordinator.build(this::retrieveBlobData, this::writeBlobData); + } + + private Map> retrieveBlobData(String instanceId) { + Map> result = new HashMap<>(); + for (Map.Entry entry : blobData(instanceId).entrySet()) { + String[] splitted = entry.getKey().split(SEPARATOR); + result + .computeIfAbsent(splitted[0], __ -> new HashMap<>()) + .put(Integer.parseInt(splitted[1]), entry.getValue()); + } + return PersistenceUtils.mapMapToMapList(result); + } + + private void writeBlobData(Map> writeInfo) { + for (Map.Entry> entry : writeInfo.entrySet()) { + Map blobData = blobData(entry.getKey()); + for (HashMappingInfo info : entry.getValue()) { + blobData.put(info.key() + SEPARATOR + info.index(), info.bytes()); + } + } } protected abstract Map blobData(String instanceId); @@ -75,7 +97,9 @@ protected BytesMapInstanceTransaction( @Override public void removeProcessInstance(WorkflowContextData workflowContext) { super.removeProcessInstance(workflowContext); - removeBlobData(workflowContext.instanceData().id()); + String instanceId = workflowContext.instanceData().id(); + removeBlobData(instanceId); + hashCoordinator.afterRemove(instanceId); } @Override @@ -254,37 +278,12 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) { buffer.readBoolean() ? buffer.readString() : null); } - protected abstract T executeNewTransaction(Callable runnable); - private void writeLargeObject( HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) { - - boolean duplicated; - String key = item.key(); - hashLock.lock(); - try { - duplicated = - executeNewTransaction( - () -> { - Map map = blobData(instanceData.id()); - byte[] value = map.get(key); - if (value == null) { - map.put(key, bytes); - return false; - } else { - return !Arrays.equals(value, bytes); - } - }); - } finally { - hashLock.unlock(); - } - - if (duplicated) { - legacyWriteObject(writer, bytes); - } else { - writer.writeByte(item.id()); - item.writeKey(writer); - } + int index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes); + writer.writeByte(item.id()); + item.writeKey(writer); + writer.writeShort((short) index); } private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { @@ -292,9 +291,12 @@ private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { .fromBuffer(buffer.readByte(), buffer) .map( item -> { - Map map = blobData(instanceId); - ByteArrayInputStream bytes = new ByteArrayInputStream(map.get(item.key())); - try (WorkflowInputBuffer input = bufferFactory.input(bytes)) { + try (WorkflowInputBuffer input = + bufferFactory.input( + new ByteArrayInputStream( + hashCoordinator + .readBytes(instanceId, item, buffer.readShort()) + .orElseThrow()))) { return input.readObject(); } }) 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 index deb5cbc3b..49dad173c 100644 --- 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 @@ -21,15 +21,12 @@ import io.serverlessworkflow.impl.persistence.bigmap.BigMapInstanceTransaction; import io.serverlessworkflow.impl.persistence.hashing.DefaultHashFactory; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; import org.h2.mvstore.MVStore; import org.h2.mvstore.tx.TransactionStore; public class MVStorePersistenceStore implements PersistenceInstanceStore { private final TransactionStore transactionStore; private final MVStore mvStore; - private final Lock hashLock = new ReentrantLock(); private final WorkflowBufferFactory bufferFactory; private final HashFactory hashFactory; @@ -56,6 +53,6 @@ public void close() { @Override public BigMapInstanceTransaction begin() { - return new MVStoreTransaction(mvStore, transactionStore, bufferFactory, hashFactory, hashLock); + return new MVStoreTransaction(mvStore, transactionStore, bufferFactory, hashFactory); } } 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 index ea839c0b8..ae63f4fce 100644 --- 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 @@ -21,34 +21,26 @@ import io.serverlessworkflow.impl.marshaller.WorkflowBufferFactory; import io.serverlessworkflow.impl.persistence.bigmap.BytesMapInstanceTransaction; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; -import java.util.ArrayDeque; -import java.util.Deque; import java.util.Map; -import java.util.concurrent.Callable; -import java.util.concurrent.locks.Lock; import org.h2.mvstore.MVStore; import org.h2.mvstore.tx.Transaction; import org.h2.mvstore.tx.TransactionMap; import org.h2.mvstore.tx.TransactionStore; public class MVStoreTransaction extends BytesMapInstanceTransaction { - - protected static final String ID_SEPARATOR = "-"; + private static final String ID_SEPARATOR = "-"; private static final String PROCESSED_PREFIX = "PROCESSED" + ID_SEPARATOR; - private final TransactionStore transactionStore; - private final Deque transactionStack = new ArrayDeque<>(); + private final Transaction transaction; private final MVStore store; public MVStoreTransaction( MVStore store, TransactionStore transactionStore, WorkflowBufferFactory bufferFactory, - HashFactory hashFactory, - Lock hashLock) { - super(bufferFactory, hashFactory, hashLock); - this.transactionStore = transactionStore; - transactionStack.add(transactionStore.begin()); + HashFactory hashFactory) { + super(bufferFactory, hashFactory); + transaction = transactionStore.begin(); this.store = store; } @@ -72,22 +64,18 @@ public Map status(WorkflowDefinitionData workflowContext) { return openMap(workflowContext, "status"); } - private Transaction currentTransaction() { - return transactionStack.peekLast(); - } - @Override public void removeTasks(String instanceId) { - currentTransaction().removeMap(taskMap(instanceId)); + transaction.removeMap(taskMap(instanceId)); } private TransactionMap taskMap(String instanceId) { - return currentTransaction().openMap(mapTaskName(instanceId)); + return transaction.openMap(mapTaskName(instanceId)); } private Map openMap(WorkflowDefinitionData workflowDefinition, String suffix) { - return currentTransaction() - .openMap(identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix); + return transaction.openMap( + identifier(workflowDefinition.workflow(), ID_SEPARATOR) + ID_SEPARATOR + suffix); } private String mapTaskName(String instanceId) { @@ -96,58 +84,46 @@ private String mapTaskName(String instanceId) { @Override public void commit(WorkflowDefinitionData definition) { - currentTransaction().commit(); + hashCoordinator.persist(); + transaction.commit(); + hashCoordinator.afterCommit(); } @Override public void rollback(WorkflowDefinitionData definition) { - currentTransaction().rollback(); + transaction.rollback(); + hashCoordinator.afterRollback(); } @Override protected Map applicationData() { - return currentTransaction().openMap("APPLICATION"); + return transaction.openMap("APPLICATION"); } @Override protected Map cloudEvents(String regId) { - return currentTransaction().openMap("CLOUDEVENTS" + ID_SEPARATOR + regId); + return transaction.openMap("CLOUDEVENTS" + ID_SEPARATOR + regId); } @Override protected Map processedCloudEvents(String regId) { - return currentTransaction().openMap(PROCESSED_PREFIX + regId); + return transaction.openMap(PROCESSED_PREFIX + regId); } @Override protected void deleteAllProcessedMaps() { store.getMapNames().stream() .filter(s -> s.startsWith(PROCESSED_PREFIX)) - .forEach(s -> currentTransaction().removeMap(currentTransaction().openMap(s))); + .forEach(s -> transaction.removeMap(transaction.openMap(s))); } @Override protected Map blobData(String instanceId) { - return currentTransaction().openMap(instanceId + ID_SEPARATOR + "blobs"); + return transaction.openMap(instanceId + ID_SEPARATOR + "blobs"); } @Override protected void removeBlobData(String instanceId) { - currentTransaction().removeMap((TransactionMap) blobData(instanceId)); - } - - @Override - protected T executeNewTransaction(Callable runnable) { - transactionStack.add(transactionStore.begin()); - try { - T result = runnable.call(); - commit(null); - return result; - } catch (Exception ex) { - rollback(null); - throw new IllegalStateException(ex); - } finally { - transactionStack.pollLast(); - } + transaction.removeMap((TransactionMap) blobData(instanceId)); } } From a1c3cad26455daf3950a1a043697eef6dcf8e7d2 Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Fri, 25 Sep 2026 13:45:08 +0200 Subject: [PATCH 3/4] [Fix #1691] Moving from int index to string generated index Signed-off-by: Francisco Javier Tirado Sarti --- .../impl/persistence/PersistenceUtils.java | 40 ----------- .../hashing/HashMappingCoordinator.java | 71 ++++++++++--------- .../persistence/hashing/HashMappingInfo.java | 2 +- .../persistence/hashing/IntegerHashItem.java | 2 +- .../impl/persistence/hashing/MD5HashItem.java | 2 +- .../bigmap/BytesMapInstanceTransaction.java | 17 ++--- .../test/AbstractHandlerPersistenceTest.java | 4 ++ 7 files changed, 51 insertions(+), 87 deletions(-) delete mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java deleted file mode 100644 index 2ddf71cf9..000000000 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.stream.Collectors; - -public class PersistenceUtils { - - private PersistenceUtils() {} - - public static Map> mapMapToMapList(Map> map) { - return map.entrySet().stream() - .collect(Collectors.toMap(Entry::getKey, e -> PersistenceUtils.mapToList(e.getValue()))); - } - - public static List mapToList(Map map) { - List list = new ArrayList<>(map.size()); - for (int i = 0; i < map.size(); i++) { - list.add(map.get(i)); - } - return list; - } -} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java index c1c28af61..c4a3c2fae 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java @@ -15,11 +15,13 @@ */ package io.serverlessworkflow.impl.persistence.hashing; +import com.github.f4b6a3.ulid.UlidFactory; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; @@ -28,8 +30,10 @@ public class HashMappingCoordinator { + private static final UlidFactory idFactory = UlidFactory.newMonotonicInstance(); + public static final HashMappingCoordinator build( - Function>> retriever, + Function>> retriever, Consumer>> writer) { return new HashMappingCoordinator(retriever, writer); } @@ -57,19 +61,19 @@ public boolean isTransient() { } } - private record PendingWrite(String key, int index, BytesWithFlag bytes) {} + private record PendingWrite(String key, String index, BytesWithFlag bytes) {} - private final Function>> retriever; + private final Function>> retriever; private final Consumer>> writer; private HashMappingCoordinator( - Function>> retriever, + Function>> retriever, Consumer>> writer) { this.retriever = retriever; this.writer = writer; } - private static Map>> mappingInfo = + private static Map>> mappingInfo = new ConcurrentHashMap<>(); private Map> pending = new HashMap<>(); @@ -78,52 +82,45 @@ private HashMappingInfo from(PendingWrite pending) { return new HashMappingInfo(pending.key, pending.index, pending.bytes.bytes); } - public int calculateIndex(String instanceId, HashItem item, byte[] bytes) { - Map> instanceMap = getInstanceMap(instanceId); + public String calculateIndex(String instanceId, HashItem item, byte[] bytes) { + Map> instanceMap = getInstanceMap(instanceId); String key = item.key(); synchronized (instanceMap) { - List list = instanceMap.computeIfAbsent(key, __ -> new ArrayList<>()); - if (list.isEmpty()) { - int index = 0; - BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes); - list.add(bytesWithFlag); - addWrite(instanceId, key, index, bytesWithFlag); - return index; - } - for (int i = 0; i < list.size(); i++) { - BytesWithFlag bytesWithFlag = list.get(i); - if (Arrays.equals(bytesWithFlag.bytes, bytes)) { - if (bytesWithFlag.isTransient()) { - addWrite(instanceId, key, i, bytesWithFlag); + Map duplicateMap = + instanceMap.computeIfAbsent(key, __ -> new HashMap<>()); + + for (Entry entry : duplicateMap.entrySet()) { + if (Arrays.equals(entry.getValue().bytes, bytes)) { + if (entry.getValue().isTransient()) { + addWrite(instanceId, key, entry.getKey(), entry.getValue()); } - return i; + return entry.getKey(); } } + String index = idFactory.create().toString(); BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes); - list.add(bytesWithFlag); - int index = list.size() - 1; + duplicateMap.put(index, bytesWithFlag); addWrite(instanceId, key, index, bytesWithFlag); return index; } } - private void addWrite(String instanceId, String key, int index, BytesWithFlag bytes) { + private void addWrite(String instanceId, String key, String index, BytesWithFlag bytes) { pending .computeIfAbsent(instanceId, __ -> new ArrayList<>()) .add(new PendingWrite(key, index, bytes)); } - public Optional readBytes(String instanceId, HashItem item, int index) { - Map> instanceMap = getInstanceMap(instanceId); + public Optional readBytes(String instanceId, HashItem item, String index) { + Map> instanceMap = getInstanceMap(instanceId); synchronized (instanceMap) { return Optional.ofNullable(instanceMap.get(item.key())) - .filter(list -> list.size() > index) - .map(list -> list.get(index)) + .map(m -> m.get(index)) .map(bytes -> bytes.bytes); } } - private Map> getInstanceMap(String instanceId) { + private Map> getInstanceMap(String instanceId) { return mappingInfo.computeIfAbsent( instanceId, k -> @@ -132,9 +129,11 @@ private Map> getInstanceMap(String instanceId) { Collectors.toMap( Map.Entry::getKey, e -> - e.getValue().stream() - .map(b -> new BytesWithFlag(b, true)) - .collect(Collectors.toCollection(ArrayList::new))))); + e.getValue().entrySet().stream() + .collect( + Collectors.toMap( + Entry::getKey, + v -> new BytesWithFlag(v.getValue(), true)))))); } public void persist() { @@ -143,7 +142,7 @@ public void persist() { String instanceId = item.getKey(); List list = new ArrayList<>(); result.put(instanceId, list); - Map> instanceMap = mappingInfo.get(instanceId); + Map> instanceMap = mappingInfo.get(instanceId); if (instanceMap != null) { synchronized (instanceMap) { item.getValue() @@ -162,7 +161,7 @@ public void persist() { public void afterCommit() { for (Map.Entry> item : pending.entrySet()) { String instanceId = item.getKey(); - Map> instanceMap = mappingInfo.get(instanceId); + Map> instanceMap = mappingInfo.get(instanceId); if (instanceMap != null) { synchronized (instanceMap) { item.getValue().forEach(v -> v.bytes.persist()); @@ -179,4 +178,8 @@ public void afterRollback() { public void afterRemove(String instanceId) { mappingInfo.remove(instanceId); } + + public static void clearAll() { + mappingInfo.clear(); + } } diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java index 70ba19acb..c0a9c68bb 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java @@ -15,4 +15,4 @@ */ package io.serverlessworkflow.impl.persistence.hashing; -public record HashMappingInfo(String key, int index, byte[] bytes) {} +public record HashMappingInfo(String key, String index, byte[] bytes) {} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java index ccd4ddf86..c554a8ad8 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java @@ -22,7 +22,7 @@ public class IntegerHashItem implements HashItem { public static final byte ID = 1; - public static final int SIZE_THRESHOLD = 4 * 4; + public static final int SIZE_THRESHOLD = 32; private final int hashCode; diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java index 1a37103e1..85a7d76e3 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java @@ -23,7 +23,7 @@ public class MD5HashItem implements HashItem { public static final byte ID = 2; - public static final int SIZE_THRESHOLD = 128 * 2; + public static final int SIZE_THRESHOLD = 150; @Override public byte id() { 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 index 193af3308..25ca66462 100644 --- 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 @@ -33,7 +33,6 @@ import io.serverlessworkflow.impl.persistence.CompletedTaskInfo; import io.serverlessworkflow.impl.persistence.PersistenceInstanceInfo; import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo; -import io.serverlessworkflow.impl.persistence.PersistenceUtils; import io.serverlessworkflow.impl.persistence.RetriedTaskInfo; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; import io.serverlessworkflow.impl.persistence.hashing.HashItem; @@ -70,15 +69,13 @@ protected BytesMapInstanceTransaction(WorkflowBufferFactory factory, HashFactory HashMappingCoordinator.build(this::retrieveBlobData, this::writeBlobData); } - private Map> retrieveBlobData(String instanceId) { - Map> result = new HashMap<>(); + private Map> retrieveBlobData(String instanceId) { + Map> result = new HashMap<>(); for (Map.Entry entry : blobData(instanceId).entrySet()) { String[] splitted = entry.getKey().split(SEPARATOR); - result - .computeIfAbsent(splitted[0], __ -> new HashMap<>()) - .put(Integer.parseInt(splitted[1]), entry.getValue()); + result.computeIfAbsent(splitted[0], __ -> new HashMap<>()).put(splitted[1], entry.getValue()); } - return PersistenceUtils.mapMapToMapList(result); + return result; } private void writeBlobData(Map> writeInfo) { @@ -280,10 +277,10 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) { private void writeLargeObject( HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) { - int index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes); + String index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes); writer.writeByte(item.id()); item.writeKey(writer); - writer.writeShort((short) index); + writer.writeString(index); } private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { @@ -295,7 +292,7 @@ private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { bufferFactory.input( new ByteArrayInputStream( hashCoordinator - .readBytes(instanceId, item, buffer.readShort()) + .readBytes(instanceId, item, buffer.readString()) .orElseThrow()))) { return input.readObject(); } diff --git a/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java b/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java index 95711eaee..f69ff90a3 100644 --- a/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java +++ b/impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java @@ -37,6 +37,7 @@ import io.serverlessworkflow.impl.executors.TransitionInfo; import io.serverlessworkflow.impl.persistence.PersistenceInstanceHandlers; import io.serverlessworkflow.impl.persistence.WorkflowPersistenceInstance; +import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator; import java.io.IOException; import java.time.Duration; import java.time.Instant; @@ -154,6 +155,8 @@ void testWorkflowInstance() throws InterruptedException { verify(parentContext).tryRetryCount(retryAttempt.capture()); assertThat(retryAttempt.getValue()).isEqualTo(numRetries); + HashMappingCoordinator.clearAll(); + // task completed handlers .writer() @@ -163,6 +166,7 @@ void testWorkflowInstance() throws InterruptedException { assertThat(stream.count()).isEqualTo(1); } definition.close(); + instance = (WorkflowPersistenceInstance) handlers.reader().find(definition, workflowInstance.id()).orElseThrow(); From 4c90bca882a04aca589dc5da47348cfc96f38cce Mon Sep 17 00:00:00 2001 From: Francisco Javier Tirado Sarti Date: Fri, 25 Sep 2026 15:59:51 +0200 Subject: [PATCH 4/4] [Fix #1691] Use Ulid for improving versatility Signed-off-by: Francisco Javier Tirado Sarti --- .../hashing/DefaultHashFactory.java | 19 +++++++ .../persistence/hashing/DefaultHashIndex.java | 51 +++++++++++++++++++ .../impl/persistence/hashing/HashFactory.java | 6 +++ .../impl/persistence/hashing/HashIndex.java | 23 +++++++++ .../hashing/HashMappingCoordinator.java | 45 ++++++++-------- .../persistence/hashing/HashMappingInfo.java | 2 +- .../bigmap/BytesMapInstanceTransaction.java | 21 +++++--- .../HashingMVStorePersistenceTest.java | 2 +- .../mvstore/MD5MVStorePersistenceTest.java | 2 +- ....java => MVStorePersistenceStoreTest.java} | 2 +- 10 files changed, 139 insertions(+), 34 deletions(-) create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashIndex.java create mode 100644 impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashIndex.java rename impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/{NonHashingMVStorePersistenceStoreTest.java => MVStorePersistenceStoreTest.java} (95%) diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java index 5498200c7..81420b434 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashFactory.java @@ -15,11 +15,15 @@ */ package io.serverlessworkflow.impl.persistence.hashing; +import com.github.f4b6a3.ulid.Ulid; +import com.github.f4b6a3.ulid.UlidFactory; import io.serverlessworkflow.impl.marshaller.WorkflowInputBuffer; import java.util.Optional; public class DefaultHashFactory implements HashFactory { + private final UlidFactory idFactory = UlidFactory.newMonotonicInstance(); + @Override public Optional fromBuffer(byte id, WorkflowInputBuffer buffer) { return Optional.ofNullable( @@ -49,4 +53,19 @@ protected boolean intCondition(byte[] data) { protected boolean md5Condition(byte[] data) { return data.length > MD5HashItem.SIZE_THRESHOLD; } + + @Override + public HashIndex indexFromBytes(byte[] bytes) { + return new DefaultHashIndex(Ulid.from(bytes)); + } + + @Override + public HashIndex indexFromString(String str) { + return new DefaultHashIndex(Ulid.from(str)); + } + + @Override + public HashIndex newIndex() { + return new DefaultHashIndex(idFactory.create()); + } } diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashIndex.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashIndex.java new file mode 100644 index 000000000..c3feb02c8 --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/DefaultHashIndex.java @@ -0,0 +1,51 @@ +/* + * 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 com.github.f4b6a3.ulid.Ulid; +import java.util.Objects; + +public class DefaultHashIndex implements HashIndex { + + private final Ulid ulid; + + public DefaultHashIndex(Ulid ulid) { + this.ulid = ulid; + } + + public String toString() { + return ulid.toString(); + } + + @Override + public byte[] toBytes() { + return ulid.toBytes(); + } + + @Override + public int hashCode() { + return Objects.hash(ulid); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) return true; + if (obj == null) return false; + if (getClass() != obj.getClass()) return false; + DefaultHashIndex other = (DefaultHashIndex) obj; + return Objects.equals(ulid, other.ulid); + } +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java index b580b311f..2c1eb262f 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashFactory.java @@ -23,4 +23,10 @@ public interface HashFactory { Optional fromData(byte[] data); Optional fromBuffer(byte id, WorkflowInputBuffer buffer); + + HashIndex indexFromBytes(byte[] bytes); + + HashIndex indexFromString(String str); + + HashIndex newIndex(); } diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashIndex.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashIndex.java new file mode 100644 index 000000000..885efba2f --- /dev/null +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashIndex.java @@ -0,0 +1,23 @@ +/* + * 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; + +public interface HashIndex { + + String toString(); + + byte[] toBytes(); +} diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java index c4a3c2fae..ebc9c1b93 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java @@ -15,7 +15,6 @@ */ package io.serverlessworkflow.impl.persistence.hashing; -import com.github.f4b6a3.ulid.UlidFactory; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -30,14 +29,16 @@ public class HashMappingCoordinator { - private static final UlidFactory idFactory = UlidFactory.newMonotonicInstance(); - public static final HashMappingCoordinator build( - Function>> retriever, + HashFactory hashFactory, + Function>> retriever, Consumer>> writer) { - return new HashMappingCoordinator(retriever, writer); + return new HashMappingCoordinator(hashFactory, retriever, writer); } + private static Map>> mappingInfo = + new ConcurrentHashMap<>(); + private static class BytesWithFlag { private final byte[] bytes; private boolean persisted; @@ -61,35 +62,35 @@ public boolean isTransient() { } } - private record PendingWrite(String key, String index, BytesWithFlag bytes) {} + private record PendingWrite(String key, HashIndex index, BytesWithFlag bytes) {} - private final Function>> retriever; + private final HashFactory hashFactory; + private final Function>> retriever; private final Consumer>> writer; private HashMappingCoordinator( - Function>> retriever, + HashFactory hashFactory, + Function>> retriever, Consumer>> writer) { + this.hashFactory = hashFactory; this.retriever = retriever; this.writer = writer; } - private static Map>> mappingInfo = - new ConcurrentHashMap<>(); - private Map> pending = new HashMap<>(); private HashMappingInfo from(PendingWrite pending) { return new HashMappingInfo(pending.key, pending.index, pending.bytes.bytes); } - public String calculateIndex(String instanceId, HashItem item, byte[] bytes) { - Map> instanceMap = getInstanceMap(instanceId); + public HashIndex calculateIndex(String instanceId, HashItem item, byte[] bytes) { + Map> instanceMap = getInstanceMap(instanceId); String key = item.key(); synchronized (instanceMap) { - Map duplicateMap = + Map duplicateMap = instanceMap.computeIfAbsent(key, __ -> new HashMap<>()); - for (Entry entry : duplicateMap.entrySet()) { + for (Entry entry : duplicateMap.entrySet()) { if (Arrays.equals(entry.getValue().bytes, bytes)) { if (entry.getValue().isTransient()) { addWrite(instanceId, key, entry.getKey(), entry.getValue()); @@ -97,7 +98,7 @@ public String calculateIndex(String instanceId, HashItem item, byte[] bytes) { return entry.getKey(); } } - String index = idFactory.create().toString(); + HashIndex index = hashFactory.newIndex(); BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes); duplicateMap.put(index, bytesWithFlag); addWrite(instanceId, key, index, bytesWithFlag); @@ -105,14 +106,14 @@ public String calculateIndex(String instanceId, HashItem item, byte[] bytes) { } } - private void addWrite(String instanceId, String key, String index, BytesWithFlag bytes) { + private void addWrite(String instanceId, String key, HashIndex index, BytesWithFlag bytes) { pending .computeIfAbsent(instanceId, __ -> new ArrayList<>()) .add(new PendingWrite(key, index, bytes)); } - public Optional readBytes(String instanceId, HashItem item, String index) { - Map> instanceMap = getInstanceMap(instanceId); + public Optional readBytes(String instanceId, HashItem item, HashIndex index) { + Map> instanceMap = getInstanceMap(instanceId); synchronized (instanceMap) { return Optional.ofNullable(instanceMap.get(item.key())) .map(m -> m.get(index)) @@ -120,7 +121,7 @@ public Optional readBytes(String instanceId, HashItem item, String index } } - private Map> getInstanceMap(String instanceId) { + private Map> getInstanceMap(String instanceId) { return mappingInfo.computeIfAbsent( instanceId, k -> @@ -142,7 +143,7 @@ public void persist() { String instanceId = item.getKey(); List list = new ArrayList<>(); result.put(instanceId, list); - Map> instanceMap = mappingInfo.get(instanceId); + Map> instanceMap = mappingInfo.get(instanceId); if (instanceMap != null) { synchronized (instanceMap) { item.getValue() @@ -161,7 +162,7 @@ public void persist() { public void afterCommit() { for (Map.Entry> item : pending.entrySet()) { String instanceId = item.getKey(); - Map> instanceMap = mappingInfo.get(instanceId); + Map> instanceMap = mappingInfo.get(instanceId); if (instanceMap != null) { synchronized (instanceMap) { item.getValue().forEach(v -> v.bytes.persist()); diff --git a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java index c0a9c68bb..859b701c4 100644 --- a/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java +++ b/impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java @@ -15,4 +15,4 @@ */ package io.serverlessworkflow.impl.persistence.hashing; -public record HashMappingInfo(String key, String index, byte[] bytes) {} +public record HashMappingInfo(String key, HashIndex index, byte[] bytes) {} 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 index 25ca66462..1f718ebc9 100644 --- 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 @@ -35,6 +35,7 @@ import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo; import io.serverlessworkflow.impl.persistence.RetriedTaskInfo; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; +import io.serverlessworkflow.impl.persistence.hashing.HashIndex; import io.serverlessworkflow.impl.persistence.hashing.HashItem; import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator; import io.serverlessworkflow.impl.persistence.hashing.HashMappingInfo; @@ -66,14 +67,17 @@ protected BytesMapInstanceTransaction(WorkflowBufferFactory factory, HashFactory this.bufferFactory = factory; this.hashFactory = hashFactory; this.hashCoordinator = - HashMappingCoordinator.build(this::retrieveBlobData, this::writeBlobData); + HashMappingCoordinator.build(hashFactory, this::retrieveBlobData, this::writeBlobData); } - private Map> retrieveBlobData(String instanceId) { - Map> result = new HashMap<>(); + private Map> retrieveBlobData(String instanceId) { + Map> result = new HashMap<>(); for (Map.Entry entry : blobData(instanceId).entrySet()) { - String[] splitted = entry.getKey().split(SEPARATOR); - result.computeIfAbsent(splitted[0], __ -> new HashMap<>()).put(splitted[1], entry.getValue()); + String key = entry.getKey(); + int indexOf = key.indexOf(SEPARATOR); + result + .computeIfAbsent(key.substring(0, indexOf), __ -> new HashMap<>()) + .put(hashFactory.indexFromString(key.substring(indexOf + 1)), entry.getValue()); } return result; } @@ -277,10 +281,10 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) { private void writeLargeObject( HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) { - String index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes); + HashIndex index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes); writer.writeByte(item.id()); item.writeKey(writer); - writer.writeString(index); + writer.writeBytes(index.toBytes()); } private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { @@ -292,7 +296,8 @@ private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) { bufferFactory.input( new ByteArrayInputStream( hashCoordinator - .readBytes(instanceId, item, buffer.readString()) + .readBytes( + instanceId, item, hashFactory.indexFromBytes(buffer.readBytes())) .orElseThrow()))) { return input.readObject(); } diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java index 61ec4b2e6..0217940de 100644 --- a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java +++ b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/HashingMVStorePersistenceTest.java @@ -18,7 +18,7 @@ import io.serverlessworkflow.impl.persistence.hashing.DefaultHashFactory; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; -public class HashingMVStorePersistenceTest extends NonHashingMVStorePersistenceStoreTest { +public class HashingMVStorePersistenceTest extends MVStorePersistenceStoreTest { @Override protected HashFactory hashFactory() { diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java index 7c764bd43..6dca19f1a 100644 --- a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java +++ b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MD5MVStorePersistenceTest.java @@ -18,7 +18,7 @@ import io.serverlessworkflow.impl.persistence.hashing.DefaultHashFactory; import io.serverlessworkflow.impl.persistence.hashing.HashFactory; -public class MD5MVStorePersistenceTest extends NonHashingMVStorePersistenceStoreTest { +public class MD5MVStorePersistenceTest extends MVStorePersistenceStoreTest { @Override protected HashFactory hashFactory() { diff --git a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/NonHashingMVStorePersistenceStoreTest.java b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java similarity index 95% rename from impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/NonHashingMVStorePersistenceStoreTest.java rename to impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java index 5858977c7..9fd1b9fca 100644 --- a/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/NonHashingMVStorePersistenceStoreTest.java +++ b/impl/persistence/mvstore/src/test/java/io/serverlessworkflow/impl/persistence/mvstore/MVStorePersistenceStoreTest.java @@ -25,7 +25,7 @@ import java.nio.file.Path; import org.junit.jupiter.api.AfterEach; -class NonHashingMVStorePersistenceStoreTest extends AbstractPersistenceTest { +class MVStorePersistenceStoreTest extends AbstractPersistenceTest { private static final String DB_NAME = "dbtest.db";