Fix startWorkflow (#234)

`startWorkflow` requires information about the @workflow executed in the
provided lambda. Previously, we added the start options to the context
and executed the workflow. The @workflow info would end up in
invokeWorkflow where the behavior was different based on the presence of
start options. This approach creates thread pool exhaustion when
starting 1000s of workflows and creates significant complexity in
`DBOSContext` and `invokeWorkflow` to account for this difference in
execution.

This PR introduces a ThreadLocal hook method that `startWorkflow` can
set and `DBOSInvocationHandler` can call. This allows `startWorkflow` to
retrieve the class/instance/workflow names and the workflow argument
from the provided lambda without requiring a thread pool thread. Once we
have this information, we can call the private `executeWorkflow` method
directly. This simplifies the context significantly and eliminates the
need for `invokeWorkflow to do dual duty.

This PR also changes `executeWorkflow` to use
`CompletableFuture.supplyAsync` instead of `executorService.submit` to
run the workflow. This will allow us to leverage different executors
available on later versions of Java. However, this work will be in a
separate PR.

Note, the only changes to the tests are
* Consistent use of DBOSAwaitedWorkflowCancelledException instead of
CancellationException
* `parentTimeoutInheritedByChild` no longer requires retries

fixes #209
fixes #227

Author: devhawk

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

@@ -679,7 +679,7 @@ public static Collection<RegisteredWorkflowInstance> getRegisteredWorkflowInstan
    */
   public static WorkflowHandle<?, ?> startWorkflow(
       RegisteredWorkflow regWorkflow, Object[] args, StartWorkflowOptions options) {
-    return executor("executeWorkflow").executeWorkflow(regWorkflow, args, options, null, null);
+    return executor("startWorkflow").startWorkflow(regWorkflow, args, options);
   }
 
   /**

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

@@ -119,23 +119,31 @@ public record EnqueueOptions(
     public EnqueueOptions {
       if (Objects.requireNonNull(workflowName, "EnqueueOptions workflowName must not be null")
           .isEmpty()) {
-        throw new IllegalArgumentException("workflowName must not be empty");
+        throw new IllegalArgumentException("EnqueueOptions workflowName must not be empty");
       }
 
       if (Objects.requireNonNull(queueName, "EnqueueOptions queueName must not be null")
           .isEmpty()) {
-        throw new IllegalArgumentException("queueName must not be empty");
+        throw new IllegalArgumentException("EnqueueOptions queueName must not be empty");
       }
 
       if (Objects.requireNonNull(className, "EnqueueOptions className must not be null")
           .isEmpty()) {
-        throw new IllegalArgumentException("className must not be empty");
+        throw new IllegalArgumentException("EnqueueOptions className must not be empty");
       }
 
       if (instanceName == null) instanceName = "";
 
-      if (timeout != null && timeout.isNegative()) {
-        throw new IllegalArgumentException("timeout must not be negative");
+      if (timeout != null) {
+        if (timeout.isNegative() || timeout.isZero()) {
+          throw new IllegalArgumentException(
+              "EnqueueOptions timeout must be a positive non-zero duration");
+        }
+
+        if (deadline != null) {
+          throw new IllegalArgumentException(
+              "EnqueueOptions timeout and deadline cannot both be set");
+        }
       }
     }
 
@@ -340,19 +348,18 @@ public <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
         Objects.requireNonNullElse(options.instanceName(), ""),
         null,
         args,
-        new StartWorkflowOptions(
+        new DBOSExecutor.ExecutionOptions(
             Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString()),
             Timeout.of(options.timeout()),
+            options.deadline,
             Objects.requireNonNull(
                 options.queueName(), "EnqueueOptions queueName must not be null"),
             options.deduplicationId,
-            options.priority,
-            options.deadline),
+            options.priority),
         null,
         null,
         options.appVersion,
-        systemDatabase,
-        null);
+        systemDatabase);
   }
 
   /**

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

@@ -27,15 +27,21 @@
 public record StartWorkflowOptions(
     String workflowId,
     Timeout timeout,
+    Instant deadline,
     String queueName,
     String deduplicationId,
-    Integer priority,
-    Instant deadline) {
+    Integer priority) {
 
   public StartWorkflowOptions {
     if (timeout instanceof Timeout.Explicit explicit) {
       if (explicit.value().isNegative() || explicit.value().isZero()) {
-        throw new IllegalArgumentException("timeout must be a positive non-zero duration");
+        throw new IllegalArgumentException(
+            "StartWorkflowOptions explicit timeout must be a positive non-zero duration");
+      }
+
+      if (deadline != null) {
+        throw new IllegalArgumentException(
+            "StartWorkflowOptions explicit timeout and deadline cannot both be set");
       }
     }
   }
@@ -55,36 +61,21 @@ public StartWorkflowOptions withWorkflowId(String workflowId) {
     return new StartWorkflowOptions(
         workflowId,
         this.timeout,
+        this.deadline,
         this.queueName,
         this.deduplicationId,
-        this.priority,
-        this.deadline);
+        this.priority);
   }
 
   /** Produces a new StartWorkflowOptions that overrides timeout value for the started workflow */
   public StartWorkflowOptions withTimeout(Timeout timeout) {
-    if (timeout != null && this.deadline != null) {
-      throw new IllegalArgumentException(
-          "should not specify a timeout if the deadline is already set");
-    }
     return new StartWorkflowOptions(
         this.workflowId,
         timeout,
+        this.deadline,
         this.queueName,
         this.deduplicationId,
-        this.priority,
-        this.deadline);
-  }
-
-  /** Produces a new StartWorkflowOptions that overrides deadline value for the started workflow */
-  public StartWorkflowOptions withDeadline(Instant deadline) {
-    return new StartWorkflowOptions(
-        this.workflowId,
-        this.timeout,
-        this.queueName,
-        this.deduplicationId,
-        this.priority,
-        deadline);
+        this.priority);
   }
 
   /** Produces a new StartWorkflowOptions that overrides timeout value for the started workflow */
@@ -102,16 +93,27 @@ public StartWorkflowOptions withNoTimeout() {
     return new StartWorkflowOptions(
         this.workflowId,
         Timeout.none(),
+        this.deadline,
         this.queueName,
         this.deduplicationId,
-        this.priority,
-        this.deadline);
+        this.priority);
+  }
+
+  /** Produces a new StartWorkflowOptions that overrides deadline value for the started workflow */
+  public StartWorkflowOptions withDeadline(Instant deadline) {
+    return new StartWorkflowOptions(
+        this.workflowId,
+        this.timeout,
+        deadline,
+        this.queueName,
+        this.deduplicationId,
+        this.priority);
   }
 
   /** Produces a new StartWorkflowOptions that assigns the started workflow to a queue */
   public StartWorkflowOptions withQueue(String queue) {
     return new StartWorkflowOptions(
-        this.workflowId, this.timeout, queue, this.deduplicationId, this.priority, this.deadline);
+        this.workflowId, this.timeout, this.deadline, queue, this.deduplicationId, this.priority);
   }
 
   /** Produces a new StartWorkflowOptions that assigns the started workflow to a queue */
@@ -127,10 +129,10 @@ public StartWorkflowOptions withDeduplicationId(String deduplicationId) {
     return new StartWorkflowOptions(
         this.workflowId,
         this.timeout,
+        this.deadline,
         this.queueName,
         deduplicationId,
-        this.priority,
-        this.deadline);
+        this.priority);
   }
 
   /**
@@ -141,23 +143,15 @@ public StartWorkflowOptions withPriority(Integer priority) {
     return new StartWorkflowOptions(
         this.workflowId,
         this.timeout,
+        this.deadline,
         this.queueName,
         this.deduplicationId,
-        priority,
-        this.deadline);
+        priority);
   }
 
   /** Get the assigned workflow ID, replacing empty with null */
   @Override
   public String workflowId() {
     return workflowId != null && workflowId.isEmpty() ? null : workflowId;
   }
-
-  /** Get timeout duration */
-  public Duration getTimeoutDuration() {
-    if (timeout instanceof Timeout.Explicit e) {
-      return e.value();
-    }
-    return null;
-  }
 }

transact/src/main/java/dev/dbos/transact/context/DBOSContext.java

@@ -5,7 +5,6 @@
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 
 public class DBOSContext {
@@ -14,8 +13,6 @@ public class DBOSContext {
   String nextWorkflowId;
   Timeout nextTimeout;
   Instant nextDeadline;
-  StartWorkflowOptions startOptions;
-  String startedWorkflowId;
 
   // current workflow fields
   private final String workflowId;
@@ -24,7 +21,6 @@ public class DBOSContext {
   private final WorkflowInfo parent;
   private final Duration timeout;
   private final Instant deadline;
-  private CompletableFuture<String> startWorkflowFuture;
 
   // private StepStatus stepStatus;
 
@@ -58,14 +54,6 @@ public DBOSContext(
     this.parent = other.parent;
     this.timeout = other.timeout;
     this.deadline = other.deadline;
-
-    if (other.startedWorkflowId != null) {
-      throw new IllegalStateException("startedWorkflowId not null");
-    }
-
-    this.startOptions = options;
-    this.startWorkflowFuture = future;
-    this.startedWorkflowId = null;
   }
 
   public boolean isInWorkflow() {
@@ -109,9 +97,6 @@ public String getNextWorkflowId() {
   }
 
   public String getNextWorkflowId(String workflowId) {
-    if (startOptions != null && startOptions.workflowId() != null) {
-      return startOptions.workflowId();
-    }
     if (nextWorkflowId != null) {
       var value = nextWorkflowId;
       this.nextWorkflowId = null;
@@ -122,10 +107,6 @@ public String getNextWorkflowId(String workflowId) {
   }
 
   public Timeout getNextTimeout() {
-    if (startOptions != null && startOptions.timeout() != null) {
-      return startOptions.timeout();
-    }
-
     return nextTimeout;
   }
 
@@ -134,66 +115,13 @@ public Duration getTimeout() {
   }
 
   public Instant getNextDeadline() {
-    if (startOptions != null && startOptions.deadline() != null) {
-      return startOptions.deadline();
-    }
-
     return nextDeadline;
   }
 
   public Instant getDeadline() {
     return deadline;
   }
 
-  public String getQueueName() {
-    if (startOptions != null) {
-      return startOptions.queueName();
-    }
-    return null;
-  }
-
-  public String getDeduplicationId() {
-    if (startOptions != null) {
-      return startOptions.deduplicationId();
-    }
-    return null;
-  }
-
-  public Integer getPriority() {
-    if (startOptions != null) {
-      return startOptions.priority();
-    }
-    return null;
-  }
-
-  public void setStartOptions(StartWorkflowOptions options, CompletableFuture<String> future) {
-    if (startedWorkflowId != null) {
-      throw new IllegalStateException();
-    }
-    startOptions = options;
-    startWorkflowFuture = future;
-  }
-
-  public boolean validateStartedWorkflow() {
-    return startOptions == null || startedWorkflowId == null;
-  }
-
-  public void setStartedWorkflowId(String workflowId) {
-    if (startOptions != null) {
-      if (startedWorkflowId != null) {
-        throw new IllegalStateException(
-            String.format(
-                "more than one workflow called from start workflow lambda: %s %s",
-                workflowId, startedWorkflowId));
-      }
-      startedWorkflowId = Objects.requireNonNull(workflowId, "workflowId must not be null");
-    }
-  }
-
-  public CompletableFuture<String> getStartWorkflowFuture() {
-    return this.startWorkflowFuture;
-  }
-
   public static String workflowId() {
     var ctx = DBOSContextHolder.get();
     return ctx == null ? null : ctx.workflowId;

transact/src/main/java/dev/dbos/transact/context/WorkflowOptions.java

@@ -19,13 +19,15 @@
 public record WorkflowOptions(String workflowId, Timeout timeout, Instant deadline) {
 
   public WorkflowOptions {
-    if (timeout != null && timeout instanceof Timeout.Explicit explicit) {
+    if (timeout instanceof Timeout.Explicit explicit) {
       if (explicit.value().isNegative() || explicit.value().isZero()) {
-        throw new IllegalArgumentException("timeout must be a positive non-zero duration");
+        throw new IllegalArgumentException(
+            "WorkflowOptions explicit timeout must be a positive non-zero duration");
       }
-      if (deadline != null && timeout instanceof Timeout.Explicit) {
+
+      if (deadline != null) {
         throw new IllegalArgumentException(
-            "WorkflowOptions may not specify both `timeout` and `deadline`");
+            "WorkflowOptions explicit timeout and deadline cannot both be set");
       }
     }
   }