non public method support + more tests (#246)

* enable invocation of non public methods in `DBOSInvocationHandler`
(fixes #108)
* Use Timeout enum in `ForkOptions.timeout` instead of Duration
* class -> record
  * GetWorkflowEventContext
  * WorkflowInitResult
  * GetPendingWorkflowsOutput
* updated workflow and events tests to have non public service
interface/implementation in same file as tests

fixes #237 (was already implemented but added tests)
fixes #238 (was already implemented but added tests)

Author: devhawk

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

@@ -617,6 +617,21 @@ public static <T, E extends Exception> WorkflowHandle<T, E> forkWorkflow(
     return executor("forkWorkflow").forkWorkflow(workflowId, startStep, options);
   }
 
+  /**
+   * Fork the workflow. Re-execute with another Id from the step provided. Steps prior to the
+   * provided step are copied over
+   *
+   * @param <T> Return type of the workflow function
+   * @param <E> Checked exception thrown by the workflow function, if any
+   * @param workflowId Original workflow Id
+   * @param startStep Start execution from this step. Prior steps copied over
+   * @return handle to the workflow
+   */
+  public static <T, E extends Exception> WorkflowHandle<T, E> forkWorkflow(
+      String workflowId, int startStep) {
+    return forkWorkflow(workflowId, startStep, new ForkOptions());
+  }
+
   /**
    * 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/StartWorkflowOptions.java

@@ -105,14 +105,7 @@ public StartWorkflowOptions withTimeout(long value, TimeUnit unit) {
 
   /** Produces a new StartWorkflowOptions that removes the timeout behavior */
   public StartWorkflowOptions withNoTimeout() {
-    return new StartWorkflowOptions(
-        this.workflowId,
-        Timeout.none(),
-        this.deadline,
-        this.queueName,
-        this.deduplicationId,
-        this.priority,
-        this.queuePartitionKey);
+    return withTimeout(Timeout.none());
   }
 
   /** Produces a new StartWorkflowOptions that overrides deadline value for the started workflow */

transact/src/main/java/dev/dbos/transact/admin/AdminServer.java

@@ -198,8 +198,8 @@ private void listWorkflows(HttpExchange exchange) throws IOException {
   private void listQueuedWorkflows(HttpExchange exchange) throws IOException {
     if (!ensurePostJson(exchange)) return;
 
-    var request = mapper.readValue(exchange.getRequestBody(), ListWorkflowsRequest.class);
-    var input = request.asInput().withQueuesOnly();
+    var request = mapper.readValue(exchange.getRequestBody(), ListQueuedWorkflowsRequest.class);
+    var input = request.asInput();
     var workflows = systemDatabase.listWorkflows(input);
     var response = workflows.stream().map(WorkflowsOutput::of).collect(Collectors.toList());
     sendMappedJson(exchange, 200, response);

transact/src/main/java/dev/dbos/transact/admin/ListQueuedWorkflowsRequest.java

@@ -0,0 +1,42 @@
+package dev.dbos.transact.admin;
+
+import dev.dbos.transact.workflow.ListWorkflowsInput;
+
+import java.time.OffsetDateTime;
+import java.util.List;
+
+public record ListQueuedWorkflowsRequest(
+    String workflow_name,
+    String start_time,
+    String end_time,
+    String status,
+    String fork_from,
+    String queue_name,
+    Integer limit,
+    Integer offset,
+    Boolean sort_desc,
+    Boolean load_input) {
+
+  public ListWorkflowsInput asInput() {
+    return new ListWorkflowsInput(
+        null, // workflow ids
+        status != null ? List.of(status) : null,
+        start_time != null ? OffsetDateTime.parse(start_time) : null,
+        end_time != null ? OffsetDateTime.parse(end_time) : null,
+        workflow_name,
+        null, // class_name,
+        null, // instance_name
+        null, // app version
+        null, // auth user
+        limit,
+        offset,
+        sort_desc,
+        null, // wf id prefix
+        load_input,
+        false, // load output
+        queue_name,
+        true, // queuesOnly: only list queued workflows
+        null, // Executor IDs
+        fork_from);
+  }
+}

transact/src/main/java/dev/dbos/transact/admin/ListWorkflowsRequest.java

@@ -7,19 +7,19 @@
 
 public record ListWorkflowsRequest(
     List<String> workflow_uuids,
+    String workflow_name,
     String authenticated_user,
     String start_time,
     String end_time,
     String status,
     String application_version,
-    String workflow_name,
+    String fork_from,
     Integer limit,
     Integer offset,
     Boolean sort_desc,
     String workflow_id_prefix,
     Boolean load_input,
-    Boolean load_output,
-    String queue_name) {
+    Boolean load_output) {
 
   public ListWorkflowsInput asInput() {
     return new ListWorkflowsInput(
@@ -38,10 +38,9 @@ public ListWorkflowsInput asInput() {
         workflow_id_prefix,
         load_input,
         load_output,
-        queue_name,
-        queue_name != null ? true : false,
+        null,
+        false,
         null, // Executor IDs
-        null // forkedFrom
-        );
+        fork_from);
   }
 }

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

@@ -1,26 +1,3 @@
 package dev.dbos.transact.database;
 
-public class GetWorkflowEventContext {
-
-  private String workflowId;
-  private int functionId;
-  private int timeoutFunctionId;
-
-  public GetWorkflowEventContext(String workflowId, int functionId, int timeoutFunctionId) {
-    this.workflowId = workflowId;
-    this.functionId = functionId;
-    this.timeoutFunctionId = timeoutFunctionId;
-  }
-
-  public String getWorkflowId() {
-    return workflowId;
-  }
-
-  public int getFunctionId() {
-    return functionId;
-  }
-
-  public int getTimeoutFunctionId() {
-    return timeoutFunctionId;
-  }
-}
+public record GetWorkflowEventContext(String workflowId, int functionId, int timeoutFunctionId) {}

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

@@ -362,23 +362,19 @@ Object getEvent(
       try (Connection conn = dataSource.getConnection()) {
         recordedOutput =
             StepsDAO.checkStepExecutionTxn(
-                callerCtx.getWorkflowId(),
-                callerCtx.getFunctionId(),
-                functionName,
-                conn,
-                this.schema);
+                callerCtx.workflowId(), callerCtx.functionId(), functionName, conn, this.schema);
       }
 
       if (recordedOutput != null) {
-        logger.debug("Replaying getEvent, id: {}, key: {}", callerCtx.getFunctionId(), key);
+        logger.debug("Replaying getEvent, id: {}, key: {}", callerCtx.functionId(), key);
         if (recordedOutput.output() != null) {
           Object[] outputArray = JSONUtil.deserializeToArray(recordedOutput.output());
           return outputArray == null ? null : outputArray[0];
         } else {
           throw new RuntimeException("No output recorded in the last getEvent");
         }
       } else {
-        logger.debug("Running getEvent, id: {}, key: {}", callerCtx.getFunctionId(), key);
+        logger.debug("Running getEvent, id: {}, key: {}", callerCtx.functionId(), key);
       }
     }
 
@@ -398,7 +394,8 @@ Object getEvent(
               .formatted(this.schema);
 
       // Wait for the notification
-      double actualTimeout = timeout.toMillis();
+      double actualTimeout =
+          Objects.requireNonNull(timeout, "getEvent timeout cannot be null").toMillis();
       var targetTime = System.currentTimeMillis() + actualTimeout;
       var checkedDBForSleep = false;
       var hasExistingNotification = false;
@@ -431,8 +428,8 @@ Object getEvent(
           actualTimeout =
               StepsDAO.sleep(
                       dataSource,
-                      callerCtx.getWorkflowId(),
-                      callerCtx.getTimeoutFunctionId(),
+                      callerCtx.workflowId(),
+                      callerCtx.timeoutFunctionId(),
                       timeout,
                       true, // skip_sleep
                       this.schema)
@@ -456,7 +453,7 @@ Object getEvent(
       // Record the output if it's in a workflow
       if (callerCtx != null) {
         StepResult output =
-            new StepResult(callerCtx.getWorkflowId(), callerCtx.getFunctionId(), functionName)
+            new StepResult(callerCtx.workflowId(), callerCtx.functionId(), functionName)
                 .withOutput(JSONUtil.serialize(value));
         StepsDAO.recordStepResultTxn(
             dataSource, output, startTime, System.currentTimeMillis(), this.schema);

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

@@ -7,14 +7,14 @@
 import dev.dbos.transact.workflow.ErrorResult;
 import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.ListWorkflowsInput;
+import dev.dbos.transact.workflow.Timeout;
 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;
 
 import java.sql.*;
-import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
 
@@ -362,7 +362,7 @@ List<WorkflowStatus> listWorkflows(ListWorkflowsInput input) throws SQLException
     sqlBuilder.append(
         """
           SELECT
-            workflow_uuid, status,  forked_from,
+            workflow_uuid, status, forked_from,
             name, class_name, config_name,
             queue_name, deduplication_id, priority, queue_partition_key,
             executor_id, application_version, application_id,
@@ -787,6 +787,8 @@ String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions option
       throw new IllegalStateException("Database is closed!");
     }
 
+    Objects.requireNonNull(options);
+
     var status = getWorkflowStatus(originalWorkflowId);
     if (status == null) {
       throw new DBOSNonExistentWorkflowException(originalWorkflowId);
@@ -801,12 +803,12 @@ String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions option
 
     String applicationVersion = options.applicationVersion();
 
-    var timeout = options.timeout();
-    if (timeout == null) {
-      timeout = status.getTimeout();
-    }
-    if (timeout == null) {
-      timeout = Duration.ZERO;
+    var timeout = Objects.requireNonNullElse(options.timeout(), Timeout.inherit());
+    Long timeoutMS = null;
+    if (timeout instanceof Timeout.Inherit) {
+      timeoutMS = status.timeoutMs();
+    } else if (timeout instanceof Timeout.Explicit explicit) {
+      timeoutMS = explicit.value().toMillis();
     }
 
     try (Connection connection = dataSource.getConnection()) {
@@ -820,7 +822,7 @@ String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions option
             forkedWorkflowId,
             status,
             applicationVersion,
-            timeout.toMillis(),
+            timeoutMS,
             this.schema);
 
         // Copy operation outputs if starting from step > 0
@@ -845,21 +847,17 @@ private static void insertForkedWorkflowStatus(
       String forkedWorkflowId,
       WorkflowStatus originalStatus,
       String applicationVersion,
-      long timeoutMs,
+      Long timeoutMS,
       String schema)
       throws SQLException {
     Objects.requireNonNull(schema);
-    long workflowDeadlineEpoch = 0;
-    if (timeoutMs > 0) {
-      workflowDeadlineEpoch = System.currentTimeMillis() + timeoutMs;
-    }
 
     String sql =
         """
           INSERT INTO %s.workflow_status (
             workflow_uuid, status, name, class_name, config_name, application_version, application_id,
-            authenticated_user, authenticated_roles, assumed_role, queue_name, inputs, workflow_deadline_epoch_ms, workflow_timeout_ms, forked_from
-          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            authenticated_user, authenticated_roles, assumed_role, queue_name, inputs, workflow_timeout_ms, forked_from
+          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """
             .formatted(schema);
 
@@ -876,9 +874,8 @@ private static void insertForkedWorkflowStatus(
       stmt.setString(10, originalStatus.assumedRole());
       stmt.setString(11, Constants.DBOS_INTERNAL_QUEUE);
       stmt.setString(12, JSONUtil.serializeArray(originalStatus.input()));
-      stmt.setLong(13, workflowDeadlineEpoch);
-      stmt.setObject(14, originalStatus.timeoutMs());
-      stmt.setString(15, originalWorkflowId);
+      stmt.setObject(13, timeoutMS);
+      stmt.setString(14, originalWorkflowId);
 
       stmt.executeUpdate();
     }

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

@@ -2,32 +2,9 @@
 
 import java.util.Objects;
 
-public class WorkflowInitResult {
-  private String workflowId;
-  private String status;
-  private Long deadlineEpochMS; // Use Long for nullable number
-  private boolean shouldExecuteOnThisExecutor;
-
-  public WorkflowInitResult(String id, String status, Long deadlineEpochMS, boolean shouldExecute) {
-    this.workflowId = id;
-    this.status = status;
-    this.deadlineEpochMS = deadlineEpochMS;
-    this.shouldExecuteOnThisExecutor = shouldExecute;
-  }
-
-  public String getStatus() {
-    return status;
-  }
-
-  public Long getDeadlineEpochMS() {
+public record WorkflowInitResult(
+    String workflowId, String status, Long deadlineEpochMS, boolean shouldExecuteOnThisExecutor) {
+  public Long deadlineEpochMS() {
     return Objects.requireNonNullElse(deadlineEpochMS, 0L);
   }
-
-  public String getWorkflowId() {
-    return workflowId;
-  }
-
-  public boolean shouldExecuteOnThisExecutor() {
-    return shouldExecuteOnThisExecutor;
-  }
 }