diff --git a/transact/src/main/java/dev/dbos/transact/DBOSClient.java b/transact/src/main/java/dev/dbos/transact/DBOSClient.java index 4abd4f50..62cbff1f 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 2b22a7f7..ed0a49e9 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; @@ -1694,9 +1695,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; + }); } /** @@ -1903,10 +1916,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( @@ -1914,22 +1964,72 @@ 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)); + /** + * Rewinds a terminal workflow to {@code startStep} as the step {@code caller} makes; see {@link + * #runTransactionalStep}. + * + *

{@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 RuntimeException refusal, + @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.rewindWorkflow", + conn -> { + if (refusal != null) { + throw refusal; + } + WorkflowDAO.rewindWorkflow(conn, ctx.schema(), workflowId, startStep, options); + logger.info("Rewinding workflow {} to step {}", workflowId, startStep); + return null; + }); } - public void checkRewindable(String workflowId, int startStep) { - dbRetry(() -> WorkflowDAO.checkRewindable(ctx, workflowId, startStep)); + /** + * 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 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) { @@ -1983,18 +2083,51 @@ public void garbageCollect(Instant cutoff, Long rowsThreshold, int batchSize) { dbRetry(() -> WorkflowDAO.runRetentionRound(ctx, cutoff, rowsThreshold, batchSize)); } - public void setWorkflowDelay(String workflowId, WorkflowDelay delay) { + /** + * Moves a DELAYED workflow's start as the step {@code caller} makes; see {@link + * #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 -> { + if (refusal != null) { + throw refusal; + } + WorkflowDAO.setWorkflowDelay(conn, ctx.schema(), workflowId, delayUntilEpochMs); + logger.info("Delaying workflow {} until {}", workflowId, delayUntilEpochMs); + return null; + }); + } + + /** + * 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"); - long delayUntilEpochMs; if (delay instanceof WorkflowDelay.Delay d) { - delayUntilEpochMs = resolveDelayUntil(d.delay()); - } else if (delay instanceof WorkflowDelay.DelayUntil du) { + return resolveDelayUntil(d.delay()); + } + 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"); + return du.delayUntil().toEpochMilli(); } - dbRetry(() -> WorkflowDAO.setWorkflowDelay(ctx, workflowId, delayUntilEpochMs)); + throw new IllegalArgumentException("Unexpected WorkflowDelay value"); } /** 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 a942701b..e8767968 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(); @@ -2171,33 +2170,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 @@ -2215,167 +2193,210 @@ 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. + * 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 void checkRewindable(DbContext ctx, String workflowId, int startStep) + 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( + String workflowId, int startStep, @Nullable WorkflowState state) { if (startStep < 0) { throw new IllegalArgumentException("startStep must be >= 0, got " + startStep); } - try (var conn = ctx.getConnection()) { - readRewindableState(conn, ctx.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); } @@ -2385,15 +2406,15 @@ private static WorkflowState readRewindableState( + " first") .formatted(workflowId, state)); } - return state; } /** - * 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, @@ -2409,48 +2430,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 63b4bc77..63d0a3d1 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,31 @@ public WorkflowHandle rewindWorkflow( Objects.requireNonNull(workflowId); Objects.requireNonNull(options); - this.runDbosFunctionAsStep( - () -> { - 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); + 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, + // 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 (precheck.runsWork()) { + try { + validateQueue(options.queueName(), options.queuePartitionKey()); + if (hasStepCheckpointStores()) { + precheck.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); } @@ -1218,14 +1213,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 9627f5fa..dd2ab23a 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 35665524..43806e01 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; @@ -14,8 +15,11 @@ 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; @@ -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); @@ -316,6 +323,121 @@ 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::counted, 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 run the target a third time, or be refused if it were still 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("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()); + } + /** * 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. @@ -460,6 +582,16 @@ interface InternalStepService { String cancelsFromStep(String targetId); + String forks(String sourceId); + + String rewinds(String targetId); + + String counted(); + + String labels(String targetId); + + String delays(String targetId); + String resumesCatchingAFailure(String targetId); String cancelsWhenReleased(String otherId) throws InterruptedException; @@ -479,6 +611,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<>(); final CountDownLatch mutualWaiting = new CountDownLatch(2); final CountDownLatch mutualRelease = new CountDownLatch(1); @@ -550,6 +683,53 @@ 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 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) { + 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 1f3bdc35..ecbf8c08 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], 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 b0836ba8..dc3908f8 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