From 4d75e0a9ec1eb1304c676b0ea810eba6fb1229ea Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 10:13:40 -0700 Subject: [PATCH 1/5] Commit fork, rewind, delay and attribute changes with their checkpoint forkWorkflow, rewindWorkflow, setWorkflowDelay and updateWorkflowAttributes now run through runTransactionalStep, so each commits its effect and its step checkpoint in one transaction under the step name it already used. A crash between the two can no longer make recovery repeat the operation: a second fork under a new random ID, or a second rewind that is refused because the first left the target ENQUEUED. Called from inside a step, each commits only while the caller still owns its workflow; from a client, it runs in a plain transaction. The executor's queue check, and a rewind's clearing of the transactional step factories' checkpoints, still run inside the step, after the checkpoint check, so a replay returns the recorded result without repeating them. A fork's ID is chosen before the transaction, so a retry that re-runs the work fails on the fork it already wrote rather than forking again. A relative delay is resolved inside the work, so a delay past the latest instant is still recorded as the step's error. The DAO methods take the caller's connection instead of opening their own transaction, and the batch fork runs through the same helper with no caller. --- .../java/dev/dbos/transact/DBOSClient.java | 11 +- .../transact/database/SystemDatabase.java | 143 +++++-- .../transact/database/dao/WorkflowDAO.java | 379 ++++++++---------- .../dbos/transact/execution/DBOSExecutor.java | 56 +-- .../transact/database/SystemDatabaseTest.java | 22 +- .../workflow/InternalStepTransactionTest.java | 86 ++++ .../workflow/PayloadTableReadTest.java | 2 +- 7 files changed, 404 insertions(+), 295 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/DBOSClient.java b/transact/src/main/java/dev/dbos/transact/DBOSClient.java index 4abd4f50b..62cbff1f6 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOSClient.java +++ b/transact/src/main/java/dev/dbos/transact/DBOSClient.java @@ -1610,7 +1610,7 @@ public void cancelWorkflows(@NonNull List workflowIds, boolean cancelChi */ public void updateWorkflowAttributes( @NonNull String workflowId, @Nullable Map attributes) { - systemDatabase.updateWorkflowAttributes(workflowId, attributes); + systemDatabase.updateWorkflowAttributes(workflowId, attributes, null); } /** @@ -1735,7 +1735,8 @@ public void deleteWorkflows(@NonNull List workflowIds, boolean deleteChi */ public @NonNull WorkflowHandle forkWorkflow( @NonNull String originalWorkflowId, int startStep, @NonNull ForkOptions options) { - var forkedWorkflowId = systemDatabase.forkWorkflow(originalWorkflowId, startStep, options); + var forkedWorkflowId = + systemDatabase.forkWorkflow(originalWorkflowId, startStep, options, null, null); return retrieveWorkflow(forkedWorkflowId); } @@ -1763,7 +1764,7 @@ public void deleteWorkflows(@NonNull List workflowIds, boolean deleteChi */ public @NonNull WorkflowHandle rewindWorkflow( @NonNull String workflowId, int startStep, @NonNull RewindOptions options) { - systemDatabase.rewindWorkflow(workflowId, startStep, options); + systemDatabase.rewindWorkflow(workflowId, startStep, options, null, null); return retrieveWorkflow(workflowId); } @@ -2039,7 +2040,7 @@ public void applySchedules(@NonNull WorkflowSchedule... schedules) { */ public void setWorkflowDelay(@NonNull String workflowId, @NonNull Duration delay) { var wfDelay = new WorkflowDelay.Delay(Objects.requireNonNull(delay, "delay must not be null")); - systemDatabase.setWorkflowDelay(workflowId, wfDelay); + systemDatabase.setWorkflowDelay(workflowId, wfDelay, null); } /** @@ -2055,7 +2056,7 @@ public void setWorkflowDelay(@NonNull String workflowId, @NonNull Instant delayU var wfDelay = new WorkflowDelay.DelayUntil( Objects.requireNonNull(delayUntil, "delayUntil must not be null")); - systemDatabase.setWorkflowDelay(workflowId, wfDelay); + systemDatabase.setWorkflowDelay(workflowId, wfDelay, null); } /** 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 b2f75b8c3..5129359f9 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1687,9 +1687,21 @@ public void resumeWorkflows( }); } - public void updateWorkflowAttributes(String workflowId, Map attributes) { - var validated = Validation.validateAttributes(attributes); - dbRetry(() -> WorkflowDAO.updateWorkflowAttributes(ctx, workflowId, validated)); + /** + * Replaces a workflow's attributes as the step {@code caller} makes; see {@link + * #runTransactionalStep}. + */ + public void updateWorkflowAttributes( + String workflowId, Map attributes, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.updateWorkflowAttributes", + conn -> { + var validated = Validation.validateAttributes(attributes); + WorkflowDAO.updateWorkflowAttributes(conn, ctx.schema(), workflowId, validated); + logger.info("Updating attributes of workflow {}", workflowId); + return null; + }); } /** @@ -1885,10 +1897,47 @@ private void recordStepError( } } - public String forkWorkflow(String originalWorkflowId, int startStep, ForkOptions options) { - return dbRetryIncludingSerializationError( - "forkWorkflow", - () -> WorkflowDAO.forkWorkflow(ctx, originalWorkflowId, startStep, options)); + /** + * Forks a workflow as the step {@code caller} makes, and returns the fork's ID; see {@link + * #runTransactionalStep}. {@code validate}, if given, runs first inside the step, so a replay + * returns the recorded fork without repeating it. + * + *

The ID is chosen before the transaction, so a retry that re-runs the work, after a commit + * whose acknowledgement was lost, fails on the fork it already wrote rather than forking again + * under a second ID. + */ + public String forkWorkflow( + String originalWorkflowId, + int startStep, + @Nullable ForkOptions options, + @Nullable Runnable validate, + @Nullable StepCaller caller) { + var resolved = Objects.requireNonNullElseGet(options, ForkOptions::new); + var forkedWorkflowId = + Objects.requireNonNullElseGet( + resolved.forkedWorkflowId(), () -> UUID.randomUUID().toString()); + return runTransactionalStep( + caller, + "DBOS.forkWorkflow", + conn -> { + if (validate != null) { + validate.run(); + } + WorkflowDAO.forkWorkflows( + ctx, + conn, + List.of(new WorkflowFork(originalWorkflowId, forkedWorkflowId, startStep)), + resolved.applicationVersion(), + resolved.queueName(), + resolved.queuePartitionKey(), + resolved.timeout()); + logger.info( + "Forking workflow {} from step {} as {}", + originalWorkflowId, + startStep, + forkedWorkflowId); + return forkedWorkflowId; + }); } public void forkWorkflows( @@ -1896,22 +1945,45 @@ public void forkWorkflows( @Nullable String applicationVersion, @Nullable String queueName, @Nullable String queuePartitionKey) { - dbRetryIncludingSerializationError( + runTransactionalStep( + null, "forkWorkflows", - () -> { + conn -> { WorkflowDAO.forkWorkflows( - ctx, forks, applicationVersion, queueName, queuePartitionKey, null); + ctx, conn, forks, applicationVersion, queueName, queuePartitionKey, null); return null; }); } - public void rewindWorkflow(String workflowId, int startStep, RewindOptions options) { - dbRetryIncludingSerializationError( - "rewindWorkflow", () -> WorkflowDAO.rewindWorkflow(ctx, workflowId, startStep, options)); - } - - public void checkRewindable(String workflowId, int startStep) { - dbRetry(() -> WorkflowDAO.checkRewindable(ctx, workflowId, startStep)); + /** + * Rewinds a terminal workflow to {@code startStep} as the step {@code caller} makes; see {@link + * #runTransactionalStep}. + * + *

{@code beforeRewind}, if given, runs inside the step once the workflow is known to be + * rewindable, and before its system database history is touched. The executor validates the queue + * there and clears the transactional step factories' checkpoints, which live in other databases + * and so commit on their own: a rewind that then rolls back leaves them cleared, and the workflow + * terminal and not re-enqueued, so nothing runs against the partial history, and since the + * deletes are idempotent the rewind can be retried. + */ + public void rewindWorkflow( + String workflowId, + int startStep, + RewindOptions options, + @Nullable Runnable beforeRewind, + @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.rewindWorkflow", + conn -> { + if (beforeRewind != null) { + WorkflowDAO.checkRewindable(conn, ctx.schema(), workflowId, startStep); + beforeRewind.run(); + } + WorkflowDAO.rewindWorkflow(conn, ctx.schema(), workflowId, startStep, options); + logger.info("Rewinding workflow {} to step {}", workflowId, startStep); + return null; + }); } public void createApplicationVersion(String versionName, @Nullable String applicationName) { @@ -1965,18 +2037,31 @@ public void garbageCollect(Instant cutoff, Long rowsThreshold, int batchSize) { dbRetry(() -> WorkflowDAO.runRetentionRound(ctx, cutoff, rowsThreshold, batchSize)); } - public void setWorkflowDelay(String workflowId, WorkflowDelay delay) { - Objects.requireNonNull(delay, "delay must not be null"); - long delayUntilEpochMs; - if (delay instanceof WorkflowDelay.Delay d) { - delayUntilEpochMs = resolveDelayUntil(d.delay()); - } else if (delay instanceof WorkflowDelay.DelayUntil du) { - // An absolute delay is the caller's instant, taken as given. - delayUntilEpochMs = du.delayUntil().toEpochMilli(); - } else { - throw new IllegalArgumentException("Unexpected WorkflowDelay value"); - } - dbRetry(() -> WorkflowDAO.setWorkflowDelay(ctx, workflowId, delayUntilEpochMs)); + /** + * Moves a DELAYED workflow's start as the step {@code caller} makes; see {@link + * #runTransactionalStep}. A relative delay is resolved in the work, so it runs from the attempt + * that commits, as a started workflow's deadline does. + */ + public void setWorkflowDelay( + String workflowId, WorkflowDelay delay, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.setWorkflowDelay", + conn -> { + Objects.requireNonNull(delay, "delay must not be null"); + long delayUntilEpochMs; + if (delay instanceof WorkflowDelay.Delay d) { + delayUntilEpochMs = resolveDelayUntil(d.delay()); + } else if (delay instanceof WorkflowDelay.DelayUntil du) { + // An absolute delay is the caller's instant, taken as given. + delayUntilEpochMs = du.delayUntil().toEpochMilli(); + } else { + throw new IllegalArgumentException("Unexpected WorkflowDelay value"); + } + WorkflowDAO.setWorkflowDelay(conn, ctx.schema(), workflowId, delayUntilEpochMs); + logger.info("Delaying workflow {} until {}", workflowId, delayUntilEpochMs); + return null; + }); } /** See {@link WorkflowDAO#cancelTimedOutWorkflows}. */ 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 ca9f5cc5b..481076104 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 @@ -27,7 +27,6 @@ import dev.dbos.transact.workflow.Debouncer.DebounceIds; import dev.dbos.transact.workflow.DeduplicationHolder; import dev.dbos.transact.workflow.ErrorResult; -import dev.dbos.transact.workflow.ForkOptions; import dev.dbos.transact.workflow.ListWorkflowsInput; import dev.dbos.transact.workflow.RewindOptions; import dev.dbos.transact.workflow.WorkflowIdReusePolicy; @@ -1286,7 +1285,8 @@ ON CONFLICT (workflow_uuid) DO UPDATE SET inputs = EXCLUDED.inputs findDeduplicationHolder(conn, ctx.schema(), queueName, deduplicationId)); } - public static void setWorkflowDelay(DbContext ctx, String workflowId, long delayUntilEpochMs) + public static void setWorkflowDelay( + Connection conn, String schema, String workflowId, long delayUntilEpochMs) throws SQLException { Objects.requireNonNull(workflowId, "workflowId must not be null"); @@ -1298,9 +1298,8 @@ public static void setWorkflowDelay(DbContext ctx, String workflowId, long delay WHERE workflow_uuid = ? AND status = ? """ - .formatted(ctx.schema(), SystemDatabase.NOW_EPOCH_MS); - try (var conn = ctx.getConnection(); - var stmt = conn.prepareStatement(sql)) { + .formatted(schema, SystemDatabase.NOW_EPOCH_MS); + try (var stmt = conn.prepareStatement(sql)) { stmt.setLong(1, delayUntilEpochMs); stmt.setString(2, workflowId); stmt.setString(3, WorkflowState.DELAYED.name()); @@ -1316,7 +1315,8 @@ private static String attributesToJson(Map attributes) { } public static void updateWorkflowAttributes( - DbContext ctx, String workflowId, Map attributes) throws SQLException { + Connection conn, String schema, String workflowId, Map attributes) + throws SQLException { Objects.requireNonNull(workflowId, "workflowId must not be null"); var attributesJson = attributesToJson(attributes); @@ -1327,9 +1327,8 @@ public static void updateWorkflowAttributes( updated_at = %2$s WHERE workflow_uuid = ? """ - .formatted(ctx.schema(), SystemDatabase.NOW_EPOCH_MS); - try (var conn = ctx.getConnection(); - var stmt = conn.prepareStatement(sql)) { + .formatted(schema, SystemDatabase.NOW_EPOCH_MS); + try (var stmt = conn.prepareStatement(sql)) { stmt.setString(1, attributesJson); stmt.setString(2, workflowId); stmt.executeUpdate(); @@ -2148,33 +2147,12 @@ public static Set getWorkflowChildren(Connection conn, String schema, St return descendants; } - public static String forkWorkflow( - DbContext ctx, String workflowId, int startStep, ForkOptions options) throws SQLException { - - options = Objects.requireNonNullElseGet(options, ForkOptions::new); - - String forkedWorkflowId = - Objects.requireNonNullElseGet( - options.forkedWorkflowId(), () -> UUID.randomUUID().toString()); - - logger.debug("forkWorkflow Original id {} forked id {}", workflowId, forkedWorkflowId); - - forkWorkflows( - ctx, - List.of(new WorkflowFork(workflowId, forkedWorkflowId, startStep)), - options.applicationVersion(), - options.queueName(), - options.queuePartitionKey(), - options.timeout()); - return forkedWorkflowId; - } - /** * Drops a terminal workflow's history from {@code startStep} on and re-enqueues it under the same * ID, so a replay re-executes everything from that step. Unlike a fork, this writes no new * workflow: peers keep addressing the same ID. * - *

In one transaction, from {@code startStep} on, it: + *

On the caller's transaction, from {@code startStep} on, it: * *

    *
  • rolls each event published past the cut back to its last value from before the cut, or @@ -2192,162 +2170,152 @@ public static String forkWorkflow( * while being rewound */ public static void rewindWorkflow( - DbContext ctx, String workflowId, int startStep, RewindOptions options) throws SQLException { + Connection conn, String schema, String workflowId, int startStep, RewindOptions options) + throws SQLException { // Function IDs start at 0, so 0 is the whole history. if (startStep < 0) { throw new IllegalArgumentException("startStep must be >= 0, got " + startStep); } Objects.requireNonNull(options, "RewindOptions must not be null"); - var schema = ctx.schema(); - try (var txConn = ctx.getConnection()) { - SqlTransaction.run( - txConn, - conn -> { - var state = readRewindableState(conn, schema, workflowId); - - // Whether a key was published at or past the cut. %2$s is the key column the - // correlated subquery compares against. - var publishedPastCut = - """ - EXISTS ( - SELECT 1 FROM "%1$s".workflow_events_history discarded - WHERE discarded.workflow_uuid = ? AND discarded.key = %2$s - AND discarded.function_id >= ? - ) - """; - - // workflow_events_history is the undo log for workflow_events, so the events are - // rolled back before the history past the cut is deleted. First unpublish every key - // the discarded steps published... - var unpublishSql = - (""" - DELETE FROM "%1$s".workflow_events - WHERE workflow_uuid = ? AND """ - + publishedPastCut) - .formatted(schema, "\"%s\".workflow_events.key".formatted(schema)); - try (var stmt = conn.prepareStatement(unpublishSql)) { - stmt.setString(1, workflowId); - stmt.setString(2, workflowId); - stmt.setInt(3, startStep); - stmt.executeUpdate(); - } + var state = readRewindableState(conn, schema, workflowId); - // ...then restore those keys to the last value published before the cut, if any. - var restoreSql = - (""" - INSERT INTO "%1$s".workflow_events (workflow_uuid, key, value, serialization) - SELECT surviving.workflow_uuid, surviving.key, surviving.value, - surviving.serialization - FROM ( - SELECT weh.workflow_uuid, weh.key, weh.value, weh.serialization, - ROW_NUMBER() OVER (PARTITION BY weh.key ORDER BY weh.function_id DESC) - AS rn - FROM "%1$s".workflow_events_history weh - WHERE weh.workflow_uuid = ? AND weh.function_id < ? AND """ - + publishedPastCut - + """ - ) surviving - WHERE surviving.rn = 1 - """) - .formatted(schema, "weh.key"); - try (var stmt = conn.prepareStatement(restoreSql)) { - stmt.setString(1, workflowId); - stmt.setInt(2, startStep); - stmt.setString(3, workflowId); - stmt.setInt(4, startStep); - stmt.executeUpdate(); - } + // Whether a key was published at or past the cut. %2$s is the key column the + // correlated subquery compares against. + var publishedPastCut = + """ + EXISTS ( + SELECT 1 FROM "%1$s".workflow_events_history discarded + WHERE discarded.workflow_uuid = ? AND discarded.key = %2$s + AND discarded.function_id >= ? + ) + """; + + // workflow_events_history is the undo log for workflow_events, so the events are + // rolled back before the history past the cut is deleted. First unpublish every key + // the discarded steps published... + var unpublishSql = + (""" + DELETE FROM "%1$s".workflow_events + WHERE workflow_uuid = ? AND """ + + publishedPastCut) + .formatted(schema, "\"%s\".workflow_events.key".formatted(schema)); + try (var stmt = conn.prepareStatement(unpublishSql)) { + stmt.setString(1, workflowId); + stmt.setString(2, workflowId); + stmt.setInt(3, startStep); + stmt.executeUpdate(); + } - StreamsDAO.deleteCloseSentinels(conn, schema, workflowId, startStep); - - for (var table : List.of("operation_outputs", "workflow_events_history")) { - var deleteSql = - """ - DELETE FROM "%s".%s WHERE workflow_uuid = ? AND function_id >= ? - """ - .formatted(schema, table); - try (var stmt = conn.prepareStatement(deleteSql)) { - stmt.setString(1, workflowId); - stmt.setInt(2, startStep); - stmt.executeUpdate(); - } - } + // ...then restore those keys to the last value published before the cut, if any. + var restoreSql = + (""" + INSERT INTO "%1$s".workflow_events (workflow_uuid, key, value, serialization) + SELECT surviving.workflow_uuid, surviving.key, surviving.value, + surviving.serialization + FROM ( + SELECT weh.workflow_uuid, weh.key, weh.value, weh.serialization, + ROW_NUMBER() OVER (PARTITION BY weh.key ORDER BY weh.function_id DESC) + AS rn + FROM "%1$s".workflow_events_history weh + WHERE weh.workflow_uuid = ? AND weh.function_id < ? AND """ + + publishedPastCut + + """ + ) surviving + WHERE surviving.rn = 1 + """) + .formatted(schema, "weh.key"); + try (var stmt = conn.prepareStatement(restoreSql)) { + stmt.setString(1, workflowId); + stmt.setInt(2, startStep); + stmt.setString(3, workflowId); + stmt.setInt(4, startStep); + stmt.executeUpdate(); + } - var notificationsSql = - """ - DELETE FROM "%s".notifications - WHERE destination_uuid = ? - AND (consumed_by_function_id >= ? OR consumed = FALSE) - """ - .formatted(schema); - try (var stmt = conn.prepareStatement(notificationsSql)) { - stmt.setString(1, workflowId); - stmt.setInt(2, startStep); - stmt.executeUpdate(); - } + StreamsDAO.deleteCloseSentinels(conn, schema, workflowId, startStep); - var outputSql = - """ - DELETE FROM "%s".workflow_output WHERE workflow_uuid = ? - """ - .formatted(schema); - try (var stmt = conn.prepareStatement(outputSql)) { - stmt.setString(1, workflowId); - stmt.executeUpdate(); - } + for (var table : List.of("operation_outputs", "workflow_events_history")) { + var deleteSql = + """ + DELETE FROM "%s".%s WHERE workflow_uuid = ? AND function_id >= ? + """ + .formatted(schema, table); + try (var stmt = conn.prepareStatement(deleteSql)) { + stmt.setString(1, workflowId); + stmt.setInt(2, startStep); + stmt.executeUpdate(); + } + } - // Re-enqueue. The legacy output and error columns are cleared too: reads fall back to - // them when there is no workflow_output row, so a value left there would be returned - // as the rewound workflow's result. Re-asserting the status read above keeps a - // workflow that moved on underneath this transaction from being resurrected. - var setVersion = - options.applicationVersion() != null ? ", application_version = ?" : ""; - var enqueueSql = - """ - UPDATE "%1$s".workflow_status - SET status = ?, owner_xid = NULL, queue_name = ?, queue_partition_key = ?, - recovery_attempts = 0, workflow_deadline_epoch_ms = NULL, - deduplication_id = NULL, started_at_epoch_ms = NULL, completed_at = NULL, - output = NULL, error = NULL, updated_at = %2$s%3$s - WHERE workflow_uuid = ? AND status = ? - """ - .formatted(schema, SystemDatabase.NOW_EPOCH_MS, setVersion); - try (var stmt = conn.prepareStatement(enqueueSql)) { - int i = 1; - stmt.setString(i++, WorkflowState.ENQUEUED.name()); - stmt.setString( - i++, - Objects.requireNonNullElse(options.queueName(), Constants.DBOS_INTERNAL_QUEUE)); - stmt.setString(i++, options.queuePartitionKey()); - if (options.applicationVersion() != null) { - stmt.setString(i++, options.applicationVersion()); - } - stmt.setString(i++, workflowId); - stmt.setString(i++, state.name()); - if (stmt.executeUpdate() != 1) { - throw new IllegalStateException( - "Workflow %s changed status while being rewound; retry the rewind" - .formatted(workflowId)); - } - } - }); + var notificationsSql = + """ + DELETE FROM "%s".notifications + WHERE destination_uuid = ? + AND (consumed_by_function_id >= ? OR consumed = FALSE) + """ + .formatted(schema); + try (var stmt = conn.prepareStatement(notificationsSql)) { + stmt.setString(1, workflowId); + stmt.setInt(2, startStep); + stmt.executeUpdate(); + } + + var outputSql = + """ + DELETE FROM "%s".workflow_output WHERE workflow_uuid = ? + """ + .formatted(schema); + try (var stmt = conn.prepareStatement(outputSql)) { + stmt.setString(1, workflowId); + stmt.executeUpdate(); + } + + // Re-enqueue. The legacy output and error columns are cleared too: reads fall back to + // them when there is no workflow_output row, so a value left there would be returned + // as the rewound workflow's result. Re-asserting the status read above keeps a + // workflow that moved on underneath this transaction from being resurrected. + var setVersion = options.applicationVersion() != null ? ", application_version = ?" : ""; + var enqueueSql = + """ + UPDATE "%1$s".workflow_status + SET status = ?, owner_xid = NULL, queue_name = ?, queue_partition_key = ?, + recovery_attempts = 0, workflow_deadline_epoch_ms = NULL, + deduplication_id = NULL, started_at_epoch_ms = NULL, completed_at = NULL, + output = NULL, error = NULL, updated_at = %2$s%3$s + WHERE workflow_uuid = ? AND status = ? + """ + .formatted(schema, SystemDatabase.NOW_EPOCH_MS, setVersion); + try (var stmt = conn.prepareStatement(enqueueSql)) { + int i = 1; + stmt.setString(i++, WorkflowState.ENQUEUED.name()); + stmt.setString( + i++, Objects.requireNonNullElse(options.queueName(), Constants.DBOS_INTERNAL_QUEUE)); + stmt.setString(i++, options.queuePartitionKey()); + if (options.applicationVersion() != null) { + stmt.setString(i++, options.applicationVersion()); + } + stmt.setString(i++, workflowId); + stmt.setString(i++, state.name()); + if (stmt.executeUpdate() != 1) { + throw new IllegalStateException( + "Workflow %s changed status while being rewound; retry the rewind" + .formatted(workflowId)); + } } } /** * Checks that a workflow can be rewound, without changing anything: it must exist and be in a - * terminal state. A rewind repeats this check in its own transaction; this one lets a caller - * refuse before touching anything outside the system database. + * terminal state. A rewind repeats this check; this one lets a caller refuse before touching + * anything outside the system database. */ - public static void checkRewindable(DbContext ctx, String workflowId, int startStep) - throws SQLException { + public static void checkRewindable( + Connection conn, String schema, String workflowId, int startStep) throws SQLException { if (startStep < 0) { throw new IllegalArgumentException("startStep must be >= 0, got " + startStep); } - try (var conn = ctx.getConnection()) { - readRewindableState(conn, ctx.schema(), workflowId); - } + readRewindableState(conn, schema, workflowId); } private static WorkflowState readRewindableState( @@ -2366,11 +2334,12 @@ private static WorkflowState readRewindableState( } /** - * Forks workflows in one transaction, each under its fork's ID and from its fork's start step. - * Fails, writing nothing, if any of them does not exist. + * Forks workflows on the caller's transaction, each under its fork's ID and from its fork's start + * step. Fails if any of them does not exist. */ public static void forkWorkflows( DbContext ctx, + Connection conn, List forks, @Nullable String applicationVersion, @Nullable String queueName, @@ -2386,48 +2355,36 @@ public static void forkWorkflows( var timeoutMs = timeout != null ? timeout.toMillis() : null; final var forkQueueName = Objects.requireNonNullElse(queueName, Constants.DBOS_INTERNAL_QUEUE); - try (var txConn = ctx.getConnection()) { - SqlTransaction.run( - txConn, - conn -> { - var wfDataMap = fetchForkWorkflowData(conn, ctx.schema(), workflowIds); - for (String id : workflowIds) { - if (!wfDataMap.containsKey(id)) { - throw new DBOSNonExistentWorkflowException(id); - } - } - var rows = - forks.stream() - .map( - fork -> { - var original = wfDataMap.get(fork.workflowId()); - // One app name per fork, shared by its status row and its copied steps: - // the source's, or this application claiming an unclaimed one. - var appName = - original.applicationName() != null - ? original.applicationName() - : ctx.appName(); - return new ForkRow(fork, original, appName); - }) - .toList(); - - batchInsertForkedStatuses( - conn, - ctx.schema(), - rows, - applicationVersion, - forkQueueName, - queuePartitionKey, - timeoutMs); - - markWasForkedFrom(conn, ctx.schema(), workflowIds); - - // A fork from step 0 has nothing before its start step to copy. - var rowsToCopy = rows.stream().filter(row -> row.fork().startStep() > 0).toList(); - if (!rowsToCopy.isEmpty()) { - batchCopyWorkflowData(conn, ctx.schema(), rowsToCopy); - } - }); + var wfDataMap = fetchForkWorkflowData(conn, ctx.schema(), workflowIds); + for (String id : workflowIds) { + if (!wfDataMap.containsKey(id)) { + throw new DBOSNonExistentWorkflowException(id); + } + } + var rows = + forks.stream() + .map( + fork -> { + var original = wfDataMap.get(fork.workflowId()); + // One app name per fork, shared by its status row and its copied steps: + // the source's, or this application claiming an unclaimed one. + var appName = + original.applicationName() != null + ? original.applicationName() + : ctx.appName(); + return new ForkRow(fork, original, appName); + }) + .toList(); + + batchInsertForkedStatuses( + conn, ctx.schema(), rows, applicationVersion, forkQueueName, queuePartitionKey, timeoutMs); + + markWasForkedFrom(conn, ctx.schema(), workflowIds); + + // A fork from step 0 has nothing before its start step to copy. + var rowsToCopy = rows.stream().filter(row -> row.fork().startStep() > 0).toList(); + if (!rowsToCopy.isEmpty()) { + batchCopyWorkflowData(conn, ctx.schema(), rowsToCopy); } } 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 4b0347e43..c5dd9a42b 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1094,14 +1094,7 @@ public List> resumeWorkflows( public void updateWorkflowAttributes(String workflowId, Map attributes) { Objects.requireNonNull(workflowId); - this.runDbosFunctionAsStep( - () -> { - logger.info("Updating attributes of workflow: {}", workflowId); - systemDatabase.updateWorkflowAttributes(workflowId, attributes); - return null; // void - }, - "DBOS.updateWorkflowAttributes", - null); + systemDatabase.updateWorkflowAttributes(workflowId, attributes, stepCaller()); } /** Whether any transactional step factory has registered its checkpoints. */ @@ -1157,16 +1150,12 @@ public WorkflowHandle forkWorkflow( String workflowId, int startStep, ForkOptions options) { String forkedId = - this.runDbosFunctionAsStep( - () -> { - logger.info("Forking workflow:{} from step:{} ", workflowId, startStep); - - validateQueue(options.queueName(), options.queuePartitionKey()); - - return systemDatabase.forkWorkflow(workflowId, startStep, options); - }, - "DBOS.forkWorkflow", - null); + systemDatabase.forkWorkflow( + workflowId, + startStep, + options, + () -> validateQueue(options.queueName(), options.queuePartitionKey()), + stepCaller()); return retrieveWorkflow(forkedId); } @@ -1189,25 +1178,21 @@ public WorkflowHandle rewindWorkflow( Objects.requireNonNull(workflowId); Objects.requireNonNull(options); - this.runDbosFunctionAsStep( + // The step factories' checkpoints go first and the system database last. If a delete fails, + // the workflow keeps its terminal status and is not re-enqueued, so nothing runs against a + // partly cleared history, and since a delete is idempotent the rewind can simply be retried. + // The rewindable check runs first so a running workflow keeps its checkpoints. + systemDatabase.rewindWorkflow( + workflowId, + startStep, + options, () -> { - logger.info("Rewinding workflow: {} to step: {}", workflowId, startStep); - validateQueue(options.queueName(), options.queuePartitionKey()); - - // The step factories' checkpoints go first and the system database last. If a delete - // fails, the workflow keeps its terminal status and is not re-enqueued, so nothing runs - // against a partly cleared history, and since a delete is idempotent the rewind can - // simply be retried. The checks run first so a running workflow keeps its checkpoints. if (hasStepCheckpointStores()) { - systemDatabase.checkRewindable(workflowId, startStep); deleteStepCheckpoints(workflowId, startStep); } - systemDatabase.rewindWorkflow(workflowId, startStep, options); - return null; // void }, - "DBOS.rewindWorkflow", - null); + stepCaller()); return retrieveWorkflow(workflowId); } @@ -1218,14 +1203,7 @@ public WorkflowHandle retrieveWorkflow(String wor public void setWorkflowDelay(@NonNull String workflowId, @NonNull WorkflowDelay delay) { Objects.requireNonNull(workflowId); - this.runDbosFunctionAsStep( - () -> { - logger.info("setWorkflowDelay workflow {} delay {}", workflowId, delay); - systemDatabase.setWorkflowDelay(workflowId, delay); - return null; // void - }, - "DBOS.setWorkflowDelay", - null); + systemDatabase.setWorkflowDelay(workflowId, delay, stepCaller()); } public void writeStream(String key, Object value, SerializationStrategy serialization) { 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 9627f5fa2..dd2ab23aa 100644 --- a/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java @@ -1215,7 +1215,7 @@ public void testForkWorkflowWithAuthenticationFields() throws Exception { // Fork the workflow var forkOptions = new ForkOptions().withApplicationVersion("1.0.0"); - var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions); + var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions, null, null); assertNotNull(forkedWorkflowId); assertNotEquals(originalWorkflowId, forkedWorkflowId); @@ -1270,7 +1270,7 @@ public void testForkWorkflowWithNullAuthenticationFields() throws Exception { // Fork the workflow var forkOptions = new ForkOptions().withApplicationVersion("1.0.0"); - var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions); + var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions, null, null); assertNotNull(forkedWorkflowId); // Retrieve forked workflow and validate null authentication fields are preserved @@ -1325,7 +1325,7 @@ public void testForkWorkflowWithEmptyAuthenticatedRoles() throws Exception { // Fork the workflow var forkOptions = new ForkOptions().withApplicationVersion("1.0.0"); - var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions); + var forkedWorkflowId = sysdb.forkWorkflow(originalWorkflowId, 0, forkOptions, null, null); assertNotNull(forkedWorkflowId); assertNotEquals(originalWorkflowId, forkedWorkflowId); @@ -1705,7 +1705,7 @@ public void testSetWorkflowDelayWithDuration() throws Exception { null); long before = System.currentTimeMillis(); - sysdb.setWorkflowDelay(wfid, new WorkflowDelay.Delay(Duration.ofSeconds(30))); + sysdb.setWorkflowDelay(wfid, new WorkflowDelay.Delay(Duration.ofSeconds(30)), null); var row = DBUtils.getWorkflowRow(dataSource, wfid); assertNotNull(row.delayUntilEpochMs()); @@ -1725,7 +1725,7 @@ public void testSetWorkflowDelayWithInstant() throws Exception { null); var targetInstant = Instant.now().plusSeconds(120); - sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(targetInstant)); + sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(targetInstant), null); var row = DBUtils.getWorkflowRow(dataSource, wfid); assertEquals(targetInstant.toEpochMilli(), row.delayUntilEpochMs()); @@ -1747,7 +1747,7 @@ public void testSetWorkflowDelayIgnoresNonDelayedWorkflow() throws Exception { WorkflowStatusInternalBuilder.create(wfid).queueName("test-queue").build(), 5, null); DBUtils.setWorkflowState(dataSource, wfid, state.name()); - sysdb.setWorkflowDelay(wfid, targetDelay); + sysdb.setWorkflowDelay(wfid, targetDelay, null); var row = DBUtils.getWorkflowRow(dataSource, wfid); assertNull(row.delayUntilEpochMs(), "Expected no delay for status " + state); @@ -1765,7 +1765,7 @@ public void testTransitionDelayedWorkflowsPastDeadline() throws Exception { 5, null); - sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(Instant.now().minusSeconds(5))); + sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(Instant.now().minusSeconds(5)), null); sysdb.transitionDelayedWorkflows(); assertEquals(WorkflowState.ENQUEUED.name(), DBUtils.getWorkflowRow(dataSource, wfid).status()); @@ -1782,7 +1782,7 @@ public void testTransitionDelayedWorkflowsFutureDeadlineNotTransitioned() throws 5, null); - sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(Instant.now().plusSeconds(60))); + sysdb.setWorkflowDelay(wfid, new WorkflowDelay.DelayUntil(Instant.now().plusSeconds(60)), null); sysdb.transitionDelayedWorkflows(); assertEquals(WorkflowState.DELAYED.name(), DBUtils.getWorkflowRow(dataSource, wfid).status()); @@ -1803,8 +1803,10 @@ public void testTransitionDelayedWorkflowsMixed() throws Exception { null); } - sysdb.setWorkflowDelay(pastWfid, new WorkflowDelay.DelayUntil(Instant.now().minusSeconds(5))); - sysdb.setWorkflowDelay(futureWfid, new WorkflowDelay.DelayUntil(Instant.now().plusSeconds(60))); + sysdb.setWorkflowDelay( + pastWfid, new WorkflowDelay.DelayUntil(Instant.now().minusSeconds(5)), null); + sysdb.setWorkflowDelay( + futureWfid, new WorkflowDelay.DelayUntil(Instant.now().plusSeconds(60)), null); sysdb.transitionDelayedWorkflows(); diff --git a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java index 796fd532e..a5f96e6b9 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -3,6 +3,7 @@ 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.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import dev.dbos.transact.DBOS; @@ -175,6 +176,74 @@ public void aLostCommitAcknowledgementReplaysTheCheckpoint() throws Exception { assertTrue(dbos.getWorkflowStatus(deletedId).isEmpty(), "deleted"); } + private List forksOf(String workflowId) { + return dbos.listWorkflows(new ListWorkflowsInput().withForkedFrom(workflowId)); + } + + @Test + public void aForkThatFailsBeforeCommitLeavesNoFork() throws Exception { + var sourceId = "source-%d".formatted(System.nanoTime()); + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(sourceId)).getResult(); + 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.forks(sourceId), new StartWorkflowOptions(callerId)); + + assertThrows(RuntimeException.class, caller::getResult); + assertTrue(fired.get()); + assertTrue(forksOf(sourceId).isEmpty(), "the fork rolled back"); + assertTrue(stepsNamed(callerId, "DBOS.forkWorkflow").isEmpty(), "and so did its checkpoint"); + } + + @Test + public void aLostCommitAcknowledgementReplaysTheFork() throws Exception { + var sourceId = "source-%d".formatted(System.nanoTime()); + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(sourceId)).getResult(); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED, + callerId, + new SQLTransientException()); + + var caller = + dbos.startWorkflow(() -> proxy.forks(sourceId), new StartWorkflowOptions(callerId)); + + var forkedId = caller.getResult(); + assertTrue(fired.get()); + var forks = forksOf(sourceId); + assertEquals(1, forks.size(), "the retry found its checkpoint and did not fork again"); + assertEquals(forkedId, forks.get(0).workflowId()); + assertEquals(1, stepsNamed(callerId, "DBOS.forkWorkflow").size()); + } + + @Test + public void aLostCommitAcknowledgementDoesNotRewindAgain() throws Exception { + var targetId = "target-%d".formatted(System.nanoTime()); + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(targetId)).getResult(); + var callerId = "caller-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED, + callerId, + new SQLTransientException()); + + // A second rewind would be refused: the first left the target ENQUEUED. + var caller = + dbos.startWorkflow(() -> proxy.rewinds(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("done", caller.getResult()); + assertTrue(fired.get()); + assertEquals(1, stepsNamed(callerId, "DBOS.rewindWorkflow").size()); + assertNull(stepsNamed(callerId, "DBOS.rewindWorkflow").get(0).error()); + assertEquals("quick", dbos.retrieveWorkflow(targetId).getResult(), "the target ran again"); + } + /** * 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. @@ -319,6 +388,10 @@ interface InternalStepService { String cancelsFromStep(String targetId); + String forks(String sourceId); + + String rewinds(String targetId); + String startsChild(String childId) throws Exception; String startsChildWhenReleased(String childId) throws Exception; @@ -401,6 +474,19 @@ public String startsChildWhenReleased(String childId) throws Exception { return "started"; } + @Override + @Workflow + public String forks(String sourceId) { + return dbos.forkWorkflow(sourceId, 0).workflowId(); + } + + @Override + @Workflow + public String rewinds(String targetId) { + dbos.rewindWorkflow(targetId, 0); + return "done"; + } + @Override @Workflow public String cancelsFromStep(String targetId) { diff --git a/transact/src/test/java/dev/dbos/transact/workflow/PayloadTableReadTest.java b/transact/src/test/java/dev/dbos/transact/workflow/PayloadTableReadTest.java index 1f3bdc356..ecbf8c086 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/PayloadTableReadTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/PayloadTableReadTest.java @@ -189,7 +189,7 @@ private void assertEveryReadSees(String workflowId, int expected) throws Excepti // The fork reads the original's inputs and copies them to the new workflow: reading the child // back shows what the fork saw. - var forkedId = systemDatabase.forkWorkflow(workflowId, 0, new ForkOptions()); + var forkedId = systemDatabase.forkWorkflow(workflowId, 0, new ForkOptions(), null, null); assertEquals( expected, (int) (Integer) systemDatabase.getWorkflowStatus(forkedId).input()[0], From a30f09f4fcef6af951d1eb4bed4d483cee14d28a Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 10:14:09 -0700 Subject: [PATCH 2/5] Say a rewind's step-checkpoint deletes can borrow a second pooled connection --- .../java/dev/dbos/transact/database/SystemDatabase.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index 5129359f9..29ab0c73f 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1961,10 +1961,11 @@ public void forkWorkflows( * *

    {@code beforeRewind}, if given, runs inside the step once the workflow is known to be * rewindable, and before its system database history is touched. The executor validates the queue - * there and clears the transactional step factories' checkpoints, which live in other databases - * and so commit on their own: a rewind that then rolls back leaves them cleared, and the workflow - * terminal and not re-enqueued, so nothing runs against the partial history, and since the - * deletes are idempotent the rewind can be retried. + * there and clears the transactional step factories' checkpoints. Those deletes go through the + * factories' own connections, a second one from this pool if a factory shares it, and so commit + * on their own: a rewind that then rolls back leaves them cleared, and the workflow terminal and + * not re-enqueued, so nothing runs against the partial history, and since the deletes are + * idempotent the rewind can be retried. */ public void rewindWorkflow( String workflowId, From d1dc416c58e3d9330bcb04df503ec13512a60df4 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 11:22:38 -0700 Subject: [PATCH 3/5] Resolve a workflow delay once, and pin delay, attributes and rewind counts setWorkflowDelay resolved a relative delay inside the step's work, so every retry of the transaction pushed it later by that attempt's backoff, contradicting resolveDelayUntil's own contract. It is now resolved once, before the transaction; a delay that cannot be resolved is thrown from inside the work, so the refusal is still recorded as the step's error. The rewind comments no longer claim nothing can run against a partly cleared history: a rewind that rolls back after the step factories' deletes leaves the workflow terminal, and a resume would still run it without those checkpoints. New tests: attributes and a delay whose step fails before commit change nothing and record nothing, and the lost-acknowledgement rewind test counts the target's runs instead of relying on its status. --- .../transact/database/SystemDatabase.java | 48 ++++++--- .../dbos/transact/execution/DBOSExecutor.java | 6 +- .../workflow/InternalStepTransactionTest.java | 101 +++++++++++++++++- 3 files changed, 135 insertions(+), 20 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index dde92ea04..70c382e26 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1972,9 +1972,9 @@ public void forkWorkflows( * rewindable, and before its system database history is touched. The executor validates the queue * there and clears the transactional step factories' checkpoints. Those deletes go through the * factories' own connections, a second one from this pool if a factory shares it, and so commit - * on their own: a rewind that then rolls back leaves them cleared, and the workflow terminal and - * not re-enqueued, so nothing runs against the partial history, and since the deletes are - * idempotent the rewind can be retried. + * on their own: a rewind that then rolls back leaves them cleared and the workflow terminal and + * not re-enqueued. Retrying the rewind is safe, as the deletes are idempotent; resuming the + * workflow instead would run it without those checkpoints. */ public void rewindWorkflow( String workflowId, @@ -2049,24 +2049,29 @@ public void garbageCollect(Instant cutoff, Long rowsThreshold, int batchSize) { /** * Moves a DELAYED workflow's start as the step {@code caller} makes; see {@link - * #runTransactionalStep}. A relative delay is resolved in the work, so it runs from the attempt - * that commits, as a started workflow's deadline does. + * #runTransactionalStep}. A relative delay is resolved once, before the transaction and its + * retries, as {@link #resolveDelayUntil(Duration)} requires. A delay that cannot be resolved is + * refused inside the step, so the refusal is recorded as the step's error. */ public void setWorkflowDelay( String workflowId, WorkflowDelay delay, @Nullable StepCaller caller) { + Long resolved; + RuntimeException refused; + try { + resolved = resolveDelayUntil(delay); + refused = null; + } catch (RuntimeException e) { + resolved = null; + refused = e; + } + final Long delayUntilEpochMs = resolved; + final RuntimeException refusal = refused; runTransactionalStep( caller, "DBOS.setWorkflowDelay", conn -> { - Objects.requireNonNull(delay, "delay must not be null"); - long delayUntilEpochMs; - if (delay instanceof WorkflowDelay.Delay d) { - delayUntilEpochMs = resolveDelayUntil(d.delay()); - } else if (delay instanceof WorkflowDelay.DelayUntil du) { - // An absolute delay is the caller's instant, taken as given. - delayUntilEpochMs = du.delayUntil().toEpochMilli(); - } else { - throw new IllegalArgumentException("Unexpected WorkflowDelay value"); + if (refusal != null) { + throw refusal; } WorkflowDAO.setWorkflowDelay(conn, ctx.schema(), workflowId, delayUntilEpochMs); logger.info("Delaying workflow {} until {}", workflowId, delayUntilEpochMs); @@ -2074,6 +2079,21 @@ public void setWorkflowDelay( }); } + /** + * The absolute end of {@code delay}: a relative one from this JVM's clock, an absolute as given. + */ + private static long resolveDelayUntil(WorkflowDelay delay) { + Objects.requireNonNull(delay, "delay must not be null"); + if (delay instanceof WorkflowDelay.Delay d) { + return resolveDelayUntil(d.delay()); + } + if (delay instanceof WorkflowDelay.DelayUntil du) { + // An absolute delay is the caller's instant, taken as given. + return du.delayUntil().toEpochMilli(); + } + throw new IllegalArgumentException("Unexpected WorkflowDelay value"); + } + /** See {@link WorkflowDAO#cancelTimedOutWorkflows}. */ public List cancelTimedOutWorkflows(int limit) { return dbRetry(() -> WorkflowDAO.cancelTimedOutWorkflows(ctx, limit)); 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 c5dd9a42b..bc39ef5db 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1179,9 +1179,9 @@ public WorkflowHandle rewindWorkflow( Objects.requireNonNull(options); // The step factories' checkpoints go first and the system database last. If a delete fails, - // the workflow keeps its terminal status and is not re-enqueued, so nothing runs against a - // partly cleared history, and since a delete is idempotent the rewind can simply be retried. - // The rewindable check runs first so a running workflow keeps its checkpoints. + // the workflow keeps its terminal status and is not re-enqueued, and since a delete is + // idempotent the rewind can simply be retried. The rewindable check runs first so a running + // workflow keeps its checkpoints. systemDatabase.rewindWorkflow( workflowId, startStep, diff --git a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java index a5f96e6b9..bb791e777 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -15,11 +15,15 @@ import java.sql.SQLException; import java.sql.SQLTransientException; +import java.time.Duration; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; 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; @@ -51,8 +55,11 @@ void beforeEach() { proxy = dbos.registerProxy(InternalStepService.class, impl); impl.self = proxy; dbos.launch(); + dbos.registerQueue(DELAYED_QUEUE, new QueueOptions()); } + private static final String DELAYED_QUEUE = "internal-step-delayed"; + @AfterEach void afterEach() { DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT); @@ -225,7 +232,7 @@ public void aLostCommitAcknowledgementReplaysTheFork() throws Exception { @Test public void aLostCommitAcknowledgementDoesNotRewindAgain() throws Exception { var targetId = "target-%d".formatted(System.nanoTime()); - dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(targetId)).getResult(); + dbos.startWorkflow(proxy::counted, new StartWorkflowOptions(targetId)).getResult(); var callerId = "caller-%d".formatted(System.nanoTime()); var fired = failOnce( @@ -233,7 +240,7 @@ public void aLostCommitAcknowledgementDoesNotRewindAgain() throws Exception { callerId, new SQLTransientException()); - // A second rewind would be refused: the first left the target ENQUEUED. + // A second rewind would run the target a third time, or be refused if it were still ENQUEUED. var caller = dbos.startWorkflow(() -> proxy.rewinds(targetId), new StartWorkflowOptions(callerId)); @@ -241,7 +248,54 @@ public void aLostCommitAcknowledgementDoesNotRewindAgain() throws Exception { assertTrue(fired.get()); assertEquals(1, stepsNamed(callerId, "DBOS.rewindWorkflow").size()); assertNull(stepsNamed(callerId, "DBOS.rewindWorkflow").get(0).error()); - assertEquals("quick", dbos.retrieveWorkflow(targetId).getResult(), "the target ran again"); + assertEquals("counted", dbos.retrieveWorkflow(targetId).getResult()); + assertEquals(2, impl.runsOf(targetId), "the target ran once, then once more for one rewind"); + } + + @Test + public void attributesThatFailBeforeCommitAreNotSet() 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")); + + var caller = + dbos.startWorkflow(() -> proxy.labels(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("failed", caller.getResult()); + assertTrue(fired.get()); + assertNull(dbos.retrieveWorkflow(targetId).getStatus().attributes(), "the update rolled back"); + assertTrue(stepsNamed(callerId, "DBOS.updateWorkflowAttributes").isEmpty()); + } + + @Test + public void aDelayThatFailsBeforeCommitIsNotSet() throws Exception { + var targetId = "delayed-%d".formatted(System.nanoTime()); + dbos.startWorkflow( + proxy::quick, + new StartWorkflowOptions(targetId).withQueue(DELAYED_QUEUE).withDelay(Duration.ofDays(1))); + var before = dbos.retrieveWorkflow(targetId).getStatus(); + assertEquals(WorkflowState.DELAYED, before.status()); + 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.delays(targetId), new StartWorkflowOptions(callerId)); + + assertEquals("failed", caller.getResult()); + assertTrue(fired.get()); + assertEquals( + before.delayUntil(), + dbos.retrieveWorkflow(targetId).getStatus().delayUntil(), + "the delay rolled back"); + assertTrue(stepsNamed(callerId, "DBOS.setWorkflowDelay").isEmpty()); } /** @@ -392,6 +446,12 @@ interface InternalStepService { String rewinds(String targetId); + String counted(); + + String labels(String targetId); + + String delays(String targetId); + String startsChild(String childId) throws Exception; String startsChildWhenReleased(String childId) throws Exception; @@ -405,6 +465,7 @@ class InternalStepServiceImpl implements InternalStepService { final CountDownLatch parentWaiting = new CountDownLatch(1); final CountDownLatch startChild = new CountDownLatch(1); final CountDownLatch childAttempted = new CountDownLatch(1); + private final Map runs = new ConcurrentHashMap<>(); InternalStepServiceImpl(DBOS dbos) { this.dbos = dbos; @@ -480,6 +541,40 @@ public String forks(String sourceId) { return dbos.forkWorkflow(sourceId, 0).workflowId(); } + @Override + @Workflow + public String counted() { + runs.computeIfAbsent(DBOSContext.workflowId(), id -> new AtomicInteger()).incrementAndGet(); + return "counted"; + } + + int runsOf(String workflowId) { + var count = runs.get(workflowId); + return count == null ? 0 : count.get(); + } + + @Override + @Workflow + public String labels(String targetId) { + try { + dbos.updateWorkflowAttributes(targetId, Map.of("label", "set")); + return "set"; + } catch (RuntimeException e) { + return "failed"; + } + } + + @Override + @Workflow + public String delays(String targetId) { + try { + dbos.setWorkflowDelay(targetId, Duration.ofDays(2)); + return "set"; + } catch (RuntimeException e) { + return "failed"; + } + } + @Override @Workflow public String rewinds(String targetId) { From 92b45e7f2c0942664d8dec3b909dcbcf14baa60d Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 11:40:37 -0700 Subject: [PATCH 4/5] Delete a rewind's step-factory checkpoints before its transaction The transactional step factories' checkpoints were deleted inside the rewind step's transaction. A factory sharing the system database's pool then asked that pool for a second connection while the step held one, so enough concurrent rewinds could exhaust it, and a rewind from a workflow would record the pool timeout as its permanent error. The deletes now run before the transaction opens, as they did before the rewind became a transactional step, with the queue and rewindable checks ahead of them. A step already recorded skips them, so a replay does not delete again; one recorded by a rival meanwhile is found by the step and replayed. A refusal reached before the transaction is thrown from inside the step, so it is still recorded as the step's error. RewindTest now counts the deletes and checks that recovering a workflow that rewound another does not repeat them. --- .../transact/database/SystemDatabase.java | 31 +++++++++----- .../dbos/transact/execution/DBOSExecutor.java | 41 ++++++++++++------- .../dbos/transact/workflow/RewindTest.java | 8 +++- 3 files changed, 53 insertions(+), 27 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index 70c382e26..b06dd4e93 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1968,27 +1968,23 @@ public void forkWorkflows( * Rewinds a terminal workflow to {@code startStep} as the step {@code caller} makes; see {@link * #runTransactionalStep}. * - *

    {@code beforeRewind}, if given, runs inside the step once the workflow is known to be - * rewindable, and before its system database history is touched. The executor validates the queue - * there and clears the transactional step factories' checkpoints. Those deletes go through the - * factories' own connections, a second one from this pool if a factory shares it, and so commit - * on their own: a rewind that then rolls back leaves them cleared and the workflow terminal and - * not re-enqueued. Retrying the rewind is safe, as the deletes are idempotent; resuming the - * workflow instead would run it without those checkpoints. + *

    {@code refusal}, if given, is a refusal the caller reached before the transaction, while + * checking the call or clearing what lives outside the system database. It is thrown from inside + * the step, so it is recorded as the step's error, unless the step is already recorded, which + * then replays. */ public void rewindWorkflow( String workflowId, int startStep, RewindOptions options, - @Nullable Runnable beforeRewind, + @Nullable RuntimeException refusal, @Nullable StepCaller caller) { runTransactionalStep( caller, "DBOS.rewindWorkflow", conn -> { - if (beforeRewind != null) { - WorkflowDAO.checkRewindable(conn, ctx.schema(), workflowId, startStep); - beforeRewind.run(); + if (refusal != null) { + throw refusal; } WorkflowDAO.rewindWorkflow(conn, ctx.schema(), workflowId, startStep, options); logger.info("Rewinding workflow {} to step {}", workflowId, startStep); @@ -1996,6 +1992,19 @@ public void rewindWorkflow( }); } + /** + * Checks that a workflow can be rewound, without changing anything; see {@link + * WorkflowDAO#checkRewindable}. + */ + public void checkRewindable(String workflowId, int startStep) { + dbRetry( + () -> { + try (var conn = ctx.getConnection()) { + WorkflowDAO.checkRewindable(conn, ctx.schema(), workflowId, startStep); + } + }); + } + public void createApplicationVersion(String versionName, @Nullable String applicationName) { dbRetryIncludingSerializationError( "createApplicationVersion", 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 bc39ef5db..44429a608 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1178,21 +1178,32 @@ public WorkflowHandle rewindWorkflow( Objects.requireNonNull(workflowId); Objects.requireNonNull(options); - // The step factories' checkpoints go first and the system database last. If a delete fails, - // the workflow keeps its terminal status and is not re-enqueued, and since a delete is - // idempotent the rewind can simply be retried. The rewindable check runs first so a running - // workflow keeps its checkpoints. - systemDatabase.rewindWorkflow( - workflowId, - startStep, - options, - () -> { - validateQueue(options.queueName(), options.queuePartitionKey()); - if (hasStepCheckpointStores()) { - deleteStepCheckpoints(workflowId, startStep); - } - }, - stepCaller()); + var caller = stepCaller(); + // The step factories' checkpoints go first, before the step's transaction opens, and the + // system database last, in that transaction. Deleting them inside it would hold one pooled + // connection while a factory sharing the pool asked for another. A step already recorded is + // replayed, so its deletes are not repeated against a workflow the rewind has since restarted; + // one recorded by a rival meanwhile is found by the step itself, which replays it instead of + // throwing the refusal below. If a delete fails, the workflow keeps its terminal status and is + // not re-enqueued, and since a delete is idempotent the rewind can simply be retried. The + // rewindable check runs first so a running workflow keeps its checkpoints. + RuntimeException refusal = null; + if (!(caller instanceof StepCaller.FromWorkflow step + && systemDatabase.checkStepResult( + step.workflowId(), step.stepId(), "DBOS.rewindWorkflow", step.ownerXid()) + != null)) { + try { + validateQueue(options.queueName(), options.queuePartitionKey()); + if (hasStepCheckpointStores()) { + systemDatabase.checkRewindable(workflowId, startStep); + deleteStepCheckpoints(workflowId, startStep); + } + } catch (RuntimeException e) { + // Thrown from inside the step, so it is recorded as the step's error. + refusal = e; + } + } + systemDatabase.rewindWorkflow(workflowId, startStep, options, refusal, caller); return retrieveWorkflow(workflowId); } diff --git a/transact/src/test/java/dev/dbos/transact/workflow/RewindTest.java b/transact/src/test/java/dev/dbos/transact/workflow/RewindTest.java index b0836ba80..dc3908f8b 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/RewindTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/RewindTest.java @@ -29,6 +29,7 @@ import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import com.zaxxer.hikari.HikariConfig; @@ -57,6 +58,8 @@ public class RewindTest { // When set, the extra checkpoint store registered in beforeEach fails its delete. private final AtomicBoolean failCheckpointDelete = new AtomicBoolean(false); + // How many times a rewind has asked the extra checkpoint store to delete. + private final AtomicInteger checkpointDeletes = new AtomicInteger(); @BeforeEach void beforeEach() throws SQLException { @@ -86,6 +89,7 @@ void beforeEach() throws SQLException { new StepCheckpointStore() { @Override public void deleteCheckpoints(String workflowId, int fromStepId) throws SQLException { + checkpointDeletes.incrementAndGet(); if (failCheckpointDelete.get()) { throw new SQLException("datasource down"); } @@ -633,6 +637,7 @@ public void rewindFromInsideAWorkflowIsCheckpointed() throws Exception { assertEquals(2, dbos.retrieveWorkflow(repairerId).getResult()); assertEquals(2, impl.runsOf("repaired")); stepIdOf(repairerId, "DBOS.rewindWorkflow", 0); + assertEquals(1, checkpointDeletes.get()); // Crash and recover the repairer with its checkpoints intact. A second rewind would either // run the target again or be refused, because the first left it ENQUEUED. @@ -642,9 +647,10 @@ public void rewindFromInsideAWorkflowIsCheckpointed() throws Exception { assertTrue(recovered.contains(repairerId), "repairer was not recovered"); assertEquals(2, dbos.retrieveWorkflow(repairerId).getResult()); - // The repairer's body ran again, but the rewind did not. + // The repairer's body ran again, but the rewind did not, nor its checkpoint deletes. assertEquals(2, impl.runsOf("repairer")); assertEquals(2, impl.runsOf("repaired")); + assertEquals(1, checkpointDeletes.get(), "the replay deleted the target's checkpoints again"); } @Test From db04138384f1d2ebc37ce5500de3731aed82760d Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 11:48:04 -0700 Subject: [PATCH 5/5] Read a rewind's pre-checks in one statement Before its step's transaction, a rewind asked whether the step was already recorded, which took two round trips through the full step check, and then read the target's status in a third. One statement now reads whether the calling step has a row, whether the caller still owns its workflow, and the target's status. The full step check still runs in the step's transaction, which replays a recorded step and refuses a caller that has lost its workflow; the pre-check only decides whether to clear the step factories' checkpoints first, which a caller that has lost its workflow now leaves alone. --- .../transact/database/SystemDatabase.java | 36 ++++++++--- .../transact/database/dao/WorkflowDAO.java | 64 +++++++++++++++++-- .../dbos/transact/execution/DBOSExecutor.java | 13 ++-- 3 files changed, 91 insertions(+), 22 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index b06dd4e93..fab374516 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -37,6 +37,7 @@ import dev.dbos.transact.workflow.WorkflowDelay; import dev.dbos.transact.workflow.WorkflowIdReusePolicy; import dev.dbos.transact.workflow.WorkflowSchedule; +import dev.dbos.transact.workflow.WorkflowState; import dev.dbos.transact.workflow.WorkflowStatus; import dev.dbos.transact.workflow.internal.StepResult; import dev.dbos.transact.workflow.internal.WorkflowStatusInternal; @@ -1993,16 +1994,33 @@ public void rewindWorkflow( } /** - * Checks that a workflow can be rewound, without changing anything; see {@link - * WorkflowDAO#checkRewindable}. + * What a rewind of {@code workflowId} by {@code caller} must know before its step's transaction, + * in one round trip: whether the step would run its work, rather than replay its checkpoint or be + * refused because the caller has lost its workflow, and the status of the workflow to rewind. */ - public void checkRewindable(String workflowId, int startStep) { - dbRetry( - () -> { - try (var conn = ctx.getConnection()) { - WorkflowDAO.checkRewindable(conn, ctx.schema(), workflowId, startStep); - } - }); + public record RewindPrecheck(boolean runsWork, @Nullable WorkflowState targetState) { + /** Refuses the rewind unless the workflow exists and is in a terminal state. */ + public void checkRewindable(String workflowId, int startStep) { + WorkflowDAO.checkRewindable(workflowId, startStep, targetState); + } + } + + public RewindPrecheck rewindPrecheck(@Nullable StepCaller caller, String workflowId) { + var read = + dbRetry( + () -> { + try (var conn = ctx.getConnection()) { + return WorkflowDAO.rewindPrecheck( + conn, + ctx.schema(), + caller == null ? null : caller.workflowId(), + caller instanceof StepCaller.FromWorkflow step ? step.stepId() : -1, + caller == null ? null : caller.ownerXid(), + workflowId); + } + }); + boolean runsWork = caller == null || (!read.recorded() && read.callerOwns()); + return new RewindPrecheck(runsWork, read.targetState()); } public void createApplicationVersion(String versionName, @Nullable String applicationName) { 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 43bfe54e3..1e1116b33 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 @@ -2337,21 +2337,74 @@ public static void rewindWorkflow( } /** - * Checks that a workflow can be rewound, without changing anything: it must exist and be in a - * terminal state. A rewind repeats this check; this one lets a caller refuse before touching - * anything outside the system database. + * What a rewind finds before its step's transaction, read in one statement: whether the calling + * step is already recorded, whether the caller still owns its workflow, and the status of the + * workflow to rewind, null if it does not exist. + */ + public record RewindPrecheck( + boolean recorded, boolean callerOwns, @Nullable WorkflowState targetState) {} + + /** + * Reads a {@link RewindPrecheck} for a rewind of {@code workflowId} made as step {@code stepId} + * of {@code callerWorkflowId} by the execution holding {@code ownerXid}. A caller with no step of + * its own passes a negative {@code stepId}, which no step has. Changes nothing. + */ + public static RewindPrecheck rewindPrecheck( + Connection conn, + String schema, + String callerWorkflowId, + int stepId, + String ownerXid, + String workflowId) + throws SQLException { + var sql = + """ + SELECT + EXISTS (SELECT 1 FROM "%1$s".operation_outputs + WHERE workflow_uuid = ? AND function_id = ?) AS recorded, + EXISTS (SELECT 1 FROM "%1$s".workflow_status + WHERE workflow_uuid = ? AND owner_xid = ?) AS caller_owns, + (SELECT status FROM "%1$s".workflow_status WHERE workflow_uuid = ?) AS target_status + """ + .formatted(schema); + try (var stmt = conn.prepareStatement(sql)) { + stmt.setString(1, callerWorkflowId); + stmt.setInt(2, stepId); + stmt.setString(3, callerWorkflowId); + stmt.setString(4, ownerXid); + stmt.setString(5, workflowId); + try (var rs = stmt.executeQuery()) { + rs.next(); + var status = rs.getString("target_status"); + return new RewindPrecheck( + rs.getBoolean("recorded"), + rs.getBoolean("caller_owns"), + status == null ? null : WorkflowState.valueOf(status)); + } + } + } + + /** + * Refuses a rewind from {@code startStep} of a workflow whose status is {@code state}, null if it + * does not exist, unless it exists and is in a terminal state. A rewind repeats this check; a + * caller checks first to refuse before touching anything outside the system database. */ public static void checkRewindable( - Connection conn, String schema, String workflowId, int startStep) throws SQLException { + String workflowId, int startStep, @Nullable WorkflowState state) { if (startStep < 0) { throw new IllegalArgumentException("startStep must be >= 0, got " + startStep); } - readRewindableState(conn, schema, workflowId); + refuseUnlessTerminal(workflowId, state); } private static WorkflowState readRewindableState( Connection conn, String schema, String workflowId) throws SQLException { var state = getWorkflowState(conn, schema, workflowId); + refuseUnlessTerminal(workflowId, state); + return state; + } + + private static void refuseUnlessTerminal(String workflowId, @Nullable WorkflowState state) { if (state == null) { throw new DBOSNonExistentWorkflowException(workflowId); } @@ -2361,7 +2414,6 @@ private static WorkflowState readRewindableState( + " first") .formatted(workflowId, state)); } - return state; } /** 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 44429a608..7313d95b9 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1182,20 +1182,19 @@ public WorkflowHandle rewindWorkflow( // The step factories' checkpoints go first, before the step's transaction opens, and the // system database last, in that transaction. Deleting them inside it would hold one pooled // connection while a factory sharing the pool asked for another. A step already recorded is - // replayed, so its deletes are not repeated against a workflow the rewind has since restarted; - // one recorded by a rival meanwhile is found by the step itself, which replays it instead of + // replayed, so its deletes are not repeated against a workflow the rewind has since restarted, + // and a caller that has lost its workflow leaves them alone for its step to refuse. One + // recorded by a rival meanwhile is found by the step itself, which replays it instead of // throwing the refusal below. If a delete fails, the workflow keeps its terminal status and is // not re-enqueued, and since a delete is idempotent the rewind can simply be retried. The // rewindable check runs first so a running workflow keeps its checkpoints. + var precheck = systemDatabase.rewindPrecheck(caller, workflowId); RuntimeException refusal = null; - if (!(caller instanceof StepCaller.FromWorkflow step - && systemDatabase.checkStepResult( - step.workflowId(), step.stepId(), "DBOS.rewindWorkflow", step.ownerXid()) - != null)) { + if (precheck.runsWork()) { try { validateQueue(options.queueName(), options.queuePartitionKey()); if (hasStepCheckpointStores()) { - systemDatabase.checkRewindable(workflowId, startStep); + precheck.checkRewindable(workflowId, startStep); deleteStepCheckpoints(workflowId, startStep); } } catch (RuntimeException e) {