From 8be290ea0f0888558b914476daae704b0199907e Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 10:06:38 -0800 Subject: [PATCH 1/3] Refuse a workflow cancelling, resuming or deleting itself Called from a workflow, or from one of its steps, a cancel, resume or delete whose targets include the calling workflow now throws IllegalArgumentException before anything is written. The workflow can catch it and carry on. Each of these clears or removes the caller's own row, so under the ownership fence the caller's next checkpoint fails. A self-cancel ended CANCELLED with no step recorded, a self-delete ended with the workflow gone, and a self-resume put the workflow back on the queue before its own step was recorded: each dispatch replayed to the same resume and did it again, with no end short of deleting the workflow. The check runs inside the call's step, so from a workflow body the refusal is recorded as that step's error and a replay or fork throws it again rather than deciding afresh. From a step body the call takes no step of its own, and the refusal is thrown into the enclosing step. This covers a caller that names itself. Refusing a cancel or delete of a tree the caller belongs to (an ancestor with cancelChildren or deleteChildren) needs the cascade and its step record in one transaction, and comes with that change. --- .../src/main/java/dev/dbos/transact/DBOS.java | 24 ++ .../dbos/transact/execution/DBOSExecutor.java | 26 ++ .../transact/workflow/OwnershipFenceTest.java | 71 ---- .../transact/workflow/SelfTargetTest.java | 366 ++++++++++++++++++ 4 files changed, 416 insertions(+), 71 deletions(-) create mode 100644 transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index a9b607154..4af7494a1 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOS.java +++ b/transact/src/main/java/dev/dbos/transact/DBOS.java @@ -872,6 +872,8 @@ public void runStep( * will be resumed in the default execution context * @return A handle to the resumed workflow * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being resumed, or one of + * its steps: a workflow can't resume itself. Nothing is resumed. */ @SuppressWarnings("unchecked") public @NonNull WorkflowHandle resumeWorkflow( @@ -892,6 +894,8 @@ public void runStep( * @param workflowId ID of the workflow to resume; must not be null * @return A handle to the resumed workflow * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being resumed, or one of + * its steps: a workflow can't resume itself. Nothing is resumed. */ public @NonNull WorkflowHandle resumeWorkflow( @NonNull String workflowId) { @@ -909,6 +913,8 @@ public void runStep( * workflows will be resumed in the default execution context * @return A list of handles to the resumed workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being resumed, or + * one of its steps: a workflow can't resume itself. Nothing is resumed. */ public @NonNull List> resumeWorkflows( @NonNull List workflowIds, @Nullable String queueName) { @@ -924,6 +930,8 @@ public void runStep( * @param workflowIds a list of workflow IDs to resume; must not be null * @return A list of handles to the resumed workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being resumed, or + * one of its steps: a workflow can't resume itself. Nothing is resumed. */ public @NonNull List> resumeWorkflows( @NonNull List workflowIds) { @@ -937,6 +945,8 @@ public void runStep( * * @param workflowId ID of the workflow to cancel * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being cancelled, or one of + * its steps: a workflow can't cancel itself. Nothing is cancelled. */ public void cancelWorkflow(@NonNull String workflowId) { cancelWorkflows(List.of(workflowId), false); @@ -948,6 +958,8 @@ public void cancelWorkflow(@NonNull String workflowId) { * @param workflowId ID of the workflow to cancel * @param cancelChildren if true, recursively cancel all descendant workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being cancelled, or one of + * its steps: a workflow can't cancel itself. Nothing is cancelled. */ public void cancelWorkflow(@NonNull String workflowId, boolean cancelChildren) { cancelWorkflows(List.of(workflowId), cancelChildren); @@ -959,6 +971,8 @@ public void cancelWorkflow(@NonNull String workflowId, boolean cancelChildren) { * * @param workflowIds a list of workflow IDs to cancel; must not be null * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being cancelled, or + * one of its steps: a workflow can't cancel itself. Nothing is cancelled. */ public void cancelWorkflows(@NonNull List workflowIds) { cancelWorkflows(workflowIds, false); @@ -970,6 +984,8 @@ public void cancelWorkflows(@NonNull List workflowIds) { * @param workflowIds a list of workflow IDs to cancel; must not be null * @param cancelChildren if true, recursively cancel all descendant workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being cancelled, or + * one of its steps: a workflow can't cancel itself. Nothing is cancelled. */ public void cancelWorkflows(@NonNull List workflowIds, boolean cancelChildren) { ensureLaunched("cancelWorkflows").cancelWorkflows(workflowIds, cancelChildren); @@ -981,6 +997,8 @@ public void cancelWorkflows(@NonNull List workflowIds, boolean cancelChi * * @param workflowId ID of the workflow to delete; must not be null * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being deleted, or one of + * its steps: a workflow can't delete itself. Nothing is deleted. */ public void deleteWorkflow(@NonNull String workflowId) { deleteWorkflows(List.of(workflowId), false); @@ -994,6 +1012,8 @@ public void deleteWorkflow(@NonNull String workflowId) { * @param deleteChildren if true, also delete any child workflows; if false, preserve child * workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside the workflow being deleted, or one of + * its steps: a workflow can't delete itself. Nothing is deleted. */ public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) { deleteWorkflows(List.of(workflowId), deleteChildren); @@ -1005,6 +1025,8 @@ public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) { * * @param workflowIds a list of workflow IDs to delete; must not be null * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being deleted, or + * one of its steps: a workflow can't delete itself. Nothing is deleted. */ public void deleteWorkflows(@NonNull List workflowIds) { deleteWorkflows(workflowIds, false); @@ -1018,6 +1040,8 @@ public void deleteWorkflows(@NonNull List workflowIds) { * @param deleteChildren if true, also delete any child workflows; if false, preserve child * workflows * @throws IllegalStateException if called before DBOS is launched + * @throws IllegalArgumentException if called from inside one of the workflows being deleted, or + * one of its steps: a workflow can't delete itself. Nothing is deleted. */ public void deleteWorkflows(@NonNull List workflowIds, boolean deleteChildren) { ensureLaunched("deleteWorkflows").deleteWorkflows(workflowIds, deleteChildren); 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 5c39b8f58..5ed84b31b 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1083,9 +1083,11 @@ public void sleep(Duration duration) { public void cancelWorkflows(List workflowIds, boolean cancelChildren) { Objects.requireNonNull(workflowIds); + var callerWorkflowId = DBOSContextHolder.get().getWorkflowId(); this.runDbosFunctionAsStep( () -> { + refuseSelfTarget(callerWorkflowId, workflowIds, "cancel"); logger.info( "Cancelling workflow(s) {}{}", workflowIds, @@ -1099,8 +1101,10 @@ public void cancelWorkflows(List workflowIds, boolean cancelChildren) { public void deleteWorkflows(List workflowIds, boolean deleteChildren) { Objects.requireNonNull(workflowIds); + var callerWorkflowId = DBOSContextHolder.get().getWorkflowId(); this.runDbosFunctionAsStep( () -> { + refuseSelfTarget(callerWorkflowId, workflowIds, "delete"); logger.info( "Deleting workflow(s) {}{}", workflowIds, @@ -1115,10 +1119,12 @@ public void deleteWorkflows(List workflowIds, boolean deleteChildren) { public List> resumeWorkflows( List workflowIds, String queueName) { Objects.requireNonNull(workflowIds); + var callerWorkflowId = DBOSContextHolder.get().getWorkflowId(); // Execute the resume operation as a workflow step this.runDbosFunctionAsStep( () -> { + refuseSelfTarget(callerWorkflowId, workflowIds, "resume"); logger.info("Resuming workflow(s) {}", workflowIds); systemDatabase.resumeWorkflows(workflowIds, queueName); return null; // void @@ -1129,6 +1135,26 @@ public List> resumeWorkflows( return workflowIds.stream().map(this::retrieveWorkflow).toList(); } + /** + * Refuses a cancel, resume or delete that names the workflow making the call, from its own body + * or from one of its steps, before anything is written. Each clears or removes the caller's own + * row, so the ownership check on the caller's next checkpoint would fail: a self-cancel would + * record no step, a self-delete would leave the workflow running with no row, and a self-resume + * would re-dispatch the workflow again and again. A workflow that wants to stop returns or throws + * instead. + * + *

It runs inside the call's step, so from a workflow body the refusal is recorded as that + * step's error and a replay throws it again; from a step body, where the call takes no step of + * its own, it is thrown into that step. + */ + private static void refuseSelfTarget( + @Nullable String callerWorkflowId, Collection targets, String action) { + if (callerWorkflowId != null && targets.contains(callerWorkflowId)) { + throw new IllegalArgumentException( + "workflow %s cannot %s itself".formatted(callerWorkflowId, action)); + } + } + public void updateWorkflowAttributes(String workflowId, Map attributes) { Objects.requireNonNull(workflowId); this.runDbosFunctionAsStep( diff --git a/transact/src/test/java/dev/dbos/transact/workflow/OwnershipFenceTest.java b/transact/src/test/java/dev/dbos/transact/workflow/OwnershipFenceTest.java index 4e0b90fdc..eb429897e 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/OwnershipFenceTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/OwnershipFenceTest.java @@ -7,8 +7,6 @@ import dev.dbos.transact.StartWorkflowOptions; import dev.dbos.transact.context.DBOSContextHolder; import dev.dbos.transact.database.DBTestAccess; -import dev.dbos.transact.exceptions.DBOSAwaitedWorkflowCancelledException; -import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; import dev.dbos.transact.json.SerializationUtil; import dev.dbos.transact.txstep.JdbcStepFactory; import dev.dbos.transact.utils.DBUtils; @@ -280,44 +278,6 @@ public void aStaleExecutionsDeadlineDoesNotCancelTheNewOwner() throws Exception assertEquals("PENDING", readStatus(workflowId), "the stale timer must not cancel it"); impl.gates.get(workflowId).countDown(); } - - @Test - public void aWorkflowCancellingItselfEndsCancelled() throws Exception { - var workflowId = "fence-self-cancel-%d".formatted(System.nanoTime()); - var handle = dbos.startWorkflow(proxy::cancelSelf, new StartWorkflowOptions(workflowId)); - - assertThrows(DBOSAwaitedWorkflowCancelledException.class, handle::getResult); - assertEquals("CANCELLED", readStatus(workflowId)); - assertEquals(0, impl.afterStepRuns.get(), "nothing runs after the cancel"); - } - - @Test - public void aWorkflowDeletingItselfEndsMissing() throws Exception { - var workflowId = "fence-self-delete-%d".formatted(System.nanoTime()); - var handle = dbos.startWorkflow(proxy::deleteSelf, new StartWorkflowOptions(workflowId)); - - assertThrows(DBOSNonExistentWorkflowException.class, handle::getResult); - assertEquals(0, impl.afterStepRuns.get(), "nothing runs after the delete"); - } - - // Recorded as it behaves today: nothing refuses a workflow resuming itself. The resume hands the - // workflow back to the queue before its own step is recorded, so the run is refused there and - // waits, and the dispatch the resume caused replays to the same resume and does it again. Only - // removing the workflow ends it. - @Test - public void aWorkflowResumingItselfIsDispatchedAgainAndAgain() throws Exception { - var workflowId = "fence-self-resume-%d".formatted(System.nanoTime()); - var handle = dbos.startWorkflow(proxy::resumeSelf, new StartWorkflowOptions(workflowId)); - - var deadline = Instant.now().plusSeconds(30); - while (impl.resumeSelfRuns.get() < 3 && Instant.now().isBefore(deadline)) { - Thread.sleep(50); - } - assertTrue(impl.resumeSelfRuns.get() >= 3, "each dispatch resumes the workflow again"); - - dbos.deleteWorkflow(workflowId); - assertThrows(DBOSNonExistentWorkflowException.class, handle::getResult); - } } interface FenceService { @@ -328,12 +288,6 @@ interface FenceService { String child(); String quick(); - - String cancelSelf(); - - String deleteSelf(); - - String resumeSelf(); } class FenceServiceImpl implements FenceService { @@ -348,7 +302,6 @@ class FenceServiceImpl implements FenceService { final Map gates = new ConcurrentHashMap<>(); final AtomicInteger afterStepRuns = new AtomicInteger(); final AtomicInteger handedOffRuns = new AtomicInteger(); - final AtomicInteger resumeSelfRuns = new AtomicInteger(); FenceServiceImpl(DBOS dbos, JdbcStepFactory stepFactory) { this.dbos = dbos; @@ -435,28 +388,4 @@ public String child() { public String quick() { return "quick"; } - - @Override - @Workflow - public String cancelSelf() { - dbos.cancelWorkflow(workflowId()); - dbos.runStep(() -> afterStepRuns.incrementAndGet(), "afterStep"); - return "not cancelled"; - } - - @Override - @Workflow - public String deleteSelf() { - dbos.deleteWorkflow(workflowId()); - dbos.runStep(() -> afterStepRuns.incrementAndGet(), "afterStep"); - return "not deleted"; - } - - @Override - @Workflow - public String resumeSelf() { - resumeSelfRuns.incrementAndGet(); - dbos.resumeWorkflow(workflowId()); - return "resumed"; - } } diff --git a/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java new file mode 100644 index 000000000..a32314561 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java @@ -0,0 +1,366 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.config.DBOSConfig; +import dev.dbos.transact.context.DBOSContext; +import dev.dbos.transact.utils.PgContainer; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * A workflow may not cancel, resume or delete itself, from its own body or from one of its steps. + * Each clears or removes its own row, so the ownership check on its next checkpoint would fail: a + * self-resume, for one, would re-dispatch the workflow again and again. The call is refused before + * anything is written, and the workflow can catch the refusal and carry on. Targeting other + * workflows, its own children's trees included, is unaffected. + */ +public class SelfTargetTest { + @AutoClose final PgContainer pgContainer = new PgContainer(); + + @AutoClose DBOS dbos; + + private SelfTargetService proxy; + private SelfTargetServiceImpl impl; + + @BeforeEach + void beforeEach() { + launch(pgContainer.dbosConfig()); + } + + private void launch(DBOSConfig config) { + dbos = new DBOS(config); + impl = new SelfTargetServiceImpl(dbos); + proxy = dbos.registerProxy(SelfTargetService.class, impl); + impl.self = proxy; + dbos.launch(); + } + + private StepInfo stepNamed(String workflowId, String name) { + return dbos.listWorkflowSteps(workflowId).stream() + .filter(step -> step.functionName().equals(name)) + .findFirst() + .orElseThrow(() -> new AssertionError("no step " + name + " in " + workflowId)); + } + + @Test + public void aWorkflowCannotCancelItself() throws Exception { + var workflowId = "self-cancel-%d".formatted(System.nanoTime()); + var handle = dbos.startWorkflow(proxy::cancelSelf, new StartWorkflowOptions(workflowId)); + + assertEquals( + "refused: workflow %s cannot cancel itself".formatted(workflowId), handle.getResult()); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + assertEquals(1, impl.afterRefusal.get(), "the workflow carries on after the refusal"); + // Recorded as the call's step, so a replay is refused the same way. + var step = stepNamed(workflowId, "DBOS.cancelWorkflow"); + assertEquals(IllegalArgumentException.class.getName(), step.error().className()); + } + + @Test + public void aRecordedRefusalReplaysRatherThanBeingDecidedAgain() throws Exception { + var workflowId = "self-cancel-replay-%d".formatted(System.nanoTime()); + var original = dbos.startWorkflow(proxy::cancelSelf, new StartWorkflowOptions(workflowId)); + var refusal = original.getResult(); + + // A fork from the step after the cancel copies the cancel's recorded refusal and replays it. + // Decided again, the refusal would name the fork, not the original. + var cancelStep = stepNamed(workflowId, "DBOS.cancelWorkflow").functionId(); + var fork = dbos.forkWorkflow(workflowId, cancelStep + 1); + assertEquals(refusal, fork.getResult()); + assertTrue(refusal.contains(workflowId), refusal); + } + + @Test + public void aBulkCancelNamingItsCallerLeavesTheOthersUntouched() throws Exception { + var otherId = "bulk-other-%d".formatted(System.nanoTime()); + var other = dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(otherId)); + assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); + + var callerId = "bulk-caller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.cancelsOtherAndSelf(otherId), new StartWorkflowOptions(callerId)); + + assertEquals( + "refused: workflow %s cannot cancel itself".formatted(callerId), caller.getResult()); + assertEquals( + WorkflowState.PENDING, + dbos.retrieveWorkflow(otherId).getStatus().status(), + "the refusal came before any write"); + impl.release.countDown(); + assertEquals(otherId, other.getResult()); + } + + @Test + public void aWorkflowCannotResumeItself() throws Exception { + var workflowId = "self-resume-%d".formatted(System.nanoTime()); + var handle = dbos.startWorkflow(proxy::resumeSelf, new StartWorkflowOptions(workflowId)); + + assertEquals( + "refused: workflow %s cannot resume itself".formatted(workflowId), handle.getResult()); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + assertEquals(1, impl.afterRefusal.get(), "dispatched once, not again and again"); + var step = stepNamed(workflowId, "DBOS.resumeWorkflow"); + assertEquals(IllegalArgumentException.class.getName(), step.error().className()); + } + + @Test + public void aWorkflowCannotDeleteItself() throws Exception { + var workflowId = "self-delete-%d".formatted(System.nanoTime()); + var handle = dbos.startWorkflow(proxy::deleteSelf, new StartWorkflowOptions(workflowId)); + + assertEquals( + "refused: workflow %s cannot delete itself".formatted(workflowId), handle.getResult()); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + var step = stepNamed(workflowId, "DBOS.deleteWorkflow"); + assertEquals(IllegalArgumentException.class.getName(), step.error().className()); + } + + @Test + public void aStepCannotCancelOrResumeItsOwnWorkflow() throws Exception { + var cancelId = "step-self-cancel-%d".formatted(System.nanoTime()); + var cancel = dbos.startWorkflow(proxy::cancelSelfInStep, new StartWorkflowOptions(cancelId)); + assertEquals( + "refused: workflow %s cannot cancel itself".formatted(cancelId), cancel.getResult()); + assertEquals(WorkflowState.SUCCESS, cancel.getStatus().status()); + + var resumeId = "step-self-resume-%d".formatted(System.nanoTime()); + var resume = dbos.startWorkflow(proxy::resumeSelfInStep, new StartWorkflowOptions(resumeId)); + assertEquals( + "refused: workflow %s cannot resume itself".formatted(resumeId), resume.getResult()); + assertEquals(WorkflowState.SUCCESS, resume.getStatus().status()); + + var deleteId = "step-self-delete-%d".formatted(System.nanoTime()); + var delete = dbos.startWorkflow(proxy::deleteSelfInStep, new StartWorkflowOptions(deleteId)); + assertEquals( + "refused: workflow %s cannot delete itself".formatted(deleteId), delete.getResult()); + assertEquals(WorkflowState.SUCCESS, delete.getStatus().status()); + } + + @Test + public void aRefusalAStepDoesNotCatchIsThatStepsError() throws Exception { + var workflowId = "step-uncaught-%d".formatted(System.nanoTime()); + var handle = + dbos.startWorkflow(proxy::cancelSelfInUncaughtStep, new StartWorkflowOptions(workflowId)); + + assertEquals( + "step failed: workflow %s cannot cancel itself".formatted(workflowId), handle.getResult()); + // Nothing of the cancel's own is recorded inside a step; the refusal is the step's result. + var step = stepNamed(workflowId, "uncaughtCancel"); + assertEquals(IllegalArgumentException.class.getName(), step.error().className()); + } + + @Test + public void aParentCanCancelItsOwnChildsTree() throws Exception { + var parentId = "cancels-child-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + var handle = + dbos.startWorkflow( + () -> proxy.cancelsChildTree(childId), new StartWorkflowOptions(parentId)); + + assertEquals("cancelled " + childId, handle.getResult()); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + assertEquals(WorkflowState.CANCELLED, dbos.retrieveWorkflow(childId).getStatus().status()); + impl.release.countDown(); + } + + @Test + public void aWorkflowCanStillCancelAndResumeAnother() throws Exception { + var otherId = "other-%d".formatted(System.nanoTime()); + var other = dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(otherId)); + assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); + + var callerId = "canceller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.cancelThenResume(otherId), new StartWorkflowOptions(callerId)); + assertEquals("done", caller.getResult()); + + assertNull(stepNamed(callerId, "DBOS.cancelWorkflow").error()); + assertNull(stepNamed(callerId, "DBOS.resumeWorkflow").error()); + impl.release.countDown(); + assertEquals(otherId, other.getResult()); + } +} + +interface SelfTargetService { + String cancelSelf(); + + String resumeSelf(); + + String cancelSelfInStep(); + + String resumeSelfInStep(); + + String deleteSelf(); + + String deleteSelfInStep(); + + String cancelSelfInUncaughtStep(); + + String cancelsChildTree(String childId) throws Exception; + + String cancelsOtherAndSelf(String otherId); + + String blocked() throws InterruptedException; + + String cancelThenResume(String otherId); +} + +class SelfTargetServiceImpl implements SelfTargetService { + private final DBOS dbos; + SelfTargetService self; + final AtomicInteger afterRefusal = new AtomicInteger(); + final CountDownLatch blockedStarted = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + + SelfTargetServiceImpl(DBOS dbos) { + this.dbos = dbos; + } + + @Override + @Workflow + public String cancelSelf() { + String result; + try { + dbos.cancelWorkflow(DBOSContext.workflowId()); + result = "not refused"; + } catch (IllegalArgumentException e) { + result = "refused: " + e.getMessage(); + } + dbos.runStep(() -> afterRefusal.incrementAndGet(), "afterRefusal"); + return result; + } + + @Override + @Workflow + public String resumeSelf() { + String result; + try { + dbos.resumeWorkflow(DBOSContext.workflowId()); + result = "not refused"; + } catch (IllegalArgumentException e) { + result = "refused: " + e.getMessage(); + } + dbos.runStep(() -> afterRefusal.incrementAndGet(), "afterRefusal"); + return result; + } + + @Override + @Workflow + public String cancelSelfInStep() { + var workflowId = DBOSContext.workflowId(); + return dbos.runStep( + () -> { + try { + dbos.cancelWorkflow(workflowId); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + }, + "cancelFromStep"); + } + + @Override + @Workflow + public String resumeSelfInStep() { + var workflowId = DBOSContext.workflowId(); + return dbos.runStep( + () -> { + try { + dbos.resumeWorkflow(workflowId); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + }, + "resumeFromStep"); + } + + @Override + @Workflow + public String deleteSelf() { + try { + dbos.deleteWorkflow(DBOSContext.workflowId()); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + } + + @Override + @Workflow + public String deleteSelfInStep() { + var workflowId = DBOSContext.workflowId(); + return dbos.runStep( + () -> { + try { + dbos.deleteWorkflow(workflowId); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + }, + "deleteFromStep"); + } + + @Override + @Workflow + public String cancelSelfInUncaughtStep() { + var workflowId = DBOSContext.workflowId(); + try { + dbos.runStep(() -> dbos.cancelWorkflow(workflowId), "uncaughtCancel"); + return "step succeeded"; + } catch (IllegalArgumentException e) { + return "step failed: " + e.getMessage(); + } + } + + @Override + @Workflow + public String cancelsChildTree(String childId) throws Exception { + dbos.startWorkflow(() -> self.blocked(), new StartWorkflowOptions(childId)); + dbos.cancelWorkflow(childId, true); + return "cancelled " + childId; + } + + @Override + @Workflow + public String cancelsOtherAndSelf(String otherId) { + try { + dbos.cancelWorkflows(List.of(otherId, DBOSContext.workflowId())); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + } + + @Override + @Workflow + public String blocked() throws InterruptedException { + blockedStarted.countDown(); + release.await(); + return DBOSContext.workflowId(); + } + + @Override + @Workflow + public String cancelThenResume(String otherId) { + dbos.cancelWorkflow(otherId); + dbos.resumeWorkflow(otherId); + return "done"; + } +} From f3568052dc84a671a74f09c75cb959343f12c9d0 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 10:40:19 -0800 Subject: [PATCH 2/3] Tighten the self-target tests Assert that a refusal inside a step records no cancel step of its own, name the step test for all three operations it covers, inline the one-use launch helper, and release blocked workflows in a finally so a failed assertion doesn't leave them parked until shutdown. --- .../transact/workflow/SelfTargetTest.java | 79 ++++++++++--------- 1 file changed, 43 insertions(+), 36 deletions(-) diff --git a/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java index a32314561..cc5d4ba93 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java @@ -6,7 +6,6 @@ import dev.dbos.transact.DBOS; import dev.dbos.transact.StartWorkflowOptions; -import dev.dbos.transact.config.DBOSConfig; import dev.dbos.transact.context.DBOSContext; import dev.dbos.transact.utils.PgContainer; @@ -36,11 +35,7 @@ public class SelfTargetTest { @BeforeEach void beforeEach() { - launch(pgContainer.dbosConfig()); - } - - private void launch(DBOSConfig config) { - dbos = new DBOS(config); + dbos = new DBOS(pgContainer.dbosConfig()); impl = new SelfTargetServiceImpl(dbos); proxy = dbos.registerProxy(SelfTargetService.class, impl); impl.self = proxy; @@ -86,20 +81,23 @@ public void aRecordedRefusalReplaysRatherThanBeingDecidedAgain() throws Exceptio public void aBulkCancelNamingItsCallerLeavesTheOthersUntouched() throws Exception { var otherId = "bulk-other-%d".formatted(System.nanoTime()); var other = dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(otherId)); - assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); - - var callerId = "bulk-caller-%d".formatted(System.nanoTime()); - var caller = - dbos.startWorkflow( - () -> proxy.cancelsOtherAndSelf(otherId), new StartWorkflowOptions(callerId)); - - assertEquals( - "refused: workflow %s cannot cancel itself".formatted(callerId), caller.getResult()); - assertEquals( - WorkflowState.PENDING, - dbos.retrieveWorkflow(otherId).getStatus().status(), - "the refusal came before any write"); - impl.release.countDown(); + try { + assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); + + var callerId = "bulk-caller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.cancelsOtherAndSelf(otherId), new StartWorkflowOptions(callerId)); + + assertEquals( + "refused: workflow %s cannot cancel itself".formatted(callerId), caller.getResult()); + assertEquals( + WorkflowState.PENDING, + dbos.retrieveWorkflow(otherId).getStatus().status(), + "the refusal came before any write"); + } finally { + impl.release.countDown(); + } assertEquals(otherId, other.getResult()); } @@ -129,7 +127,7 @@ public void aWorkflowCannotDeleteItself() throws Exception { } @Test - public void aStepCannotCancelOrResumeItsOwnWorkflow() throws Exception { + public void aStepCannotCancelResumeOrDeleteItsOwnWorkflow() throws Exception { var cancelId = "step-self-cancel-%d".formatted(System.nanoTime()); var cancel = dbos.startWorkflow(proxy::cancelSelfInStep, new StartWorkflowOptions(cancelId)); assertEquals( @@ -160,6 +158,9 @@ public void aRefusalAStepDoesNotCatchIsThatStepsError() throws Exception { // Nothing of the cancel's own is recorded inside a step; the refusal is the step's result. var step = stepNamed(workflowId, "uncaughtCancel"); assertEquals(IllegalArgumentException.class.getName(), step.error().className()); + assertTrue( + dbos.listWorkflowSteps(workflowId).stream() + .noneMatch(s -> s.functionName().equals("DBOS.cancelWorkflow"))); } @Test @@ -170,27 +171,33 @@ public void aParentCanCancelItsOwnChildsTree() throws Exception { dbos.startWorkflow( () -> proxy.cancelsChildTree(childId), new StartWorkflowOptions(parentId)); - assertEquals("cancelled " + childId, handle.getResult()); - assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); - assertEquals(WorkflowState.CANCELLED, dbos.retrieveWorkflow(childId).getStatus().status()); - impl.release.countDown(); + try { + assertEquals("cancelled " + childId, handle.getResult()); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + assertEquals(WorkflowState.CANCELLED, dbos.retrieveWorkflow(childId).getStatus().status()); + } finally { + impl.release.countDown(); + } } @Test public void aWorkflowCanStillCancelAndResumeAnother() throws Exception { var otherId = "other-%d".formatted(System.nanoTime()); var other = dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(otherId)); - assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); - - var callerId = "canceller-%d".formatted(System.nanoTime()); - var caller = - dbos.startWorkflow( - () -> proxy.cancelThenResume(otherId), new StartWorkflowOptions(callerId)); - assertEquals("done", caller.getResult()); - - assertNull(stepNamed(callerId, "DBOS.cancelWorkflow").error()); - assertNull(stepNamed(callerId, "DBOS.resumeWorkflow").error()); - impl.release.countDown(); + try { + assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); + + var callerId = "canceller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.cancelThenResume(otherId), new StartWorkflowOptions(callerId)); + assertEquals("done", caller.getResult()); + + assertNull(stepNamed(callerId, "DBOS.cancelWorkflow").error()); + assertNull(stepNamed(callerId, "DBOS.resumeWorkflow").error()); + } finally { + impl.release.countDown(); + } assertEquals(otherId, other.getResult()); } } From fc553961ca3ba88ec747ee843e2fda8ee6265dc3 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Thu, 12 Nov 2026 11:02:09 -0800 Subject: [PATCH 3/3] Record DBOS's own step errors in the native format A built-in step recorded what it threw with the configured serializer, so a custom serializer that doesn't keep exception types rebuilt a different type on replay. A workflow that caught a refused self-cancel's IllegalArgumentException on its first run would then fail when recovered or forked, instead of taking the same branch. DBOS's own steps now record their errors natively whatever serializer is configured, as a child start's refusal already is. DBOS.getResult keeps the configured serializer: what it records is the awaited workflow's own error, which belongs to the application. User steps are unchanged, and replay reads the recorded format, so it needs no change. --- .../dbos/transact/execution/DBOSExecutor.java | 18 +++- .../transact/workflow/SelfTargetTest.java | 93 ++++++++++++++++++- 2 files changed, 106 insertions(+), 5 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 5ed84b31b..3ddcb55e8 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1748,7 +1748,7 @@ public T runStep( if (!ctx.isInWorkflow()) { return Objects.requireNonNull(step, "step must not be null").execute(); } - return runStepInternal(step, options, childWorkflowId); + return runStepInternal(step, options, childWorkflowId, false); } public T runDbosFunctionAsStep( @@ -1767,14 +1767,21 @@ public T runDbosFunctionAsStep( throw new IllegalArgumentException("DBOS step name must start with DBOS."); } // DBOS methods are never retried - return runStepInternal(step, new StepOptions(stepName), childWorkflowId); + // DBOS's own steps record what they throw in the native format, whatever serializer is + // configured: a replay must rethrow the exact type the workflow caught, such as a refused + // self-cancel's IllegalArgumentException, which a custom serializer need not preserve. + // DBOS.getResult is the exception: what it throws is the awaited workflow's own error, which + // keeps the configured serializer, as that workflow's outcome does. + boolean recordErrorNatively = !"DBOS.getResult".equals(stepName); + return runStepInternal(step, new StepOptions(stepName), childWorkflowId, recordErrorNatively); } @SuppressWarnings("unchecked") private T runStepInternal( @NonNull ThrowingSupplier step, @NonNull StepOptions options, - @Nullable String childWorkflowId) + @Nullable String childWorkflowId, + boolean recordErrorNatively) throws E { Objects.requireNonNull(step, "step must not be null"); @@ -1877,7 +1884,10 @@ private T runStepInternal( if (exception != null) { logger.debug("executeStep #{} for workflow {} threw {}", stepId, workflowId, exception); - var serializedException = SerializationUtil.serializeError(exception, null, serializer); + var serializedException = + recordErrorNatively + ? SerializationUtil.serializeError(exception, SerializationUtil.NATIVE, null) + : SerializationUtil.serializeError(exception, null, serializer); var stepResult = new StepResult( workflowId, diff --git a/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java index cc5d4ba93..8eb655bdd 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java @@ -6,7 +6,10 @@ import dev.dbos.transact.DBOS; import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.config.DBOSConfig; import dev.dbos.transact.context.DBOSContext; +import dev.dbos.transact.json.DBOSJavaSerializer; +import dev.dbos.transact.json.DBOSSerializer; import dev.dbos.transact.utils.PgContainer; import java.util.List; @@ -35,13 +38,48 @@ public class SelfTargetTest { @BeforeEach void beforeEach() { - dbos = new DBOS(pgContainer.dbosConfig()); + launch(pgContainer.dbosConfig()); + } + + private void launch(DBOSConfig config) { + dbos = new DBOS(config); impl = new SelfTargetServiceImpl(dbos); proxy = dbos.registerProxy(SelfTargetService.class, impl); impl.self = proxy; dbos.launch(); } + /** + * A serializer that keeps values as the native one does but loses an error's type, as a custom + * serializer may: every error reads back as a RuntimeException. + */ + private static final class TypeLosingSerializer implements DBOSSerializer { + @Override + public String name() { + return "type-losing"; + } + + @Override + public String serialize(Object value) { + return DBOSJavaSerializer.INSTANCE.serialize(value); + } + + @Override + public Object deserialize(String text) { + return DBOSJavaSerializer.INSTANCE.deserialize(text); + } + + @Override + public String serializeThrowable(Throwable throwable) { + return throwable.getMessage(); + } + + @Override + public Throwable deserializeThrowable(String text) { + return new RuntimeException(text); + } + } + private StepInfo stepNamed(String workflowId, String name) { return dbos.listWorkflowSteps(workflowId).stream() .filter(step -> step.functionName().equals(name)) @@ -77,6 +115,38 @@ public void aRecordedRefusalReplaysRatherThanBeingDecidedAgain() throws Exceptio assertTrue(refusal.contains(workflowId), refusal); } + @Test + public void aRecordedRefusalReplaysAsItsOwnTypeUnderACustomSerializer() throws Exception { + dbos.close(); + launch(pgContainer.dbosConfig().withSerializer(new TypeLosingSerializer())); + var workflowId = "self-cancel-custom-%d".formatted(System.nanoTime()); + var original = dbos.startWorkflow(proxy::cancelSelf, new StartWorkflowOptions(workflowId)); + var refusal = original.getResult(); + assertEquals("refused: workflow %s cannot cancel itself".formatted(workflowId), refusal); + + // The replay must rethrow the IllegalArgumentException the workflow catches, not the + // serializer's stand-in for it. + var cancelStep = stepNamed(workflowId, "DBOS.cancelWorkflow").functionId(); + var fork = dbos.forkWorkflow(workflowId, cancelStep + 1); + assertEquals(refusal, fork.getResult()); + } + + @Test + public void anAwaitedWorkflowsErrorKeepsTheConfiguredSerializer() throws Exception { + dbos.close(); + launch(pgContainer.dbosConfig().withSerializer(new TypeLosingSerializer())); + var parentId = "awaits-failing-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + var handle = + dbos.startWorkflow( + () -> proxy.awaitsFailingChild(childId), new StartWorkflowOptions(parentId)); + + assertEquals("child failed", handle.getResult()); + // DBOS's own steps record their errors natively; getResult records the child's error, which + // is the application's, with the serializer the application configured. + assertEquals("type-losing", stepNamed(parentId, "DBOS.getResult").serialization()); + } + @Test public void aBulkCancelNamingItsCallerLeavesTheOthersUntouched() throws Exception { var otherId = "bulk-other-%d".formatted(System.nanoTime()); @@ -223,6 +293,10 @@ interface SelfTargetService { String blocked() throws InterruptedException; + String failingChild(); + + String awaitsFailingChild(String childId) throws Exception; + String cancelThenResume(String otherId); } @@ -355,6 +429,23 @@ public String cancelsOtherAndSelf(String otherId) { } } + @Override + @Workflow + public String failingChild() { + throw new IllegalStateException("child failed"); + } + + @Override + @Workflow + public String awaitsFailingChild(String childId) throws Exception { + var child = dbos.startWorkflow(() -> self.failingChild(), new StartWorkflowOptions(childId)); + try { + return child.getResult(); + } catch (RuntimeException e) { + return e.getMessage(); + } + } + @Override @Workflow public String blocked() throws InterruptedException {