feat: Add workflow versioning

Signed-off-by: Javier Aliaga <javier@diagrid.io>

Author: javier-aliaga

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -17,6 +17,7 @@
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService.TaskFailureDetails;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
 import io.grpc.Channel;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
@@ -42,7 +43,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
   private static final Logger logger = Logger.getLogger(DurableTaskGrpcWorker.class.getPackage().getName());
   private static final Duration DEFAULT_MAXIMUM_TIMER_INTERVAL = Duration.ofDays(3);
 
-  private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
+  private final TaskOrchestrationFactories orchestrationFactories;
+
   private final HashMap<String, TaskActivityFactory> activityFactories = new HashMap<>();
 
   private final ManagedChannel managedSidecarChannel;
@@ -57,7 +59,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
   private Thread workerThread;
 
   DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder) {
-    this.orchestrationFactories.putAll(builder.orchestrationFactories);
+    this.orchestrationFactories = builder.orchestrationFactories;
     this.activityFactories.putAll(builder.activityFactories);
     this.appId = builder.appId;
 

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -13,6 +13,8 @@
 
 package io.dapr.durabletask;
 
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
 import io.grpc.Channel;
 
 import java.time.Duration;
@@ -24,7 +26,7 @@
  *
  */
 public final class DurableTaskGrpcWorkerBuilder {
-  final HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
+  TaskOrchestrationFactories orchestrationFactories = new TaskOrchestrationFactories();
   final HashMap<String, TaskActivityFactory> activityFactories = new HashMap<>();
   int port;
   Channel channel;
@@ -40,20 +42,11 @@ public final class DurableTaskGrpcWorkerBuilder {
    * @return this builder object
    */
   public DurableTaskGrpcWorkerBuilder addOrchestration(TaskOrchestrationFactory factory) {
-    String key = factory.getName();
-    if (key == null || key.length() == 0) {
-      throw new IllegalArgumentException("A non-empty task orchestration name is required.");
-    }
-
-    if (this.orchestrationFactories.containsKey(key)) {
-      throw new IllegalArgumentException(
-          String.format("A task orchestration factory named %s is already registered.", key));
-    }
-
-    this.orchestrationFactories.put(key, factory);
+    this.orchestrationFactories.addOrchestration(factory);
     return this;
   }
 
+
   /**
    * Adds an activity factory to be used by the constructed {@link DurableTaskGrpcWorker}.
    *
@@ -161,4 +154,6 @@ public DurableTaskGrpcWorkerBuilder appId(String appId) {
   public DurableTaskGrpcWorker build() {
     return new DurableTaskGrpcWorker(this);
   }
+
+
 }

durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRunner.java

@@ -16,10 +16,11 @@
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.StringValue;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
 
 import java.time.Duration;
 import java.util.Base64;
-import java.util.HashMap;
 import java.util.logging.Logger;
 
 /**
@@ -134,8 +135,8 @@ public static byte[] loadAndRun(byte[] orchestratorRequestBytes, TaskOrchestrati
     }
 
     // Register the passed orchestration as the default ("*") orchestration
-    HashMap<String, TaskOrchestrationFactory> orchestrationFactories = new HashMap<>();
-    orchestrationFactories.put("*", new TaskOrchestrationFactory() {
+    TaskOrchestrationFactories orchestrationFactories = new TaskOrchestrationFactories();
+    orchestrationFactories.addOrchestration(new TaskOrchestrationFactory() {
       @Override
       public String getName() {
         return "*";
@@ -145,6 +146,16 @@ public String getName() {
       public TaskOrchestration create() {
         return orchestration;
       }
+
+      @Override
+      public String getVersionName() {
+        return "";
+      }
+
+      @Override
+      public Boolean isLatestVersion() {
+        return false;
+      }
     });
 
     TaskOrchestrationExecutor taskOrchestrationExecutor = new TaskOrchestrationExecutor(

durabletask-client/src/main/java/io/dapr/durabletask/OrchestrationRuntimeStatus.java

@@ -68,7 +68,12 @@ public enum OrchestrationRuntimeStatus {
   /**
    * The orchestration is in a suspended state.
    */
-  SUSPENDED;
+  SUSPENDED,
+
+  /**
+   * The orchestration is in a stalled state.
+   */
+  STALLED;
 
   static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.OrchestrationStatus status) {
     switch (status) {
@@ -88,6 +93,8 @@ static OrchestrationRuntimeStatus fromProtobuf(OrchestratorService.Orchestration
         return PENDING;
       case ORCHESTRATION_STATUS_SUSPENDED:
         return SUSPENDED;
+      case ORCHESTRATION_STATUS_STALLED:
+        return STALLED;
       default:
         throw new IllegalArgumentException(String.format("Unknown status value: %s", status));
     }

durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationContext.java

@@ -326,6 +326,15 @@ default void continueAsNew(Object input) {
     this.continueAsNew(input, true);
   }
 
+  /**
+   * Check if the given patch name can be applied to the orchestration.
+   *
+   * @param patchName The name of the patch to check.
+   * @return True if the given patch name can be applied to the orchestration, False otherwise.
+   */
+
+  boolean isPatched(String patchName);
+
   /**
    * Restarts the orchestration with a new input and clears its history.
    *

durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java

@@ -19,7 +19,11 @@
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService.ScheduleTaskAction.Builder;
 import io.dapr.durabletask.interruption.ContinueAsNewInterruption;
 import io.dapr.durabletask.interruption.OrchestratorBlockedException;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactories;
+import io.dapr.durabletask.orchestration.TaskOrchestrationFactory;
+import io.dapr.durabletask.orchestration.exception.VersionNotRegisteredException;
 import io.dapr.durabletask.util.UuidGenerator;
+import io.opentelemetry.api.internal.StringUtils;
 
 import javax.annotation.Nullable;
 import java.time.Duration;
@@ -47,14 +51,14 @@
 final class TaskOrchestrationExecutor {
 
   private static final String EMPTY_STRING = "";
-  private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories;
+  private final TaskOrchestrationFactories orchestrationFactories;
   private final DataConverter dataConverter;
   private final Logger logger;
   private final Duration maximumTimerInterval;
   private final String appId;
 
   public TaskOrchestrationExecutor(
-      HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
+      TaskOrchestrationFactories orchestrationFactories,
       DataConverter dataConverter,
       Duration maximumTimerInterval,
       Logger logger,
@@ -79,6 +83,9 @@ public TaskOrchestratorResult execute(List<OrchestratorService.HistoryEvent> pas
       }
       completed = true;
       logger.finest("The orchestrator execution completed normally");
+    } catch (VersionNotRegisteredException versionNotRegisteredException) {
+      logger.warning("The orchestrator version is not registered: " + versionNotRegisteredException.toString());
+      context.setVersionNotRegistered();
     } catch (OrchestratorBlockedException orchestratorBlockedException) {
       logger.fine("The orchestrator has yielded and will await for new events.");
     } catch (ContinueAsNewInterruption continueAsNewInterruption) {
@@ -87,7 +94,7 @@ public TaskOrchestratorResult execute(List<OrchestratorService.HistoryEvent> pas
     } catch (Exception e) {
       // The orchestrator threw an unhandled exception - fail it
       // TODO: What's the right way to log this?
-      logger.warning("The orchestrator failed with an unhandled exception: " + e.toString());
+      logger.warning("The orchestrator failed with an unhandled exception: " + e);
       context.fail(new FailureDetails(e));
     }
 
@@ -97,12 +104,16 @@ public TaskOrchestratorResult execute(List<OrchestratorService.HistoryEvent> pas
       context.complete(null);
     }
 
-    return new TaskOrchestratorResult(context.pendingActions.values(), context.getCustomStatus());
+    return new TaskOrchestratorResult(context.pendingActions.values(),
+        context.getCustomStatus(),
+        context.version,
+        context.encounteredPatches);
   }
 
   private class ContextImplTask implements TaskOrchestrationContext {
 
     private String orchestratorName;
+    private final List<String> encounteredPatches = new ArrayList<>();
     private String rawInput;
     private String instanceId;
     private Instant currentInstant;
@@ -127,6 +138,11 @@ private class ContextImplTask implements TaskOrchestrationContext {
     private Object continuedAsNewInput;
     private boolean preserveUnprocessedEvents;
     private Object customStatus;
+    private final Map<String, Boolean> appliedPatches = new HashMap<>();
+    private final Map<String, Boolean> historyPatches = new HashMap<>();
+
+    private OrchestratorService.OrchestrationVersion orchestratorStartedVersion;
+    private String version;
 
     public ContextImplTask(List<OrchestratorService.HistoryEvent> pastEvents,
                            List<OrchestratorService.HistoryEvent> newEvents) {
@@ -144,6 +160,7 @@ private void setName(String name) {
       this.orchestratorName = name;
     }
 
+
     private void setInput(String rawInput) {
       this.rawInput = rawInput;
     }
@@ -363,6 +380,34 @@ public <V> Task<V> callActivity(
       return this.createAppropriateTask(taskFactory, options);
     }
 
+    @Override
+    public boolean isPatched(String patchName) {
+      var isPatched = this.checkPatch(patchName);
+      if (isPatched) {
+        this.encounteredPatches.add(patchName);
+      }
+
+      return isPatched;
+    }
+
+    public boolean checkPatch(String patchName) {
+      if (this.appliedPatches.containsKey(patchName)) {
+        return this.appliedPatches.get(patchName);
+      }
+
+      if (this.historyPatches.containsKey(patchName)) {
+        this.appliedPatches.put(patchName, true);
+        return true;
+      }
+
+      if (this.isReplaying) {
+        this.appliedPatches.put(patchName, false);
+        return false;
+      }
+      this.appliedPatches.put(patchName, true);
+      return true;
+    }
+
     @Override
     public void continueAsNew(Object input, boolean preserveUnprocessedEvents) {
       Helpers.throwIfOrchestratorComplete(this.isComplete);
@@ -924,6 +969,7 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
           case ORCHESTRATORSTARTED:
             Instant instant = DataConverter.getInstantFromTimestamp(e.getTimestamp());
             this.setCurrentInstant(instant);
+            this.orchestratorStartedVersion = e.getOrchestratorStarted().getVersion();
             this.logger.fine(() -> this.instanceId + ": Workflow orchestrator started");
             break;
           case ORCHESTRATORCOMPLETED:
@@ -938,18 +984,32 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
             this.logger.fine(() -> this.instanceId + ": Workflow execution started");
             this.setAppId(e.getRouter().getSourceAppID());
 
+            if (this.orchestratorStartedVersion != null
+                && this.orchestratorStartedVersion.getPatchesCount() > 0) {
+              for (var patch : this.orchestratorStartedVersion.getPatchesList()) {
+                this.historyPatches.put(patch, true);
+              }
+            }
+
+            var versionName = "";
+            if (this.orchestratorStartedVersion != null && !StringUtils.isNullOrEmpty(this.orchestratorStartedVersion.getName())) {
+              versionName = this.orchestratorStartedVersion.getName();
+            }
+
             // Create and invoke the workflow orchestrator
-            TaskOrchestrationFactory factory = TaskOrchestrationExecutor.this.orchestrationFactories
-                .get(executionStarted.getName());
+            TaskOrchestrationFactory factory = TaskOrchestrationExecutor.this.orchestrationFactories.getOrchestrationFactory(executionStarted.getName(), versionName);
+
             if (factory == null) {
               // Try getting the default orchestrator
-              factory = TaskOrchestrationExecutor.this.orchestrationFactories.get("*");
+              factory = TaskOrchestrationExecutor.this.orchestrationFactories.getOrchestrationFactory("*");
             }
             // TODO: Throw if the factory is null (orchestration by that name doesn't exist)
             if (factory == null) {
               throw new IllegalStateException("No factory found for orchestrator: " + executionStarted.getName());
             }
 
+            this.version = factory.getVersionName();
+
             TaskOrchestration orchestrator = factory.create();
             orchestrator.run(this);
             break;
@@ -959,6 +1019,9 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
           case EXECUTIONTERMINATED:
             this.handleExecutionTerminated(e);
             break;
+          case EXECUTIONSTALLED:
+            this.logger.fine(() -> this.instanceId + ": Workflow execution stalled");
+            break;
           case TASKSCHEDULED:
             this.handleTaskScheduled(e);
             break;
@@ -998,6 +1061,22 @@ private void processEvent(OrchestratorService.HistoryEvent e) {
       }
     }
 
+    public void setVersionNotRegistered() {
+      this.pendingActions.clear();
+
+      OrchestratorService.CompleteOrchestrationAction.Builder builder = OrchestratorService.CompleteOrchestrationAction
+          .newBuilder();
+      builder.setOrchestrationStatus(OrchestratorService.OrchestrationStatus.ORCHESTRATION_STATUS_STALLED);
+
+      int id = this.sequenceNumber++;
+      OrchestratorService.OrchestratorAction action = OrchestratorService.OrchestratorAction.newBuilder()
+          .setId(id)
+          .setCompleteOrchestration(builder.build())
+          .build();
+      this.pendingActions.put(id, action);
+
+    }
+
     private class TaskRecord<V> {
       private final CompletableTask<V> task;
       private final String taskName;

durabletask-client/src/main/java/io/dapr/durabletask/TaskOrchestratorResult.java

@@ -17,17 +17,23 @@
 
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 
 final class TaskOrchestratorResult {
 
   private final Collection<OrchestratorService.OrchestratorAction> actions;
 
   private final String customStatus;
 
-  public TaskOrchestratorResult(Collection<OrchestratorService.OrchestratorAction> actions, String customStatus) {
+  private final String version;
+
+  private final List<String> patches;
+
+  public TaskOrchestratorResult(Collection<OrchestratorService.OrchestratorAction> actions, String customStatus, String version, List<String> patches) {
     this.actions = Collections.unmodifiableCollection(actions);
-    ;
     this.customStatus = customStatus;
+    this.version = version;
+    this.patches = patches;
   }
 
   public Collection<OrchestratorService.OrchestratorAction> getActions() {