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..3ddcb55e8 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( @@ -1722,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( @@ -1741,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"); @@ -1851,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/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..8eb655bdd --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java @@ -0,0 +1,464 @@ +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.json.DBOSJavaSerializer; +import dev.dbos.transact.json.DBOSSerializer; +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(); + } + + /** + * 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)) + .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 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()); + var other = dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(otherId)); + 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()); + } + + @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 aStepCannotCancelResumeOrDeleteItsOwnWorkflow() 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()); + assertTrue( + dbos.listWorkflowSteps(workflowId).stream() + .noneMatch(s -> s.functionName().equals("DBOS.cancelWorkflow"))); + } + + @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)); + + 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)); + 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()); + } +} + +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 failingChild(); + + String awaitsFailingChild(String childId) throws Exception; + + 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 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 { + blockedStarted.countDown(); + release.await(); + return DBOSContext.workflowId(); + } + + @Override + @Workflow + public String cancelThenResume(String otherId) { + dbos.cancelWorkflow(otherId); + dbos.resumeWorkflow(otherId); + return "done"; + } +}