Event receiver infra and scheduler improvements (#200)

**Infrastructure for pluggable event receivers**
Add lifecycle listeners and workflow execution entrypoints.
Allow deadline to be specified in start options (consolidates code also)
Test for same.

**Scheduler improvements:**
Make scheduler less special (uses event receivers except for its
validator)
Use user-specified queues.
Add mode where missed executions are not ignored, but made up
Fix scheduler's key calculation to be on second boundaries (as it should
be)
Check all things in one test, rather than one per run, saving over 10
seconds of test time

Author: chuck-dbos

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

@@ -5,6 +5,9 @@
 import dev.dbos.transact.context.DBOSContextHolder;
 import dev.dbos.transact.database.ExternalState;
 import dev.dbos.transact.execution.DBOSExecutor;
+import dev.dbos.transact.execution.DBOSLifecycleListener;
+import dev.dbos.transact.execution.RegisteredWorkflow;
+import dev.dbos.transact.execution.RegisteredWorkflowInstance;
 import dev.dbos.transact.execution.ThrowingRunnable;
 import dev.dbos.transact.execution.ThrowingSupplier;
 import dev.dbos.transact.internal.DBOSInvocationHandler;
@@ -17,9 +20,13 @@
 
 import java.lang.reflect.Method;
 import java.time.Duration;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
@@ -38,6 +45,7 @@ private DBOS() {}
   public static class Instance {
     private final WorkflowRegistry workflowRegistry = new WorkflowRegistry();
     private final QueueRegistry queueRegistry = new QueueRegistry();
+    private final Set<DBOSLifecycleListener> lifecycleRegistry = ConcurrentHashMap.newKeySet();
 
     private DBOSConfig config;
 
@@ -61,33 +69,40 @@ private void registerClassWorkflows(
         throw new IllegalStateException("Cannot register workflow after DBOS is launched");
       }
 
+      String className = implementation.getClass().getName();
+      workflowRegistry.register(interfaceClass, implementation, className, instanceName);
+
       Method[] methods = implementation.getClass().getDeclaredMethods();
       for (Method method : methods) {
         Workflow wfAnnotation = method.getAnnotation(Workflow.class);
         if (wfAnnotation != null) {
           method.setAccessible(true); // In case it's not public
-          registerWorkflowMethod(wfAnnotation, implementation, instanceName, method);
+          registerWorkflowMethod(wfAnnotation, implementation, className, instanceName, method);
         }
       }
     }
 
     private String registerWorkflowMethod(
-        Workflow wfTag, Object target, String instanceName, Method method) {
+        Workflow wfTag, Object target, String className, String instanceName, Method method) {
       if (dbosExecutor.get() != null) {
         throw new IllegalStateException("Cannot register workflow after DBOS is launched");
       }
 
       String name = wfTag.name().isEmpty() ? method.getName() : wfTag.name();
       workflowRegistry.register(
-          target.getClass().getName(),
-          name,
-          target,
-          instanceName,
-          method,
-          wfTag.maxRecoveryAttempts());
+          className, name, target, instanceName, method, wfTag.maxRecoveryAttempts());
       return name;
     }
 
+    void registerLifecycleListener(DBOSLifecycleListener l) {
+      if (dbosExecutor.get() != null) {
+        throw new IllegalStateException(
+            "Cannot register lifecycle listener after DBOS is launched");
+      }
+
+      lifecycleRegistry.add(l);
+    }
+
     void registerQueue(Queue queue) {
       if (dbosExecutor.get() != null) {
         throw new IllegalStateException("Cannot build a queue after DBOS is launched");
@@ -117,6 +132,7 @@ private void registerInternals() {
     void clearRegistry() {
       workflowRegistry.clear();
       queueRegistry.clear();
+      lifecycleRegistry.clear();
 
       registerInternals();
     }
@@ -153,7 +169,12 @@ public void launch() {
         var executor = new DBOSExecutor(config);
 
         if (dbosExecutor.compareAndSet(null, executor)) {
-          executor.start(this, workflowRegistry.getSnapshot(), queueRegistry.getSnapshot());
+          executor.start(
+              this,
+              new HashSet<DBOSLifecycleListener>(this.lifecycleRegistry),
+              workflowRegistry.getWorkflowSnapshot(),
+              workflowRegistry.getInstanceSnapshot(),
+              queueRegistry.getSnapshot());
         }
       }
     }
@@ -211,6 +232,15 @@ public static Queue registerQueue(Queue queue) {
     return queue;
   }
 
+  /**
+   * Register a lifecycle listener that receives callbacks when DBOS is launched or shut down
+   *
+   * @param listener
+   */
+  public static void registerLifecycleListener(DBOSLifecycleListener listener) {
+    ensureInstance().registerLifecycleListener(listener);
+  }
+
   /**
    * Reinitializes the singleton instance of DBOS with config. For use in tests that reinitialize
    * DBOS @DBOSConfig config dbos configuration
@@ -620,6 +650,38 @@ public static List<StepInfo> listWorkflowSteps(String workflowId) {
     return executor("listWorkflowSteps").listWorkflowSteps(workflowId);
   }
 
+  /**
+   * Get all workflows registered with DBOS.
+   *
+   * @return list of all registered workflow methods
+   */
+  public static Collection<RegisteredWorkflow> getRegisteredWorkflows() {
+    return executor("getRegisteredWorkflows").getWorkflows();
+  }
+
+  /**
+   * Get all workflow classes registered with DBOS.
+   *
+   * @return list of all class instances containing registered workflow methods
+   */
+  public static Collection<RegisteredWorkflowInstance> getRegisteredWorkflowInstances() {
+    return executor("getRegisteredWorkflowInstances").getInstances();
+  }
+
+  /**
+   * Execute a workflow based on registration and arguments. This is expected to be used by generic
+   * callers, not app code.
+   *
+   * @param regWorkflow Registration of the workflow. @see getRegisteredWorkflows
+   * @param args Workflow function arguments
+   * @param options Execution options, such as ID, queue, and timeout/deadline
+   * @return WorkflowHandle to the executed workflow
+   */
+  public static WorkflowHandle<?, ?> startWorkflow(
+      RegisteredWorkflow regWorkflow, Object[] args, StartWorkflowOptions options) {
+    return executor("executeWorkflow").executeWorkflow(regWorkflow, args, options, null, null);
+  }
+
   /**
    * Get a system database record stored by an external service A unique value is stored per
    * combination of service, workflowName, and key

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

@@ -2,16 +2,17 @@
 
 import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.execution.DBOSExecutor;
-import dev.dbos.transact.execution.DBOSExecutor.ExecuteWorkflowOptions;
 import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.ListWorkflowsInput;
 import dev.dbos.transact.workflow.StepInfo;
+import dev.dbos.transact.workflow.Timeout;
 import dev.dbos.transact.workflow.WorkflowHandle;
 import dev.dbos.transact.workflow.WorkflowState;
 import dev.dbos.transact.workflow.WorkflowStatus;
 import dev.dbos.transact.workflow.internal.WorkflowStatusInternal;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -89,6 +90,7 @@ public record EnqueueOptions(
       String workflowId,
       String appVersion,
       Duration timeout,
+      Instant deadline,
       String deduplicationId,
       Integer priority) {
 
@@ -117,7 +119,7 @@ public record EnqueueOptions(
 
     /** Construct `EnqueueOptions` with a minimum set of required options */
     public EnqueueOptions(String className, String workflowName, String queueName) {
-      this(workflowName, queueName, className, "", null, null, null, null, null);
+      this(workflowName, queueName, className, "", null, null, null, null, null, null);
     }
 
     /**
@@ -135,6 +137,7 @@ public EnqueueOptions withClassName(String className) {
           this.workflowId,
           this.appVersion,
           this.timeout,
+          this.deadline,
           this.deduplicationId,
           this.priority);
     }
@@ -155,6 +158,7 @@ public EnqueueOptions withWorkflowId(String workflowId) {
           workflowId,
           this.appVersion,
           this.timeout,
+          this.deadline,
           this.deduplicationId,
           this.priority);
     }
@@ -175,6 +179,7 @@ public EnqueueOptions withAppVersion(String appVersion) {
           this.workflowId,
           appVersion,
           this.timeout,
+          this.deadline,
           this.deduplicationId,
           this.priority);
     }
@@ -195,6 +200,28 @@ public EnqueueOptions withTimeout(Duration timeout) {
           this.workflowId,
           this.appVersion,
           timeout,
+          this.deadline,
+          this.deduplicationId,
+          this.priority);
+    }
+
+    /**
+     * Specify a deadline for the workflow. This is an absolute time, regardless of when the
+     * workflow starts.
+     *
+     * @param deadline Instant after which the workflow will be canceled.
+     * @return New `EnqueueOptions` with the deadline set
+     */
+    public EnqueueOptions withDeadline(Instant deadline) {
+      return new EnqueueOptions(
+          this.workflowName,
+          this.queueName,
+          this.className,
+          this.instanceName,
+          this.workflowId,
+          this.appVersion,
+          this.timeout,
+          deadline,
           this.deduplicationId,
           this.priority);
     }
@@ -215,6 +242,7 @@ public EnqueueOptions withDeduplicationId(String deduplicationId) {
           this.workflowId,
           this.appVersion,
           this.timeout,
+          this.deadline,
           deduplicationId,
           this.priority);
     }
@@ -235,6 +263,7 @@ public EnqueueOptions withInstanceName(String instName) {
           this.workflowId,
           this.appVersion,
           this.timeout,
+          this.deadline,
           this.deduplicationId,
           this.priority);
     }
@@ -254,6 +283,7 @@ public EnqueueOptions withPriority(Integer priority) {
           this.workflowId,
           this.appVersion,
           this.timeout,
+          this.deadline,
           this.deduplicationId,
           priority);
     }
@@ -288,14 +318,14 @@ public <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
         Objects.requireNonNullElse(options.instanceName(), ""),
         null,
         args,
-        new ExecuteWorkflowOptions(
+        new StartWorkflowOptions(
             Objects.requireNonNullElseGet(options.workflowId(), () -> UUID.randomUUID().toString()),
-            options.timeout(),
-            null,
+            Timeout.of(options.timeout()),
             Objects.requireNonNull(
                 options.queueName(), "EnqueueOptions queueName must not be null"),
             options.deduplicationId,
-            options.priority),
+            options.priority,
+            options.deadline),
         null,
         null,
         options.appVersion,