End a timed-out run's wait at its deadline, whoever cancelled the row (#635)

Fixes #627.

## The problem

A workflow past its deadline is cancelled by whichever comes first: its
own timer or the database-side timeout sweep, which runs on every
executor. The timer cancelled the run's future, which is what ends a
local handle's `getResult()`, only when `cancelOwnedWorkflow(workflowId,
ownerXid)` succeeded, that is, only while the run still owned the row.
If the sweep (on any executor) settled the row first, the timer found
nothing to cancel and left the future alone. `getResult()` then waited
until the run's next write was refused: through a whole `dbos.sleep`, 30
s for a 1 s timeout in #627.

## The fix

- **The timer always ends the wait at the deadline.** It still cancels
the row only while the run owns it, but it cancels the run's future
either way. That happens in a `finally`, so a cancel that fails against
the database still ends the wait.
- **`getResult()` reports what the row records.** When the future was
cancelled, it awaits the row instead of assuming a cancellation. That
gives:
- cancelled, whether by this run, the sweep on any executor, or an
explicit cancel;
- if the workflow was resumed after the timeout, the resumed run's
outcome. The handle used to report that case as a cancellation.
- **A run whose body has finished is left alone.** Such a run only
records or adopts its outcome, so it completes its own future promptly.
Cancelling it would make `getResult()` return a copy deserialized from
the row instead of the value the run holds: for a portable workflow
returning a record, a `Map`.
- **The timer is dropped when the run finishes first.** It used to stay
queued until the deadline, holding the run's future and so its result.
The run's future now cancels its timer on completion, and the timeout
scheduler (`setRemoveOnCancelPolicy(true)`) removes cancelled tasks
instead of keeping them until their delay runs out.

As before, cancelling the future doesn't stop the run's thread, because
`CompletableFuture.cancel` doesn't interrupt. The run keeps going until
its next write is refused. What changes is that `getResult()` no longer
waits for it.

The first commit took a different approach: the sweep completed the
handles of local runs whose rows it cancelled. It only helped the
executor whose sweep won the row, so the second commit reverts it.

## Behaviour changes for a handle whose deadline passed

- If the row has since been deleted, `getResult()` throws
`DBOSNonExistentWorkflowException` instead of reporting a cancellation.
- If the workflow was resumed, `getResult()` waits for the resumed run.

## Tests

- `WorkflowTimeoutSweepTest`:
- `aRunTheSweepCancelsFirstStillEndsAtItsDeadline`: the sweep cancels
the row before the run's timer fires.
- `aRowAnotherExecutorCancelledEndsTheWaitAtTheDeadline`: the row is
cancelled the way another executor's sweep would.
  - `aHandleReportsTheOutcomeOfARunResumedAfterTheTimeout`.
  
Each waits until the run is inside its sleep, where nothing else would
end the wait. All three fail against main's `DBOSExecutor`.
- `TimeoutTest.aFinishedRunLeavesNoTimerQueued`: fails without either
the timer cancel or the remove-on-cancel policy.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: devhawk

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

@@ -93,6 +93,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -325,7 +326,11 @@ public void start(
           };
 
       executorService = executorServiceSupplier.get();
-      timeoutScheduler = Executors.newScheduledThreadPool(2);
+      var scheduler = new ScheduledThreadPoolExecutor(2);
+      // A workflow's timeout timer is cancelled when the workflow finishes; without this, the
+      // cancelled timer, and the result it references, stays queued until the deadline.
+      scheduler.setRemoveOnCancelPolicy(true);
+      timeoutScheduler = scheduler;
 
       systemDatabase = SystemDatabase.create(config, this.executorId, this.appName);
       systemDatabase.start();
@@ -1599,7 +1604,9 @@ public <T, E extends Exception> T getResult(String workflowId, Future<T> futureR
           } catch (DBOSWorkflowExecutionConflictException e) {
             return awaitWorkflowResult(workflowId, false);
           } catch (CancellationException e) {
-            throw new DBOSAwaitedWorkflowCancelledException(workflowId);
+            // The run's deadline passed. The row has the outcome: cancelled by this run, by the
+            // timeout sweep or explicitly, or, if the workflow was resumed, the new run's.
+            return awaitWorkflowResult(workflowId, true);
           } catch (ExecutionException e) {
             if (e.getCause() instanceof Exception cause) {
               if (cause instanceof DBOSWorkflowExecutionConflictException) {
@@ -2597,6 +2604,11 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
     final String notRecordedWarning =
         "Workflow outcome was not recorded: the workflow is no longer owned by this execution."
             + " Waiting for the recorded outcome.";
+    // Set as the run starts writing its outcome. From then on it completes its own future
+    // promptly, with the result it holds or the outcome it adopts, so the deadline timer leaves it
+    // to: cancelling it could make getResult read back a deserialized copy of that result. Work
+    // before the write, such as deleting step checkpoints, stays under the deadline.
+    var recordingOutcome = new AtomicBoolean(false);
     Supplier<T> task =
         () -> {
           DBOSContextHolder.clear();
@@ -2650,6 +2662,7 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
             deleteCompletedStepCheckpoints(workflowId);
 
             active.release();
+            recordingOutcome.set(true);
             if (persistWorkflowOutput(workflowId, output, initResult.serialization(), ownerXid)) {
               return output;
             }
@@ -2694,6 +2707,7 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
               if (!Thread.currentThread().isInterrupted()) {
                 deleteCompletedStepCheckpoints(workflowId);
               }
+              recordingOutcome.set(true);
               if (persistWorkflowError(workflowId, actual, initResult.serialization(), ownerXid)) {
                 // active is already closed here: try-with-resources closes before catch runs,
                 // so the entry is released before this terminal write becomes durable.
@@ -2731,21 +2745,39 @@ private <T, E extends Exception> WorkflowHandle<T, E> executeWorkflow(
 
     var future = CompletableFuture.supplyAsync(task, executorService);
     if (timeLeft != null) {
-      timeoutScheduler.schedule(
-          () -> {
-            // A run that lost the workflow is left to reach the fence and report the owner's
-            // outcome; cancelling its future would report a cancellation that never happened.
-            if (!future.isDone() && systemDatabase.cancelOwnedWorkflow(workflowId, ownerXid)) {
-              future.cancel(true);
-            }
-          },
-          timeLeft.toMillis(),
-          TimeUnit.MILLISECONDS);
+      var timer =
+          timeoutScheduler.schedule(
+              () -> {
+                // The deadline has passed. Unless the run is already writing its outcome, the
+                // handle stops waiting on it, even if the cancel fails. The run cancels the row
+                // only while it still owns it; if it doesn't, something else settled or took over
+                // the row (the timeout sweep on any executor, an explicit cancel, a resume), and
+                // getResult reports what the row records.
+                if (future.isDone() || recordingOutcome.get()) {
+                  return;
+                }
+                try {
+                  systemDatabase.cancelOwnedWorkflow(workflowId, ownerXid);
+                } finally {
+                  // Checked again: the run may have started writing during the cancel.
+                  if (!recordingOutcome.get()) {
+                    future.cancel(true);
+                  }
+                }
+              },
+              timeLeft.toMillis(),
+              TimeUnit.MILLISECONDS);
+      // A run that ends before its deadline needs no timer.
+      future.whenComplete((result, error) -> timer.cancel(false));
     }
 
     return new WorkflowHandleFuture<>(this, workflowId, future);
   }
 
+  ScheduledExecutorService timeoutScheduler() {
+    return timeoutScheduler;
+  }
+
   // Persist an queued workflow to the database without execution
   // used by DBOSExecutor and DBOSClient
   public static void enqueueWorkflow(

transact/src/test/java/dev/dbos/transact/execution/DBOSExecutorTestAccess.java

@@ -6,6 +6,8 @@
 import dev.dbos.transact.database.SystemDatabase;
 import dev.dbos.transact.workflow.WorkflowHandle;
 
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
 /**
  * Note that this class is in the `dev.dbos.transact.execution` package and allows access to
  * package-scoped executor methods as a result.
@@ -23,6 +25,11 @@ public static SchedulerService getSchedulerService(DBOSExecutor exec) {
     return exec.getSchedulerService();
   }
 
+  /** Tasks queued on the executor's timeout scheduler: workflow timers and periodic upkeep. */
+  public static int queuedTimeoutTasks(DBOSExecutor exec) {
+    return ((ScheduledThreadPoolExecutor) exec.timeoutScheduler()).getQueue().size();
+  }
+
   public static ConductorOptions getConductorOptions(DBOSExecutor exec) {
     return exec.conductorOptions();
   }

transact/src/test/java/dev/dbos/transact/txstep/CompletedStepCheckpointsTest.java

@@ -2,6 +2,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import dev.dbos.transact.DBOS;
@@ -20,6 +21,7 @@
 
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
@@ -49,6 +51,8 @@ interface CompletionService {
   String stepResultNotRecorded() throws SQLException;
 
   String interruptedThenFails() throws SQLException;
+
+  String cleanupOutlastsTheDeadline() throws SQLException;
 }
 
 class CompletionServiceImpl implements CompletionService {
@@ -58,6 +62,7 @@ class CompletionServiceImpl implements CompletionService {
   final CountDownLatch gate = new CountDownLatch(1);
   StepCheckpointStore brokenStore;
   StepCheckpointStore recordingStore;
+  StepCheckpointStore blockingStore;
   final AtomicInteger unrecordedRuns = new AtomicInteger();
 
   CompletionServiceImpl(DBOS dbos, JdbcStepFactory factory) {
@@ -118,6 +123,14 @@ public String interruptedThenFails() throws SQLException {
     throw new IllegalStateException("interrupted");
   }
 
+  @Override
+  @Workflow
+  public String cleanupOutlastsTheDeadline() throws SQLException {
+    dbos.integration().useStepCheckpointStore(blockingStore);
+    factory.txStep((Connection conn) -> {}, "one");
+    return "done";
+  }
+
   @Override
   @Workflow
   public String alsoUsesABrokenStore() throws SQLException {
@@ -257,6 +270,44 @@ public void anExecutionThatLostTheWorkflowDeletesNothing() throws Exception {
     assertEquals(1, checkpoints(workflowId));
   }
 
+  /**
+   * Deleting the checkpoints comes before the outcome is written, so the deadline still applies to
+   * it: a cleanup that outlasts the deadline doesn't hold the handle past it.
+   */
+  @Test
+  public void aCleanupThatOutlastsTheDeadlineStillEndsTheWaitAtIt() throws Exception {
+    launch(false);
+    var cleanupGate = new CountDownLatch(1);
+    impl.blockingStore =
+        new StepCheckpointStore() {
+          @Override
+          public void deleteCheckpoints(String workflowId, int fromStepId) {}
+
+          @Override
+          public void deleteCheckpoints(Collection<String> workflowIds) {}
+
+          @Override
+          public void deleteCompletedCheckpoints(String workflowId) {
+            try {
+              cleanupGate.await(60, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+              Thread.currentThread().interrupt();
+            }
+          }
+        };
+    try {
+      var handle =
+          dbos.startWorkflow(
+              proxy::cleanupOutlastsTheDeadline,
+              new StartWorkflowOptions("wf-slow-cleanup").withTimeout(Duration.ofSeconds(2)));
+      assertThrows(
+          DBOSAwaitedWorkflowCancelledException.class,
+          () -> assertTimeoutPreemptively(Duration.ofSeconds(15), () -> handle.getResult()));
+    } finally {
+      cleanupGate.countDown();
+    }
+  }
+
   @Test
   public void aFailedDeleteDoesNotChangeTheOutcomeOrSkipTheOtherFactories() throws Exception {
     launch(false);

transact/src/test/java/dev/dbos/transact/workflow/TimeoutTest.java

@@ -237,6 +237,31 @@ public void queuedTimedOut() {
     assertEquals(WorkflowState.CANCELLED, s.status());
   }
 
+  /** A run that finishes well before its deadline doesn't leave its timer queued until then. */
+  @Test
+  public void aFinishedRunLeavesNoTimerQueued() throws Exception {
+    SimpleServiceImpl impl = new SimpleServiceImpl(dbos);
+    SimpleService simpleService = dbos.registerProxy(SimpleService.class, impl);
+    impl.setSelf(simpleService);
+    dbos.launch();
+    var executor = DBOSTestAccess.getDbosExecutor(dbos);
+    int before = DBOSExecutorTestAccess.queuedTimeoutTasks(executor);
+
+    for (int i = 0; i < 5; i++) {
+      try (var id = new WorkflowOptions("timer-" + i).withTimeout(1, TimeUnit.HOURS).setContext()) {
+        assertEquals("Processed: ab", simpleService.workWithString("ab"));
+      }
+    }
+
+    // The timer is cancelled as the run's future completes, which may be just after getResult
+    // returns.
+    long giveUp = System.nanoTime() + Duration.ofSeconds(5).toNanos();
+    while (DBOSExecutorTestAccess.queuedTimeoutTasks(executor) != before) {
+      assertTrue(System.nanoTime() < giveUp, "the finished runs' timers are still queued");
+      Thread.sleep(20);
+    }
+  }
+
   @Test
   public void sync() throws Exception {