diff --git a/transact/src/main/java/dev/dbos/transact/DBOS.java b/transact/src/main/java/dev/dbos/transact/DBOS.java index a9b607154..bc3a83eb8 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,9 @@ 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, or, when {@code cancelChildren}, from inside one of its descendants: a workflow + * can't cancel itself or a tree it belongs to. Nothing is cancelled. */ public void cancelWorkflow(@NonNull String workflowId, boolean cancelChildren) { cancelWorkflows(List.of(workflowId), cancelChildren); @@ -959,6 +972,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 +985,9 @@ 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, or, when {@code cancelChildren}, from inside one of their descendants: a + * workflow can't cancel itself or a tree it belongs to. Nothing is cancelled. */ public void cancelWorkflows(@NonNull List workflowIds, boolean cancelChildren) { ensureLaunched("cancelWorkflows").cancelWorkflows(workflowIds, cancelChildren); @@ -981,6 +999,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 +1014,9 @@ 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, or, when {@code deleteChildren}, from inside one of its descendants: a workflow + * can't delete itself or a tree it belongs to. Nothing is deleted. */ public void deleteWorkflow(@NonNull String workflowId, boolean deleteChildren) { deleteWorkflows(List.of(workflowId), deleteChildren); @@ -1005,6 +1028,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 +1043,9 @@ 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, or, when {@code deleteChildren}, from inside one of their descendants: a + * workflow can't delete itself or a tree it belongs to. 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/DBOSClient.java b/transact/src/main/java/dev/dbos/transact/DBOSClient.java index 199115fda..4abd4f50b 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOSClient.java +++ b/transact/src/main/java/dev/dbos/transact/DBOSClient.java @@ -1597,7 +1597,7 @@ public void cancelWorkflows(@NonNull List workflowIds) { * @param cancelChildren if true, recursively cancel all descendant workflows */ public void cancelWorkflows(@NonNull List workflowIds, boolean cancelChildren) { - systemDatabase.cancelWorkflows(workflowIds, cancelChildren); + systemDatabase.cancelWorkflows(workflowIds, cancelChildren, null); } /** @@ -1627,7 +1627,7 @@ public void updateWorkflowAttributes( */ public @NonNull WorkflowHandle resumeWorkflow( @NonNull String workflowId, @Nullable String queueName) { - systemDatabase.resumeWorkflows(List.of(workflowId), queueName); + systemDatabase.resumeWorkflows(List.of(workflowId), queueName, null); return retrieveWorkflow(workflowId); } @@ -1674,7 +1674,7 @@ public void updateWorkflowAttributes( */ public @NonNull List> resumeWorkflows( @NonNull List workflowIds, @Nullable String queueName) { - systemDatabase.resumeWorkflows(workflowIds, queueName); + systemDatabase.resumeWorkflows(workflowIds, queueName, null); return workflowIds.stream().map(this::retrieveWorkflow).toList(); } @@ -1719,7 +1719,7 @@ public void deleteWorkflows(@NonNull List workflowIds) { * workflows */ public void deleteWorkflows(@NonNull List workflowIds, boolean deleteChildren) { - systemDatabase.deleteWorkflows(workflowIds, deleteChildren); + systemDatabase.deleteWorkflows(workflowIds, deleteChildren, null); } /** diff --git a/transact/src/main/java/dev/dbos/transact/database/SendCaller.java b/transact/src/main/java/dev/dbos/transact/database/SendCaller.java deleted file mode 100644 index 6393d7785..000000000 --- a/transact/src/main/java/dev/dbos/transact/database/SendCaller.java +++ /dev/null @@ -1,22 +0,0 @@ -package dev.dbos.transact.database; - -/** - * The workflow execution a send is made from. The send lands only while that execution still owns - * the workflow. - */ -public sealed interface SendCaller { - /** The sending workflow. */ - String workflowId(); - - /** The token of the sending workflow's execution. */ - String ownerXid(); - - /** A send from the workflow body, recorded as the workflow's step {@code stepId}. */ - record FromWorkflow(String workflowId, int stepId, String ownerXid) implements SendCaller {} - - /** - * A send from inside a step, which records nothing of its own but is still checked against the - * owner. - */ - record FromStep(String workflowId, String ownerXid) implements SendCaller {} -} diff --git a/transact/src/main/java/dev/dbos/transact/database/StepCaller.java b/transact/src/main/java/dev/dbos/transact/database/StepCaller.java new file mode 100644 index 000000000..7326e1ea6 --- /dev/null +++ b/transact/src/main/java/dev/dbos/transact/database/StepCaller.java @@ -0,0 +1,22 @@ +package dev.dbos.transact.database; + +/** + * The workflow execution a DBOS operation, such as a send or a cancel, is made from. The operation + * lands only while that execution still owns the workflow. + */ +public sealed interface StepCaller { + /** The calling workflow. */ + String workflowId(); + + /** The token of the calling workflow's execution. */ + String ownerXid(); + + /** A call from the workflow body, recorded as the workflow's step {@code stepId}. */ + record FromWorkflow(String workflowId, int stepId, String ownerXid) implements StepCaller {} + + /** + * A call from inside a step, which records nothing of its own but is still checked against the + * owner. + */ + record FromStep(String workflowId, String ownerXid) implements StepCaller {} +} 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 2abc4bbf8..2b22a7f7a 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -18,6 +18,7 @@ import dev.dbos.transact.database.signal.SignalMap; import dev.dbos.transact.database.signal.Subscription; import dev.dbos.transact.exceptions.*; +import dev.dbos.transact.internal.DebugTriggers; import dev.dbos.transact.internal.Validation; import dev.dbos.transact.json.DBOSSerializer; import dev.dbos.transact.json.SerializationUtil; @@ -47,6 +48,7 @@ import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Predicate; @@ -1575,7 +1577,7 @@ public Optional checkChildWorkflow(String workflowUuid, int function */ public void sendBulk( List messages, - @Nullable SendCaller caller, + @Nullable StepCaller caller, String functionName, boolean sendToForks, String serialization) { @@ -1641,8 +1643,28 @@ public void sleep(String workflowId, int functionId, Duration duration, String o "sleep", () -> StepsDAO.sleep(ctx, workflowId, functionId, duration, ownerXid)); } - public void cancelWorkflows(List workflowIds, boolean cancelChildren) { - dbRetry(() -> WorkflowDAO.cancelWorkflows(ctx, workflowIds, cancelChildren)); + /** + * Cancels the workflows, and their descendants when {@code cancelChildren}, as the step {@code + * caller} makes; see {@link #runTransactionalStep}. + */ + public void cancelWorkflows( + List workflowIds, boolean cancelChildren, @Nullable StepCaller caller) { + var ran = new AtomicBoolean(); + runTransactionalStep( + caller, + "DBOS.cancelWorkflow", + conn -> { + WorkflowDAO.cancelWorkflows( + conn, ctx.schema(), workflowIds, cancelChildren, callerWorkflowId(caller)); + ran.set(true); + return null; + }); + // Once, after the commit, and only when this call ran the work: a call that only replays a + // checkpoint logs nothing. + if (ran.get()) { + logger.info( + "Cancelled workflow(s) {}{}", workflowIds, cancelChildren ? " and their children" : ""); + } } /** @@ -1654,8 +1676,22 @@ public boolean cancelOwnedWorkflow(String workflowId, String ownerXid) { return dbRetry(() -> WorkflowDAO.cancelOwnedWorkflow(ctx, workflowId, ownerXid)); } - public void resumeWorkflows(List workflowIds, String queueName) { - dbRetry(() -> WorkflowDAO.resumeWorkflows(ctx, workflowIds, queueName)); + /** Resumes the workflows as the step {@code caller} makes; see {@link #runTransactionalStep}. */ + public void resumeWorkflows( + List workflowIds, String queueName, @Nullable StepCaller caller) { + var ran = new AtomicBoolean(); + runTransactionalStep( + caller, + "DBOS.resumeWorkflow", + conn -> { + WorkflowDAO.resumeWorkflows( + conn, ctx.schema(), workflowIds, queueName, callerWorkflowId(caller)); + ran.set(true); + return null; + }); + if (ran.get()) { + logger.info("Resumed workflow(s) {}", workflowIds); + } } public void updateWorkflowAttributes(String workflowId, Map attributes) { @@ -1663,9 +1699,208 @@ public void updateWorkflowAttributes(String workflowId, Map attr dbRetry(() -> WorkflowDAO.updateWorkflowAttributes(ctx, workflowId, validated)); } - public void deleteWorkflows(List workflowIds, boolean deleteChildren) { - dbRetryIncludingSerializationError( - "deleteWorkflows", () -> WorkflowDAO.deleteWorkflows(ctx, workflowIds, deleteChildren)); + /** + * Deletes the workflows, and their descendants when {@code deleteChildren}, as the step {@code + * caller} makes; see {@link #runTransactionalStep}. + */ + public void deleteWorkflows( + List workflowIds, boolean deleteChildren, @Nullable StepCaller caller) { + var ran = new AtomicBoolean(); + runTransactionalStep( + caller, + "DBOS.deleteWorkflow", + conn -> { + WorkflowDAO.deleteWorkflows( + conn, ctx.schema(), workflowIds, deleteChildren, callerWorkflowId(caller)); + ran.set(true); + return null; + }); + if (ran.get()) { + logger.info( + "Deleted workflow(s) {}{}", workflowIds, deleteChildren ? " and their children" : ""); + } + } + + /** The workflow a call is made from, from its body or a step, or null outside any workflow. */ + private static @Nullable String callerWorkflowId(@Nullable StepCaller caller) { + return caller == null ? null : caller.workflowId(); + } + + /** + * Runs {@code work} as a DBOS operation made by {@code caller}, committing its effect and its + * checkpoint together, so a crash between the two can never make a replay run it a second time. + * + *
    + *
  • From the workflow body ({@link StepCaller.FromWorkflow}), one transaction checks whether + * the step is already recorded (and that the caller still owns the workflow), replays the + * recorded output or error if it is, and otherwise runs {@code work} and records its output + * as the step. If {@code work} fails, the transaction rolls back and the failure is + * recorded as the step's error in a fresh one, so a replay is refused the same way. + *
  • From inside a step ({@link StepCaller.FromStep}), nothing is recorded, but {@code work} + * commits only while the caller still owns the workflow. + *
  • Outside a workflow ({@code null}), {@code work} runs in a plain transaction. + *
+ * + *

The whole transaction is retried as {@link #dbRetryIncludingSerializationError} does. From + * the workflow body, a retry after a commit whose acknowledgement was lost finds its own + * checkpoint and replays it; from a step or outside a workflow there is none, so {@code work} + * runs again. + * + *

Whatever {@code work} throws once the retries end is recorded, so a workflow that catches it + * takes the same branch on replay. So is a conflict that outlasted the retries, wherever the + * database raised it: in {@code work}, in the step check, or in the record and owner check after + * it. Which statement a deadlock picks as its victim is the database's choice, and the + * transaction is rolled back whichever it was. + * + *

No other failure outside {@code work} is recorded, and a workflow that catches one may take + * another branch on replay: a failure of the step check or of the record that is not a conflict, + * a lost ownership, or a checkpoint that disagrees with this call. Nor are a failed commit, whose + * outcome is unknown and which a replay settles by finding the checkpoint or not, and a failure + * an interrupt cut short, which stops the execution rather than answering the step. + */ + @Nullable T runTransactionalStep( + @Nullable StepCaller caller, String stepName, SqlTransaction.SqlCall work) { + var step = caller instanceof StepCaller.FromWorkflow fromWorkflow ? fromWorkflow : null; + long startTime = System.currentTimeMillis(); + // What work itself threw on the attempt that ended the retries, as against a failure of the + // check or the record around it, which is not the step's answer. + var workFailure = new AtomicReference(); + StepResult replayed; + T output; + try { + record Outcome(@Nullable StepResult replayed, @Nullable T output) {} + var outcome = + dbRetryIncludingSerializationError( + stepName, + () -> { + workFailure.set(null); + Outcome attempt; + try (var conn = ctx.getConnection()) { + // Set rather than inherited, since a pool can hand out another level. A tree walk + // relies on it: each level's children read must see a child committed while the + // level's write waited, which a REPEATABLE READ snapshot would hide. + conn.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED); + attempt = + SqlTransaction.call( + conn, + c -> { + if (step != null) { + var recorded = + StepsDAO.checkStepResult( + c, + ctx.schema(), + step.workflowId(), + step.stepId(), + stepName, + step.ownerXid()); + if (recorded != null) { + return new Outcome(recorded, null); + } + } + T result; + try { + result = work.run(c); + } catch (SQLException | RuntimeException e) { + workFailure.set(e); + throw e; + } + // The owner is checked last, as every fenced write locks the status + // row. + if (step != null) { + var serialized = + SerializationUtil.serializeValue(result, null, ctx.serializer()); + StepsDAO.recordStepResult( + ctx, + c, + new StepResult( + step.workflowId(), + step.stepId(), + stepName, + serialized.serializedValue(), + null, + null, + serialized.serialization()), + startTime, + System.currentTimeMillis(), + step.ownerXid()); + } else if (caller != null) { + WorkflowDAO.checkOwner( + c, ctx.schema(), caller.workflowId(), caller.ownerXid()); + } + DebugTriggers.debugTriggerPoint( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT); + return new Outcome(null, result); + }); + } + DebugTriggers.debugTriggerPoint( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED); + return attempt; + }); + replayed = outcome.replayed(); + output = outcome.output(); + } catch (RuntimeException e) { + // The work's own failure is its answer: the exception that escaped must be it, or the + // retries' wrapping of it, not something that ended the retries after it was caught. A + // conflict that outlasted the retries is the answer too, whichever statement raised it, + // since the database rolled the transaction back. + var failure = workFailure.get(); + boolean fromWork = failure != null && (e == failure || e.getCause() == failure); + if (step != null + && (fromWork || isSerializationError(e)) + && !Thread.currentThread().isInterrupted()) { + recordStepError(step, stepName, e, startTime); + } + throw e; + } + + if (replayed == null) { + return output; + } + if (replayed.error() == null && replayed.output() == null) { + throw new IllegalStateException( + "Recorded output and error are both null for workflow %s step %d (%s)" + .formatted(step.workflowId(), step.stepId(), stepName)); + } + if (replayed.error() != null) { + var t = + SerializationUtil.deserializeError( + replayed.error(), replayed.serialization(), ctx.serializer()); + throw t instanceof RuntimeException re ? re : new RuntimeException(t.getMessage(), t); + } + @SuppressWarnings("unchecked") + var value = + (T) + SerializationUtil.deserializeValue( + replayed.output(), replayed.serialization(), ctx.serializer()); + return value; + } + + /** + * Records {@code error} as the step {@code step} after its work's transaction rolled back. A + * failure to record it, such as the workflow having changed hands, is thrown in its place, with + * {@code error} attached. + */ + private void recordStepError( + StepCaller.FromWorkflow step, String stepName, RuntimeException error, long startTime) { + // In the native format whatever serializer is configured: a replay must rethrow the exact + // type the workflow caught, which a custom serializer need not preserve. + var serialized = SerializationUtil.serializeError(error, SerializationUtil.NATIVE, null); + try { + recordStepResult( + new StepResult( + step.workflowId(), + step.stepId(), + stepName, + null, + serialized.serializedValue(), + null, + serialized.serialization()), + startTime, + step.ownerXid()); + } catch (RuntimeException recordFailure) { + recordFailure.addSuppressed(error); + throw recordFailure; + } } public String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions options) { @@ -1851,7 +2086,12 @@ public boolean deprecatePatch(String workflowId, int functionId, String patchNam } public Set getWorkflowChildren(String workflowId) { - return dbRetry(() -> WorkflowDAO.getWorkflowChildren(ctx, workflowId)); + return dbRetry( + () -> { + try (var conn = ctx.getConnection()) { + return WorkflowDAO.getWorkflowChildren(conn, ctx.schema(), workflowId); + } + }); } /** @@ -1872,7 +2112,9 @@ public String schema() { @Override public Set workflowChildren(String workflowId) throws SQLException { - return WorkflowDAO.getWorkflowChildren(ctx, workflowId); + try (var conn = ctx.getConnection()) { + return WorkflowDAO.getWorkflowChildren(conn, ctx.schema(), workflowId); + } } @Override diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java index bbfddd828..e2fdf2468 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java @@ -5,8 +5,8 @@ import dev.dbos.transact.Constants; import dev.dbos.transact.database.DbContext; import dev.dbos.transact.database.GetEventCaller; -import dev.dbos.transact.database.SendCaller; import dev.dbos.transact.database.SqlTransaction; +import dev.dbos.transact.database.StepCaller; import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.database.signal.SignalKey; import dev.dbos.transact.database.signal.SignalMap; @@ -98,7 +98,7 @@ WHERE forked_from IN (%s) public static void sendBulk( DbContext ctx, List messages, - @Nullable SendCaller caller, + @Nullable StepCaller caller, String functionName, boolean sendToForks, String serialization) @@ -130,7 +130,7 @@ record SerializedPair(SendMessage msg, SerializationUtil.SerializedResult serial txConn, conn -> { // A send from the workflow body replays as the step it was recorded as. - if (caller instanceof SendCaller.FromWorkflow step) { + if (caller instanceof StepCaller.FromWorkflow step) { StepResult recorded = StepsDAO.checkStepResult( conn, @@ -216,7 +216,7 @@ ON CONFLICT (message_uuid) DO NOTHING // The owner is checked after the messages, so the status row is locked last, as every // fenced write locks it. Recording the step checks it first; a send from inside a step // has no step to record, so it is checked on its own. - if (caller instanceof SendCaller.FromWorkflow step) { + if (caller instanceof StepCaller.FromWorkflow step) { var output = new StepResult( step.workflowId(), step.stepId(), functionName, null, null, null, null); 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 c031d36dd..a942701bb 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 @@ -234,6 +234,7 @@ public static WorkflowInitResult initWorkflowStatus( null) .withChildWorkflowId(initStatus.workflowId()); StepsDAO.recordStepResult(ctx, conn, childStep, null, null, parent.ownerXid()); + DebugTriggers.debugTriggerPoint(DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED); } // The insert's conflict arm changes nothing, so committing a row this call found is @@ -798,8 +799,10 @@ public static void checkWorkflow( * ends, so no hand-off can land between this check and the write it guards: a cancel, resume or * claim waits for that write to commit. The caller must be inside a transaction, or the lock * would end with this statement. A write that changes other rows in the same transaction makes - * them first and checks last, so every fenced write takes the status row last and none can - * deadlock with another. + * them first and checks last. That does not rule out a deadlock: a cancel or delete of a tree + * holds each target's row as it walks, so two workflows cancelling each other, or a delete racing + * a fenced write from one of the tree's descendants, can each wait on a row the other holds. The + * database aborts one with 40P01, and its retry runs as if the other had committed first. */ public static void checkOwner(Connection conn, String schema, String workflowId, String ownerXid) throws SQLException { @@ -1832,26 +1835,78 @@ private static Collection filterNullsAndBlanks(Collection workfl return workflowIds.stream().filter(id -> id != null && !id.isBlank()).toList(); } + /** + * Cancels the workflows and, when {@code cancelChildren}, their descendants, on the caller's + * transaction. + * + *

Each level is cancelled before its children are read. A child whose insert holds its + * parent's row lock commits before that level's update goes ahead, and is then read with the next + * level; one inserted after it finds its parent no longer owned and is rolled back. Reading the + * tree first would miss a child committed in between. Each read sees what committed before it + * only under READ COMMITTED, so the transaction must run at that level. + * + * @param callerWorkflowId the workflow making the call, when it is made from inside one, which + * may not cancel itself or a tree it belongs to + * @throws IllegalArgumentException if the caller is a target, before anything is cancelled, or is + * reached among a target's descendants, after which the caller's transaction must roll back + * the levels already cancelled + */ public static void cancelWorkflows( - DbContext ctx, List workflowIds, boolean cancelChildren) throws SQLException { + Connection conn, + String schema, + List workflowIds, + boolean cancelChildren, + @Nullable String callerWorkflowId) + throws SQLException { var roots = filterNullsAndBlanks(workflowIds); if (roots.isEmpty()) { return; } + refuseSelfTarget(callerWorkflowId, roots, "cancel"); if (!cancelChildren) { - cancelBatch(ctx, roots); + applyLevel(conn, cancelStatements(schema), roots, null); return; } - // Cancel level-by-level so newly-spawned children are also caught - var visited = new HashSet<>(roots); + walkTree(conn, schema, roots, callerWorkflowId, "cancel", cancelStatements(schema)); + } + + /** + * Runs {@code levelStatements} on the roots and then on each level of their descendants in turn, + * each level before its children are read, on the caller's transaction. A level's statements and + * the read of its children go to the database in one round trip; see {@link #applyLevel}. + * + *

A walk that reaches the calling workflow is an {@code action} on a tree the caller is + * inside, and is refused there: the levels above have been acted on by now, and the refusal rolls + * them back with the transaction. Each workflow's root is carried down the walk, so the refusal + * names the tree without reading rows the walk may already have changed. + */ + private static void walkTree( + Connection conn, + String schema, + Collection roots, + @Nullable String callerWorkflowId, + String action, + List levelStatements) + throws SQLException { + var rootOf = new HashMap(); + roots.forEach(root -> rootOf.put(root, root)); List frontier = new ArrayList<>(roots); while (!frontier.isEmpty()) { - cancelBatch(ctx, frontier); - var children = getDirectChildren(ctx, frontier); - frontier = children.stream().filter(c -> !visited.contains(c)).toList(); - visited.addAll(frontier); + if (callerWorkflowId != null && frontier.contains(callerWorkflowId)) { + throw new IllegalArgumentException( + "workflow %s cannot %s the tree of %s, which contains it" + .formatted(callerWorkflowId, action, rootOf.get(callerWorkflowId))); + } + var children = applyLevel(conn, levelStatements, frontier, childrenSql(schema)); + var next = new ArrayList(); + for (var child : children.entrySet()) { + if (rootOf.putIfAbsent(child.getKey(), rootOf.get(child.getValue())) == null) { + next.add(child.getKey()); + } + } + frontier = next; } } @@ -1888,12 +1943,12 @@ AND status NOT IN (?, ?) } } - private static void cancelBatch(DbContext ctx, Collection workflowIds) - throws SQLException { - String sql = + /** What a cancel does to a level of workflows, given as {@code ANY(?)}. */ + private static List cancelStatements(String schema) { + return List.of( """ UPDATE "%1$s".workflow_status - SET status = ?, + SET status = '%3$s', queue_name = NULL, deduplication_id = NULL, started_at_epoch_ms = NULL, @@ -1901,31 +1956,106 @@ private static void cancelBatch(DbContext ctx, Collection workflowIds) updated_at = %2$s, completed_at = %2$s WHERE workflow_uuid = ANY(?) - AND status NOT IN (?, ?) + AND status NOT IN ('%4$s', '%5$s') """ - .formatted(ctx.schema(), SystemDatabase.NOW_EPOCH_MS); + .formatted( + schema, + SystemDatabase.NOW_EPOCH_MS, + WorkflowState.CANCELLED.name(), + WorkflowState.SUCCESS.name(), + WorkflowState.ERROR.name())); + } - try (Connection conn = ctx.getConnection(); - PreparedStatement stmt = conn.prepareStatement(sql)) { - Array array = conn.createArrayOf("text", workflowIds.toArray(String[]::new)); + /** + * What a delete does to a level of workflows, given as {@code ANY(?)}: every row they have. + * + *

workflow_input and workflow_output (migration 109) have no foreign key, so a status delete + * was never going to take them. Migration 112 then drops the operation_outputs cascade, leaving + * all three the same: nothing follows a status row out on its own, so each is cleared by ID. + */ + private static List deleteStatements(String schema) { + return List.of("workflow_status", "operation_outputs", "workflow_input", "workflow_output") + .stream() + .map(table -> "DELETE FROM \"%s\".%s WHERE workflow_uuid = ANY(?)".formatted(schema, table)) + .toList(); + } + + /** Reads a level's direct children, given as {@code ANY(?)}, each with its parent. */ + private static String childrenSql(String schema) { + return """ + SELECT workflow_uuid, parent_workflow_id + FROM "%s".workflow_status + WHERE parent_workflow_id = ANY(?) + """ + .formatted(schema); + } + + /** + * Runs {@code statements}, each taking {@code level} as its only parameter, and then {@code + * readChildren} if given, in one round trip on the caller's transaction, and returns the children + * it read, each mapped to its parent. + * + *

One prepared statement carries them all, and the driver sends them as separate statements in + * a single flush. The server still runs them one after another, each on a snapshot of its own + * under READ COMMITTED, so the read sees a child whose insert held a parent's lock until the + * writes before it could proceed. One statement would not: a single snapshot, taken before any + * wait, would hide that child. + */ + private static Map applyLevel( + Connection conn, + List statements, + Collection level, + @Nullable String readChildren) + throws SQLException { + var sql = new ArrayList<>(statements); + if (readChildren != null) { + sql.add(readChildren); + } + var children = new HashMap(); + try (var stmt = conn.prepareStatement(String.join(";\n", sql))) { + var array = conn.createArrayOf("text", level.toArray(String[]::new)); try { - stmt.setString(1, WorkflowState.CANCELLED.name()); - stmt.setArray(2, array); - stmt.setString(3, WorkflowState.SUCCESS.name()); - stmt.setString(4, WorkflowState.ERROR.name()); - stmt.executeUpdate(); + for (int i = 1; i <= sql.size(); i++) { + stmt.setArray(i, array); + } + boolean isResultSet = stmt.execute(); + while (isResultSet || stmt.getUpdateCount() != -1) { + if (isResultSet) { + try (var rs = stmt.getResultSet()) { + while (rs.next()) { + children.put(rs.getString(1), rs.getString(2)); + } + } + } + isResultSet = stmt.getMoreResults(); + } } finally { array.free(); } } + return children; } - public static void resumeWorkflows(DbContext ctx, List workflowIds, String queueName) + /** + * Resumes the workflows onto {@code queueName}, or the internal queue, on the caller's + * transaction. + * + * @param callerWorkflowId the workflow making the call, when it is made from inside one, which + * may not resume itself + * @throws IllegalArgumentException if the caller is a target, before anything is resumed + */ + public static void resumeWorkflows( + Connection conn, + String schema, + List workflowIds, + String queueName, + @Nullable String callerWorkflowId) throws SQLException { var filtered = filterNullsAndBlanks(workflowIds); if (filtered.isEmpty()) { return; } + refuseSelfTarget(callerWorkflowId, filtered, "resume"); String sql = """ @@ -1942,10 +2072,9 @@ public static void resumeWorkflows(DbContext ctx, List workflowIds, Stri WHERE workflow_uuid = ANY(?) AND status NOT IN (?, ?) """ - .formatted(ctx.schema(), SystemDatabase.NOW_EPOCH_MS); + .formatted(schema, SystemDatabase.NOW_EPOCH_MS); - try (Connection conn = ctx.getConnection(); - PreparedStatement stmt = conn.prepareStatement(sql)) { + try (PreparedStatement stmt = conn.prepareStatement(sql)) { Array array = conn.createArrayOf("text", filtered.toArray(String[]::new)); try { stmt.setString(1, WorkflowState.ENQUEUED.name()); @@ -1960,69 +2089,53 @@ AND status NOT IN (?, ?) } } + /** + * Deletes the workflows and, when {@code deleteChildren}, their descendants, on the caller's + * transaction. + * + * @param callerWorkflowId the workflow making the call, when it is made from inside one, which + * may not delete itself or a tree it belongs to + * @throws IllegalArgumentException if the caller is a target, before anything is deleted, or is + * reached among a target's descendants, after which the caller's transaction must roll back + * the levels already deleted + */ public static void deleteWorkflows( - DbContext ctx, List workflowIds, boolean deleteChildren) throws SQLException { + Connection conn, + String schema, + List workflowIds, + boolean deleteChildren, + @Nullable String callerWorkflowId) + throws SQLException { var filtered = filterNullsAndBlanks(workflowIds); if (filtered.isEmpty()) { return; } + refuseSelfTarget(callerWorkflowId, filtered, "delete"); - var wfIdSet = new HashSet(filtered); - if (deleteChildren) { - for (var wfid : filtered) { - var children = getWorkflowChildren(ctx, wfid); - wfIdSet.addAll(children); - } - } - - var sql = - """ - DELETE FROM "%s".workflow_status - WHERE workflow_uuid = ANY(?); - """ - .formatted(ctx.schema()); - - var ids = wfIdSet.toArray(String[]::new); - try (var conn = ctx.getConnection()) { - SqlTransaction.run( - conn, - c -> { - try (var stmt = c.prepareStatement(sql)) { - var array = c.createArrayOf("text", ids); - try { - stmt.setArray(1, array); - stmt.executeUpdate(); - } finally { - array.free(); - } - } - deleteWorkflowChildRows(c, ctx.schema(), ids); - }); + if (!deleteChildren) { + applyLevel(conn, deleteStatements(schema), filtered, null); + return; } + // Each level is deleted before its children are read, as a cancel's is, so a child whose + // insert holds its parent's lock is read once it commits, and one inserted later finds its + // parent gone. The children are still found by their own parent_workflow_id. + walkTree(conn, schema, filtered, callerWorkflowId, "delete", deleteStatements(schema)); } - private static List getDirectChildren(DbContext ctx, Collection workflowIds) - throws SQLException { + /** The direct children of the given workflows, each mapped to its parent. */ + private static Map getDirectChildren( + Connection conn, String schema, Collection workflowIds) throws SQLException { if (workflowIds.isEmpty()) { - return List.of(); + return Map.of(); } - var sql = - """ - SELECT workflow_uuid - FROM "%s".workflow_status - WHERE parent_workflow_id = ANY(?) - """ - .formatted(ctx.schema()); - - try (var conn = ctx.getConnection(); - var stmt = conn.prepareStatement(sql)) { + try (var stmt = conn.prepareStatement(childrenSql(schema))) { var array = conn.createArrayOf("text", workflowIds.toArray(String[]::new)); try { stmt.setArray(1, array); try (var rs = stmt.executeQuery()) { - var result = new ArrayList(); + var result = new HashMap(); while (rs.next()) { - result.add(rs.getString(1)); + result.put(rs.getString(1), rs.getString(2)); } return result; } @@ -2032,12 +2145,26 @@ private static List getDirectChildren(DbContext ctx, Collection } } - public static Set getWorkflowChildren(DbContext ctx, String workflowId) + /** + * Refuses a cancel, resume or delete, made from inside a workflow, that targets that workflow. + * Each clears or removes the workflow's row, so the call's own checkpoint, or the enclosing + * step's, would then fail the ownership check, rolling the call back and leaving the workflow + * waiting on an outcome nothing writes. + */ + 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 static Set getWorkflowChildren(Connection conn, String schema, String workflowId) throws SQLException { var descendants = new HashSet(); List frontier = List.of(workflowId); while (!frontier.isEmpty()) { - var children = getDirectChildren(ctx, frontier); + var children = getDirectChildren(conn, schema, frontier).keySet(); frontier = children.stream().filter(c -> !descendants.contains(c)).toList(); descendants.addAll(frontier); } @@ -2579,29 +2706,6 @@ private static void batchCopyWorkflowData(Connection conn, String schema, List workflowIds, @Nullable String queueName @Override public void deleteWorkflows(List workflowIds, boolean deleteChildren) { - systemDatabase.deleteWorkflows(workflowIds, deleteChildren); + systemDatabase.deleteWorkflows(workflowIds, deleteChildren, null); } @Override 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..63b4bc775 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -20,7 +20,7 @@ import dev.dbos.transact.database.ExternalState; import dev.dbos.transact.database.GetEventCaller; import dev.dbos.transact.database.Result; -import dev.dbos.transact.database.SendCaller; +import dev.dbos.transact.database.StepCaller; import dev.dbos.transact.database.StreamIterator; import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.database.TimedWorkflowStatus; @@ -941,18 +941,9 @@ private void sendBulkAs( @Nullable String serializationFormat, String functionName) { - DBOSContext ctx = DBOSContextHolder.get(); // A send from a workflow is its step; one from inside a step records nothing, but is still // made only while the step's execution owns the workflow. - SendCaller caller = null; - if (ctx.isInWorkflow()) { - caller = - ctx.isInStep() - ? new SendCaller.FromStep(ctx.getWorkflowId(), ctx.getOwnerXid()) - : new SendCaller.FromWorkflow( - ctx.getWorkflowId(), ctx.getAndIncrementFunctionId(), ctx.getOwnerXid()); - } - systemDatabase.sendBulk(messages, caller, functionName, sendToForks, serializationFormat); + systemDatabase.sendBulk(messages, stepCaller(), functionName, sendToForks, serializationFormat); } /** @@ -1084,47 +1075,19 @@ public void sleep(Duration duration) { public void cancelWorkflows(List workflowIds, boolean cancelChildren) { Objects.requireNonNull(workflowIds); - this.runDbosFunctionAsStep( - () -> { - logger.info( - "Cancelling workflow(s) {}{}", - workflowIds, - cancelChildren ? " and their children" : ""); - systemDatabase.cancelWorkflows(workflowIds, cancelChildren); - return null; // void - }, - "DBOS.cancelWorkflow", - null); + systemDatabase.cancelWorkflows(workflowIds, cancelChildren, stepCaller()); } public void deleteWorkflows(List workflowIds, boolean deleteChildren) { Objects.requireNonNull(workflowIds); - this.runDbosFunctionAsStep( - () -> { - logger.info( - "Deleting workflow(s) {}{}", - workflowIds, - deleteChildren ? " and their children" : ""); - systemDatabase.deleteWorkflows(workflowIds, deleteChildren); - return null; - }, - "DBOS.deleteWorkflow", - null); + systemDatabase.deleteWorkflows(workflowIds, deleteChildren, stepCaller()); } public List> resumeWorkflows( List workflowIds, String queueName) { Objects.requireNonNull(workflowIds); - // Execute the resume operation as a workflow step - this.runDbosFunctionAsStep( - () -> { - logger.info("Resuming workflow(s) {}", workflowIds); - systemDatabase.resumeWorkflows(workflowIds, queueName); - return null; // void - }, - "DBOS.resumeWorkflow", - null); + systemDatabase.resumeWorkflows(workflowIds, queueName, stepCaller()); return workflowIds.stream().map(this::retrieveWorkflow).toList(); } @@ -1725,6 +1688,26 @@ public T runStep( return runStepInternal(step, options, childWorkflowId); } + /** + * Who a DBOS operation is being called by, for one that commits its effect and its checkpoint + * together: the workflow body, which takes the next step ID; a step, which records nothing of its + * own; or nothing outside a workflow. Mirrors the choice {@link #runDbosFunctionAsStep} makes. + */ + private @Nullable StepCaller stepCaller() { + var ctx = DBOSContextHolder.get(); + if (!ctx.isInWorkflow()) { + return null; + } + if (ctx.isInStep()) { + return new StepCaller.FromStep(ctx.getWorkflowId(), ctx.getOwnerXid()); + } + if (HOOK_HOLDER.get() != null) { + throw new RuntimeException("DBOS operations cannot be called from the startWorkflow lambda"); + } + return new StepCaller.FromWorkflow( + ctx.getWorkflowId(), ctx.getAndIncrementFunctionId(), ctx.getOwnerXid()); + } + public T runDbosFunctionAsStep( @NonNull ThrowingSupplier step, @NonNull String stepName, diff --git a/transact/src/main/java/dev/dbos/transact/internal/DebugTriggers.java b/transact/src/main/java/dev/dbos/transact/internal/DebugTriggers.java index 510c887af..1e2c9fdb6 100644 --- a/transact/src/main/java/dev/dbos/transact/internal/DebugTriggers.java +++ b/transact/src/main/java/dev/dbos/transact/internal/DebugTriggers.java @@ -76,6 +76,14 @@ public static void clearDebugTrigger(String name) { // "DEBUG_TRIGGER_WORKFLOW_ENQUEUE"; public static final String DEBUG_TRIGGER_STEP_COMMIT = "DEBUG_TRIGGER_STEP_COMMIT"; public static final String DEBUG_TRIGGER_INITWF_COMMIT = "DEBUG_TRIGGER_INITWF_COMMIT"; + // Inside a child's insert transaction, once the parent's step is recorded and its row locked. + public static final String DEBUG_TRIGGER_CHILD_RECORDED = "DEBUG_TRIGGER_CHILD_RECORDED"; + // Inside an internal step's transaction, after its work and checkpoint, before the commit. + public static final String DEBUG_TRIGGER_INTERNAL_STEP_COMMIT = + "DEBUG_TRIGGER_INTERNAL_STEP_COMMIT"; + // After an internal step's transaction has committed, before the call returns. + public static final String DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED = + "DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED"; // Inside a debouncer's takeover transaction, after it cancels a stranded debouncer workflow and // before it creates the promised workflow. public static final String DEBUG_TRIGGER_DEBOUNCE_TAKEOVER = "DEBUG_TRIGGER_DEBOUNCE_TAKEOVER"; diff --git a/transact/src/test/java/dev/dbos/transact/database/IdleTransactionTimeoutTest.java b/transact/src/test/java/dev/dbos/transact/database/IdleTransactionTimeoutTest.java index 791db8e15..56eda0bda 100644 --- a/transact/src/test/java/dev/dbos/transact/database/IdleTransactionTimeoutTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/IdleTransactionTimeoutTest.java @@ -120,7 +120,8 @@ public void aStrandedOwnerLockIsReleased() throws Exception { // The stalled session holds the row; the cancel waits for the server to end it. var cancel = - CompletableFuture.runAsync(() -> sysdb.cancelWorkflows(List.of(workflowId), false)); + CompletableFuture.runAsync( + () -> sysdb.cancelWorkflows(List.of(workflowId), false, null)); cancel.get(30, TimeUnit.SECONDS); assertNull(WorkflowDAO.getWorkflowOwner(ctx, workflowId), "the cancel released it"); diff --git a/transact/src/test/java/dev/dbos/transact/database/RecordStepResultTest.java b/transact/src/test/java/dev/dbos/transact/database/RecordStepResultTest.java index d46e659c6..96224085f 100644 --- a/transact/src/test/java/dev/dbos/transact/database/RecordStepResultTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/RecordStepResultTest.java @@ -198,7 +198,7 @@ public void aStaleOwnerIsAConflictAndRecordsNothing() throws Exception { public void aReleasedWorkflowIsAConflict() throws Exception { var workflowId = "record-step-released-%d".formatted(System.nanoTime()); var owner = createWorkflow(workflowId); - sysdb.cancelWorkflows(List.of(workflowId), false); + sysdb.cancelWorkflows(List.of(workflowId), false, null); assertThrows( DBOSWorkflowExecutionConflictException.class, @@ -210,7 +210,7 @@ public void aReleasedWorkflowIsAConflict() throws Exception { public void aDeletedWorkflowIsAConflict() throws Exception { var workflowId = "record-step-deleted-%d".formatted(System.nanoTime()); var owner = createWorkflow(workflowId); - sysdb.deleteWorkflows(List.of(workflowId), false); + sysdb.deleteWorkflows(List.of(workflowId), false, null); assertThrows( DBOSWorkflowExecutionConflictException.class, @@ -248,7 +248,7 @@ public void theOwnerCheckHoldsOffAHandOffUntilTheWriteCommits() throws Exception assertTrue(checked.await(10, TimeUnit.SECONDS)); var cancel = - CompletableFuture.runAsync(() -> sysdb.cancelWorkflows(List.of(workflowId), false)); + CompletableFuture.runAsync(() -> sysdb.cancelWorkflows(List.of(workflowId), false, null)); assertThrows( TimeoutException.class, () -> cancel.get(1, TimeUnit.SECONDS), diff --git a/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java b/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java index 9395174da..9627f5fa2 100644 --- a/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java @@ -119,7 +119,7 @@ public void testDeleteWorkflows() throws Exception { var rows = DBUtils.getWorkflowRows(dataSource); assertEquals(5, rows.size()); - sysdb.deleteWorkflows(List.of("wfid-1", "wfid-3"), false); + sysdb.deleteWorkflows(List.of("wfid-1", "wfid-3"), false, null); rows = DBUtils.getWorkflowRows(dataSource); assertEquals(3, rows.size()); @@ -199,7 +199,7 @@ public void testDeleteWorkflowsList() throws Exception { sysdb.initWorkflowStatus(status, 5, null); } - sysdb.deleteWorkflows(List.of("wfid-0", "wfid-2", "wfid-4"), false); + sysdb.deleteWorkflows(List.of("wfid-0", "wfid-2", "wfid-4"), false, null); var rows = DBUtils.getWorkflowRows(dataSource); assertEquals(2, rows.size()); @@ -223,7 +223,9 @@ public void testCancelWorkflows() throws Exception { // Cancel all five IDs in one call sysdb.cancelWorkflows( - List.of("wf-pending-1", "wf-pending-2", "wf-pending-3", "wf-success", "wf-error"), false); + List.of("wf-pending-1", "wf-pending-2", "wf-pending-3", "wf-success", "wf-error"), + false, + null); // PENDING ones become CANCELLED, and updated_at is refreshed for (var wfid : List.of("wf-pending-1", "wf-pending-2", "wf-pending-3")) { @@ -276,7 +278,7 @@ public void testResumeWorkflows() throws Exception { long beforeResume = System.currentTimeMillis(); // Resume all four IDs in one call - sysdb.resumeWorkflows(workflowIds, null); + sysdb.resumeWorkflows(workflowIds, null, null); // CANCELLED ones become ENQUEUED; updated_at advances; deadline and completed_at are cleared for (var wfid : List.of("wf-cancelled-1", "wf-cancelled-2")) { @@ -302,7 +304,7 @@ public void testResumeWorkflowsCustomQueue() throws Exception { long beforeResume = System.currentTimeMillis(); // Resume all four IDs in one call - sysdb.resumeWorkflows(workflowIds, "customQueue"); + sysdb.resumeWorkflows(workflowIds, "customQueue", null); // CANCELLED ones become ENQUEUED; updated_at advances; deadline and completed_at are cleared for (var wfid : List.of("wf-cancelled-1", "wf-cancelled-2")) { @@ -326,7 +328,7 @@ public void testCancelWorkflowsNullInList() throws Exception { sysdb.initWorkflowStatus(WorkflowStatusInternalBuilder.create("wf-id").build(), 5, null); long beforeCancel = System.currentTimeMillis(); - sysdb.cancelWorkflows(Arrays.asList("wf-id", null), false); + sysdb.cancelWorkflows(Arrays.asList("wf-id", null), false, null); var row = DBUtils.getWorkflowRow(dataSource, "wf-id"); assertEquals(WorkflowState.CANCELLED.name(), row.status()); @@ -355,7 +357,7 @@ public void testCancelWorkflowsWithChildren() throws Exception { } // Without cancelChildren=true, only the parent is cancelled - sysdb.cancelWorkflows(List.of("parent"), false); + sysdb.cancelWorkflows(List.of("parent"), false, null); assertEquals( WorkflowState.CANCELLED.name(), DBUtils.getWorkflowRow(dataSource, "parent").status()); for (var i = 0; i < 3; i++) { @@ -373,7 +375,7 @@ public void testCancelWorkflowsWithChildren() throws Exception { DBUtils.setWorkflowState(dataSource, "parent", WorkflowState.PENDING.name()); // With cancelChildren=true, parent + all descendants are cancelled - sysdb.cancelWorkflows(List.of("parent"), true); + sysdb.cancelWorkflows(List.of("parent"), true, null); assertEquals( WorkflowState.CANCELLED.name(), DBUtils.getWorkflowRow(dataSource, "parent").status()); for (var i = 0; i < 3; i++) { @@ -394,7 +396,7 @@ public void testResumeWorkflowsNullInList() throws Exception { DBUtils.setWorkflowState(dataSource, "wf-id", WorkflowState.CANCELLED.name()); long beforeResume = System.currentTimeMillis(); - sysdb.resumeWorkflows(Arrays.asList("wf-id", null), null); + sysdb.resumeWorkflows(Arrays.asList("wf-id", null), null, null); var row = DBUtils.getWorkflowRow(dataSource, "wf-id"); assertEquals(WorkflowState.ENQUEUED.name(), row.status()); @@ -406,7 +408,7 @@ public void testResumeWorkflowsNullInList() throws Exception { public void testDeleteWorkflowsNullInList() throws Exception { sysdb.initWorkflowStatus(WorkflowStatusInternalBuilder.create("wf-id").build(), 5, null); - sysdb.deleteWorkflows(Arrays.asList("wf-id", null), false); + sysdb.deleteWorkflows(Arrays.asList("wf-id", null), false, null); assertNull(DBUtils.getWorkflowRow(dataSource, "wf-id")); } diff --git a/transact/src/test/java/dev/dbos/transact/database/TransactionalStepTest.java b/transact/src/test/java/dev/dbos/transact/database/TransactionalStepTest.java new file mode 100644 index 000000000..87dc7eb70 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/database/TransactionalStepTest.java @@ -0,0 +1,184 @@ +package dev.dbos.transact.database; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.DBOSTestAccess; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.database.dao.WorkflowDAO; +import dev.dbos.transact.exceptions.DBOSWorkflowExecutionConflictException; +import dev.dbos.transact.utils.PgContainer; +import dev.dbos.transact.workflow.Workflow; +import dev.dbos.transact.workflow.WorkflowState; + +import java.sql.SQLException; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * What an internal step records when its work fails, and that a caller who no longer owns its + * workflow commits nothing, whether it calls from the workflow body or from inside a step. + */ +public class TransactionalStepTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + + @AutoClose DBOS dbos; + + private SystemDatabase sysdb; + private BlockingServiceImpl impl; + private BlockingService proxy; + + @BeforeEach + void beforeEach() { + dbos = new DBOS(pgContainer.dbosConfig()); + impl = new BlockingServiceImpl(); + proxy = dbos.registerProxy(BlockingService.class, impl); + dbos.launch(); + sysdb = DBOSTestAccess.getSystemDatabase(dbos); + } + + @AfterEach + void afterEach() { + impl.release.countDown(); + } + + private String startBlocked(String prefix) throws Exception { + var id = "%s-%d".formatted(prefix, System.nanoTime()); + dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(id)); + assertTrue(impl.started.tryAcquire(10, TimeUnit.SECONDS)); + return id; + } + + private String ownerOf(String workflowId) throws SQLException { + var sql = + "SELECT owner_xid FROM \"%s\".workflow_status WHERE workflow_uuid = ?" + .formatted(sysdb.context().schema()); + try (var conn = sysdb.context().getConnection(); + var stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + try (var rs = stmt.executeQuery()) { + assertTrue(rs.next()); + return rs.getString(1); + } + } + } + + private WorkflowState statusOf(String workflowId) { + return dbos.getWorkflowStatus(workflowId).orElseThrow().status(); + } + + @Test + public void aConflictThatOutlastsItsRetriesIsRecordedAndReplayed() throws Exception { + var callerId = startBlocked("caller"); + var step = new StepCaller.FromWorkflow(callerId, 1000, ownerOf(callerId)); + var attempts = new AtomicInteger(); + + assertThrows( + RuntimeException.class, + () -> + sysdb.runTransactionalStep( + step, + "conflicted", + conn -> { + attempts.incrementAndGet(); + throw new SQLException("conflict", "40001"); + })); + assertEquals(10, attempts.get(), "retried until the retries ran out"); + + var recorded = dbos.listWorkflowSteps(callerId); + assertEquals(1, recorded.size()); + assertNotNull(recorded.get(0).error(), "the failure is the step's answer"); + + // A replay that would now succeed is refused the same way, so it takes the same branch. + assertThrows( + RuntimeException.class, + () -> + sysdb.runTransactionalStep( + step, + "conflicted", + conn -> { + attempts.incrementAndGet(); + return "succeeded"; + })); + assertEquals(10, attempts.get(), "the replay did not run the work"); + } + + @Test + public void aCallerThatLostItsWorkflowCommitsNothing() throws Exception { + var callerId = startBlocked("caller"); + var targetId = startBlocked("target"); + var staleOwner = ownerOf(callerId); + DBTestAccess.takeOwnership(sysdb, callerId); + + assertThrows( + DBOSWorkflowExecutionConflictException.class, + () -> + sysdb.cancelWorkflows( + List.of(targetId), false, new StepCaller.FromStep(callerId, staleOwner))); + assertEquals(WorkflowState.PENDING, statusOf(targetId), "the step's cancel rolled back"); + + assertThrows( + DBOSWorkflowExecutionConflictException.class, + () -> + sysdb.cancelWorkflows( + List.of(targetId), false, new StepCaller.FromWorkflow(callerId, 1000, staleOwner))); + assertEquals(WorkflowState.PENDING, statusOf(targetId), "the body's cancel rolled back"); + assertTrue(dbos.listWorkflowSteps(callerId).isEmpty(), "nothing recorded"); + } + + /** + * The step check passes, then the workflow changes hands while the work runs: the record at the + * end of the transaction is refused, and the work rolls back with it. + */ + @Test + public void aWorkflowLostDuringTheWorkCommitsNothing() throws Exception { + var callerId = startBlocked("caller"); + var targetId = startBlocked("target"); + var step = new StepCaller.FromWorkflow(callerId, 1000, ownerOf(callerId)); + + assertThrows( + DBOSWorkflowExecutionConflictException.class, + () -> + sysdb.runTransactionalStep( + step, + "DBOS.cancelWorkflow", + conn -> { + WorkflowDAO.cancelWorkflows( + conn, sysdb.context().schema(), List.of(targetId), false, callerId); + DBTestAccess.takeOwnership(sysdb, callerId); + return null; + })); + + assertEquals(WorkflowState.PENDING, statusOf(targetId), "the cancel rolled back"); + assertTrue(dbos.listWorkflowSteps(callerId).isEmpty(), "nothing recorded"); + } +} + +interface BlockingService { + String blocked() throws InterruptedException; +} + +class BlockingServiceImpl implements BlockingService { + final Semaphore started = new Semaphore(0); + final CountDownLatch release = new CountDownLatch(1); + + @Override + @Workflow + public String blocked() throws InterruptedException { + started.release(); + release.await(); + return "released"; + } +} diff --git a/transact/src/test/java/dev/dbos/transact/database/WorkflowInsertOwnershipTest.java b/transact/src/test/java/dev/dbos/transact/database/WorkflowInsertOwnershipTest.java index 982a57d62..706d4630b 100644 --- a/transact/src/test/java/dev/dbos/transact/database/WorkflowInsertOwnershipTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/WorkflowInsertOwnershipTest.java @@ -114,7 +114,7 @@ public void aRetriedStartFindingItsOwnRowCancelledOwnsNothing() throws Exception var token = UUID.randomUUID().toString(); var status = WorkflowStatusInternalBuilder.create(workflowId).build(); insert(status, token, null); - sysdb.cancelWorkflows(List.of(workflowId), false); + sysdb.cancelWorkflows(List.of(workflowId), false, null); var retried = insert(status, token, null); diff --git a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java new file mode 100644 index 000000000..356655240 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -0,0 +1,597 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.context.DBOSContext; +import dev.dbos.transact.context.DBOSContextHolder; +import dev.dbos.transact.internal.DebugTriggers; +import dev.dbos.transact.utils.PgContainer; + +import java.sql.SQLException; +import java.sql.SQLTransientException; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.ResourceLock; + +/** + * A DBOS operation called from a workflow commits its effect and its checkpoint together. A failure + * before the commit leaves neither, so the replay does the operation once; a commit whose + * acknowledgement is lost is found on retry and replayed, not done again. + */ +@ResourceLock(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT) +@ResourceLock(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED) +@ResourceLock(DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED) +public class InternalStepTransactionTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + + @AutoClose DBOS dbos; + + private InternalStepService proxy; + private InternalStepServiceImpl impl; + + @BeforeEach + void beforeEach() { + dbos = new DBOS(pgContainer.dbosConfig()); + impl = new InternalStepServiceImpl(dbos); + proxy = dbos.registerProxy(InternalStepService.class, impl); + impl.self = proxy; + dbos.launch(); + } + + @AfterEach + void afterEach() { + DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT); + DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED); + DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED); + impl.release.countDown(); + impl.startChild.countDown(); + impl.mutualRelease.countDown(); + } + + /** + * Arms {@code trigger} to throw {@code failure} once, on the calling workflow {@code callerId}'s + * own thread. Test classes run concurrently and the triggers are global, so one armed for any + * caller could be tripped, and spent, by another class's cancel. + */ + private static AtomicBoolean failOnce(String trigger, String callerId, SQLException failure) { + var fired = new AtomicBoolean(); + DebugTriggers.setDebugTrigger( + trigger, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (callerId.equals(DBOSContextHolder.get().getWorkflowId()) + && fired.compareAndSet(false, true)) { + sneakyThrow(failure); + } + })); + return fired; + } + + /** As {@link #failOnce}, but throws {@code failure} the first {@code times} times. */ + private static AtomicInteger failTimes( + String trigger, String callerId, int times, SQLException failure) { + var fired = new AtomicInteger(); + DebugTriggers.setDebugTrigger( + trigger, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (callerId.equals(DBOSContextHolder.get().getWorkflowId()) + && fired.getAndIncrement() < times) { + sneakyThrow(failure); + } + })); + return fired; + } + + @SuppressWarnings("unchecked") + private static void sneakyThrow(Throwable t) throws E { + throw (E) t; + } + + private String startBlocked(String prefix) throws Exception { + var id = "%s-%d".formatted(prefix, System.nanoTime()); + dbos.startWorkflow(proxy::blocked, new StartWorkflowOptions(id)); + assertTrue(impl.blockedStarted.await(10, TimeUnit.SECONDS)); + return id; + } + + private List stepsNamed(String workflowId, String name) { + return dbos.listWorkflowSteps(workflowId).stream() + .filter(step -> step.functionName().equals(name)) + .toList(); + } + + private WorkflowState statusOf(String workflowId) { + return dbos.retrieveWorkflow(workflowId).getStatus().status(); + } + + @Test + public void aFailureBeforeCommitLeavesNeitherEffectNorCheckpoint() throws Exception { + var targetId = startBlocked("target"); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT, + callerId, + new SQLException("crash", "XX000")); + + // The workflow catches the failed cancel, looks at the target, and cancels again. + var caller = + dbos.startWorkflow( + () -> proxy.cancelsAfterAFailure(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("first failed, target PENDING", caller.getResult(), "the cancel rolled back"); + assertTrue(fired.get()); + assertEquals(WorkflowState.CANCELLED, statusOf(targetId)); + // Neither the failure, which the rollback undid, nor its checkpoint was recorded: only the + // second cancel's. + var recorded = stepsNamed(callerId, "DBOS.cancelWorkflow"); + assertEquals(1, recorded.size()); + assertNull(recorded.get(0).error()); + } + + /** + * A conflict the database raises after the work, in the record or the owner check, and that + * outlasts the retries is the step's answer as one raised in the work is: which statement a + * deadlock picks is not the caller's doing. + */ + @Test + public void aConflictOutsideTheWorkThatOutlastsItsRetriesIsRecorded() throws Exception { + var targetId = startBlocked("target"); + var callerId = "caller-%d".formatted(System.nanoTime()); + // Ten is every attempt the first cancel gets; the second cancel is left alone. + var fired = + failTimes( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT, + callerId, + 10, + new SQLException("conflict", "40001")); + + var caller = + dbos.startWorkflow( + () -> proxy.cancelsAfterAFailure(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("first failed, target PENDING", caller.getResult()); + assertEquals(11, fired.get(), "ten conflicts, then the second cancel"); + assertEquals(WorkflowState.CANCELLED, statusOf(targetId)); + var recorded = stepsNamed(callerId, "DBOS.cancelWorkflow"); + assertEquals(2, recorded.size()); + assertTrue(recorded.get(0).error() != null, "the conflict is the first cancel's answer"); + assertNull(recorded.get(1).error()); + + // A replay, which would now succeed at once, is refused the same way and takes the same branch. + var fork = + dbos.forkWorkflow(callerId, recorded.get(1).functionId() + 1); + assertEquals("first failed, target PENDING", fork.getResult()); + } + + @Test + public void aResumeThatFailsBeforeCommitLeavesNeitherEffectNorCheckpoint() throws Exception { + var targetId = startBlocked("target"); + dbos.cancelWorkflow(targetId); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT, + callerId, + new SQLException("crash", "XX000")); + + var caller = + dbos.startWorkflow( + () -> proxy.resumesCatchingAFailure(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("failed, target CANCELLED", caller.getResult(), "the resume rolled back"); + assertTrue(fired.get()); + assertTrue(stepsNamed(callerId, "DBOS.resumeWorkflow").isEmpty()); + } + + @Test + public void aResumeWhoseCommitAcknowledgementIsLostReplaysTheCheckpoint() throws Exception { + var targetId = startBlocked("target"); + dbos.cancelWorkflow(targetId); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED, + callerId, + new SQLTransientException()); + + var caller = + dbos.startWorkflow( + () -> proxy.resumesCatchingAFailure(targetId), new StartWorkflowOptions(callerId)); + + var result = caller.getResult(); + assertTrue( + result.equals("resumed, target ENQUEUED") || result.equals("resumed, target PENDING"), + result); + assertTrue(fired.get()); + var recorded = stepsNamed(callerId, "DBOS.resumeWorkflow"); + assertEquals(1, recorded.size()); + assertNull(recorded.get(0).error()); + } + + /** + * Two workflows cancelling each other lock each other's row and then their own. Whether one + * simply commits first or the database breaks a deadlock between them, one cancel wins and the + * other workflow, by then cancelled, commits nothing: never both, and neither hangs. + */ + @Test + public void twoWorkflowsCancellingEachOtherLeaveExactlyOneCancelled() throws Exception { + var firstId = "mutual-a-%d".formatted(System.nanoTime()); + var secondId = "mutual-b-%d".formatted(System.nanoTime()); + dbos.startWorkflow( + () -> proxy.cancelsWhenReleased(secondId), new StartWorkflowOptions(firstId)); + dbos.startWorkflow( + () -> proxy.cancelsWhenReleased(firstId), new StartWorkflowOptions(secondId)); + assertTrue(impl.mutualWaiting.await(10, TimeUnit.SECONDS)); + impl.mutualRelease.countDown(); + + var terminal = List.of(WorkflowState.SUCCESS, WorkflowState.CANCELLED, WorkflowState.ERROR); + var deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60); + while (!(terminal.contains(statusOf(firstId)) && terminal.contains(statusOf(secondId)))) { + assertTrue(System.nanoTime() < deadline, "both workflows end"); + Thread.sleep(100); + } + var states = List.of(statusOf(firstId), statusOf(secondId)); + assertTrue( + states.contains(WorkflowState.CANCELLED) && states.contains(WorkflowState.SUCCESS), + "one cancelled the other and finished: " + states); + } + + @Test + public void aSendInAStartWorkflowLambdaIsRefused() throws Exception { + var targetId = startBlocked("target"); + var callerId = "caller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.sendsInStartLambda(targetId), new StartWorkflowOptions(callerId)); + + var result = caller.getResult(); + assertTrue( + result.contains("DBOS operations cannot be called from the startWorkflow lambda"), result); + assertTrue(stepsNamed(callerId, "DBOS.send").isEmpty()); + } + + @Test + public void aTransientFailureBeforeCommitIsRetriedNotRecorded() throws Exception { + var targetId = startBlocked("target"); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT, + callerId, + new SQLTransientException()); + + var caller = + dbos.startWorkflow(() -> proxy.cancels(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("done", caller.getResult()); + assertTrue(fired.get()); + assertEquals(WorkflowState.CANCELLED, statusOf(targetId)); + var recorded = stepsNamed(callerId, "DBOS.cancelWorkflow"); + assertEquals(1, recorded.size()); + assertNull(recorded.get(0).error(), "the transient failure is not the step's answer"); + } + + @Test + public void aLostCommitAcknowledgementReplaysTheCheckpoint() throws Exception { + var targetId = startBlocked("target"); + var deletedId = "deleted-%d".formatted(System.nanoTime()); + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(deletedId)).getResult(); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED, + callerId, + new SQLTransientException()); + + // The cancel commits, its acknowledgement is lost, and the retry finds the checkpoint it wrote. + var caller = + dbos.startWorkflow( + () -> proxy.cancelsThenDeletes(targetId, deletedId), + new StartWorkflowOptions(callerId)); + + assertEquals("done", caller.getResult()); + assertTrue(fired.get()); + assertEquals(WorkflowState.CANCELLED, statusOf(targetId)); + assertEquals(1, stepsNamed(callerId, "DBOS.cancelWorkflow").size()); + assertEquals(1, stepsNamed(callerId, "DBOS.deleteWorkflow").size()); + assertTrue(dbos.getWorkflowStatus(deletedId).isEmpty(), "deleted"); + } + + /** + * A child whose insert holds its parent's row lock when a cancel of the parent's tree arrives: + * the cancel waits for the insert to commit, then reads the parent's children and finds it. + */ + @Test + public void aChildWhoseInsertHoldsItsParentIsCancelledWithTheTree() throws Exception { + var parentId = "insert-first-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + var holding = new CountDownLatch(1); + var releaseInsert = new CountDownLatch(1); + DebugTriggers.setDebugTrigger( + DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (parentId.equals(DBOSContextHolder.get().getWorkflowId())) { + holding.countDown(); + try { + releaseInsert.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + })); + + dbos.startWorkflow(() -> proxy.startsChild(childId), new StartWorkflowOptions(parentId)); + assertTrue(holding.await(10, TimeUnit.SECONDS), "the child's insert holds the parent"); + + var cancel = CompletableFuture.runAsync(() -> dbos.cancelWorkflow(parentId, true)); + Thread.sleep(500); + assertFalse(cancel.isDone(), "the cancel waits on the parent's row"); + releaseInsert.countDown(); + cancel.get(30, TimeUnit.SECONDS); + + assertEquals(WorkflowState.CANCELLED, statusOf(parentId)); + assertEquals(WorkflowState.CANCELLED, statusOf(childId), "the new child did not escape"); + } + + /** + * A child inserted after a cancel of its parent's tree has committed finds its parent no longer + * owned, and its insert rolls back: no child is left running. + */ + @Test + public void aChildInsertedAfterItsParentIsCancelledIsNeverCreated() throws Exception { + var parentId = "cancel-first-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + dbos.startWorkflow( + () -> proxy.startsChildWhenReleased(childId), new StartWorkflowOptions(parentId)); + assertTrue(impl.parentWaiting.await(10, TimeUnit.SECONDS)); + + dbos.cancelWorkflow(parentId, true); + impl.startChild.countDown(); + assertTrue(impl.childAttempted.await(30, TimeUnit.SECONDS)); + + assertEquals(WorkflowState.CANCELLED, statusOf(parentId)); + assertTrue(dbos.getWorkflowStatus(childId).isEmpty(), "the child's insert rolled back"); + } + + /** + * A child whose insert holds its parent's row lock when a delete of the parent's tree arrives: + * the delete waits for the insert to commit, then reads the parent's children and finds it. + */ + @Test + public void aChildWhoseInsertHoldsItsParentIsDeletedWithTheTree() throws Exception { + var parentId = "delete-insert-first-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + var holding = new CountDownLatch(1); + var releaseInsert = new CountDownLatch(1); + DebugTriggers.setDebugTrigger( + DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (parentId.equals(DBOSContextHolder.get().getWorkflowId())) { + holding.countDown(); + try { + releaseInsert.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + })); + + dbos.startWorkflow(() -> proxy.startsChild(childId), new StartWorkflowOptions(parentId)); + assertTrue(holding.await(10, TimeUnit.SECONDS), "the child's insert holds the parent"); + + var delete = CompletableFuture.runAsync(() -> dbos.deleteWorkflow(parentId, true)); + Thread.sleep(500); + assertFalse(delete.isDone(), "the delete waits on the parent's row"); + releaseInsert.countDown(); + delete.get(30, TimeUnit.SECONDS); + + assertTrue(dbos.getWorkflowStatus(parentId).isEmpty(), "the parent is deleted"); + assertTrue(dbos.getWorkflowStatus(childId).isEmpty(), "the new child did not escape"); + } + + /** + * A child inserted after a delete of its parent's tree has committed finds its parent gone, and + * its insert rolls back: no orphan is left running. + */ + @Test + public void aChildInsertedAfterItsParentIsDeletedIsNeverCreated() throws Exception { + var parentId = "delete-first-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + dbos.startWorkflow( + () -> proxy.startsChildWhenReleased(childId), new StartWorkflowOptions(parentId)); + assertTrue(impl.parentWaiting.await(10, TimeUnit.SECONDS)); + + dbos.deleteWorkflow(parentId, true); + impl.startChild.countDown(); + assertTrue(impl.childAttempted.await(30, TimeUnit.SECONDS)); + + assertTrue(dbos.getWorkflowStatus(parentId).isEmpty(), "the parent is deleted"); + assertTrue(dbos.getWorkflowStatus(childId).isEmpty(), "the child's insert rolled back"); + } + + @Test + public void aCallFromInsideAStepRecordsNoCheckpointOfItsOwn() throws Exception { + var targetId = startBlocked("target"); + var callerId = "caller-%d".formatted(System.nanoTime()); + var caller = + dbos.startWorkflow( + () -> proxy.cancelsFromStep(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("done", caller.getResult()); + assertEquals(WorkflowState.CANCELLED, statusOf(targetId)); + assertTrue(stepsNamed(callerId, "DBOS.cancelWorkflow").isEmpty()); + assertEquals(1, stepsNamed(callerId, "cancelStep").size()); + } +} + +interface InternalStepService { + String blocked() throws InterruptedException; + + String quick(); + + String cancels(String targetId); + + String cancelsAfterAFailure(String targetId); + + String cancelsThenDeletes(String targetId, String deletedId); + + String cancelsFromStep(String targetId); + + String resumesCatchingAFailure(String targetId); + + String cancelsWhenReleased(String otherId) throws InterruptedException; + + String sendsInStartLambda(String targetId); + + String startsChild(String childId) throws Exception; + + String startsChildWhenReleased(String childId) throws Exception; +} + +class InternalStepServiceImpl implements InternalStepService { + private final DBOS dbos; + InternalStepService self; + final CountDownLatch blockedStarted = new CountDownLatch(1); + final CountDownLatch release = new CountDownLatch(1); + final CountDownLatch parentWaiting = new CountDownLatch(1); + final CountDownLatch startChild = new CountDownLatch(1); + final CountDownLatch childAttempted = new CountDownLatch(1); + final CountDownLatch mutualWaiting = new CountDownLatch(2); + final CountDownLatch mutualRelease = new CountDownLatch(1); + + InternalStepServiceImpl(DBOS dbos) { + this.dbos = dbos; + } + + @Override + @Workflow + public String blocked() throws InterruptedException { + blockedStarted.countDown(); + release.await(); + return DBOSContext.workflowId(); + } + + @Override + @Workflow + public String quick() { + return "quick"; + } + + @Override + @Workflow + public String cancels(String targetId) { + dbos.cancelWorkflow(targetId); + return "done"; + } + + @Override + @Workflow + public String cancelsAfterAFailure(String targetId) { + String first; + try { + dbos.cancelWorkflow(targetId); + first = "first committed"; + } catch (RuntimeException e) { + first = "first failed"; + } + var between = dbos.getWorkflowStatus(targetId).orElseThrow().status(); + dbos.cancelWorkflow(targetId); + return first + ", target " + between; + } + + @Override + @Workflow + public String cancelsThenDeletes(String targetId, String deletedId) { + dbos.cancelWorkflow(targetId); + dbos.deleteWorkflow(deletedId); + return "done"; + } + + @Override + @Workflow + public String startsChild(String childId) throws Exception { + dbos.startWorkflow(() -> self.blocked(), new StartWorkflowOptions(childId)); + return "started"; + } + + @Override + @Workflow + public String startsChildWhenReleased(String childId) throws Exception { + parentWaiting.countDown(); + startChild.await(); + try { + dbos.startWorkflow(() -> self.blocked(), new StartWorkflowOptions(childId)); + } finally { + childAttempted.countDown(); + } + return "started"; + } + + @Override + @Workflow + public String cancelsFromStep(String targetId) { + dbos.runStep(() -> dbos.cancelWorkflow(targetId), "cancelStep"); + return "done"; + } + + @Override + @Workflow + public String resumesCatchingAFailure(String targetId) { + String outcome; + try { + dbos.resumeWorkflow(targetId); + outcome = "resumed"; + } catch (RuntimeException e) { + outcome = "failed"; + } + return outcome + ", target " + dbos.getWorkflowStatus(targetId).orElseThrow().status(); + } + + @Override + @Workflow + public String cancelsWhenReleased(String otherId) throws InterruptedException { + mutualWaiting.countDown(); + mutualRelease.await(); + dbos.cancelWorkflow(otherId); + return "cancelled " + otherId; + } + + @Override + @Workflow + public String sendsInStartLambda(String targetId) { + try { + dbos.startWorkflow( + () -> { + dbos.send(targetId, "message", "topic"); + return self.quick(); + }, + new StartWorkflowOptions(DBOSContext.workflowId() + "-child")); + return "not refused"; + } catch (RuntimeException e) { + return String.valueOf(e.getCause() == null ? e.getMessage() : e.getCause().getMessage()); + } + } +} 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/RepeatableReadPoolTest.java b/transact/src/test/java/dev/dbos/transact/workflow/RepeatableReadPoolTest.java new file mode 100644 index 000000000..463d1008e --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/RepeatableReadPoolTest.java @@ -0,0 +1,102 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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.DBOSContextHolder; +import dev.dbos.transact.internal.DebugTriggers; +import dev.dbos.transact.utils.PgContainer; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.ResourceLock; + +/** + * A tree cancel run on a pool whose connections default to REPEATABLE READ. Each level must read + * its children after its own write has waited out a child's insert, which only READ COMMITTED + * shows: a REPEATABLE READ snapshot, taken at the transaction's first statement, would hide the + * child and let it escape. The step sets the level itself, so the pool's default does not matter. + */ +@ResourceLock(DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED) +public class RepeatableReadPoolTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + + @AutoClose DBOS dbos; + @AutoClose HikariDataSource dataSource; + + private InternalStepService proxy; + private InternalStepServiceImpl impl; + + @BeforeEach + void beforeEach() { + var hikari = new HikariConfig(); + hikari.setJdbcUrl(pgContainer.jdbcUrl()); + hikari.setUsername(pgContainer.username()); + hikari.setPassword(pgContainer.password()); + hikari.setTransactionIsolation("TRANSACTION_REPEATABLE_READ"); + dataSource = new HikariDataSource(hikari); + + dbos = new DBOS(DBOSConfig.defaults("repeatable-read-pool").withDataSource(dataSource)); + impl = new InternalStepServiceImpl(dbos); + proxy = dbos.registerProxy(InternalStepService.class, impl); + impl.self = proxy; + dbos.launch(); + } + + @AfterEach + void afterEach() { + DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED); + impl.release.countDown(); + } + + @Test + public void aChildWhoseInsertHoldsItsParentIsCancelledWithTheTree() throws Exception { + var parentId = "repeatable-read-%d".formatted(System.nanoTime()); + var childId = parentId + "-child"; + var holding = new CountDownLatch(1); + var releaseInsert = new CountDownLatch(1); + DebugTriggers.setDebugTrigger( + DebugTriggers.DEBUG_TRIGGER_CHILD_RECORDED, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (parentId.equals(DBOSContextHolder.get().getWorkflowId())) { + holding.countDown(); + try { + releaseInsert.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + })); + + dbos.startWorkflow(() -> proxy.startsChild(childId), new StartWorkflowOptions(parentId)); + assertTrue(holding.await(10, TimeUnit.SECONDS), "the child's insert holds the parent"); + + var cancel = CompletableFuture.runAsync(() -> dbos.cancelWorkflow(parentId, true)); + Thread.sleep(500); + assertFalse(cancel.isDone(), "the cancel waits on the parent's row"); + releaseInsert.countDown(); + cancel.get(30, TimeUnit.SECONDS); + + assertEquals(WorkflowState.CANCELLED, statusOf(parentId)); + assertEquals(WorkflowState.CANCELLED, statusOf(childId), "the new child did not escape"); + } + + private WorkflowState statusOf(String workflowId) { + return dbos.retrieveWorkflow(workflowId).getStatus().status(); + } +} 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..acf834ee7 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/SelfTargetTest.java @@ -0,0 +1,553 @@ +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, + * nor cancel or delete a tree it belongs to. Each clears or removes its own row, so the ownership + * check on the call's checkpoint, in the same transaction, would roll the call back and leave the + * workflow waiting on an outcome nothing writes. The call is refused instead, 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 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 aChildCannotCancelATreeItBelongsTo() throws Exception { + var parentId = "tree-parent-%d".formatted(System.nanoTime()); + var handle = dbos.startWorkflow(proxy::parentOfCanceller, new StartWorkflowOptions(parentId)); + + var result = handle.getResult(); + assertTrue(result.startsWith("refused: workflow "), result); + assertTrue( + result.endsWith("cannot cancel the tree of %s, which contains it".formatted(parentId)), + result); + // The walk cancelled the parent before reaching the child, and the refusal rolled it back. + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + } + + @Test + public void aStepCannotCancelATreeItsWorkflowBelongsTo() throws Exception { + var parentId = "step-tree-parent-%d".formatted(System.nanoTime()); + var handle = + dbos.startWorkflow(proxy::parentOfStepCanceller, new StartWorkflowOptions(parentId)); + + var result = handle.getResult(); + assertTrue( + result.endsWith("cannot cancel the tree of %s, which contains it".formatted(parentId)), + result); + // The walk cancelled the parent before reaching the child, and the refusal rolled it back. + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + } + + @Test + public void aChildCannotDeleteATreeItBelongsTo() throws Exception { + var parentId = "delete-tree-parent-%d".formatted(System.nanoTime()); + var handle = dbos.startWorkflow(proxy::parentOfDeleter, new StartWorkflowOptions(parentId)); + + var result = handle.getResult(); + assertTrue( + result.endsWith("cannot delete the tree of %s, which contains it".formatted(parentId)), + result); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + } + + @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 aGrandchildCannotCancelItsGrandparentsTree() throws Exception { + var rootId = "tree-root-%d".formatted(System.nanoTime()); + var handle = + dbos.startWorkflow(proxy::grandparentOfCanceller, new StartWorkflowOptions(rootId)); + + var result = handle.getResult(); + assertTrue( + result.endsWith("cannot cancel the tree of %s, which contains it".formatted(rootId)), + result); + assertEquals(WorkflowState.SUCCESS, handle.getStatus().status()); + } + + @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 parentOfCanceller(); + + String deleteSelf(); + + String deleteSelfInStep(); + + String cancelSelfInUncaughtStep(); + + String parentOfDeleter(); + + String deletesTree(String rootId); + + String cancelsChildTree(String childId) throws Exception; + + String grandparentOfCanceller(); + + String middleOfTree(String rootId); + + String cancelsTree(String rootId); + + String parentOfStepCanceller(); + + String cancelsTreeInStep(String rootId); + + 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 parentOfDeleter() { + return self.deletesTree(DBOSContext.workflowId()); + } + + @Override + @Workflow + public String deletesTree(String rootId) { + try { + dbos.deleteWorkflow(rootId, true); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + 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 parentOfCanceller() { + return self.cancelsTree(DBOSContext.workflowId()); + } + + @Override + @Workflow + public String grandparentOfCanceller() { + return self.middleOfTree(DBOSContext.workflowId()); + } + + @Override + @Workflow + public String middleOfTree(String rootId) { + return self.cancelsTree(rootId); + } + + @Override + @Workflow + public String cancelsTree(String rootId) { + try { + dbos.cancelWorkflow(rootId, true); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + } + + @Override + @Workflow + public String parentOfStepCanceller() { + return self.cancelsTreeInStep(DBOSContext.workflowId()); + } + + @Override + @Workflow + public String cancelsTreeInStep(String rootId) { + return dbos.runStep( + () -> { + try { + dbos.cancelWorkflow(rootId, true); + return "not refused"; + } catch (IllegalArgumentException e) { + return "refused: " + e.getMessage(); + } + }, + "cancelTreeFromStep"); + } + + @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"; + } +}