DBOSExecutor refactoring (#355)

The primary purpose of this PR is to move proxy workflow/step execution
from `DBOSInvocationHandler` to `DBOSExecutor` so that it can be
leveraged by alternative proxy infrastructure (like Spring Boot).
However, I took the opportunity to cleanup and refactor several aspects
of `DBOSExecutor` to make future changes easier.

> Note, the only test changes were due to internal method signature
changes

* reordered `DBOSExecutor` methods to group similar functionality
together
* added `DBOSExecutor.runStep` to be invoked by `DBOS.runStep` and
`DBOSInvocationHandler.invoke`
* renamed `callFunctionAsStep` to `runDbosFunctionAsStep` to better
indicate its usage
* `runStepInternal` is now private and is used by both `runStep` and
`runDbosFunctionAsStep`, eliminating some duplicate code
* moved invocation hook declarations and types into DBOSExecutor
(Invocation and hookHolder ThreadLocal)
* moved WorkflowHandleFuture.getResult implementation into DBOSExecutor
* removed StepOptions.retriesAllowed and strongly typed retryInterval
* assorted renames for clarity
* Changed SystemDatabase.getPendingWorkflows to use listWorkflows under
the hood. method now returns WorkflowStatus so I dropped
GetPendingWorkflowsOutput

Author: devhawk

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

@@ -568,7 +568,7 @@ public void setEvent(
   public <T, E extends Exception> T runStep(
       @NonNull ThrowingSupplier<T, E> stepfunc, @NonNull StepOptions opts) throws E {
 
-    return ensureLaunched("runStep").runStepInternal(stepfunc, opts, null);
+    return ensureLaunched("runStep").runStep(stepfunc, opts, null);
   }
 
   /**

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

@@ -7,6 +7,7 @@
 import dev.dbos.transact.database.StreamIterator;
 import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.execution.DBOSExecutor;
+import dev.dbos.transact.execution.ExecutionOptions;
 import dev.dbos.transact.json.DBOSSerializer;
 import dev.dbos.transact.json.PortableWorkflowException;
 import dev.dbos.transact.json.SerializationUtil;
@@ -537,33 +538,34 @@ public EnqueueOptions(
     }
 
     var workflowId =
-        DBOSExecutor.enqueueWorkflow(
-            options.workflowName(),
-            options.className(),
-            options.instanceName(),
-            null,
-            positionalArgs,
-            namedArgs,
-            new DBOSExecutor.ExecutionOptions(
-                Objects.requireNonNullElseGet(
-                    options.workflowId(), () -> UUID.randomUUID().toString()),
-                Timeout.of(options.timeout()),
-                options.deadline,
-                options.queueName(),
-                options.deduplicationId,
-                options.priority,
-                options.queuePartitionKey,
-                options.delay,
-                options.appVersion,
-                false,
-                false,
-                serializationFormat),
-            null,
-            null,
-            null,
-            null,
-            systemDatabase,
-            this.serializer);
+        Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString());
+
+    DBOSExecutor.enqueueWorkflow(
+        options.workflowName(),
+        options.className(),
+        options.instanceName(),
+        null,
+        positionalArgs,
+        namedArgs,
+        new ExecutionOptions(
+            workflowId,
+            Timeout.of(options.timeout()),
+            options.deadline,
+            options.queueName(),
+            options.deduplicationId,
+            options.priority,
+            options.queuePartitionKey,
+            options.delay,
+            options.appVersion,
+            false,
+            false,
+            serializationFormat),
+        null,
+        null,
+        null,
+        null,
+        systemDatabase,
+        this.serializer);
 
     return new WorkflowHandleClient<>(workflowId);
   }

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

@@ -11,7 +11,6 @@
 import dev.dbos.transact.workflow.WorkflowHandle;
 import dev.dbos.transact.workflow.WorkflowSchedule;
 import dev.dbos.transact.workflow.WorkflowStatus;
-import dev.dbos.transact.workflow.internal.GetPendingWorkflowsOutput;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
@@ -920,9 +919,9 @@ static CompletableFuture<BaseResponse> handleExistPendingWorkflows(
         () -> {
           ExistPendingWorkflowsRequest request = (ExistPendingWorkflowsRequest) message;
           try {
-            List<GetPendingWorkflowsOutput> pending =
+            var pending =
                 conductor.systemDatabase.getPendingWorkflows(
-                    request.executor_id, request.application_version);
+                    List.of(request.executor_id), request.application_version);
             return new ExistPendingWorkflowsResponse(request, !pending.isEmpty());
           } catch (Exception e) {
             logger.error("Exception encountered when checking for pending workflows", e);

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

@@ -9,7 +9,8 @@ public class DBOSContextHolder {
       ThreadLocal.withInitial(DBOSContext::new);
 
   public static @NonNull DBOSContext get() {
-    return contextHolder.get();
+    // contextHolder is never null. Even if cleared, it gets set back to the initial value.
+    return Objects.requireNonNull(contextHolder.get());
   }
 
   public static void clear() {

transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java

@@ -22,7 +22,6 @@
 import dev.dbos.transact.workflow.WorkflowSchedule;
 import dev.dbos.transact.workflow.WorkflowStatus;
 import dev.dbos.transact.workflow.WorkflowStream;
-import dev.dbos.transact.workflow.internal.GetPendingWorkflowsOutput;
 import dev.dbos.transact.workflow.internal.StepResult;
 import dev.dbos.transact.workflow.internal.WorkflowStatusInternal;
 
@@ -303,8 +302,8 @@ public List<WorkflowAggregateRow> getWorkflowAggregates(GetWorkflowAggregatesInp
     return dbRetry(() -> workflowDAO.getWorkflowAggregates(input));
   }
 
-  public List<GetPendingWorkflowsOutput> getPendingWorkflows(String executorId, String appVersion) {
-    return dbRetry(() -> workflowDAO.getPendingWorkflows(executorId, appVersion));
+  public List<WorkflowStatus> getPendingWorkflows(List<String> executorIds, String appVersion) {
+    return dbRetry(() -> workflowDAO.getPendingWorkflows(executorIds, appVersion));
   }
 
   public boolean clearQueueAssignment(String workflowId) {

transact/src/main/java/dev/dbos/transact/database/WorkflowDAO.java

@@ -19,7 +19,6 @@
 import dev.dbos.transact.workflow.WorkflowDelay;
 import dev.dbos.transact.workflow.WorkflowState;
 import dev.dbos.transact.workflow.WorkflowStatus;
-import dev.dbos.transact.workflow.internal.GetPendingWorkflowsOutput;
 import dev.dbos.transact.workflow.internal.StepResult;
 import dev.dbos.transact.workflow.internal.WorkflowStatusInternal;
 
@@ -808,38 +807,14 @@ private static WorkflowStatus resultsToWorkflowStatus(
     return info;
   }
 
-  List<GetPendingWorkflowsOutput> getPendingWorkflows(String executorId, String appVersion)
+  List<WorkflowStatus> getPendingWorkflows(List<String> executorIds, String appVersion)
       throws SQLException {
-
-    final String sql =
-        """
-          SELECT workflow_uuid, queue_name
-          FROM "%s".workflow_status
-          WHERE status = ?
-            AND executor_id = ?
-            AND application_version = ?
-        """
-            .formatted(this.schema);
-
-    List<GetPendingWorkflowsOutput> results = new ArrayList<>();
-
-    try (Connection connection = dataSource.getConnection();
-        PreparedStatement stmt = connection.prepareStatement(sql)) {
-
-      stmt.setString(1, WorkflowState.PENDING.name());
-      stmt.setString(2, executorId);
-      stmt.setString(3, appVersion);
-
-      try (ResultSet rs = stmt.executeQuery()) {
-        while (rs.next()) {
-          results.add(
-              new GetPendingWorkflowsOutput(
-                  rs.getString("workflow_uuid"), rs.getString("queue_name")));
-        }
-      }
-    }
-
-    return results;
+    var input =
+        new ListWorkflowsInput()
+            .withStatus(WorkflowState.PENDING)
+            .withExecutorIds(executorIds)
+            .withApplicationVersion(appVersion);
+    return listWorkflows(input);
   }
 
   @SuppressWarnings("unchecked")