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

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

@@ -366,19 +366,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));
   }
 
   /**
@@ -469,7 +473,19 @@ public List<StepInfo> listWorkflowSteps(
   }
 
   public <T> Result<T> awaitWorkflowResult(String workflowId) {
-    return dbRetry(() -> WorkflowDAO.<T>awaitWorkflowResult(ctx, dbPollingInterval, workflowId));
+    return awaitWorkflowResult(workflowId, false);
+  }
+
+  /**
+   * 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) {
+    return dbRetry(
+        () ->
+            WorkflowDAO.<T>awaitWorkflowResult(ctx, dbPollingInterval, 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 = ?
         """
@@ -1246,9 +1254,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

@@ -1062,8 +1062,12 @@ public WorkflowStatus getWorkflowStatus(String workflowId) {
   }
 
   public <T, E extends Exception> T getResult(String workflowId) throws E {
+    return getResult(workflowId, false);
+  }
+
+  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")
@@ -1095,8 +1099,18 @@ public <T, E extends Exception> T getResult(String workflowId, Future<T> futureR
         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) throws E {
-    var result = systemDatabase.<T>awaitWorkflowResult(workflowId);
+    return awaitWorkflowResult(workflowId, false);
+  }
+
+  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,7 +1770,11 @@ 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)) {
@@ -1804,12 +1822,30 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
             }
 
             active.release();
-            persistWorkflowOutput(workflowId, output, initResult.serialization());
+            if (!persistWorkflowOutput(workflowId, output, initResult.serialization())) {
+              // 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. Park the execution and wait for the
+              // recorded outcome to become visible. The row is known to have existed (this run
+              // just tried to write to it), so failIfMissing: a missing row means it was
+              // deleted, and the park surfaces DBOSNonExistentWorkflowException.
+              logger.warn(
+                  "Workflow outcome was not recorded: the workflow is no longer owned by this execution. Waiting for the recorded outcome. workflowId {}",
+                  workflowId);
+              return awaitWorkflowResult(workflowId, true);
+            }
 
             return output;
           } 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 and deliver the recorded outcome
+            // through this run's own future. The row is known to have existed, so
+            // failIfMissing: a missing row means it was deleted.
+            logger.warn(
+                "Aborting duplicate execution of workflow. Waiting for the recorded outcome. workflowId {}",
+                workflowId);
+            return awaitWorkflowResult(workflowId, true);
           } catch (Exception e) {
             Throwable actual = e;
 
@@ -1825,19 +1861,39 @@ 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.
+            // 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 awaitWorkflowResult throws DBOSAwaitedWorkflowCancelledException, but a
+            // concurrent resume may have taken the workflow back, in which case the recorded
+            // outcome is the truth. The row is known to have existed (the cancellation was
+            // read from it), so failIfMissing: a missing row means it was deleted.
             if (actual instanceof DBOSWorkflowCancelledException cancelled
                 && cancelled.workflowId().equals(workflowId)) {
-              throw cancelled;
+              logger.warn(
+                  "Workflow was cancelled during execution. Waiting for the recorded outcome. workflowId {}",
+                  workflowId);
+              return awaitWorkflowResult(workflowId, true);
+            }
+
+            // The park after a refused outcome write found no workflow_status row at all (the
+            // workflow was deleted or garbage collected): deliver the error as the workflow's
+            // outcome; there is nothing left to record onto.
+            if (actual instanceof DBOSNonExistentWorkflowException nonExistent
+                && workflowId.equals(nonExistent.workflowId())) {
+              throw nonExistent;
             }
 
             // 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());
+            if (!persistWorkflowError(workflowId, actual, initResult.serialization())) {
+              // The row was not PENDING: this run no longer owns the workflow's outcome
+              // (see the equivalent refusal on the success path above).
+              logger.warn(
+                  "Workflow outcome was not recorded: the workflow is no longer owned by this execution. Waiting for the recorded outcome. workflowId {}",
+                  workflowId);
+              return awaitWorkflowResult(workflowId, true);
+            }
             throw e;
           } finally {
             DBOSContextHolder.clear();
@@ -2007,14 +2063,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());
   }
 
   /**

transact/src/main/java/dev/dbos/transact/workflow/internal/WorkflowHandleDBPoll.java

@@ -7,10 +7,19 @@
 public class WorkflowHandleDBPoll<T, E extends Exception> implements WorkflowHandle<T, E> {
   private final DBOSExecutor executor;
   private final String workflowId;
+  private final boolean failIfMissing;
 
   public WorkflowHandleDBPoll(DBOSExecutor executor, String workflowId) {
+    this(executor, workflowId, false);
+  }
+
+  // failIfMissing is for handles built from a workflow_status row that was just read: a
+  // missing row means it was deleted, so getResult fails fast with
+  // DBOSNonExistentWorkflowException instead of polling for a row that will never reappear.
+  public WorkflowHandleDBPoll(DBOSExecutor executor, String workflowId, boolean failIfMissing) {
     this.executor = executor;
     this.workflowId = workflowId;
+    this.failIfMissing = failIfMissing;
   }
 
   @Override
@@ -20,7 +29,7 @@ public String workflowId() {
 
   @Override
   public T getResult() throws E {
-    return executor.getResult(this.workflowId);
+    return executor.getResult(this.workflowId, failIfMissing);
   }
 
   @Override