Single Execution (#236)

Use database to limit workflows to one executor globally.
Use memory map to limit to one execution locally.
Close commit holes in WF status and step outputs.
Test infrastructure and tests.
Fixes #226

Author: chuck-dbos

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

@@ -355,7 +355,9 @@ public <T, E extends Exception> WorkflowHandle<T, E> enqueueWorkflow(
             Objects.requireNonNull(
                 options.queueName(), "EnqueueOptions queueName must not be null"),
             options.deduplicationId,
-            options.priority),
+            options.priority,
+            false,
+            false),
         null,
         null,
         options.appVersion,
@@ -379,7 +381,7 @@ public void send(String destinationId, Object message, String topic, String idem
     var status =
         new WorkflowStatusInternal(workflowId, WorkflowState.SUCCESS)
             .withName("temp_workflow-send-client");
-    systemDatabase.initWorkflowStatus(status, null);
+    systemDatabase.initWorkflowStatus(status, null, false, false);
     systemDatabase.send(status.workflowId(), 0, destinationId, message, topic);
   }
 

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

@@ -18,7 +18,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class NotificationsDAO {
+class NotificationsDAO {
 
   private static final Logger logger = LoggerFactory.getLogger(NotificationsDAO.class);
 
@@ -37,7 +37,7 @@ void speedUpPollingForTest() {
     dbPollingIntervalEventMs = 100;
   }
 
-  public void send(
+  void send(
       String workflowUuid, int functionId, String destinationUuid, Object message, String topic)
       throws SQLException {
 
@@ -96,7 +96,8 @@ public void send(
 
         // Record operation result
         var output = new StepResult(workflowUuid, functionId, functionName);
-        StepsDAO.recordStepResultTxn(output, startTime, conn, this.schema);
+        StepsDAO.recordStepResultTxn(
+            output, startTime, System.currentTimeMillis(), conn, this.schema);
 
         conn.commit();
 
@@ -111,7 +112,7 @@ public void send(
     }
   }
 
-  public Object recv(
+  Object recv(
       String workflowUuid, int functionId, int timeoutFunctionId, String topic, Duration timeout)
       throws SQLException, InterruptedException {
 
@@ -245,7 +246,8 @@ WITH oldest_entry AS (
         StepResult output =
             new StepResult(workflowUuid, functionId, functionName)
                 .withOutput(JSONUtil.serialize(toSave));
-        StepsDAO.recordStepResultTxn(output, startTime, conn, this.schema);
+        StepsDAO.recordStepResultTxn(
+            output, startTime, System.currentTimeMillis(), conn, this.schema);
 
         conn.commit();
         return toSave;
@@ -294,8 +296,7 @@ ON CONFLICT (workflow_uuid, key, function_id)
     }
   }
 
-  public void setEvent(
-      String workflowId, int functionId, String key, Object message, boolean asStep)
+  void setEvent(String workflowId, int functionId, String key, Object message, boolean asStep)
       throws SQLException {
     if (dataSource.isClosed()) {
       throw new IllegalStateException("Database is closed!");
@@ -329,7 +330,8 @@ public void setEvent(
         if (asStep) {
           // Record the operation result
           StepResult output = new StepResult(workflowId, functionId, functionName);
-          StepsDAO.recordStepResultTxn(output, startTime, conn, this.schema);
+          StepsDAO.recordStepResultTxn(
+              output, startTime, System.currentTimeMillis(), conn, this.schema);
         }
 
         conn.commit();
@@ -342,7 +344,7 @@ public void setEvent(
     }
   }
 
-  public Object getEvent(
+  Object getEvent(
       String targetUuid, String key, Duration timeout, GetWorkflowEventContext callerCtx)
       throws SQLException {
     if (dataSource.isClosed()) {
@@ -456,7 +458,8 @@ public Object getEvent(
         StepResult output =
             new StepResult(callerCtx.getWorkflowId(), callerCtx.getFunctionId(), functionName)
                 .withOutput(JSONUtil.serialize(value));
-        StepsDAO.recordStepResultTxn(dataSource, output, startTime, this.schema);
+        StepsDAO.recordStepResultTxn(
+            dataSource, output, startTime, System.currentTimeMillis(), this.schema);
       }
 
       return value;

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

@@ -14,7 +14,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class QueuesDAO {
+class QueuesDAO {
   private static final Logger logger = LoggerFactory.getLogger(QueuesDAO.class);
 
   private final HikariDataSource dataSource;
@@ -33,7 +33,7 @@ public class QueuesDAO {
    * @param appVersion The application version
    * @return List of workflow UUIDs that are due for execution
    */
-  public List<String> getAndStartQueuedWorkflows(Queue queue, String executorId, String appVersion)
+  List<String> getAndStartQueuedWorkflows(Queue queue, String executorId, String appVersion)
       throws SQLException {
     if (dataSource.isClosed()) {
       throw new IllegalStateException("Database is closed!");
@@ -236,7 +236,7 @@ SELECT executor_id, COUNT(*) as task_count
     }
   }
 
-  public boolean clearQueueAssignment(String workflowId) throws SQLException {
+  boolean clearQueueAssignment(String workflowId) throws SQLException {
     if (dataSource.isClosed()) {
       throw new IllegalStateException("Database is closed!");
     }

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

@@ -1,6 +1,7 @@
 package dev.dbos.transact.database;
 
 import dev.dbos.transact.exceptions.*;
+import dev.dbos.transact.internal.DebugTriggers;
 import dev.dbos.transact.json.JSONUtil;
 import dev.dbos.transact.workflow.ErrorResult;
 import dev.dbos.transact.workflow.StepInfo;
@@ -17,7 +18,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class StepsDAO {
+class StepsDAO {
 
   private static final Logger logger = LoggerFactory.getLogger(StepsDAO.class);
 
@@ -29,20 +30,29 @@ public class StepsDAO {
     this.schema = Objects.requireNonNull(schema);
   }
 
-  public static void recordStepResultTxn(
-      HikariDataSource dataSource, StepResult result, long startTimeEpochMs, String schema)
+  static void recordStepResultTxn(
+      HikariDataSource dataSource,
+      StepResult result,
+      long startTimeEpochMs,
+      long endTimeEpochMs,
+      String schema)
       throws SQLException {
     if (dataSource.isClosed()) {
       throw new IllegalStateException("Database is closed!");
     }
 
     try (Connection connection = dataSource.getConnection(); ) {
-      recordStepResultTxn(result, startTimeEpochMs, connection, schema);
+      recordStepResultTxn(result, startTimeEpochMs, endTimeEpochMs, connection, schema);
     }
+    DebugTriggers.debugTriggerPoint(DebugTriggers.DEBUG_TRIGGER_STEP_COMMIT);
   }
 
-  public static void recordStepResultTxn(
-      StepResult result, Long startTimeEpochMs, Connection connection, String schema)
+  static void recordStepResultTxn(
+      StepResult result,
+      Long startTimeEpochMs,
+      Long endTimeEpochMs,
+      Connection connection,
+      String schema)
       throws SQLException {
 
     Objects.requireNonNull(schema);
@@ -51,6 +61,7 @@ public static void recordStepResultTxn(
           INSERT INTO %s.operation_outputs
             (workflow_uuid, function_id, function_name, output, error, child_workflow_id, started_at_epoch_ms, completed_at_epoch_ms)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+          ON CONFLICT DO NOTHING RETURNING completed_at_epoch_ms
         """
             .formatted(schema);
 
@@ -78,11 +89,20 @@ public static void recordStepResultTxn(
       }
 
       pstmt.setObject(7, startTimeEpochMs);
-      Long endTime = startTimeEpochMs == null ? null : System.currentTimeMillis();
-      pstmt.setObject(8, endTime);
-
-      pstmt.executeUpdate();
+      pstmt.setObject(8, endTimeEpochMs);
 
+      try (ResultSet rs = pstmt.executeQuery()) {
+        if (rs.next() && endTimeEpochMs != null) {
+          long completedAt = rs.getLong("completed_at_epoch_ms");
+          logger.warn(
+              String.format(
+                  "Step output for %s:%d-%s was already recorded",
+                  result.workflowId(), result.stepId(), result.functionName()));
+          if (completedAt != endTimeEpochMs) {
+            throw new DBOSWorkflowExecutionConflictException(result.workflowId());
+          }
+        }
+      }
     } catch (SQLException e) {
       logger.debug("recordStepResultTxn error", e);
       if ("23505".equals(e.getSQLState())) {
@@ -108,7 +128,7 @@ public static void recordStepResultTxn(
    *     match the provided name.
    * @throws SQLException For other database access errors.
    */
-  public static StepResult checkStepExecutionTxn(
+  static StepResult checkStepExecutionTxn(
       String workflowId, int functionId, String functionName, Connection connection, String schema)
       throws SQLException, DBOSWorkflowCancelledException, DBOSUnexpectedStepException {
 
@@ -175,7 +195,7 @@ public static StepResult checkStepExecutionTxn(
     return recordedResult;
   }
 
-  public List<StepInfo> listWorkflowSteps(String workflowId) throws SQLException {
+  List<StepInfo> listWorkflowSteps(String workflowId) throws SQLException {
 
     if (dataSource.isClosed()) {
       throw new IllegalStateException("Database is closed!");
@@ -250,12 +270,12 @@ public List<StepInfo> listWorkflowSteps(String workflowId) throws SQLException {
     return steps;
   }
 
-  public Duration sleep(String workflowUuid, int functionId, Duration duration, boolean skipSleep)
+  Duration sleep(String workflowUuid, int functionId, Duration duration, boolean skipSleep)
       throws SQLException {
     return StepsDAO.sleep(dataSource, workflowUuid, functionId, duration, skipSleep, this.schema);
   }
 
-  public static Duration sleep(
+  static Duration sleep(
       HikariDataSource dataSource,
       String workflowUuid,
       int functionId,
@@ -298,7 +318,7 @@ public static Duration sleep(
         StepResult output =
             new StepResult(workflowUuid, functionId, functionName)
                 .withOutput(JSONUtil.serialize(endTime));
-        recordStepResultTxn(dataSource, output, startTime, schema);
+        recordStepResultTxn(dataSource, output, startTime, (long) endTime, schema);
       } catch (DBOSWorkflowExecutionConflictException e) {
         logger.error("Error recording sleep", e);
       }

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

@@ -85,15 +85,29 @@ void speedUpPollingForTest() {
    *
    * @param initStatus The initial workflow status details.
    * @param maxRetries Optional maximum number of retries.
+   * @param isRecoveryRequest True if this is a recovery request, indicating that this node is told
+   *     it owns the workflow even if the ID already exists
+   * @param isDequeuedRequest True if this is a dequeue request, indicating that this node is told
+   *     it owns the workflow (provided it is in the enqueued state)
    * @return An object containing the current status and optionally the deadline epoch milliseconds.
    * @throws DBOSConflictingWorkflowException If a conflicting workflow already exists.
    * @throws DBOSMaxRecoveryAttemptsExceededException If the workflow exceeds max retries.
    */
   public WorkflowInitResult initWorkflowStatus(
-      WorkflowStatusInternal initStatus, Integer maxRetries) {
+      WorkflowStatusInternal initStatus,
+      Integer maxRetries,
+      boolean isRecoveryRequest,
+      boolean isDequeuedRequest) {
+
+    // This ID will be used to tell if we are the first writer of the record, or if there is an
+    // existing one
+    // Note that it is generated outside of the DB retry loop, in case commit acks get lost and we
+    // do not know if we committed or not
+    String ownerXid = UUID.randomUUID().toString();
     return DbRetry.call(
         () -> {
-          return workflowDAO.initWorkflowStatus(initStatus, maxRetries);
+          return workflowDAO.initWorkflowStatus(
+              initStatus, maxRetries, isRecoveryRequest, isDequeuedRequest, ownerXid);
         });
   }
 
@@ -163,9 +177,10 @@ public StepResult checkStepExecutionTxn(String workflowId, int functionId, Strin
   }
 
   public void recordStepResultTxn(StepResult result, long startTime) {
+    var et = System.currentTimeMillis();
     DbRetry.run(
         () -> {
-          StepsDAO.recordStepResultTxn(dataSource, result, startTime, this.schema);
+          StepsDAO.recordStepResultTxn(dataSource, result, startTime, et, this.schema);
         });
   }