diff --git a/transact/src/main/java/dev/dbos/transact/DBOSClient.java b/transact/src/main/java/dev/dbos/transact/DBOSClient.java index 62cbff1f6..af74d71e1 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOSClient.java +++ b/transact/src/main/java/dev/dbos/transact/DBOSClient.java @@ -1912,7 +1912,7 @@ public void setLatestApplicationVersion( * @param schedule the schedule configuration */ public void createSchedule(@NonNull WorkflowSchedule schedule) { - systemDatabase.createSchedule(schedule); + systemDatabase.createSchedule(schedule, null); } /** @@ -1960,7 +1960,7 @@ public void createSchedule(@NonNull WorkflowSchedule schedule) { * @param name schedule name */ public void deleteSchedule(@NonNull String name) { - systemDatabase.deleteSchedule(name); + systemDatabase.deleteSchedule(name, null); } /** @@ -1969,7 +1969,7 @@ public void deleteSchedule(@NonNull String name) { * @param name schedule name */ public void pauseSchedule(@NonNull String name) { - systemDatabase.pauseSchedule(name); + systemDatabase.pauseSchedule(name, null); } /** @@ -1978,7 +1978,7 @@ public void pauseSchedule(@NonNull String name) { * @param name schedule name */ public void resumeSchedule(@NonNull String name) { - systemDatabase.resumeSchedule(name); + systemDatabase.resumeSchedule(name, 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 ed0a49e95..e55521a7d 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -18,6 +18,7 @@ import dev.dbos.transact.database.signal.SignalMap; import dev.dbos.transact.database.signal.Subscription; import dev.dbos.transact.exceptions.*; +import dev.dbos.transact.exceptions.DBOSQueueDuplicatedException; import dev.dbos.transact.internal.DebugTriggers; import dev.dbos.transact.internal.Validation; import dev.dbos.transact.json.DBOSSerializer; @@ -1373,17 +1374,52 @@ public Object takeOverDebouncerWorkflow( @Nullable WorkflowStatusInternal promised, @Nullable DebounceCaller caller) { // The delay is resolved outside the retry, as in initWorkflowStatus, so a retry cannot push it - // later. An attempt that committed but lost its ack finds the holder cancelled on the retry and - // returns null; the caller then goes on to extend the promised workflow like any other call. - // Safe to replay after a peer committed, as a serialization error means one did: the holder is - // re-checked under its lock, the step is re-checked, and the insert keeps a row already there. + // later. An attempt that committed but lost its ack replays its checkpoint on the retry; with + // no caller, it finds the holder cancelled and returns null, and the caller then goes on to + // extend the promised workflow like any other call. Safe to replay after a peer committed, as + // a serialization error means one did: the holder is re-checked under its lock, the step is + // re-checked, and the insert keeps a row already there. String ownerXid = UUID.randomUUID().toString(); Long delayUntilEpochMs = promised == null ? null : resolveDelayUntil(promised); - return dbRetryIncludingSerializationError( - "takeOverDebouncerWorkflow", - () -> - WorkflowDAO.takeOverDebouncerWorkflow( - ctx, debouncerWorkflowId, promised, delayUntilEpochMs, ownerXid, caller)); + return runRecordedStep( + stepOf(caller), + caller != null ? caller.stepName() : "takeOverDebouncerWorkflow", + System.currentTimeMillis(), + conn -> { + String created; + var savepoint = conn.setSavepoint(); + try { + created = + WorkflowDAO.takeOverDebouncerWorkflow( + ctx, conn, debouncerWorkflowId, promised, delayUntilEpochMs, ownerXid); + } catch (DBOSQueueDuplicatedException e) { + // Something else holds the key where the promised workflow would go: a debounced + // workflow of a newer node, on a user queue, which the debouncer workflow's key on the + // internal queue never collided with. Undo the takeover, cancel and all, and record + // that it took nothing over, which is how the caller treats it: it goes on to extend + // that workflow as any other call on the key would. + conn.rollback(savepoint); + logger.debug( + "Could not take over debouncer workflow {}: {} holds its key on queue {}", + debouncerWorkflowId, + e.deduplicationId(), + e.queueName()); + created = null; + } + if (caller == null) { + return new Recorded(created, null, null, null); + } + // The step names the workflow it created as its child. + var serialized = SerializationUtil.serializeValue(created, null, ctx.serializer()); + return new Recorded( + created, serialized.serializedValue(), serialized.serialization(), created); + }); + } + + private static @Nullable StepCaller stepOf(@Nullable DebounceCaller caller) { + return caller == null + ? null + : new StepCaller.FromWorkflow(caller.workflowId(), caller.stepId(), caller.ownerXid()); } /** @@ -1406,11 +1442,13 @@ public Object debounceDelayedWorkflow( // caller's, so it can deadlock with a bulk cancel, resume or recovery taking both the other // way. Safe to replay: the step is re-checked first, and the delay and inputs it writes are // fixed before the first attempt. - return dbRetryIncludingSerializationError( - "debounceDelayedWorkflow", - () -> + return runTransactionalStep( + stepOf(caller), + caller != null ? caller.stepName() : "debounceDelayedWorkflow", + conn -> WorkflowDAO.debounceDelayedWorkflow( ctx, + conn, workflowName, className, instanceName, @@ -1419,8 +1457,7 @@ public Object debounceDelayedWorkflow( delayUntilEpochMs, args, serializationFormat, - ids, - caller)); + ids)); } /** Returns the given executors' PENDING workflows to their queues; reports which rows moved. */ @@ -1582,13 +1619,29 @@ public void sendBulk( String functionName, boolean sendToForks, String serialization) { - dbRetryIncludingSerializationError( - "sendBulk", - () -> - NotificationsDAO.sendBulk( - ctx, messages, caller, functionName, sendToForks, serialization)); + // Nothing to send is nothing to record: every run, replays included, does the same nothing. + if (messages.isEmpty()) { + return; + } + runRecordedStep( + caller, + functionName, + System.currentTimeMillis(), + conn -> { + NotificationsDAO.sendBulk(ctx, conn, messages, sendToForks, serialization); + return Recorded.nothing(); + }); } + /** + * Receives a message on {@code topic} as step {@code stepId}, waiting up to {@code timeout}. + * + *

The wait and the claim are retried separately: the claim alone is the step's transaction, so + * a conflict on it replays the claim, not the wait. The wait's deadline is durable, so a wait + * that is retried keeps it. A peer that took the message between the wait and the claim leaves + * the claim nothing, and the step records null. The step is recorded as starting when the wait + * did. + */ public Object recv( String workflowId, int stepId, @@ -1596,18 +1649,40 @@ public Object recv( String topic, Duration timeout, String ownerXid) { - return dbRetry( - () -> - NotificationsDAO.recv( - ctx, - workflowId, - stepId, - timeout, - timeoutStepId, - topic, - notificationRecheckInterval(), - createSubscription, - ownerXid)); + var waited = + dbRetry( + () -> + NotificationsDAO.awaitMessage( + ctx, + workflowId, + stepId, + timeout, + timeoutStepId, + topic, + notificationRecheckInterval(), + createSubscription, + ownerXid)); + if (waited instanceof NotificationsDAO.RecvWait.Replayed replayed) { + return replayed.message(); + } + var step = new StepCaller.FromWorkflow(workflowId, stepId, ownerXid); + if (waited instanceof NotificationsDAO.RecvWait.TimedOut timedOut) { + return runRecordedStep( + step, + "DBOS.recv", + timedOut.startedAtEpochMs(), + conn -> { + var serialized = SerializationUtil.serializeValue(null, null, ctx.serializer()); + return new Recorded<>( + null, serialized.serializedValue(), serialized.serialization(), null); + }); + } + var ready = (NotificationsDAO.RecvWait.Ready) waited; + return runRecordedStep( + step, + "DBOS.recv", + ready.startedAtEpochMs(), + conn -> NotificationsDAO.claimMessage(ctx, conn, workflowId, stepId, topic)); } public void setEvent( @@ -1618,11 +1693,28 @@ public void setEvent( boolean asStep, String serialization, String ownerXid) { - dbRetryIncludingSerializationError( - "setEvent", - () -> - NotificationsDAO.setEvent( - ctx, workflowId, functionId, key, message, asStep, serialization, ownerXid)); + // Even from inside a step the history row names a step, the one setting the event. + StepCaller caller = + asStep + ? new StepCaller.FromWorkflow(workflowId, functionId, ownerXid) + : new StepCaller.FromStep(workflowId, ownerXid); + runRecordedStep( + caller, + "DBOS.setEvent", + System.currentTimeMillis(), + conn -> { + var serialized = + SerializationUtil.serializeValue(message, serialization, ctx.serializer()); + NotificationsDAO.setEvent( + conn, + ctx.schema(), + workflowId, + functionId, + key, + serialized.serializedValue(), + serialized.serialization()); + return Recorded.nothing(); + }); signal(new SignalKey.Event(workflowId, key)); } @@ -1773,8 +1865,53 @@ public void deleteWorkflows( */ @Nullable T runTransactionalStep( @Nullable StepCaller caller, String stepName, SqlTransaction.SqlCall work) { + boolean recorded = caller instanceof StepCaller.FromWorkflow; + return runRecordedStep( + caller, + stepName, + System.currentTimeMillis(), + conn -> { + T result = work.run(conn); + if (!recorded) { + return new Recorded<>(result, null, null, null); + } + var serialized = SerializationUtil.serializeValue(result, null, ctx.serializer()); + return new Recorded<>( + result, serialized.serializedValue(), serialized.serialization(), null); + }); + } + + /** + * What a step's work hands back, with what its checkpoint records: the output as serialized, and + * the child workflow the step names, if any. Recorded only from the workflow body. + */ + public record Recorded( + @Nullable T value, + @Nullable String output, + @Nullable String serialization, + @Nullable String childWorkflowId) { + + /** + * A step that returns nothing and records neither an output nor a serialization, as send, + * setEvent and the stream writes always have. + */ + public static Recorded nothing() { + return new Recorded<>(null, null, null, null); + } + } + + /** + * As {@link #runTransactionalStep}, for work that says what its checkpoint records: a message + * kept in the serialization it arrived in, a step naming the child workflow it created, or a step + * that returns nothing and records no output at all. {@code startTime} is when the step began, + * which for a recv is before its wait. + */ + @Nullable T runRecordedStep( + @Nullable StepCaller caller, + String stepName, + long startTime, + SqlTransaction.SqlCall> work) { var step = caller instanceof StepCaller.FromWorkflow fromWorkflow ? fromWorkflow : null; - long startTime = System.currentTimeMillis(); // What work itself threw on the attempt that ended the retries, as against a failure of the // check or the record around it, which is not the step's answer. var workFailure = new AtomicReference(); @@ -1810,7 +1947,7 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} return new Outcome(recorded, null); } } - T result; + Recorded result; try { result = work.run(c); } catch (SQLException | RuntimeException e) { @@ -1820,8 +1957,6 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} // The owner is checked last, as every fenced write locks the status // row. if (step != null) { - var serialized = - SerializationUtil.serializeValue(result, null, ctx.serializer()); StepsDAO.recordStepResult( ctx, c, @@ -1829,10 +1964,10 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} step.workflowId(), step.stepId(), stepName, - serialized.serializedValue(), + result.output(), null, - null, - serialized.serialization()), + result.childWorkflowId(), + result.serialization()), startTime, System.currentTimeMillis(), step.ownerXid()); @@ -1842,7 +1977,7 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} } DebugTriggers.debugTriggerPoint( DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT); - return new Outcome(null, result); + return new Outcome(null, result.value()); }); } DebugTriggers.debugTriggerPoint( @@ -1869,11 +2004,8 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} if (replayed == null) { return output; } - if (replayed.error() == null && replayed.output() == null) { - throw new IllegalStateException( - "Recorded output and error are both null for workflow %s step %d (%s)" - .formatted(step.workflowId(), step.stepId(), stepName)); - } + // A row with neither output nor error is a step that returned nothing, as send, setEvent and + // the stream writes recorded theirs before they ran through here: the row is the record. if (replayed.error() != null) { var t = SerializationUtil.deserializeError( @@ -2139,8 +2271,15 @@ public void transitionDelayedWorkflows() { dbRetry(() -> WorkflowDAO.transitionDelayedWorkflows(ctx)); } - public void createSchedule(WorkflowSchedule schedule) { - dbRetry(() -> SchedulesDAO.createSchedule(ctx, schedule)); + /** Creates a schedule as the step {@code caller} makes; see {@link #runTransactionalStep}. */ + public void createSchedule(WorkflowSchedule schedule, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.createSchedule", + conn -> { + SchedulesDAO.createSchedule(ctx, conn, schedule); + return null; + }); } public Optional getSchedule(String name) { @@ -2178,20 +2317,41 @@ public List listScheduleRecords( return ctx.serializer(); } - public void pauseSchedule(String name) { - dbRetry(() -> SchedulesDAO.pauseSchedule(ctx, name)); + /** Pauses a schedule as the step {@code caller} makes; see {@link #runTransactionalStep}. */ + public void pauseSchedule(String name, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.pauseSchedule", + conn -> { + SchedulesDAO.pauseSchedule(conn, ctx.schema(), name); + return null; + }); } - public void resumeSchedule(String name) { - dbRetry(() -> SchedulesDAO.resumeSchedule(ctx, name)); + /** Resumes a schedule as the step {@code caller} makes; see {@link #runTransactionalStep}. */ + public void resumeSchedule(String name, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.resumeSchedule", + conn -> { + SchedulesDAO.resumeSchedule(conn, ctx.schema(), name); + return null; + }); } public void updateScheduleLastFiredAt(String name, Instant lastFiredAt) { dbRetry(() -> SchedulesDAO.updateScheduleLastFiredAt(ctx, name, lastFiredAt)); } - public void deleteSchedule(String name) { - dbRetry(() -> SchedulesDAO.deleteSchedule(ctx, name)); + /** Deletes a schedule as the step {@code caller} makes; see {@link #runTransactionalStep}. */ + public void deleteSchedule(String name, @Nullable StepCaller caller) { + runTransactionalStep( + caller, + "DBOS.deleteSchedule", + conn -> { + SchedulesDAO.deleteSchedule(conn, ctx.schema(), name); + return null; + }); } public void applySchedules(List schedules) { @@ -2278,11 +2438,16 @@ public void writeStreamFromStep( Object value, String serializationFormat, String ownerXid) { - dbRetryIncludingSerializationError( - "writeStreamFromStep", - () -> - StreamsDAO.writeStreamFromStep( - ctx, workflowId, functionId, key, value, serializationFormat, ownerXid)); + // Even from inside a step the entry names a step, the one writing it. + runRecordedStep( + new StepCaller.FromStep(workflowId, ownerXid), + "DBOS.writeStream", + System.currentTimeMillis(), + conn -> { + StreamsDAO.writeStream( + conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); + return Recorded.nothing(); + }); signal(new SignalKey.Stream(workflowId, key)); } @@ -2293,17 +2458,27 @@ public void writeStreamFromWorkflow( Object value, String serializationFormat, String ownerXid) { - dbRetryIncludingSerializationError( - "writeStreamFromWorkflow", - () -> - StreamsDAO.writeStreamFromWorkflow( - ctx, workflowId, functionId, key, value, serializationFormat, ownerXid)); + runRecordedStep( + new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), + StreamsDAO.writeStepName(value), + System.currentTimeMillis(), + conn -> { + StreamsDAO.writeStream( + conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); + return Recorded.nothing(); + }); signal(new SignalKey.Stream(workflowId, key)); } public void closeStream(String workflowId, int functionId, String key, String ownerXid) { - dbRetryIncludingSerializationError( - "closeStream", () -> StreamsDAO.closeStream(ctx, workflowId, functionId, key, ownerXid)); + runRecordedStep( + new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), + "DBOS.closeStream", + System.currentTimeMillis(), + conn -> { + StreamsDAO.closeStream(conn, ctx.schema(), workflowId, functionId, key); + return Recorded.nothing(); + }); // Closing writes the sentinel entry, so readers need the same wake-up as any other write. signal(new SignalKey.Stream(workflowId, key)); } diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java index e2fdf2468..9c2847059 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java @@ -5,14 +5,11 @@ import dev.dbos.transact.Constants; import dev.dbos.transact.database.DbContext; import dev.dbos.transact.database.GetEventCaller; -import dev.dbos.transact.database.SqlTransaction; -import dev.dbos.transact.database.StepCaller; import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.database.signal.SignalKey; import dev.dbos.transact.database.signal.SignalMap; import dev.dbos.transact.database.signal.Subscription; import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; -import dev.dbos.transact.exceptions.DBOSWorkflowExecutionConflictException; import dev.dbos.transact.json.DBOSSerializer; import dev.dbos.transact.json.SerializationUtil; import dev.dbos.transact.workflow.SendMessage; @@ -94,12 +91,11 @@ WHERE forked_from IN (%s) return result; } - /** Sends messages from {@code caller}, or from outside any workflow when it is null. */ + /** Sends messages on the caller's transaction. */ public static void sendBulk( DbContext ctx, + Connection conn, List messages, - @Nullable StepCaller caller, - String functionName, boolean sendToForks, String serialization) throws SQLException { @@ -114,7 +110,6 @@ public static void sendBulk( } DBOSSerializer serializer = ctx.serializer(); - var startTime = System.currentTimeMillis(); // Serialize each message once record SerializedPair(SendMessage msg, SerializationUtil.SerializedResult serialized) {} @@ -125,111 +120,88 @@ record SerializedPair(SendMessage msg, SerializationUtil.SerializedResult serial msg, SerializationUtil.serializeValue(msg.message(), serialization, serializer))); } - try (var txConn = ctx.getConnection()) { - SqlTransaction.run( - txConn, - conn -> { - // A send from the workflow body replays as the step it was recorded as. - if (caller instanceof StepCaller.FromWorkflow step) { - StepResult recorded = - StepsDAO.checkStepResult( - conn, - ctx.schema(), - step.workflowId(), - step.stepId(), - functionName, - step.ownerXid()); - if (recorded != null) { - logger.debug( - "Replaying sendBulk, workflowId: {}, stepId: {}", - step.workflowId(), - step.stepId()); - return; - } - } + // Collect all destination IDs for fork resolution + Map> forkDescendants = Map.of(); + if (sendToForks) { + List destIds = pairs.stream().map(p -> p.msg().destinationId()).distinct().toList(); + forkDescendants = findForkDescendantsTxn(conn, ctx.schema(), destIds); + } - // Collect all destination IDs for fork resolution - Map> forkDescendants = Map.of(); - if (sendToForks) { - List destIds = - pairs.stream().map(p -> p.msg().destinationId()).distinct().toList(); - forkDescendants = findForkDescendantsTxn(conn, ctx.schema(), destIds); - } + // Build insert rows: base dest + sorted descendants + record InsertRow( + String destId, + SerializationUtil.SerializedResult serialized, + String topic, + String messageUuid) {} + List rows = new ArrayList<>(); + for (var pair : pairs) { + var msg = pair.msg(); + String baseDest = msg.destinationId(); + String finalTopic = (msg.topic() != null) ? msg.topic() : Constants.DBOS_NULL_TOPIC; + + List destinations = new ArrayList<>(); + destinations.add(baseDest); + if (sendToForks) { + var desc = forkDescendants.getOrDefault(baseDest, Set.of()); + desc.stream().sorted().forEach(destinations::add); + } - // Build insert rows: base dest + sorted descendants - record InsertRow( - String destId, - SerializationUtil.SerializedResult serialized, - String topic, - String messageUuid) {} - List rows = new ArrayList<>(); - for (var pair : pairs) { - var msg = pair.msg(); - String baseDest = msg.destinationId(); - String finalTopic = (msg.topic() != null) ? msg.topic() : Constants.DBOS_NULL_TOPIC; - - List destinations = new ArrayList<>(); - destinations.add(baseDest); - if (sendToForks) { - var desc = forkDescendants.getOrDefault(baseDest, Set.of()); - desc.stream().sorted().forEach(destinations::add); - } - - for (String dest : destinations) { - var wfid = - msg.idempotencyKey() != null - ? msg.idempotencyKey() + "::" + dest - : UUID.randomUUID().toString(); - rows.add(new InsertRow(dest, pair.serialized(), finalTopic, wfid)); - } - } + for (String dest : destinations) { + var wfid = + msg.idempotencyKey() != null + ? msg.idempotencyKey() + "::" + dest + : UUID.randomUUID().toString(); + rows.add(new InsertRow(dest, pair.serialized(), finalTopic, wfid)); + } + } - // Batch-insert all rows - final String sql = - """ - INSERT INTO "%s".notifications - (destination_uuid, topic, message, serialization, message_uuid) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT (message_uuid) DO NOTHING - """ - .formatted(ctx.schema()); - - try (PreparedStatement stmt = conn.prepareStatement(sql)) { - for (var row : rows) { - stmt.setString(1, row.destId()); - stmt.setString(2, row.topic()); - stmt.setString(3, row.serialized().serializedValue()); - stmt.setString(4, row.serialized().serialization()); - stmt.setString(5, row.messageUuid()); - stmt.addBatch(); - } - stmt.executeBatch(); - } catch (SQLException e) { - if ("23503".equals(e.getSQLState())) { - var distinctDests = rows.stream().map(InsertRow::destId).distinct().toList(); - throw new DBOSNonExistentWorkflowException( - distinctDests.size() == 1 ? distinctDests.get(0) : null); - } - throw e; - } + // Batch-insert all rows + final String sql = + """ + INSERT INTO "%s".notifications + (destination_uuid, topic, message, serialization, message_uuid) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT (message_uuid) DO NOTHING + """ + .formatted(ctx.schema()); - // The owner is checked after the messages, so the status row is locked last, as every - // fenced write locks it. Recording the step checks it first; a send from inside a step - // has no step to record, so it is checked on its own. - if (caller instanceof StepCaller.FromWorkflow step) { - var output = - new StepResult( - step.workflowId(), step.stepId(), functionName, null, null, null, null); - StepsDAO.recordStepResult( - ctx, conn, output, startTime, System.currentTimeMillis(), step.ownerXid()); - } else if (caller != null) { - WorkflowDAO.checkOwner(conn, ctx.schema(), caller.workflowId(), caller.ownerXid()); - } - }); + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + for (var row : rows) { + stmt.setString(1, row.destId()); + stmt.setString(2, row.topic()); + stmt.setString(3, row.serialized().serializedValue()); + stmt.setString(4, row.serialized().serialization()); + stmt.setString(5, row.messageUuid()); + stmt.addBatch(); + } + stmt.executeBatch(); + } catch (SQLException e) { + if ("23503".equals(e.getSQLState())) { + var distinctDests = rows.stream().map(InsertRow::destId).distinct().toList(); + throw new DBOSNonExistentWorkflowException( + distinctDests.size() == 1 ? distinctDests.get(0) : null); + } + throw e; } } - public static Object recv( + /** How a wait for a message ended. */ + public sealed interface RecvWait { + /** The step already ran: what it received, or null if it received nothing. */ + record Replayed(@Nullable Object message) implements RecvWait {} + + /** The timeout passed with no message; the wait began at {@code startedAtEpochMs}. */ + record TimedOut(long startedAtEpochMs) implements RecvWait {} + + /** A message is waiting to be claimed; the wait began at {@code startedAtEpochMs}. */ + record Ready(long startedAtEpochMs) implements RecvWait {} + } + + /** + * Waits for a message on {@code topic} for the recv that is step {@code stepId}, until one + * arrives or the timeout passes. A step already recorded is replayed instead, without waiting. + */ + public static RecvWait awaitMessage( DbContext ctx, String workflowId, int stepId, @@ -246,17 +218,19 @@ public static Object recv( } Objects.requireNonNull(dbPollingInterval); - var stepName = "DBOS.recv"; final var recvTopic = Objects.requireNonNullElse(topic, Constants.DBOS_NULL_TOPIC); - var result = StepsDAO.checkStepResult(ctx, workflowId, stepId, stepName, ownerXid); + var result = StepsDAO.checkStepResult(ctx, workflowId, stepId, "DBOS.recv", ownerXid); if (result != null) { logger.debug( "Replaying recv, workflowId: {}, stepId: {}, topic: {}", workflowId, stepId, recvTopic); // A NULL output is a recv that claimed no message (a peer took it) and returned null. // Replay that rather than receiving again: a second run would record a different result // for this step. - return result.output() != null ? result.toResult(ctx.serializer()) : null; + return new RecvWait.Replayed( + result.output() != null || result.error() != null + ? result.toResult(ctx.serializer()) + : null); } logger.debug( @@ -282,8 +256,7 @@ public static Object recv( stmt.setString(2, recvTopic); try (var rs = stmt.executeQuery()) { if (rs.next()) { - // query for results - break; + return new RecvWait.Ready(startTime); } } WorkflowDAO.checkWorkflow(conn, ctx.schema(), workflowId, ownerXid); @@ -291,11 +264,7 @@ public static Object recv( var duration = Duration.between(Instant.now(), timeoutAt); if (duration.isNegative() || duration.isZero()) { - var output = SerializationUtil.serializeValue(null, null, ctx.serializer()); - var stepResult = StepResult.ofOutput(workflowId, stepId, stepName, output); - StepsDAO.recordStepResult( - ctx, stepResult, startTime, System.currentTimeMillis(), ownerXid); - return null; + return new RecvWait.TimedOut(startTime); } var loopDuration = @@ -304,10 +273,17 @@ public static Object recv( SignalMap.awaitAny(loopDuration, messageSignal); } } + } - ctx.checkClosed(); - WorkflowDAO.checkWorkflow(ctx, workflowId, ownerXid); - + /** + * Claims the oldest unconsumed message on {@code topic} for the recv that is step {@code stepId}, + * on the caller's transaction, and returns it with its checkpoint: the message as it was stored. + * A peer that claimed it first leaves none, and the step records null. + */ + public static SystemDatabase.Recorded claimMessage( + DbContext ctx, Connection conn, String workflowId, int stepId, @Nullable String topic) + throws SQLException { + final var recvTopic = Objects.requireNonNullElse(topic, Constants.DBOS_NULL_TOPIC); var updateSql = """ UPDATE "%1$s".notifications @@ -327,61 +303,34 @@ public static Object recv( """ .formatted(ctx.schema()); - try (var txConn = ctx.getConnection()) { - // Only the claim is replayed, never the wait above it: re-running the whole method would - // register a fresh subscription and restart the caller's timeout from zero. A peer that won - // the message between attempts lands this on the null-message branch below, which is a case - // it already handles. - return SystemDatabase.retryOnSerializationError( - "recv claim", - () -> - SqlTransaction.call( - txConn, - conn -> { - String serializedMessage = null; - String serialization = null; - try (PreparedStatement stmt = conn.prepareStatement(updateSql)) { - // The consuming step is recorded so a rewind can delete exactly the - // messages consumed by the steps it discards. - stmt.setInt(1, stepId); - stmt.setString(2, workflowId); - stmt.setString(3, recvTopic); - stmt.setString(4, workflowId); - stmt.setString(5, recvTopic); - - // Recording the step checks that this execution still owns the workflow, so - // a stale execution's claim rolls back and leaves the message for the - // owner. - try (ResultSet rs = stmt.executeQuery()) { - if (rs.next()) { - serializedMessage = rs.getString("message"); - serialization = rs.getString("serialization"); - } - } - } - - var deserializedMessage = - SerializationUtil.deserializeValue( - serializedMessage, serialization, ctx.serializer()); - - var output = - new StepResult( - workflowId, - stepId, - stepName, - serializedMessage, - null, - null, - serialization); - StepsDAO.recordStepResult( - ctx, conn, output, startTime, System.currentTimeMillis(), ownerXid); - - return deserializedMessage; - })); + String serializedMessage = null; + String serialization = null; + try (PreparedStatement stmt = conn.prepareStatement(updateSql)) { + // The consuming step is recorded so a rewind can delete exactly the messages consumed by + // the steps it discards. + stmt.setInt(1, stepId); + stmt.setString(2, workflowId); + stmt.setString(3, recvTopic); + stmt.setString(4, workflowId); + stmt.setString(5, recvTopic); + try (ResultSet rs = stmt.executeQuery()) { + if (rs.next()) { + serializedMessage = rs.getString("message"); + serialization = rs.getString("serialization"); + } + } } + + var message = + SerializationUtil.deserializeValue(serializedMessage, serialization, ctx.serializer()); + return new SystemDatabase.Recorded<>(message, serializedMessage, serialization, null); } - private static void setEvent( + /** + * Publishes {@code key} on the caller's transaction, and records it in the event history under + * {@code functionId}, the step that set it. + */ + public static void setEvent( Connection conn, String schema, String workflowId, @@ -426,85 +375,6 @@ ON CONFLICT (workflow_uuid, key, function_id) } } - public static void setEvent( - DbContext ctx, - String workflowId, - int functionId, - String key, - Object message, - boolean asStep, - String serialization, - String ownerXid) - throws SQLException { - - DBOSSerializer serializer = ctx.serializer(); - var startTime = System.currentTimeMillis(); - String functionName = "DBOS.setEvent"; - - SerializationUtil.SerializedResult serializedResult = - SerializationUtil.serializeValue(message, serialization, serializer); - - try (var txConn = ctx.getConnection()) { - try { - SqlTransaction.run( - txConn, - conn -> { - if (asStep) { - var recordedOutput = - StepsDAO.checkStepResult( - conn, ctx.schema(), workflowId, functionId, functionName, ownerXid); - if (recordedOutput != null) { - logger.debug( - "Replaying setEvent, workflow: {}, step: {}, key: {}", - workflowId, - functionId, - key); - return; - } else { - logger.debug( - "Running setEvent, workflow: {}, step: {}, key: {}", - workflowId, - functionId, - key); - } - } - - setEvent( - conn, - ctx.schema(), - workflowId, - functionId, - key, - serializedResult.serializedValue(), - serializedResult.serialization()); - - // The owner is checked after the event, so the status row is locked last, as every - // fenced write locks it. Recording the step checks it first; from inside a step there - // is no step of its own to record, so it is checked on its own: an execution that has - // lost its workflow sets nothing. - if (asStep) { - StepResult output = - new StepResult(workflowId, functionId, functionName, null, null, null, null); - StepsDAO.recordStepResult( - ctx, conn, output, startTime, System.currentTimeMillis(), ownerXid); - } else { - WorkflowDAO.checkOwner(conn, ctx.schema(), workflowId, ownerXid); - } - }); - } catch (DBOSWorkflowExecutionConflictException e) { - // Refused because this execution no longer owns the workflow: expected, not an error. - logger.debug( - "setEvent refused, workflow: {} id: {}, key: {}", workflowId, functionId, key, e); - throw e; - } catch (Exception e) { - // The helper has already rolled back; this only records which call lost its work. - logger.error( - "setEvent rollback, workflow: {} id: {}, key: {}", workflowId, functionId, key, e); - throw e; - } - } - } - private record GetEventResult(String value, String serialization) {} private static Optional getEvent( diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/SchedulesDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/SchedulesDAO.java index d5b702dec..c98c67320 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/SchedulesDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/SchedulesDAO.java @@ -37,10 +37,11 @@ private SchedulesDAO() {} return schedule.applicationName() != null ? schedule.applicationName() : ctx.appName(); } - public static void createSchedule(DbContext ctx, WorkflowSchedule schedule) throws SQLException { - try (Connection conn = ctx.getConnection()) { - createSchedule(conn, ctx.schema(), ctx.serializer(), schedule, scheduleOwner(ctx, schedule)); - } + /** Creates a schedule on the caller's transaction. */ + public static void createSchedule(DbContext ctx, Connection conn, WorkflowSchedule schedule) + throws SQLException { + Objects.requireNonNull(schedule, "schedule must not be null"); + createSchedule(conn, ctx.schema(), ctx.serializer(), schedule, scheduleOwner(ctx, schedule)); } static void createSchedule( @@ -250,24 +251,25 @@ public static Optional getScheduleRecord(DbContext ctx, String n } } - public static void pauseSchedule(DbContext ctx, String name) throws SQLException { - setScheduleStatus(ctx, name, ScheduleStatus.PAUSED); + public static void pauseSchedule(Connection conn, String schema, String name) + throws SQLException { + setScheduleStatus(conn, schema, name, ScheduleStatus.PAUSED); } - public static void resumeSchedule(DbContext ctx, String name) throws SQLException { - setScheduleStatus(ctx, name, ScheduleStatus.ACTIVE); + public static void resumeSchedule(Connection conn, String schema, String name) + throws SQLException { + setScheduleStatus(conn, schema, name, ScheduleStatus.ACTIVE); } - private static void setScheduleStatus(DbContext ctx, String name, ScheduleStatus status) - throws SQLException { + private static void setScheduleStatus( + Connection conn, String schema, String name, ScheduleStatus status) throws SQLException { String sql = """ UPDATE "%s".workflow_schedules SET status = ? WHERE schedule_name = ? """ - .formatted(ctx.schema()); + .formatted(schema); - try (Connection conn = ctx.getConnection(); - PreparedStatement ps = conn.prepareStatement(sql)) { + try (PreparedStatement ps = conn.prepareStatement(sql)) { ps.setString(1, status.name()); ps.setString(2, name); ps.executeUpdate(); @@ -290,13 +292,8 @@ public static void updateScheduleLastFiredAt(DbContext ctx, String name, Instant } } - public static void deleteSchedule(DbContext ctx, String name) throws SQLException { - try (var conn = ctx.getConnection()) { - deleteSchedule(conn, ctx.schema(), name); - } - } - - static void deleteSchedule(Connection conn, String schema, String name) throws SQLException { + public static void deleteSchedule(Connection conn, String schema, String name) + throws SQLException { String sql = """ DELETE FROM "%s".workflow_schedules WHERE schedule_name = ? diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/StreamsDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/StreamsDAO.java index b61310e3f..8623ea610 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/StreamsDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/StreamsDAO.java @@ -1,14 +1,12 @@ package dev.dbos.transact.database.dao; import dev.dbos.transact.database.DbContext; -import dev.dbos.transact.database.SqlTransaction; import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.database.signal.SignalKey; import dev.dbos.transact.database.signal.SignalMap; import dev.dbos.transact.database.signal.Subscription; import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; import dev.dbos.transact.json.SerializationUtil; -import dev.dbos.transact.workflow.internal.StepResult; import java.sql.Connection; import java.sql.SQLException; @@ -19,69 +17,18 @@ public class StreamsDAO { private StreamsDAO() {} - public static void writeStreamFromStep( - DbContext ctx, - String workflowId, - int functionId, - String key, - Object value, - String serializationFormat, - String ownerXid) - throws SQLException { - // A step records nothing of its own, but its write lands only while its execution still owns - // the workflow. The owner is checked after the insert, so the status row is locked last, as - // every fenced write locks it. - try (var txConn = ctx.getConnection()) { - SqlTransaction.run( - txConn, - conn -> { - insertStream( - conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); - WorkflowDAO.checkOwner(conn, ctx.schema(), workflowId, ownerXid); - }); - } - } - - public static void writeStreamFromWorkflow( - DbContext ctx, - String workflowId, - int functionId, - String key, - Object value, - String serializationFormat, - String ownerXid) - throws SQLException { - String functionName = - STREAM_CLOSED_SENTINEL.equals(value) ? "DBOS.closeStream" : "DBOS.writeStream"; - long startTime = System.currentTimeMillis(); - - try (var txConn = ctx.getConnection()) { - SqlTransaction.run( - txConn, - conn -> { - StepResult recordedOutput = - StepsDAO.checkStepResult( - conn, ctx.schema(), workflowId, functionId, functionName, ownerXid); - - if (recordedOutput != null) { - logger.debug("Replaying writeStream, id: {}, key: {}", functionId, key); - return; - } else { - logger.debug("Running writeStream, id: {}, key: {}", functionId, key); - } - - insertStream( - conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); - - var output = - new StepResult(workflowId, functionId, functionName, null, null, null, null); - StepsDAO.recordStepResult( - ctx, conn, output, startTime, System.currentTimeMillis(), ownerXid); - }); - } + /** + * The step a write from the workflow body is recorded as: a close when it writes the close + * sentinel, however it was called. + */ + public static String writeStepName(Object value) { + return STREAM_CLOSED_SENTINEL.equals(value) ? "DBOS.closeStream" : "DBOS.writeStream"; } - private static void insertStream( + /** + * Appends {@code value} to the stream on the caller's transaction, as step {@code functionId}. + */ + public static void writeStream( Connection conn, String schema, String workflowId, @@ -156,11 +103,11 @@ SELECT COALESCE(MAX("offset"), -1) + 1 } } + /** Closes the stream on the caller's transaction, as step {@code functionId}. */ public static void closeStream( - DbContext ctx, String workflowId, int functionId, String key, String ownerXid) + Connection conn, String schema, String workflowId, int functionId, String key) throws SQLException { - writeStreamFromWorkflow( - ctx, workflowId, functionId, key, STREAM_CLOSED_SENTINEL, "portable_json", ownerXid); + writeStream(conn, schema, workflowId, functionId, key, STREAM_CLOSED_SENTINEL, "portable_json"); } public static Object readStream( 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 e8767968c..5dbf34d3c 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 @@ -5,7 +5,6 @@ import dev.dbos.transact.database.ChildOf; import dev.dbos.transact.database.DatabaseTime; import dev.dbos.transact.database.DbContext; -import dev.dbos.transact.database.DebounceCaller; import dev.dbos.transact.database.Result; import dev.dbos.transact.database.SqlTransaction; import dev.dbos.transact.database.SystemDatabase; @@ -938,59 +937,18 @@ public static void checkOwner(Connection conn, String schema, String workflowId, * it, before it went silent or, on a node that was only slow, while this call waited -- or {@code * promised} is null because its inputs named none. * - *

Returns the promised workflow's id if this call created it, otherwise null. With a {@code - * caller}, it is that caller's step: if the step already ran, what it recorded is returned and - * nothing is touched; otherwise the takeover and its checkpoint commit together. The return is - * {@code Object}, as a replay hands back whatever the serializer preserved. + *

Returns the promised workflow's id if this call created it, otherwise null. Runs on the + * caller's transaction, which commits the takeover with the caller's step, if any. */ - public static Object takeOverDebouncerWorkflow( + public static @Nullable String takeOverDebouncerWorkflow( DbContext ctx, + Connection conn, String debouncerWorkflowId, @Nullable WorkflowStatusInternal promised, @Nullable Long delayUntilEpochMs, - String creatorXid, - @Nullable DebounceCaller caller) + String creatorXid) throws SQLException { - long startTime = System.currentTimeMillis(); - try (var conn = ctx.getConnection()) { - return SqlTransaction.call( - conn, - c -> { - if (caller != null) { - var prev = - StepsDAO.checkStepResult( - c, - ctx.schema(), - caller.workflowId(), - caller.stepId(), - caller.stepName(), - caller.ownerXid()); - if (prev != null) { - return prev.toResult(ctx.serializer()); - } - } - String created = - takeOver(ctx, c, debouncerWorkflowId, promised, delayUntilEpochMs, creatorXid); - if (caller != null) { - var serialized = SerializationUtil.serializeValue(created, null, ctx.serializer()); - StepsDAO.recordStepResult( - ctx, - c, - new StepResult( - caller.workflowId(), - caller.stepId(), - caller.stepName(), - serialized.serializedValue(), - null, - created, - serialized.serialization()), - startTime, - System.currentTimeMillis(), - caller.ownerXid()); - } - return created; - }); - } + return takeOver(ctx, conn, debouncerWorkflowId, promised, delayUntilEpochMs, creatorXid); } private static @Nullable String takeOver( @@ -1085,10 +1043,9 @@ public static Object takeOverDebouncerWorkflow( *

The new inputs are {@code args} serialized in {@code serializationFormat}, the workflow's * registered format (null for the default), with this database's serializer. * - *

With a {@code caller}, the bounce is that caller's step: if the step already ran, what it - * recorded is returned and nothing is touched; otherwise the bounce and its checkpoint commit - * together, so a crash can never leave the row extended but the step unrecorded, which on replay - * would bounce again. + *

Runs on the caller's transaction, which commits the bounce with the caller's step, if any, + * so a crash can never leave the row extended but the step unrecorded, which on replay would + * bounce again. * *

With {@code ids}, the bounce is the debouncer's first step, which records the ids it assigns * with the outcome, as that step has always recorded them. When nothing was extended, the same @@ -1096,11 +1053,12 @@ public static Object takeOverDebouncerWorkflow( * internal queue -- what a debouncer from before debounced workflows leaves there -- so the * caller can forward to it rather than create a second workflow beside it. * - *

The return is what the step records -- the {@link DebounceResult}, or the ids completed with - * it -- as {@code Object}, since a replay hands back whatever the serializer preserved. + *

The return is what the caller's step records -- the {@link DebounceResult}, or the ids + * completed with it. Runs on the caller's transaction, which commits the bounce with that step. */ public static Object debounceDelayedWorkflow( DbContext ctx, + Connection conn, String workflowName, String className, @Nullable String instanceName, @@ -1109,69 +1067,30 @@ public static Object debounceDelayedWorkflow( long delayUntilEpochMs, Object[] args, @Nullable String serializationFormat, - @Nullable DebounceIds ids, - @Nullable DebounceCaller caller) + @Nullable DebounceIds ids) throws SQLException { - long startTime = System.currentTimeMillis(); - try (var conn = ctx.getConnection()) { - return SqlTransaction.call( - conn, - c -> { - if (caller != null) { - var prev = - StepsDAO.checkStepResult( - c, - ctx.schema(), - caller.workflowId(), - caller.stepId(), - caller.stepName(), - caller.ownerXid()); - if (prev != null) { - return prev.toResult(ctx.serializer()); - } - } - // Serialize after the replay check, so a step that already ran never serializes - // arguments it will not use. - var serializedArgs = - SerializationUtil.serializeArgs(args, null, serializationFormat, ctx.serializer()); - var result = - bounce( - ctx, - c, - workflowName, - className, - instanceName, - queueName, - deduplicationId, - delayUntilEpochMs, - serializedArgs.serializedValue(), - serializedArgs.serialization()); - Object recorded = - ids == null - ? result - : ids.withBounced( - result, findDebouncerWorkflow(ctx, c, queueName, deduplicationId, result)); - if (caller == null) { - return recorded; - } - var serialized = SerializationUtil.serializeValue(recorded, null, ctx.serializer()); - StepsDAO.recordStepResult( - ctx, - c, - new StepResult( - caller.workflowId(), - caller.stepId(), - caller.stepName(), - serialized.serializedValue(), - null, - null, - serialized.serialization()), - startTime, - System.currentTimeMillis(), - caller.ownerXid()); - return recorded; - }); - } + // Serialize after the replay check, so a step that already ran never serializes + // arguments it will not use. + var serializedArgs = + SerializationUtil.serializeArgs(args, null, serializationFormat, ctx.serializer()); + var result = + bounce( + ctx, + conn, + workflowName, + className, + instanceName, + queueName, + deduplicationId, + delayUntilEpochMs, + serializedArgs.serializedValue(), + serializedArgs.serialization()); + Object recorded = + ids == null + ? result + : ids.withBounced( + result, findDebouncerWorkflow(ctx, conn, queueName, deduplicationId, result)); + return recorded; } /** diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSConductorEngine.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSConductorEngine.java index 0c7424da4..e00fd54f8 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSConductorEngine.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSConductorEngine.java @@ -141,12 +141,12 @@ public List listSchedules( @Override public void pauseSchedule(String name) { - systemDatabase.pauseSchedule(name); + systemDatabase.pauseSchedule(name, null); } @Override public void resumeSchedule(String name) { - systemDatabase.resumeSchedule(name); + systemDatabase.resumeSchedule(name, null); } @Override diff --git a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java index 63d0a3d13..d2c55641e 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1305,13 +1305,7 @@ public void createSchedule(@NonNull WorkflowSchedule schedule) { validateQueue(schedule.queueName()); validateWorkflow(schedule.workflowName(), schedule.className()); - this.runDbosFunctionAsStep( - () -> { - systemDatabase.createSchedule(schedule); - return null; - }, - "DBOS.createSchedule", - null); + systemDatabase.createSchedule(schedule, stepCaller()); } public Optional getSchedule(String name) { @@ -1336,33 +1330,15 @@ public List listSchedules( } public void deleteSchedule(String name) { - this.runDbosFunctionAsStep( - () -> { - systemDatabase.deleteSchedule(name); - return null; - }, - "DBOS.deleteSchedule", - null); + systemDatabase.deleteSchedule(name, stepCaller()); } public void pauseSchedule(String name) { - this.runDbosFunctionAsStep( - () -> { - systemDatabase.pauseSchedule(name); - return null; - }, - "DBOS.pauseSchedule", - null); + systemDatabase.pauseSchedule(name, stepCaller()); } public void resumeSchedule(String name) { - this.runDbosFunctionAsStep( - () -> { - systemDatabase.resumeSchedule(name); - return null; - }, - "DBOS.resumeSchedule", - null); + systemDatabase.resumeSchedule(name, stepCaller()); } public void applySchedules(List schedules) { diff --git a/transact/src/main/java/dev/dbos/transact/workflow/internal/InternalWorkflows.java b/transact/src/main/java/dev/dbos/transact/workflow/internal/InternalWorkflows.java index 49761a594..372485b4e 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/internal/InternalWorkflows.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/internal/InternalWorkflows.java @@ -5,7 +5,6 @@ import dev.dbos.transact.StartWorkflowOptions; import dev.dbos.transact.database.DebounceCaller; import dev.dbos.transact.database.SystemDatabase; -import dev.dbos.transact.exceptions.DBOSQueueDuplicatedException; import dev.dbos.transact.exceptions.DBOSWorkflowFunctionNotFoundException; import dev.dbos.transact.execution.DBOSExecutor; import dev.dbos.transact.execution.RegisteredWorkflow; @@ -71,8 +70,9 @@ public static Method debouncerWorkflowMethod() { * another call took it over. Returns null after cancelling when its user workflow already exists * -- it started that workflow and then went silent, which the cancel cannot undo, but left alone * it would hold the key forever once its node is gone -- or when the inputs do not name one. - * Either way the caller goes on as for any call on the key, and extends the promised workflow if - * it is still waiting. + * Returns null, cancelling nothing, when something else holds the key where the promised workflow + * would go. Either way the caller goes on as for any call on the key, and extends the promised + * workflow if it is still waiting. * *

With a {@code caller}, the takeover is that caller's step, checkpointed in the same * transaction, so a replay neither cancels again nor loses the id it returned. @@ -84,26 +84,14 @@ public static Method debouncerWorkflowMethod() { @Nullable DebounceCaller caller) { var status = systemDatabase.getWorkflowStatus(debouncerWorkflowId); var userWorkflowId = status == null ? null : getUserWorkflowId(status.input()); - try { - return systemDatabase.takeOverDebouncerWorkflow( - debouncerWorkflowId, - userWorkflowId == null ? null : promised.apply(userWorkflowId), - caller) - instanceof String id - ? id - : null; - } catch (DBOSQueueDuplicatedException e) { - // Something else holds the key where the promised workflow would go: a debounced workflow of - // a newer node, on a user queue, which the debouncer workflow's key on the internal queue - // never collided with. The transaction rolled back, cancel and all, and the caller extends - // that workflow as any other call on the key would. - logger.debug( - "Could not take over debouncer workflow {}: {} holds its key on queue {}", - debouncerWorkflowId, - e.deduplicationId(), - e.queueName()); - return null; - } + // A collision with whatever else holds the key comes back as null, the takeover undone. + return systemDatabase.takeOverDebouncerWorkflow( + debouncerWorkflowId, + userWorkflowId == null ? null : promised.apply(userWorkflowId), + caller) + instanceof String id + ? id + : null; } /** 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 dd2ab23aa..5e69d010d 100644 --- a/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java +++ b/transact/src/test/java/dev/dbos/transact/database/SystemDatabaseTest.java @@ -792,7 +792,7 @@ private static WorkflowSchedule makeSchedule(String name) { @Test public void testCreateAndGetSchedule() { - sysdb.createSchedule(makeSchedule("sched-1")); + sysdb.createSchedule(makeSchedule("sched-1"), null); var result = sysdb.getSchedule("sched-1"); assertTrue(result.isPresent()); @@ -818,10 +818,11 @@ public void testGetScheduleNotFound() { @Test public void testCreateScheduleDuplicate() { - sysdb.createSchedule(makeSchedule("sched-dup")); + sysdb.createSchedule(makeSchedule("sched-dup"), null); // SchedulesDAO catches 23505 and raises its own message, so this never reaches dbRetry's // terminal path. What dbRetry throws is pinned by dbRetryThrowsTypedOnANonRetryableFailure. - assertThrows(RuntimeException.class, () -> sysdb.createSchedule(makeSchedule("sched-dup"))); + assertThrows( + RuntimeException.class, () -> sysdb.createSchedule(makeSchedule("sched-dup"), null)); } @Test @@ -918,8 +919,8 @@ public void testCreateScheduleNullStatusThrows() { @Test public void testListSchedules() { - sysdb.createSchedule(makeSchedule("alpha-1")); - sysdb.createSchedule(makeSchedule("alpha-2")); + sysdb.createSchedule(makeSchedule("alpha-1"), null); + sysdb.createSchedule(makeSchedule("alpha-2"), null); sysdb.createSchedule( new WorkflowSchedule( null, @@ -933,8 +934,9 @@ public void testListSchedules() { false, null, null, - null)); - sysdb.pauseSchedule("beta-1"); + null), + null); + sysdb.pauseSchedule("beta-1", null); // list all var all = sysdb.listSchedules(null, null, null); @@ -987,11 +989,11 @@ public void testListSchedules() { @Test public void testListSchedulesWithSqlLikeWildcards() { // Test that SQL LIKE wildcards in prefixes are escaped correctly - sysdb.createSchedule(makeSchedule("test%s Sched")); - sysdb.createSchedule(makeSchedule("test%sother")); - sysdb.createSchedule(makeSchedule("test_schedule")); - sysdb.createSchedule(makeSchedule("test_sched")); - sysdb.createSchedule(makeSchedule("testA_schedB")); + sysdb.createSchedule(makeSchedule("test%s Sched"), null); + sysdb.createSchedule(makeSchedule("test%sother"), null); + sysdb.createSchedule(makeSchedule("test_schedule"), null); + sysdb.createSchedule(makeSchedule("test_sched"), null); + sysdb.createSchedule(makeSchedule("testA_schedB"), null); // "test%s" should only match "test%s Sched" and "test%sother" (literal %) var byPercent = sysdb.listSchedules(null, null, List.of("test%s")); @@ -1010,18 +1012,18 @@ public void testListSchedulesWithSqlLikeWildcards() { @Test public void testPauseAndResumeSchedule() { - sysdb.createSchedule(makeSchedule("sched-pause")); + sysdb.createSchedule(makeSchedule("sched-pause"), null); - sysdb.pauseSchedule("sched-pause"); + sysdb.pauseSchedule("sched-pause", null); assertEquals(ScheduleStatus.PAUSED, sysdb.getSchedule("sched-pause").get().status()); - sysdb.resumeSchedule("sched-pause"); + sysdb.resumeSchedule("sched-pause", null); assertEquals(ScheduleStatus.ACTIVE, sysdb.getSchedule("sched-pause").get().status()); } @Test public void testUpdateLastFiredAt() { - sysdb.createSchedule(makeSchedule("sched-fired")); + sysdb.createSchedule(makeSchedule("sched-fired"), null); assertNull(sysdb.getSchedule("sched-fired").get().lastFiredAt()); sysdb.updateScheduleLastFiredAt("sched-fired", Instant.parse("2026-03-26T10:00:00Z")); @@ -1032,15 +1034,15 @@ public void testUpdateLastFiredAt() { @Test public void testDeleteSchedule() { - sysdb.createSchedule(makeSchedule("sched-del-1")); - sysdb.createSchedule(makeSchedule("sched-del-2")); + sysdb.createSchedule(makeSchedule("sched-del-1"), null); + sysdb.createSchedule(makeSchedule("sched-del-2"), null); assertEquals(2, sysdb.listSchedules(null, null, null).size()); - sysdb.deleteSchedule("sched-del-1"); + sysdb.deleteSchedule("sched-del-1", null); assertFalse(sysdb.getSchedule("sched-del-1").isPresent()); assertEquals(1, sysdb.listSchedules(null, null, null).size()); - sysdb.deleteSchedule("sched-del-2"); + sysdb.deleteSchedule("sched-del-2", null); assertFalse(sysdb.getSchedule("sched-del-2").isPresent()); assertEquals(0, sysdb.listSchedules(null, null, null).size()); } @@ -1061,7 +1063,7 @@ public void testCreateScheduleWithAllFields() { ZoneId.of("America/New_York"), "my-queue", null); - sysdb.createSchedule(schedule); + sysdb.createSchedule(schedule, null); var s = sysdb.getSchedule("sched-full").get(); assertEquals("my-id-123", s.id()); diff --git a/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java b/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java new file mode 100644 index 000000000..249bfc6bb --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java @@ -0,0 +1,121 @@ +package dev.dbos.transact.scheduled; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import dev.dbos.transact.DBOS; +import dev.dbos.transact.StartWorkflowOptions; +import dev.dbos.transact.context.DBOSContextHolder; +import dev.dbos.transact.internal.DebugTriggers; +import dev.dbos.transact.utils.PgContainer; +import dev.dbos.transact.workflow.Workflow; +import dev.dbos.transact.workflow.WorkflowSchedule; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.ResourceLock; + +/** + * A schedule written from a workflow commits with its checkpoint, and a refusal is recorded as the + * step's answer, so a replay is refused the same way after the obstacle is gone. + */ +class ScheduleCheckpointTest { + + @AutoClose final PgContainer pgContainer = new PgContainer(); + @AutoClose DBOS dbos; + + private ScheduleCreator proxy; + + @BeforeEach + void beforeEach() { + dbos = new DBOS(pgContainer.dbosConfig()); + dbos.registerProxy(ScheduledWorkflowService.class, new ScheduledWorkflowImpl()); + proxy = dbos.registerProxy(ScheduleCreator.class, new ScheduleCreatorImpl(dbos)); + dbos.launch(); + } + + static WorkflowSchedule schedule(String name) { + return new WorkflowSchedule( + name, "scheduledRun", ScheduledWorkflowImpl.class.getName(), "0 0 * * * *"); + } + + @Test + public void aRefusedCreateReplaysItsRefusalOnceTheNameIsFree() throws Exception { + var name = "taken-%d".formatted(System.nanoTime()); + dbos.createSchedule(schedule(name)); + var workflowId = "creator-%d".formatted(System.nanoTime()); + + var result = + dbos.startWorkflow(() -> proxy.creates(name), new StartWorkflowOptions(workflowId)) + .getResult(); + assertEquals("refused", result); + var steps = dbos.listWorkflowSteps(workflowId); + assertEquals(1, steps.size()); + assertEquals("DBOS.createSchedule", steps.get(0).functionName()); + assertNotNull(steps.get(0).error(), "the refusal is the step's answer"); + + // With the holder gone, a create would now succeed; the fork replays the recorded refusal. + dbos.deleteSchedule(name); + assertEquals("refused", dbos.forkWorkflow(workflowId, 1).getResult()); + assertTrue(dbos.getSchedule(name).isEmpty(), "the replay created nothing"); + } + + @Test + @ResourceLock(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT) + public void aCreateThatFailsBeforeCommitLeavesNoSchedule() throws Exception { + var name = "crashed-%d".formatted(System.nanoTime()); + var workflowId = "creator-%d".formatted(System.nanoTime()); + var fired = new AtomicBoolean(); + // Armed for this workflow only: the trigger is global, and other classes' steps pass it. + DebugTriggers.setDebugTrigger( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT, + new DebugTriggers.DebugAction() + .setCallback( + () -> { + if (workflowId.equals(DBOSContextHolder.get().getWorkflowId()) + && fired.compareAndSet(false, true)) { + throw new RuntimeException("crash"); + } + })); + String result; + try { + result = + dbos.startWorkflow(() -> proxy.creates(name), new StartWorkflowOptions(workflowId)) + .getResult(); + } finally { + DebugTriggers.clearDebugTrigger(DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMIT); + } + + assertEquals("refused", result); + assertTrue(fired.get()); + assertTrue(dbos.getSchedule(name).isEmpty(), "the schedule rolled back with the step"); + assertTrue(dbos.listWorkflowSteps(workflowId).isEmpty(), "and so did its checkpoint"); + } +} + +interface ScheduleCreator { + String creates(String name); +} + +class ScheduleCreatorImpl implements ScheduleCreator { + private final DBOS dbos; + + ScheduleCreatorImpl(DBOS dbos) { + this.dbos = dbos; + } + + @Override + @Workflow + public String creates(String name) { + try { + dbos.createSchedule(ScheduleCheckpointTest.schedule(name)); + return "created"; + } catch (RuntimeException e) { + return "refused"; + } + } +} diff --git a/transact/src/test/java/dev/dbos/transact/workflow/DebouncerTest.java b/transact/src/test/java/dev/dbos/transact/workflow/DebouncerTest.java index cf76607fc..992886a34 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/DebouncerTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/DebouncerTest.java @@ -1149,6 +1149,44 @@ public void aTakeoverWhoseCreateCollidesRollsBack() throws Exception { assertTrue(dbos.getWorkflowStatus("promised-17").isEmpty()); } + /** + * The same collision inside a workflow: the takeover rolls back, and its step records that it + * took nothing over, so a replay returns null again without depending on how the serializer + * restores exceptions. + */ + @Test + @ResourceLock( + value = DebugTriggers.DEBUG_TRIGGER_DEBOUNCE_TAKEOVER, + mode = ResourceAccessMode.READ) + public void aCollidingTakeoverInsideAWorkflowRecordsNothingTakenOver() throws Exception { + dbos.registerProxy(DebouncedService.class, serviceImpl); + var runnerImpl = new TakeoverRunnerImpl(); + TakeoverRunner runner = dbos.registerProxy(TakeoverRunner.class, runnerImpl); + dbos.launch(); + var queue = "takeover-collision-in-workflow"; + dbos.registerQueue(queue, new QueueOptions()); + var stranded = plantDebouncerWorkflow("mixed", "promised-18", queue, "no-such-version"); + DebouncedRows.insert( + pgContainer.dataSource(), + debouncedRow(queue, "process", System.currentTimeMillis() + 60_000)); + runnerImpl.body = () -> takeOver(stranded, "mixed", "fresh", queue); + + var runnerId = "wf-colliding-takeover"; + try (var o = new WorkflowOptions(runnerId).setContext()) { + assertNull(runner.run()); + } + var steps = + dbos.listWorkflowSteps(runnerId).stream() + .filter(step -> step.functionName().equals("DBOS.takeOverStrandedDebouncer")) + .toList(); + assertEquals(1, steps.size()); + assertNull(steps.get(0).error(), "the collision is not an error"); + + assertNull(replay(runnerId)); + assertEquals(WorkflowState.ENQUEUED, dbos.retrieveWorkflow(stranded).getStatus().status()); + assertTrue(dbos.getWorkflowStatus("promised-18").isEmpty()); + } + /** * Inside a workflow the takeover is a step, checkpointed in the transaction that cancels the * debouncer workflow and creates the promised one, and a replay returns it rather than cancelling @@ -1846,3 +1884,17 @@ private void flipToPending(String workflowId) throws SQLException { } } } + +interface TakeoverRunner { + String run(); +} + +class TakeoverRunnerImpl implements TakeoverRunner { + java.util.function.Supplier body; + + @Override + @Workflow + public String run() { + return body.get(); + } +} 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 43806e019..396038fd5 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -2,6 +2,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -10,6 +11,7 @@ import dev.dbos.transact.StartWorkflowOptions; import dev.dbos.transact.context.DBOSContext; import dev.dbos.transact.context.DBOSContextHolder; +import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException; import dev.dbos.transact.internal.DebugTriggers; import dev.dbos.transact.utils.PgContainer; @@ -327,6 +329,81 @@ private List forksOf(String workflowId) { return dbos.listWorkflows(new ListWorkflowsInput().withForkedFrom(workflowId)); } + @Test + public void aSendToAMissingWorkflowReplaysItsRefusalOnceTheWorkflowExists() throws Exception { + var destinationId = "missing-%d".formatted(System.nanoTime()); + var callerId = "caller-%d".formatted(System.nanoTime()); + + var result = + dbos.startWorkflow(() -> proxy.sends(destinationId), new StartWorkflowOptions(callerId)) + .getResult(); + assertEquals("refused", result); + var recorded = stepsNamed(callerId, "DBOS.send"); + assertEquals(1, recorded.size()); + assertNotNull(recorded.get(0).error(), "the refusal is the step's answer"); + + // With the destination there, a send would now succeed; the fork replays the refusal. + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(destinationId)).getResult(); + assertEquals("refused", dbos.forkWorkflow(callerId, 1).getResult()); + } + + @Test + public void aLostCommitAcknowledgementReplaysTheClaimedMessage() throws Exception { + var receiverId = "receiver-%d".formatted(System.nanoTime()); + var fired = + failOnce( + DebugTriggers.DEBUG_TRIGGER_INTERNAL_STEP_COMMITTED, + receiverId, + new SQLTransientException()); + + var receiver = dbos.startWorkflow(proxy::receives, new StartWorkflowOptions(receiverId)); + dbos.send(receiverId, "first", "topic"); + dbos.send(receiverId, "second", "topic"); + + // The claim commits and its acknowledgement is lost. Claiming again would take the second. + assertEquals("first", receiver.getResult(), "the retry replayed the claim it committed"); + assertTrue(fired.get()); + assertEquals(1, stepsNamed(receiverId, "DBOS.recv").size()); + } + + /** + * A send records neither output nor serialization, as earlier releases did, and a replay of that + * row returns rather than sending again or refusing the row. + */ + @Test + public void aSendRecordsNoOutputAndReplays() throws Exception { + var destinationId = "destination-%d".formatted(System.nanoTime()); + dbos.startWorkflow(proxy::quick, new StartWorkflowOptions(destinationId)).getResult(); + var callerId = "caller-%d".formatted(System.nanoTime()); + + assertEquals( + "sent", + dbos.startWorkflow(() -> proxy.sends(destinationId), new StartWorkflowOptions(callerId)) + .getResult()); + var recorded = stepsNamed(callerId, "DBOS.send"); + assertEquals(1, recorded.size()); + assertNull(recorded.get(0).output()); + assertNull(recorded.get(0).error()); + + assertEquals("sent", dbos.forkWorkflow(callerId, 1).getResult()); + } + + @Test + public void aTimedOutRecvIsRecordedFromTheStartOfItsWait() throws Exception { + var receiverId = "receiver-%d".formatted(System.nanoTime()); + + assertEquals( + "nothing", + dbos.startWorkflow(() -> proxy.receivesFor(1000), new StartWorkflowOptions(receiverId)) + .getResult()); + var recorded = stepsNamed(receiverId, "DBOS.recv"); + assertEquals(1, recorded.size()); + var step = recorded.get(0); + assertTrue( + step.completedAtEpochMs() - step.startedAtEpochMs() >= 900, + "recorded %d ms".formatted(step.completedAtEpochMs() - step.startedAtEpochMs())); + } + @Test public void aForkThatFailsBeforeCommitLeavesNoFork() throws Exception { var sourceId = "source-%d".formatted(System.nanoTime()); @@ -584,6 +661,12 @@ interface InternalStepService { String forks(String sourceId); + String receives(); + + String receivesFor(long millis); + + String sends(String destinationId); + String rewinds(String targetId); String counted(); @@ -683,6 +766,29 @@ public String startsChildWhenReleased(String childId) throws Exception { return "started"; } + @Override + @Workflow + public String sends(String destinationId) { + try { + dbos.send(destinationId, "hello", "topic"); + return "sent"; + } catch (DBOSNonExistentWorkflowException e) { + return "refused"; + } + } + + @Override + @Workflow + public String receivesFor(long millis) { + return dbos.recv("topic", Duration.ofMillis(millis)).orElse("nothing"); + } + + @Override + @Workflow + public String receives() { + return dbos.recv("topic", Duration.ofSeconds(30)).orElse("nothing"); + } + @Override @Workflow public String forks(String sourceId) {