From 02d6265f3c37525c50d4b1d477925b2de164e4b4 Mon Sep 17 00:00:00 2001 From: maxdml Date: Mon, 27 Jul 2026 18:54:01 -0700 Subject: [PATCH 1/5] Update workflow status with a final result only when status is PENDING The terminal-outcome write (updateWorkflowOutcome) now applies only to a PENDING row and reports whether it landed: a run owns its workflow's outcome exactly as long as the row says that run is what the workflow is doing. When the write is refused, the runner parks on awaitWorkflowResult and delivers the recorded outcome through its own handle instead of its locally computed result. A missing row surfaces as DBOSNonExistentWorkflowException. awaitWorkflowResult now throws DBOSMaxRecoveryAttemptsExceededException for a dead-lettered row instead of polling forever. --- .../transact/database/SystemDatabase.java | 12 +- .../transact/database/dao/WorkflowDAO.java | 50 +++- .../dbos/transact/execution/DBOSExecutor.java | 42 ++- .../WorkflowOutcomeOwnershipTest.java | 257 ++++++++++++++++++ 4 files changed, 333 insertions(+), 28 deletions(-) create mode 100644 transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java 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..aef1c842 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)); } /** 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..ca47f88c 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,18 @@ 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, or handed to + * another execution (ENQUEUED/DELAYED, e.g. by a concurrent resume). If the row does not exist at + * all, a {@link DBOSNonExistentWorkflowException} is thrown. + */ + static boolean updateWorkflowOutcome( Connection conn, String schema, String workflowId, @@ -336,13 +347,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,11 +363,11 @@ 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. + // The guarded UPDATE matched no rows. Re-read (only on this rare no-op path) to + // distinguish a row this run no longer owns from a row that is gone. var readSql = """ SELECT status FROM "%s".workflow_status WHERE workflow_uuid = ? @@ -367,12 +376,14 @@ static void updateWorkflowOutcome( 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); + if (!rs.next()) { + throw new DBOSNonExistentWorkflowException(workflowId); } } } + return false; } + return true; } } @@ -381,12 +392,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 +408,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); } } @@ -1213,7 +1228,7 @@ public static Result awaitWorkflowResult( 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,6 +1261,13 @@ 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 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..d60a8d4b 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1804,7 +1804,16 @@ 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. + 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); + } return output; } catch (DBOSWorkflowExecutionConflictException e) { @@ -1826,18 +1835,31 @@ 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 terminal state), so the write would be refused anyway, and rethrowing here + // preserves the getResult() conversion to DBOSAwaitedWorkflowCancelledException. if (actual instanceof DBOSWorkflowCancelledException cancelled && cancelled.workflowId().equals(workflowId)) { throw cancelled; } + // The 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); + } throw e; } finally { DBOSContextHolder.clear(); @@ -2007,14 +2029,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/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..55877451 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java @@ -0,0 +1,257 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.*; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.context.DBOSContextHolder; +import dev.dbos.transact.exceptions.DBOSMaxRecoveryAttemptsExceededException; +import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; +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; + } + + 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"); + } +} + +interface OutcomeOwnershipService { + String blockedWorkflow() 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"; + } +} From fd49706a28e84b5f9cbfa21a1538c1ecfdcfb871 Mon Sep 17 00:00:00 2001 From: maxdml Date: Tue, 28 Jul 2026 14:28:40 -0700 Subject: [PATCH 2/5] Park and adopt the recorded outcome on own-cancellation A run that observes its own cancellation now parks on the recorded outcome instead of rethrowing from its local view: normally the row is CANCELLED and awaitWorkflowResult throws DBOSAwaitedWorkflowCancelledException (the same error the handle delivered before), but a concurrent resume may have taken the workflow back, in which case the recorded outcome is the truth. Safe to park because checkWorkflow only throws own-cancellation after reading CANCELLED from the DB, so the row is never left PENDING by this path. --- .../dbos/transact/execution/DBOSExecutor.java | 14 ++-- .../WorkflowOutcomeOwnershipTest.java | 64 +++++++++++++++++++ 2 files changed, 74 insertions(+), 4 deletions(-) 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 d60a8d4b..5b0155e3 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1834,12 +1834,18 @@ private WorkflowHandle executeWorkflow( logger.error("executeWorkflow {}", workflowId, actual); - // Skip persistWorkflowError for cancelled workflows: the DB already holds CANCELLED - // (the terminal state), so the write would be refused anyway, and rethrowing here - // preserves 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. 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); } // The outcome write found no workflow_status row at all (the workflow was deleted diff --git a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java index 55877451..e1dd3bae 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java @@ -6,8 +6,10 @@ 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; @@ -65,6 +67,19 @@ void beforeEach() { 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(); } @@ -233,10 +248,48 @@ public void deletedRowFailsTheRunWithNonExistentWorkflow() throws Exception { handle::getResult, "a run whose row vanished must not report a completion"); } + + @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()); + } } interface OutcomeOwnershipService { String blockedWorkflow() throws InterruptedException; + + String selfCancellingWorkflow() throws InterruptedException; } class OutcomeOwnershipServiceImpl implements OutcomeOwnershipService { @@ -254,4 +307,15 @@ public String blockedWorkflow() throws InterruptedException { 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); + } } From 71f6937615a4d640ca179527874350c3477a354b Mon Sep 17 00:00:00 2001 From: maxdml Date: Tue, 28 Jul 2026 14:48:07 -0700 Subject: [PATCH 3/5] Park and adopt the recorded outcome on execution conflict A run that loses the execution race (a concurrent run recorded a step checkpoint, or the workflow is already active on this executor) now parks on awaitWorkflowResult inside the task and delivers the recorded outcome through its own future, instead of completing the future exceptionally and relying on the handle's getResult conversion. The getResult conversions remain as backstops for futures produced before the park was in place. --- .../dbos/transact/execution/DBOSExecutor.java | 10 ++++- .../WorkflowOutcomeOwnershipTest.java | 37 +++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) 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 5b0155e3..bcd49448 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1817,8 +1817,14 @@ private WorkflowHandle executeWorkflow( 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. + logger.warn( + "Aborting duplicate execution of workflow. Waiting for the recorded outcome. workflowId {}", + workflowId); + return awaitWorkflowResult(workflowId); } catch (Exception e) { Throwable actual = e; diff --git a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java index e1dd3bae..f9676db4 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java @@ -3,6 +3,7 @@ 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; @@ -284,6 +285,42 @@ public void cancelledRunStillReportsCancellationForACancelledRow() throws Except "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 { From 9bfb285358a329f864703ad289fb3c1d6f7df05b Mon Sep 17 00:00:00 2001 From: maxdml Date: Tue, 28 Jul 2026 19:09:09 -0700 Subject: [PATCH 4/5] Detect a deleted row from the park instead of re-reading after the outcome write The PENDING-guarded outcome update now does exactly one statement and only reports whether the write landed. Runs that park on the recorded outcome know their row must already exist, so awaitWorkflowResult gains an opt-in failIfMissing that throws DBOSNonExistentWorkflowException on a missing row instead of polling forever. Tolerant callers (unchecked retrieves, debounced workflows whose rows appear later) keep the polling default. --- .../transact/database/SystemDatabase.java | 14 +++++- .../transact/database/dao/WorkflowDAO.java | 43 +++++++++---------- .../dbos/transact/execution/DBOSExecutor.java | 36 +++++++++++----- 3 files changed, 58 insertions(+), 35 deletions(-) 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 aef1c842..5468efe9 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -473,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 ca47f88c..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 @@ -325,9 +325,10 @@ ON CONFLICT (workflow_uuid) * 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, or handed to - * another execution (ENQUEUED/DELAYED, e.g. by a concurrent resume). If the row does not exist at - * all, a {@link DBOSNonExistentWorkflowException} is thrown. + *

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, @@ -365,25 +366,7 @@ static boolean updateWorkflowOutcome( stmt.setString(6, workflowId); stmt.setString(7, WorkflowState.PENDING.name()); - if (stmt.executeUpdate() == 0) { - // The guarded UPDATE matched no rows. Re-read (only on this rare no-op path) to - // distinguish a row this run no longer owns from a row that is gone. - 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()) { - throw new DBOSNonExistentWorkflowException(workflowId); - } - } - } - return false; - } - return true; + return stmt.executeUpdate() != 0; } } @@ -1221,9 +1204,19 @@ 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 = @@ -1271,6 +1264,10 @@ public static Result awaitWorkflowResult( 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 bcd49448..1dabd34f 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1095,8 +1095,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); } @@ -1808,11 +1818,13 @@ private WorkflowHandle executeWorkflow( // 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. + // 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); + return awaitWorkflowResult(workflowId, true); } return output; @@ -1820,11 +1832,12 @@ private WorkflowHandle executeWorkflow( // 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. + // 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); + return awaitWorkflowResult(workflowId, true); } catch (Exception e) { Throwable actual = e; @@ -1845,18 +1858,19 @@ private WorkflowHandle executeWorkflow( // 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. + // 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)) { logger.warn( "Workflow was cancelled during execution. Waiting for the recorded outcome. workflowId {}", workflowId); - return awaitWorkflowResult(workflowId); + return awaitWorkflowResult(workflowId, true); } - // The 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. + // 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; @@ -1870,7 +1884,7 @@ private WorkflowHandle executeWorkflow( 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); + return awaitWorkflowResult(workflowId, true); } throw e; } finally { From 24607e88996eaf4845d357e6ff33ea6548c76aa5 Mon Sep 17 00:00:00 2001 From: maxdml Date: Thu, 30 Jul 2026 08:39:46 -0700 Subject: [PATCH 5/5] failIfMissing must be passed with getresult --- .../dbos/transact/execution/DBOSExecutor.java | 12 ++++++++-- .../internal/WorkflowHandleDBPoll.java | 11 +++++++++- .../WorkflowOutcomeOwnershipTest.java | 22 +++++++++++++++++++ 3 files changed, 42 insertions(+), 3 deletions(-) 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 1dabd34f..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") @@ -1766,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)) { 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 index f9676db4..fc5a333b 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/WorkflowOutcomeOwnershipTest.java @@ -250,6 +250,28 @@ public void deletedRowFailsTheRunWithNonExistentWorkflow() throws Exception { "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