App Version Support (#327)

fixes #314

Author: devhawk

build.gradle.kts

@@ -160,8 +160,12 @@ subprojects {
     tasks.withType<Test> {
       useJUnitPlatform()
       testLogging {
-        events("passed", "skipped", "failed")
+        events("failed")
         showStandardStreams = true
+        showExceptions = true
+        showCauses = true
+        showStackTraces = true
+        exceptionFormat = org.gradle.api.tasks.testing.logging.TestExceptionFormat.FULL
       }
       addTestListener(
         object : TestListener {

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -19,6 +19,7 @@
 import dev.dbos.transact.workflow.SerializationStrategy;
 import dev.dbos.transact.workflow.StepInfo;
 import dev.dbos.transact.workflow.StepOptions;
+import dev.dbos.transact.workflow.VersionInfo;
 import dev.dbos.transact.workflow.Workflow;
 import dev.dbos.transact.workflow.WorkflowClassName;
 import dev.dbos.transact.workflow.WorkflowHandle;
@@ -731,6 +732,33 @@ public void deleteWorkflows(@NonNull List<String> workflowIds, boolean deleteChi
     return forkWorkflow(workflowId, startStep, new ForkOptions());
   }
 
+  /**
+   * List all registered application versions, ordered by timestamp descending.
+   *
+   * @return list of {@link VersionInfo} records
+   */
+  public @NonNull List<VersionInfo> listApplicationVersions() {
+    return ensureLaunched("listApplicationVersions").listApplicationVersions();
+  }
+
+  /**
+   * Get the most recently promoted application version.
+   *
+   * @return the latest {@link VersionInfo}
+   */
+  public @NonNull VersionInfo getLatestApplicationVersion() {
+    return ensureLaunched("getLatestApplicationVersion").getLatestApplicationVersion();
+  }
+
+  /**
+   * Promote a version to be the latest application version.
+   *
+   * @param versionName the version to promote
+   */
+  public void setLatestApplicationVersion(@NonNull String versionName) {
+    ensureLaunched("setLatestApplicationVersion").setLatestApplicationVersion(versionName);
+  }
+
   /**
    * Retrieve a handle to a workflow, given its ID. Note that a handle is always returned, whether
    * the workflow exists or not; getStatus() can be used to tell the difference

transact/src/main/java/dev/dbos/transact/DBOSClient.java

@@ -11,6 +11,7 @@
 import dev.dbos.transact.workflow.SerializationStrategy;
 import dev.dbos.transact.workflow.StepInfo;
 import dev.dbos.transact.workflow.Timeout;
+import dev.dbos.transact.workflow.VersionInfo;
 import dev.dbos.transact.workflow.WorkflowHandle;
 import dev.dbos.transact.workflow.WorkflowState;
 import dev.dbos.transact.workflow.WorkflowStatus;
@@ -516,13 +517,14 @@ public EnqueueOptions(
                 options.deduplicationId,
                 options.priority,
                 options.queuePartitionKey,
+                options.appVersion,
                 false,
                 false,
                 serializationFormat),
             null,
             null,
             null,
-            options.appVersion,
+            null,
             systemDatabase,
             this.serializer);
 
@@ -800,4 +802,31 @@ public void deleteWorkflows(@NonNull List<String> workflowIds, boolean deleteChi
   public @NonNull List<StepInfo> listWorkflowSteps(@NonNull String workflowId) {
     return systemDatabase.listWorkflowSteps(workflowId);
   }
+
+  /**
+   * List all registered application versions, ordered by timestamp descending.
+   *
+   * @return list of {@link VersionInfo} records
+   */
+  public @NonNull List<VersionInfo> listApplicationVersions() {
+    return systemDatabase.listApplicationVersions();
+  }
+
+  /**
+   * Get the most recently promoted application version.
+   *
+   * @return the latest {@link VersionInfo}
+   */
+  public @NonNull VersionInfo getLatestApplicationVersion() {
+    return systemDatabase.getLatestApplicationVersion();
+  }
+
+  /**
+   * Promote an existing version to be the latest application version by updating its timestamp.
+   *
+   * @param versionName the version to promote; it must already exist
+   */
+  public void setLatestApplicationVersion(@NonNull String versionName) {
+    systemDatabase.updateApplicationVersionTimestamp(versionName, Instant.now());
+  }
 }

transact/src/main/java/dev/dbos/transact/StartWorkflowOptions.java

@@ -36,7 +36,8 @@ public record StartWorkflowOptions(
     @Nullable String queueName,
     @Nullable String deduplicationId,
     @Nullable Integer priority,
-    @Nullable String queuePartitionKey) {
+    @Nullable String queuePartitionKey,
+    @Nullable String appVersion) {
 
   public StartWorkflowOptions {
     if (timeout instanceof Timeout.Explicit explicit) {
@@ -64,17 +65,17 @@ public record StartWorkflowOptions(
 
   /** Construct with default options */
   public StartWorkflowOptions() {
-    this(null, null, null, null, null, null, null);
+    this(null, null, null, null, null, null, null, null);
   }
 
   /** Construct with a specified workflow ID */
   public StartWorkflowOptions(String workflowId) {
-    this(workflowId, null, null, null, null, null, null);
+    this(workflowId, null, null, null, null, null, null, null);
   }
 
   /** Construct with a specified queue */
   public StartWorkflowOptions(@NonNull Queue queue) {
-    this(null, null, null, queue.name(), null, null, null);
+    this(null, null, null, queue.name(), null, null, null, null);
   }
 
   /** Produces a new StartWorkflowOptions that overrides the ID assigned to the started workflow */
@@ -86,7 +87,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         this.deduplicationId,
         this.priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /** Produces a new StartWorkflowOptions that overrides timeout value for the started workflow */
@@ -98,7 +100,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         this.deduplicationId,
         this.priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /** Produces a new StartWorkflowOptions that overrides timeout value for the started workflow */
@@ -125,7 +128,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         this.deduplicationId,
         this.priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /** Produces a new StartWorkflowOptions that assigns the started workflow to a queue */
@@ -137,7 +141,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         queue,
         this.deduplicationId,
         this.priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /** Produces a new StartWorkflowOptions that assigns the started workflow to a queue */
@@ -157,7 +162,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         deduplicationId,
         this.priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /**
@@ -172,7 +178,8 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         this.deduplicationId,
         priority,
-        this.queuePartitionKey);
+        this.queuePartitionKey,
+        this.appVersion);
   }
 
   /** Produces a new StartWorkflowOptions that assigns a queue partition key */
@@ -184,7 +191,21 @@ public StartWorkflowOptions(@NonNull Queue queue) {
         this.queueName,
         this.deduplicationId,
         this.priority,
-        queuePartitionKey);
+        queuePartitionKey,
+        this.appVersion);
+  }
+
+  /** Produces a new StartWorkflowOptions that assigns an app version */
+  public @NonNull StartWorkflowOptions withAppVersion(@Nullable String appVersion) {
+    return new StartWorkflowOptions(
+        this.workflowId,
+        this.timeout,
+        this.deadline,
+        this.queueName,
+        this.deduplicationId,
+        this.priority,
+        this.queuePartitionKey,
+        appVersion);
   }
 
   /** Get the assigned workflow ID, replacing empty with null */

transact/src/main/java/dev/dbos/transact/conductor/Conductor.java

@@ -17,6 +17,7 @@
 import dev.dbos.transact.conductor.protocol.GetWorkflowRequest;
 import dev.dbos.transact.conductor.protocol.GetWorkflowResponse;
 import dev.dbos.transact.conductor.protocol.ImportWorkflowRequest;
+import dev.dbos.transact.conductor.protocol.ListApplicationVersionsResponse;
 import dev.dbos.transact.conductor.protocol.ListQueuedWorkflowsRequest;
 import dev.dbos.transact.conductor.protocol.ListStepsRequest;
 import dev.dbos.transact.conductor.protocol.ListStepsResponse;
@@ -26,6 +27,7 @@
 import dev.dbos.transact.conductor.protocol.RestartRequest;
 import dev.dbos.transact.conductor.protocol.ResumeRequest;
 import dev.dbos.transact.conductor.protocol.RetentionRequest;
+import dev.dbos.transact.conductor.protocol.SetLatestApplicationVersionRequest;
 import dev.dbos.transact.conductor.protocol.SuccessResponse;
 import dev.dbos.transact.conductor.protocol.WorkflowOutputsResponse;
 import dev.dbos.transact.conductor.protocol.WorkflowsOutput;
@@ -709,13 +711,15 @@ CompletableFuture<BaseResponse> getResponseAsync(BaseMessage message, WebSocket
       case GET_METRICS -> handleGetMetrics(this, message);
       case GET_WORKFLOW -> handleGetWorkflow(this, message);
       case IMPORT_WORKFLOW -> handleImportWorkflow(this, message);
+      case LIST_APPLICATION_VERSIONS -> handleListApplicationVersions(this, message);
       case LIST_QUEUED_WORKFLOWS -> handleListQueuedWorkflows(this, message);
       case LIST_STEPS -> handleListSteps(this, message);
       case LIST_WORKFLOWS -> handleListWorkflows(this, message);
       case RECOVERY -> handleRecovery(this, message);
       case RESTART -> handleRestart(this, message);
       case RESUME -> handleResume(this, message);
       case RETENTION -> handleRetention(this, message);
+      case SET_LATEST_APPLICATION_VERSION -> handleSetLatestApplicationVersion(this, message);
     };
   }
 
@@ -878,6 +882,38 @@ static CompletableFuture<BaseResponse> handleListQueuedWorkflows(
         });
   }
 
+  static CompletableFuture<BaseResponse> handleListApplicationVersions(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          try {
+            var output = conductor.systemDatabase.listApplicationVersions();
+            return new ListApplicationVersionsResponse(message, output);
+          } catch (Exception e) {
+            logger.error("Exception encountered when listing application versions", e);
+            return new ListApplicationVersionsResponse(message, e);
+          }
+        });
+  }
+
+  static CompletableFuture<BaseResponse> handleSetLatestApplicationVersion(
+      Conductor conductor, BaseMessage message) {
+    return CompletableFuture.supplyAsync(
+        () -> {
+          SetLatestApplicationVersionRequest request = (SetLatestApplicationVersionRequest) message;
+          try {
+            conductor.dbosExecutor.setLatestApplicationVersion(request.version_name);
+            return new SuccessResponse(request, true);
+          } catch (Exception e) {
+            logger.error(
+                "Exception encountered when setting latest application version to {}",
+                request.version_name,
+                e);
+            return new SuccessResponse(request, e);
+          }
+        });
+  }
+
   static CompletableFuture<BaseResponse> handleListSteps(Conductor conductor, BaseMessage message) {
     return CompletableFuture.supplyAsync(
         () -> {

transact/src/main/java/dev/dbos/transact/conductor/protocol/BaseMessage.java

@@ -20,13 +20,19 @@
   @JsonSubTypes.Type(value = GetMetricsRequest.class, name = "get_metrics"),
   @JsonSubTypes.Type(value = GetWorkflowRequest.class, name = "get_workflow"),
   @JsonSubTypes.Type(value = ImportWorkflowRequest.class, name = "import_workflow"),
+  @JsonSubTypes.Type(
+      value = ListApplicationVersionsRequest.class,
+      name = "list_application_versions"),
   @JsonSubTypes.Type(value = ListQueuedWorkflowsRequest.class, name = "list_queued_workflows"),
   @JsonSubTypes.Type(value = ListStepsRequest.class, name = "list_steps"),
   @JsonSubTypes.Type(value = ListWorkflowsRequest.class, name = "list_workflows"),
   @JsonSubTypes.Type(value = RecoveryRequest.class, name = "recovery"),
   @JsonSubTypes.Type(value = RestartRequest.class, name = "restart"),
   @JsonSubTypes.Type(value = ResumeRequest.class, name = "resume"),
   @JsonSubTypes.Type(value = RetentionRequest.class, name = "retention"),
+  @JsonSubTypes.Type(
+      value = SetLatestApplicationVersionRequest.class,
+      name = "set_latest_application_version"),
 })
 @JsonIgnoreProperties(ignoreUnknown = true)
 public abstract class BaseMessage {

transact/src/main/java/dev/dbos/transact/conductor/protocol/GetMetricsResponse.java

@@ -4,21 +4,21 @@
 
 import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
 
 public class GetMetricsResponse extends BaseResponse {
-  public static record MetricsDataOutput(String metric_type, String metric_name, long value) {}
+  public static record MetricsDataOutput(String metric_type, String metric_name, long value) {
+    public static MetricsDataOutput fromMetricData(MetricData m) {
+      return new MetricsDataOutput(m.metricType(), m.metricName(), m.value());
+    }
+  }
 
   public List<MetricsDataOutput> metrics;
 
   public GetMetricsResponse() {}
 
   public GetMetricsResponse(BaseMessage message, List<MetricData> metrics) {
     super(message.type, message.request_id);
-    this.metrics =
-        metrics.stream()
-            .map(m -> new MetricsDataOutput(m.metricType(), m.metricName(), m.value()))
-            .collect(Collectors.toList());
+    this.metrics = metrics.stream().map(MetricsDataOutput::fromMetricData).toList();
   }
 
   public GetMetricsResponse(BaseMessage message, Exception ex) {

transact/src/main/java/dev/dbos/transact/conductor/protocol/ListApplicationVersionsRequest.java

@@ -0,0 +1,6 @@
+package dev.dbos.transact.conductor.protocol;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ListApplicationVersionsRequest extends BaseMessage {}

transact/src/main/java/dev/dbos/transact/conductor/protocol/ListApplicationVersionsResponse.java

@@ -0,0 +1,33 @@
+package dev.dbos.transact.conductor.protocol;
+
+import dev.dbos.transact.workflow.VersionInfo;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ListApplicationVersionsResponse extends BaseResponse {
+  public static record AppVersionInfo(
+      String version_id, String version_name, long version_timestamp, long created_at) {
+    public static AppVersionInfo fromVersionInfo(VersionInfo v) {
+      return new AppVersionInfo(
+          v.versionId(),
+          v.versionName(),
+          v.versionTimestamp().toEpochMilli(),
+          v.createdAt().toEpochMilli());
+    }
+  }
+
+  public List<AppVersionInfo> output;
+
+  public ListApplicationVersionsResponse() {}
+
+  public ListApplicationVersionsResponse(BaseMessage message, List<VersionInfo> versions) {
+    super(message.type, message.request_id);
+    this.output = versions.stream().map(AppVersionInfo::fromVersionInfo).toList();
+  }
+
+  public ListApplicationVersionsResponse(BaseMessage message, Exception ex) {
+    super(message.type, message.request_id, ex.getMessage());
+    this.output = Collections.emptyList();
+  }
+}