Update workflow status with a final result only when status is PENDING

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

@@ -517,7 +517,7 @@ public void sleep(@NonNull Duration duration) {
    * @throws E if the workflow threw an exception
    */
   public <T, E extends Exception> T getResult(@NonNull String workflowId) throws E {
-    return ensureLaunched("getResult").<T, E>getResult(workflowId);
+    return ensureLaunched("getResult").<T, E>getResult(workflowId, false);
   }
 
   /**

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

@@ -62,7 +62,7 @@ public WorkflowHandleClient(@NonNull String workflowId) {
 
     @Override
     public T getResult() throws E {
-      var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
+      var result = systemDatabase.<T>awaitWorkflowResult(workflowId, false);
       return Result.<T, E>process(result);
     }
 

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

@@ -450,19 +450,23 @@ public WorkflowInitResult initWorkflowStatus(
    *
    * @param workflowId id of the workflow
    * @param result output serialized as json
+   * @return true if the outcome was recorded, false if the row is no longer PENDING and this
+   *     execution no longer owns the workflow's outcome
    */
-  public void recordWorkflowOutput(String workflowId, String result) {
-    dbRetry(() -> WorkflowDAO.recordWorkflowOutput(ctx, workflowId, result));
+  public boolean recordWorkflowOutput(String workflowId, String result) {
+    return dbRetry(() -> WorkflowDAO.recordWorkflowOutput(ctx, workflowId, result));
   }
 
   /**
    * Store the error to workflow_status
    *
    * @param workflowId id of the workflow
    * @param error output serialized as json
+   * @return true if the outcome was recorded, false if the row is no longer PENDING and this
+   *     execution no longer owns the workflow's outcome
    */
-  public void recordWorkflowError(String workflowId, String error) {
-    dbRetry(() -> WorkflowDAO.recordWorkflowError(ctx, workflowId, error));
+  public boolean recordWorkflowError(String workflowId, String error) {
+    return dbRetry(() -> WorkflowDAO.recordWorkflowError(ctx, workflowId, error));
   }
 
   /**
@@ -552,10 +556,19 @@ public List<StepInfo> listWorkflowSteps(
     return dbRetry(() -> StepsDAO.listWorkflowSteps(ctx, workflowId, loadOutput, limit, offset));
   }
 
-  public <T> Result<T> awaitWorkflowResult(String workflowId) {
+  /**
+   * Awaits a workflow's recorded outcome. A missing row normally means the workflow just hasn't
+   * been inserted yet (an unchecked retrieve, or a debounced workflow whose row appears only after
+   * the debounce period), so by default it is polled for. Callers that know the row must already
+   * exist pass {@code failIfMissing} to fail fast instead.
+   */
+  public <T> Result<T> awaitWorkflowResult(String workflowId, boolean failIfMissing) {
     // Not a notification wait: no channel carries workflow completion, so this poll is the only
     // delivery mechanism and stays short whether or not a listener is running.
-    return dbRetry(() -> WorkflowDAO.<T>awaitWorkflowResult(ctx, DB_POLLING_INTERVAL, workflowId));
+    return dbRetry(
+        () ->
+            WorkflowDAO.<T>awaitWorkflowResult(
+                ctx, DB_POLLING_INTERVAL, workflowId, failIfMissing));
   }
 
   public List<String> startQueuedWorkflows(

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

@@ -318,7 +318,19 @@ ON CONFLICT (workflow_uuid)
     }
   }
 
-  static void updateWorkflowOutcome(
+  /**
+   * Record a workflow's terminal outcome, reporting whether the write landed. The write applies
+   * only to a PENDING row: a run owns its workflow's outcome exactly as long as the row says that
+   * run is what the workflow is doing. (Note: this does not prevent a write when another concurrent
+   * execution is already running and the status is PENDING. However, both executions should be
+   * deterministic and idempotent.)
+   *
+   * <p>Returning false means the row was CANCELLED, dead-lettered, already terminal, handed to
+   * another execution (ENQUEUED/DELAYED, e.g. by a concurrent resume), or gone entirely. Callers
+   * that need to distinguish a deleted row do so when they park on the recorded outcome (see {@link
+   * #awaitWorkflowResult(DbContext, Duration, String, boolean)}).
+   */
+  static boolean updateWorkflowOutcome(
       Connection conn,
       String schema,
       String workflowId,
@@ -336,13 +348,11 @@ static void updateWorkflowOutcome(
           "updateWorkflowOutcome called with non-terminal status: " + status);
     }
 
-    // Never overwrite a CANCELLED workflow: a workflow cancelled during its final step must not
-    // subsequently complete.
     var sql =
         """
           UPDATE "%s".workflow_status
           SET status = ?, output = ?, error = ?, updated_at = ?, completed_at = ?, deduplication_id = NULL
-          WHERE workflow_uuid = ? AND status != ?
+          WHERE workflow_uuid = ? AND status = ?
         """
             .formatted(schema);
 
@@ -354,25 +364,9 @@ static void updateWorkflowOutcome(
       stmt.setLong(4, now);
       stmt.setLong(5, now);
       stmt.setString(6, workflowId);
-      stmt.setString(7, WorkflowState.CANCELLED.name());
+      stmt.setString(7, WorkflowState.PENDING.name());
 
-      if (stmt.executeUpdate() == 0) {
-        // The guarded UPDATE matched no rows. Re-read status to check whether the workflow
-        // was cancelled; if so, raise so it ends as CANCELLED rather than completing.
-        var readSql =
-            """
-            SELECT status FROM "%s".workflow_status WHERE workflow_uuid = ?
-            """
-                .formatted(schema);
-        try (var readStmt = conn.prepareStatement(readSql)) {
-          readStmt.setString(1, workflowId);
-          try (var rs = readStmt.executeQuery()) {
-            if (rs.next() && WorkflowState.CANCELLED.name().equals(rs.getString(1))) {
-              throw new DBOSWorkflowCancelledException(workflowId);
-            }
-          }
-        }
-      }
+      return stmt.executeUpdate() != 0;
     }
   }
 
@@ -381,12 +375,14 @@ static void updateWorkflowOutcome(
    *
    * @param workflowId id of the workflow
    * @param result output serialized as json
+   * @return true if the outcome was recorded, false if the row is no longer PENDING
    */
-  public static void recordWorkflowOutput(DbContext ctx, String workflowId, String result)
+  public static boolean recordWorkflowOutput(DbContext ctx, String workflowId, String result)
       throws SQLException {
 
     try (var conn = ctx.getConnection()) {
-      updateWorkflowOutcome(conn, ctx.schema(), workflowId, WorkflowState.SUCCESS, result, null);
+      return updateWorkflowOutcome(
+          conn, ctx.schema(), workflowId, WorkflowState.SUCCESS, result, null);
     }
   }
 
@@ -395,12 +391,14 @@ public static void recordWorkflowOutput(DbContext ctx, String workflowId, String
    *
    * @param workflowId id of the workflow
    * @param error output serialized as json
+   * @return true if the outcome was recorded, false if the row is no longer PENDING
    */
-  public static void recordWorkflowError(DbContext ctx, String workflowId, String error)
+  public static boolean recordWorkflowError(DbContext ctx, String workflowId, String error)
       throws SQLException {
 
     try (var conn = ctx.getConnection()) {
-      updateWorkflowOutcome(conn, ctx.schema(), workflowId, WorkflowState.ERROR, null, error);
+      return updateWorkflowOutcome(
+          conn, ctx.schema(), workflowId, WorkflowState.ERROR, null, error);
     }
   }
 
@@ -1206,14 +1204,24 @@ private static WorkflowStatus resultsToWorkflowStatus(
     return info;
   }
 
+  /**
+   * Poll the workflow's row until it reaches a terminal state, then return the recorded outcome.
+   *
+   * <p>A missing row normally means the workflow just hasn't been inserted yet (an unchecked
+   * retrieve, or a debounced workflow whose row appears only after the debounce period), so polling
+   * is correct. Callers that know the row must already exist (a run parking on an outcome it just
+   * failed to write) pass {@code failIfMissing} to fail fast with {@link
+   * DBOSNonExistentWorkflowException} instead of polling forever.
+   */
   @SuppressWarnings("unchecked")
   public static <T> Result<T> awaitWorkflowResult(
-      DbContext ctx, Duration dbPollingInterval, String workflowId) throws SQLException {
+      DbContext ctx, Duration dbPollingInterval, String workflowId, boolean failIfMissing)
+      throws SQLException {
 
     DBOSSerializer serializer = ctx.serializer();
     final String sql =
         """
-          SELECT status, output, error, serialization
+          SELECT status, output, error, serialization, recovery_attempts
           FROM "%s".workflow_status
           WHERE workflow_uuid = ?
         """
@@ -1247,9 +1255,20 @@ public static <T> Result<T> awaitWorkflowResult(
               }
               case CANCELLED -> throw new DBOSAwaitedWorkflowCancelledException(workflowId);
 
+              case MAX_RECOVERY_ATTEMPTS_EXCEEDED -> {
+                // A workflow is dead-lettered by the attempt that pushes recovery_attempts
+                // past maxRetries+1, so a dead-lettered row carries maxRetries+2 attempts.
+                int maxRetries = Math.max(0, rs.getInt("recovery_attempts") - 2);
+                throw new DBOSMaxRecoveryAttemptsExceededException(workflowId, maxRetries);
+              }
+
               default -> {}
             }
             // Status is PENDING or other - continue polling
+          } else if (failIfMissing) {
+            // The caller knows the row must already exist, so a missing row means it was
+            // deleted: fail fast instead of polling forever.
+            throw new DBOSNonExistentWorkflowException(workflowId);
           }
           // Row not found - workflow hasn't appeared yet, continue polling
         }

transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

@@ -1061,9 +1061,9 @@ public WorkflowStatus getWorkflowStatus(String workflowId) {
         () -> systemDatabase.getWorkflowStatus(workflowId), "DBOS.getWorkflowStatus", null);
   }
 
-  public <T, E extends Exception> T getResult(String workflowId) throws E {
+  public <T, E extends Exception> T getResult(String workflowId, boolean failIfMissing) throws E {
     return this.runDbosFunctionAsStep(
-        () -> awaitWorkflowResult(workflowId), "DBOS.getResult", workflowId);
+        () -> awaitWorkflowResult(workflowId, failIfMissing), "DBOS.getResult", workflowId);
   }
 
   @SuppressWarnings("unchecked")
@@ -1073,13 +1073,13 @@ public <T, E extends Exception> T getResult(String workflowId, Future<T> futureR
           try {
             return futureResult.get();
           } catch (DBOSWorkflowExecutionConflictException e) {
-            return awaitWorkflowResult(workflowId);
+            return awaitWorkflowResult(workflowId, false);
           } catch (CancellationException e) {
             throw new DBOSAwaitedWorkflowCancelledException(workflowId);
           } catch (ExecutionException e) {
             if (e.getCause() instanceof Exception cause) {
               if (cause instanceof DBOSWorkflowExecutionConflictException) {
-                return awaitWorkflowResult(workflowId);
+                return awaitWorkflowResult(workflowId, false);
               }
               if (cause instanceof DBOSWorkflowCancelledException cancelled) {
                 throw new DBOSAwaitedWorkflowCancelledException(cancelled.workflowId());
@@ -1095,8 +1095,14 @@ public <T, E extends Exception> T getResult(String workflowId, Future<T> futureR
         workflowId);
   }
 
-  private <T, E extends Exception> T awaitWorkflowResult(String workflowId) throws E {
-    var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
+  // A missing row normally means the workflow just hasn't been inserted yet (an unchecked
+  // retrieve, or a debounced workflow whose row appears only after the debounce period), so
+  // polling is correct. Callers that know the row must already exist (a run parking on an
+  // outcome it just failed to write) pass failIfMissing to fail fast instead of polling
+  // forever.
+  private <T, E extends Exception> T awaitWorkflowResult(String workflowId, boolean failIfMissing)
+      throws E {
+    var result = systemDatabase.<T>awaitWorkflowResult(workflowId, failIfMissing);
     return Result.<T, E>process(result);
   }
 
@@ -1756,21 +1762,31 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
       return retrieveWorkflow(workflowId);
     }
     if (initResult.status().equals(WorkflowState.SUCCESS)) {
-      return retrieveWorkflow(workflowId);
+      // The workflow already completed: its recorded outcome is this call's result. The row
+      // is known to have existed (persistWorkflow just read this status from it), so
+      // failIfMissing: a row deleted in the meantime surfaces
+      // DBOSNonExistentWorkflowException instead of polling forever.
+      return new WorkflowHandleDBPoll<>(this, workflowId, true);
     } else if (initResult.status().equals(WorkflowState.ERROR)) {
       logger.warn("Idempotency check not impl for error");
     } else if (initResult.status().equals(WorkflowState.CANCELLED)) {
       logger.warn("Idempotency check not impl for cancelled");
     }
 
     final var finalOptions = options;
+    final String notRecordedWarning =
+        "Workflow outcome was not recorded: the workflow is no longer owned by this execution."
+            + " Waiting for the recorded outcome.";
     Supplier<T> task =
         () -> {
           DBOSContextHolder.clear();
           var bucket =
               finalOptions.isDequeuedRequest()
                   ? new QueueBucket(finalOptions.queueName(), finalOptions.queuePartitionKey())
                   : NO_QUEUE;
+          // The warning to park under, set by whichever site found that this run does not
+          // own the workflow's outcome.
+          String pendingAdopt;
           try (var active = new ActiveWorkflowGuard(workflowId, bucket)) {
             logger.debug(
                 "executeWorkflow task {}({}) {}",
@@ -1804,12 +1820,19 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
             }
 
             active.release();
-            persistWorkflowOutput(workflowId, output, initResult.serialization());
-
-            return output;
+            if (persistWorkflowOutput(workflowId, output, initResult.serialization())) {
+              return output;
+            }
+            // The row was not PENDING: this run no longer owns the workflow's outcome. It
+            // may have been cancelled, dead-lettered, completed by a concurrent execution,
+            // or handed back to the queue by a resume.
+            pendingAdopt = notRecordedWarning;
           } catch (DBOSWorkflowExecutionConflictException e) {
-            // don't persist execution conflict exception
-            throw e;
+            // Another execution owns this workflow (a concurrent run recorded a step
+            // checkpoint, or the workflow is already active on this executor). Never
+            // persist the conflict: park the execution instead.
+            pendingAdopt =
+                "Aborting duplicate execution of workflow. Waiting for the recorded outcome.";
           } catch (Exception e) {
             Throwable actual = e;
 
@@ -1823,25 +1846,38 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
               }
             }
 
-            logger.error("executeWorkflow {}", workflowId, actual);
-
-            // Skip persistWorkflowError for cancelled workflows: the DB already holds CANCELLED
-            // (the terminal state), and calling persistWorkflowError would cause
-            // updateWorkflowOutcome to throw DBOSWorkflowCancelledException from inside the
-            // catch block, bypassing the getResult() conversion to
-            // DBOSAwaitedWorkflowCancelledException.
             if (actual instanceof DBOSWorkflowCancelledException cancelled
                 && cancelled.workflowId().equals(workflowId)) {
-              throw cancelled;
+              // The run observed its own cancellation (checkWorkflow only throws this after
+              // reading CANCELLED from the DB). Skip the outcome write so it can never clobber
+              // the row, and adopt the recorded outcome: normally the row is still CANCELLED
+              // and the park throws DBOSAwaitedWorkflowCancelledException, but a concurrent
+              // resume may have taken the workflow back, in which case the recorded outcome
+              // is the truth.
+              pendingAdopt =
+                  "Workflow was cancelled during execution. Waiting for the recorded outcome.";
+            } else if (persistWorkflowError(workflowId, actual, initResult.serialization())) {
+              // active is already closed here: try-with-resources closes before catch runs,
+              // so the entry is released before this terminal write becomes durable.
+              logger.error("executeWorkflow {}", workflowId, actual);
+              throw e;
+            } else {
+              // The row was not PENDING: this run no longer owns the workflow's outcome, and
+              // the error it computed is not the workflow's error.
+              logger.debug("executeWorkflow {}", workflowId, actual);
+              pendingAdopt = notRecordedWarning;
             }
-
-            // active is already closed here: try-with-resources closes before catch runs,
-            // so the entry is released before this terminal write becomes durable.
-            persistWorkflowError(workflowId, actual, initResult.serialization());
-            throw e;
           } finally {
             DBOSContextHolder.clear();
           }
+
+          // Reached only when a refusal above set pendingAdopt: this run does not own the
+          // workflow's outcome. Park the execution and deliver the recorded outcome through
+          // this run's own future. The row is known to have existed (this run inserted or
+          // read it), so failIfMissing: a missing row means it was deleted, and the park
+          // surfaces DBOSNonExistentWorkflowException instead of polling forever.
+          logger.warn("{} workflowId {}", pendingAdopt, workflowId);
+          return awaitWorkflowResult(workflowId, true);
         };
 
     if (initResult.deadline() != null && Instant.now().isAfter(initResult.deadline())) {
@@ -2007,14 +2043,14 @@ private static WorkflowInitResult persistWorkflow(
     return initResult[0];
   }
 
-  private void persistWorkflowOutput(String workflowId, Object result, String serialization) {
+  private boolean persistWorkflowOutput(String workflowId, Object result, String serialization) {
     var serialized = SerializationUtil.serializeValue(result, serialization, this.serializer);
-    systemDatabase.recordWorkflowOutput(workflowId, serialized.serializedValue());
+    return systemDatabase.recordWorkflowOutput(workflowId, serialized.serializedValue());
   }
 
-  private void persistWorkflowError(String workflowId, Throwable error, String serialization) {
+  private boolean persistWorkflowError(String workflowId, Throwable error, String serialization) {
     var serialized = SerializationUtil.serializeError(error, serialization, this.serializer);
-    systemDatabase.recordWorkflowError(workflowId, serialized.serializedValue());
+    return systemDatabase.recordWorkflowError(workflowId, serialized.serializedValue());
   }
 
   /**