diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index ac2cd6a7..5468efe9 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -366,9 +366,11 @@ 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)); } /** @@ -376,9 +378,11 @@ public void recordWorkflowOutput(String workflowId, String result) { * * @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 listWorkflowSteps( } public Result awaitWorkflowResult(String workflowId) { - return dbRetry(() -> WorkflowDAO.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 Result awaitWorkflowResult(String workflowId, boolean failIfMissing) { + return dbRetry( + () -> + WorkflowDAO.awaitWorkflowResult(ctx, dbPollingInterval, workflowId, failIfMissing)); } public List startQueuedWorkflows( diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java index dd8dc140..5fd1654c 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/WorkflowDAO.java +++ b/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.) + * + *

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. + * + *

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 Result 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 Result 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 } diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java index 5e19fb43..7233e3b6 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1062,8 +1062,12 @@ public WorkflowStatus getWorkflowStatus(String workflowId) { } public T getResult(String workflowId) throws E { + return getResult(workflowId, false); + } + + public 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 getResult(String workflowId, Future 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 awaitWorkflowResult(String workflowId) throws E { - var result = systemDatabase.awaitWorkflowResult(workflowId); + return awaitWorkflowResult(workflowId, false); + } + + private T awaitWorkflowResult(String workflowId, boolean failIfMissing) + throws E { + var result = systemDatabase.awaitWorkflowResult(workflowId, failIfMissing); return Result.process(result); } @@ -1756,7 +1770,11 @@ private WorkflowHandle 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 WorkflowHandle 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 WorkflowHandle 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()); } /** diff --git a/transact/src/main/java/dev/dbos/transact/workflow/internal/WorkflowHandleDBPoll.java b/transact/src/main/java/dev/dbos/transact/workflow/internal/WorkflowHandleDBPoll.java index d7bd263a..723effa0 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/internal/WorkflowHandleDBPoll.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/internal/WorkflowHandleDBPoll.java @@ -7,10 +7,19 @@ public class WorkflowHandleDBPoll implements WorkflowHandle { 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 diff --git a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java new file mode 100644 index 00000000..fc5a333b --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java @@ -0,0 +1,380 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.*; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.DBOSTestAccess; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.context.DBOSContextHolder; +import dev.dbos.transact.exceptions.DBOSAwaitedWorkflowCancelledException; +import dev.dbos.transact.exceptions.DBOSMaxRecoveryAttemptsExceededException; +import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; +import dev.dbos.transact.exceptions.DBOSWorkflowCancelledException; +import dev.dbos.transact.json.SerializationUtil; +import dev.dbos.transact.utils.PgContainer; + +import java.sql.SQLException; +import java.time.Instant; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import com.zaxxer.hikari.HikariDataSource; +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * A run may record its outcome only while its workflow_status row is still PENDING: that row is + * what says "this run is what the workflow is doing". Every other status means the run lost + * ownership (a concurrent resume re-enqueued it, a recovery raced it, it was cancelled or + * dead-lettered) and the recorded outcome, not the one the run computed, is the workflow's outcome. + */ +public class WorkflowOutcomeOwnershipTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + + DBOSConfig dbosConfig; + @AutoClose DBOS dbos; + @AutoClose HikariDataSource dataSource; + + private OutcomeOwnershipService proxy; + private OutcomeOwnershipServiceImpl impl; + + @BeforeEach + void beforeEach() { + dbosConfig = pgContainer.dbosConfig().withAppVersion("v1.0.0"); + dbos = new DBOS(dbosConfig); + dataSource = pgContainer.dataSource(); + + impl = new OutcomeOwnershipServiceImpl(); + proxy = dbos.registerProxy(OutcomeOwnershipService.class, impl); + + dbos.launch(); + } + + // Starts a run and returns once it is blocked inside the workflow function, with its row + // PENDING. + private WorkflowHandle startBlockedRun(String workflowId) throws InterruptedException { + impl.startedLatches.put(workflowId, new CountDownLatch(1)); + impl.releaseLatches.put(workflowId, new CountDownLatch(1)); + var handle = + dbos.startWorkflow(() -> proxy.blockedWorkflow(), new StartWorkflowOptions(workflowId)); + impl.startedLatches.get(workflowId).await(); + return handle; + } + + // Starts a run that will observe its own cancellation once released, and returns once it is + // blocked inside the workflow function, with its row PENDING. + private WorkflowHandle startSelfCancellingRun(String workflowId) + throws InterruptedException { + impl.startedLatches.put(workflowId, new CountDownLatch(1)); + impl.releaseLatches.put(workflowId, new CountDownLatch(1)); + var handle = + dbos.startWorkflow( + () -> proxy.selfCancellingWorkflow(), new StartWorkflowOptions(workflowId)); + impl.startedLatches.get(workflowId).await(); + return handle; + } + + private void releaseRun(String workflowId) { + impl.releaseLatches.get(workflowId).countDown(); + } + + // Takes the row away from the blocked run, standing in for the concurrent + // resume/recovery/cancel that would do it in production. + private void rewriteRow(String workflowId, WorkflowState status, String output, String error) + throws SQLException { + var sql = + "UPDATE dbos.workflow_status SET status = ?, output = ?, error = ?, updated_at = ?" + + " WHERE workflow_uuid = ?"; + try (var conn = dataSource.getConnection(); + var stmt = conn.prepareStatement(sql)) { + stmt.setString(1, status.name()); + stmt.setString(2, output); + stmt.setString(3, error); + stmt.setLong(4, Instant.now().toEpochMilli()); + stmt.setString(5, workflowId); + assertEquals(1, stmt.executeUpdate()); + } + } + + private void setRecoveryAttempts(String workflowId, int attempts) throws SQLException { + var sql = "UPDATE dbos.workflow_status SET recovery_attempts = ? WHERE workflow_uuid = ?"; + try (var conn = dataSource.getConnection(); + var stmt = conn.prepareStatement(sql)) { + stmt.setInt(1, attempts); + stmt.setString(2, workflowId); + assertEquals(1, stmt.executeUpdate()); + } + } + + private void deleteRow(String workflowId) throws SQLException { + var sql = "DELETE FROM dbos.workflow_status WHERE workflow_uuid = ?"; + try (var conn = dataSource.getConnection(); + var stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + assertEquals(1, stmt.executeUpdate()); + } + } + + private record Row(String status, String output) {} + + private Row readRow(String workflowId) throws SQLException { + var sql = "SELECT status, output FROM dbos.workflow_status WHERE workflow_uuid = ?"; + try (var conn = dataSource.getConnection(); + var stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + try (var rs = stmt.executeQuery()) { + assertTrue(rs.next(), "workflow row not found: " + workflowId); + return new Row(rs.getString("status"), rs.getString("output")); + } + } + } + + // Mirrors the default workflow serializer, so rewritten outputs/errors deserialize the same + // way a recorded outcome would. + private static String serializeValue(Object value) { + return SerializationUtil.serializeValue(value, null, null).serializedValue(); + } + + private static String serializeError(Throwable error) { + return SerializationUtil.serializeError(error, null, null).serializedValue(); + } + + @Test + public void recordedSuccessSupersedesTheRunResult() throws Exception { + var workflowId = "outcome-ownership-success-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + var recorded = serializeValue("recorded-elsewhere"); + rewriteRow(workflowId, WorkflowState.SUCCESS, recorded, null); + releaseRun(workflowId); + + assertEquals( + "recorded-elsewhere", + handle.getResult(), + "the run must report the recorded output, not its own"); + + var row = readRow(workflowId); + assertEquals(WorkflowState.SUCCESS.name(), row.status()); + assertEquals(recorded, row.output(), "the recorded output must not be overwritten"); + } + + @Test + public void recordedErrorSupersedesTheRunResult() throws Exception { + var workflowId = "outcome-ownership-error-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + rewriteRow( + workflowId, + WorkflowState.ERROR, + null, + serializeError(new IllegalStateException("recorded failure"))); + releaseRun(workflowId); + + var e = + assertThrows( + IllegalStateException.class, handle::getResult, "the recorded error must be adopted"); + assertEquals("recorded failure", e.getMessage()); + assertEquals(WorkflowState.ERROR.name(), readRow(workflowId).status()); + } + + @Test + public void nonTerminalRowParksTheRunUntilAnOutcomeIsRecorded() throws Exception { + var workflowId = "outcome-ownership-parked-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + // ENQUEUED with no queue name: nothing dequeues it, so the run stays parked until this test + // records the outcome itself. + rewriteRow(workflowId, WorkflowState.ENQUEUED, null, null); + releaseRun(workflowId); + + var done = + CompletableFuture.supplyAsync( + () -> { + try { + return handle.getResult(); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + + assertThrows( + TimeoutException.class, + () -> done.get(3, TimeUnit.SECONDS), + "the run must wait for the owning execution"); + + rewriteRow(workflowId, WorkflowState.SUCCESS, serializeValue("recorded-by-owner"), null); + + assertEquals( + "recorded-by-owner", + done.get(30, TimeUnit.SECONDS), + "the parked run must adopt the recorded outcome"); + } + + @Test + public void deadLetteredRowFailsTheRun() throws Exception { + var workflowId = "outcome-ownership-dlq-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + rewriteRow(workflowId, WorkflowState.MAX_RECOVERY_ATTEMPTS_EXCEEDED, null, null); + // A workflow is dead-lettered by the attempt that pushes recovery_attempts past + // maxRetries+1, so a dead-lettered row carries maxRetries+2 attempts. + final int maxRetries = 3; + setRecoveryAttempts(workflowId, maxRetries + 2); + releaseRun(workflowId); + + var e = + assertThrows( + DBOSMaxRecoveryAttemptsExceededException.class, + handle::getResult, + "a dead-lettered workflow must not report a completion"); + assertEquals(maxRetries, e.maxRetries(), "the error must report the exhausted retry budget"); + + var row = readRow(workflowId); + assertEquals(WorkflowState.MAX_RECOVERY_ATTEMPTS_EXCEEDED.name(), row.status()); + assertNull(row.output(), "the refused outcome must not record an output"); + } + + @Test + public void deletedRowFailsTheRunWithNonExistentWorkflow() throws Exception { + var workflowId = "outcome-ownership-deleted-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + deleteRow(workflowId); + releaseRun(workflowId); + + assertThrows( + DBOSNonExistentWorkflowException.class, + handle::getResult, + "a run whose row vanished must not report a completion"); + } + + @Test + public void completedWorkflowWhoseRowVanishesFailsWithNonExistentWorkflow() throws Exception { + var workflowId = "outcome-ownership-completed-deleted-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + releaseRun(workflowId); + assertEquals("own-result", handle.getResult()); + + // A dispatch of an already-completed workflow does not re-execute it: it hands back a + // handle onto the recorded outcome. That row was just read, so a row that is gone by the + // time the outcome is read was deleted — fail fast instead of polling for a row that will + // never reappear. + var redispatched = + DBOSTestAccess.getDbosExecutor(dbos) + .executeWorkflowById(workflowId, true, false); + deleteRow(workflowId); + + assertThrows( + DBOSNonExistentWorkflowException.class, + redispatched::getResult, + "a completed workflow whose row was deleted must not be polled for"); + } + + @Test + public void cancelledRunAdoptsARecordedOutcome() throws Exception { + // A run that observes its own cancellation adopts the recorded outcome rather than trusting + // its local view: here a concurrent "resume" already rewrote the row to SUCCESS, so the + // handle reports that outcome instead of a cancellation that is no longer the workflow's + // state. + var workflowId = "outcome-ownership-cancel-adopt-%d".formatted(System.currentTimeMillis()); + var handle = startSelfCancellingRun(workflowId); + var recorded = serializeValue("recorded-after-cancel"); + rewriteRow(workflowId, WorkflowState.SUCCESS, recorded, null); + releaseRun(workflowId); + + assertEquals( + "recorded-after-cancel", + handle.getResult(), + "the run must adopt the recorded outcome, not report its cancellation"); + + var row = readRow(workflowId); + assertEquals(WorkflowState.SUCCESS.name(), row.status()); + assertEquals(recorded, row.output(), "the recorded output must not be overwritten"); + } + + @Test + public void cancelledRunStillReportsCancellationForACancelledRow() throws Exception { + var workflowId = "outcome-ownership-cancelled-%d".formatted(System.currentTimeMillis()); + var handle = startSelfCancellingRun(workflowId); + rewriteRow(workflowId, WorkflowState.CANCELLED, null, null); + releaseRun(workflowId); + + assertThrows( + DBOSAwaitedWorkflowCancelledException.class, + handle::getResult, + "a genuinely cancelled workflow must still report its cancellation"); + assertEquals(WorkflowState.CANCELLED.name(), readRow(workflowId).status()); + } + + @Test + public void conflictingExecutionParksAndAdoptsTheRecordedOutcome() throws Exception { + // A recovery dispatch of a workflow that is already active on this executor loses the + // start race: it must park and adopt the outcome recorded by the run that owns the + // workflow, not surface the conflict. + var workflowId = "outcome-ownership-conflict-%d".formatted(System.currentTimeMillis()); + var handle = startBlockedRun(workflowId); + + var duplicate = + DBOSTestAccess.getDbosExecutor(dbos) + .executeWorkflowById(workflowId, true, false); + + var done = + CompletableFuture.supplyAsync( + () -> { + try { + return duplicate.getResult(); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + + assertThrows( + TimeoutException.class, + () -> done.get(3, TimeUnit.SECONDS), + "the duplicate must wait for the owning execution"); + + releaseRun(workflowId); + + assertEquals( + "own-result", + done.get(30, TimeUnit.SECONDS), + "the duplicate must adopt the recorded outcome"); + assertEquals("own-result", handle.getResult()); + } +} + +interface OutcomeOwnershipService { + String blockedWorkflow() throws InterruptedException; + + String selfCancellingWorkflow() throws InterruptedException; +} + +class OutcomeOwnershipServiceImpl implements OutcomeOwnershipService { + + // Per-workflow latches, keyed by workflow ID: each run blocks until the test has rewritten its + // row, then returns a result the test can tell apart from anything recorded out-of-band. + final ConcurrentHashMap startedLatches = new ConcurrentHashMap<>(); + final ConcurrentHashMap releaseLatches = new ConcurrentHashMap<>(); + + @Override + @Workflow + public String blockedWorkflow() throws InterruptedException { + var wfId = DBOSContextHolder.get().getWorkflowId(); + startedLatches.get(wfId).countDown(); + releaseLatches.get(wfId).await(); + return "own-result"; + } + + // Stands in for a run that observes its own cancellation mid-flight: the cancellation is + // thrown only after the test has rewritten the row. + @Override + @Workflow + public String selfCancellingWorkflow() throws InterruptedException { + var wfId = DBOSContextHolder.get().getWorkflowId(); + startedLatches.get(wfId).countDown(); + releaseLatches.get(wfId).await(); + throw new DBOSWorkflowCancelledException(wfId); + } +}