[Fix #1296] Prevent double loading of the same process instance (#1297)

Signed-off-by: fjtirado <ftirados@redhat.com>

Author: fjtirado

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

@@ -39,6 +39,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData {
 
@@ -56,6 +57,7 @@ public class WorkflowDefinition implements AutoCloseable, WorkflowDefinitionData
   private Cancellable everySchedule;
   private Cancellable cronSchedule;
   private Collection<WorkflowInstance> scheduledInstances = new ArrayList<>();
+  private Map<String, WorkflowInstance> activeInstances = new ConcurrentHashMap<>();
 
   private WorkflowDefinition(
       WorkflowApplication application, Workflow workflow, ResourceLoader resourceLoader) {
@@ -177,6 +179,18 @@ public void addScheduledInstance(WorkflowInstance workflowInstance) {
     scheduledInstances.add(workflowInstance);
   }
 
+  void removeInstance(WorkflowInstance instance) {
+    activeInstances.remove(instance.id());
+  }
+
+  void addInstance(WorkflowInstance instance) {
+    activeInstances.put(instance.id(), instance);
+  }
+
+  public Optional<WorkflowInstance> activeInstance(String instanceId) {
+    return Optional.ofNullable(activeInstances.get(instanceId));
+  }
+
   @Override
   public WorkflowDefinitionId id() {
     return definitionId;
@@ -194,6 +208,7 @@ public void close() {
       cronSchedule.cancel();
     }
     scheduledInstances.clear();
+    activeInstances.clear();
   }
 
   @Override

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

@@ -63,6 +63,7 @@ protected WorkflowMutableInstance(WorkflowDefinition definition, String id, Work
     this.input = input;
     this.status = new AtomicReference<>(WorkflowStatus.PENDING);
     this.workflowContext = new WorkflowContext(definition, this);
+    definition.addInstance(this);
   }
 
   @Override
@@ -120,6 +121,7 @@ private void whenCompleted(WorkflowModel result, Throwable ex) {
     if (ex != null) {
       handleException(ex instanceof CompletionException ? ex = ex.getCause() : ex);
     }
+    workflowContext.definition().removeInstance(this);
   }
 
   private void handleException(Throwable ex) {

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

@@ -28,14 +28,14 @@ protected final Stream<WorkflowInstance> scanAll(
       String applicationId) {
     return operations
         .scanAll(applicationId, definition)
-        .map(v -> new WorkflowPersistenceInstance(definition, v));
+        .map(v -> WorkflowPersistenceInstance.of(definition, v));
   }
 
   protected final Optional<WorkflowInstance> find(
       PersistenceInstanceOperations operations, WorkflowDefinition definition, String instanceId) {
     return operations
         .readWorkflowInfo(definition, instanceId)
-        .map(i -> new WorkflowPersistenceInstance(definition, i));
+        .map(i -> WorkflowPersistenceInstance.of(definition, i));
   }
 
   @Override

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

@@ -18,6 +18,7 @@
 import io.serverlessworkflow.impl.TaskContext;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowMutableInstance;
 import io.serverlessworkflow.impl.WorkflowStatus;
@@ -28,7 +29,13 @@ public class WorkflowPersistenceInstance extends WorkflowMutableInstance {
 
   private final PersistenceWorkflowInfo info;
 
-  public WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWorkflowInfo info) {
+  public static WorkflowInstance of(WorkflowDefinition definition, PersistenceWorkflowInfo info) {
+    return definition
+        .activeInstance(info.id())
+        .orElseGet(() -> new WorkflowPersistenceInstance(definition, info));
+  }
+
+  private WorkflowPersistenceInstance(WorkflowDefinition definition, PersistenceWorkflowInfo info) {
     super(definition, info.id(), info.input());
     this.info = info;
     this.startedAt = info.startedAt();

impl/persistence/tests/src/main/java/io/serverlessworkflow/impl/persistence/test/AbstractHandlerPersistenceTest.java

@@ -152,6 +152,7 @@ void testWorkflowInstance() throws InterruptedException {
     try (Stream<WorkflowInstance> stream = handlers.reader().scanAll(definition)) {
       assertThat(stream.count()).isEqualTo(1);
     }
+    definition.close();
     instance =
         (WorkflowPersistenceInstance)
             handlers.reader().find(definition, workflowInstance.id()).orElseThrow();