Skip to content

Refuse a workflow cancelling, resuming or deleting itself #661

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
devhawk merged 3 commits into from
Nov 12, 2026
Merged

Refuse a workflow cancelling, resuming or deleting itself #661

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8be290e
Refuse a workflow cancelling, resuming or deleting itself
devhawk Nov 12, 2026
f356805
Tighten the self-target tests
devhawk Nov 12, 2026
fc55396
Record DBOS's own step errors in the native format
devhawk Nov 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions transact/src/main/java/dev/dbos/transact/DBOS.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -872,6 +872,8 @@ public <E extends Exception> 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 <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
Expand All @@ -892,6 +894,8 @@ public <E extends Exception> 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 <T, E extends Exception> @NonNull WorkflowHandle<T, E> resumeWorkflow(
@NonNull String workflowId) {
Expand All @@ -909,6 +913,8 @@ public <E extends Exception> 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<WorkflowHandle<Object, Exception>> resumeWorkflows(
@NonNull List<String> workflowIds, @Nullable String queueName) {
Expand All @@ -924,6 +930,8 @@ public <E extends Exception> 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<WorkflowHandle<Object, Exception>> resumeWorkflows(
@NonNull List<String> workflowIds) {
Expand All @@ -937,6 +945,8 @@ public <E extends Exception> 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);
Expand All @@ -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);
Expand All @@ -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<String> workflowIds) {
cancelWorkflows(workflowIds, false);
Expand All @@ -970,6 +984,8 @@ public void cancelWorkflows(@NonNull List<String> 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<String> workflowIds, boolean cancelChildren) {
ensureLaunched("cancelWorkflows").cancelWorkflows(workflowIds, cancelChildren);
Expand All @@ -981,6 +997,8 @@ public void cancelWorkflows(@NonNull List<String> 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);
Expand All @@ -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);
Expand All @@ -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<String> workflowIds) {
deleteWorkflows(workflowIds, false);
Expand All @@ -1018,6 +1040,8 @@ public void deleteWorkflows(@NonNull List<String> 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<String> workflowIds, boolean deleteChildren) {
ensureLaunched("deleteWorkflows").deleteWorkflows(workflowIds, deleteChildren);
Expand Down
44 changes: 40 additions & 4 deletions transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1083,9 +1083,11 @@ public void sleep(Duration duration) {

public void cancelWorkflows(List<String> workflowIds, boolean cancelChildren) {
Objects.requireNonNull(workflowIds);
var callerWorkflowId = DBOSContextHolder.get().getWorkflowId();

this.runDbosFunctionAsStep(
() -> {
refuseSelfTarget(callerWorkflowId, workflowIds, "cancel");
logger.info(
"Cancelling workflow(s) {}{}",
workflowIds,
Expand All @@ -1099,8 +1101,10 @@ public void cancelWorkflows(List<String> workflowIds, boolean cancelChildren) {

public void deleteWorkflows(List<String> workflowIds, boolean deleteChildren) {
Objects.requireNonNull(workflowIds);
var callerWorkflowId = DBOSContextHolder.get().getWorkflowId();
this.runDbosFunctionAsStep(
() -> {
refuseSelfTarget(callerWorkflowId, workflowIds, "delete");
logger.info(
"Deleting workflow(s) {}{}",
workflowIds,
Expand All @@ -1115,10 +1119,12 @@ public void deleteWorkflows(List<String> workflowIds, boolean deleteChildren) {
public List<WorkflowHandle<Object, Exception>> resumeWorkflows(
List<String> 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
Expand All @@ -1129,6 +1135,26 @@ public List<WorkflowHandle<Object, Exception>> 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.
*
* <p>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<String> targets, String action) {
if (callerWorkflowId != null && targets.contains(callerWorkflowId)) {
throw new IllegalArgumentException(
"workflow %s cannot %s itself".formatted(callerWorkflowId, action));
Comment thread
devhawk marked this conversation as resolved.
Resolved
}
}

public void updateWorkflowAttributes(String workflowId, Map<String, Object> attributes) {
Objects.requireNonNull(workflowId);
this.runDbosFunctionAsStep(
Expand Down Expand Up @@ -1722,7 +1748,7 @@ public <T, E extends Exception> 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, E extends Exception> T runDbosFunctionAsStep(
Expand All @@ -1741,14 +1767,21 @@ public <T, E extends Exception> 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, E extends Exception> T runStepInternal(
@NonNull ThrowingSupplier<T, E> step,
@NonNull StepOptions options,
@Nullable String childWorkflowId)
@Nullable String childWorkflowId,
boolean recordErrorNatively)
throws E {

Objects.requireNonNull(step, "step must not be null");
Expand Down Expand Up @@ -1851,7 +1884,10 @@ private <T, E extends Exception> 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,
Expand Down
71 changes: 0 additions & 71 deletions transact/src/test/java/dev/dbos/transact/workflow/OwnershipFenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 {
Expand All @@ -328,12 +288,6 @@ interface FenceService {
String child();

String quick();

String cancelSelf();

String deleteSelf();

String resumeSelf();
}

class FenceServiceImpl implements FenceService {
Expand All @@ -348,7 +302,6 @@ class FenceServiceImpl implements FenceService {
final Map<String, CountDownLatch> 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;
Expand Down Expand Up @@ -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";
}
}
Loading
Loading