From 66f05542fbc6f19fc2b3e7fbbcda811d317ab1f2 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 10:17:45 -0700 Subject: [PATCH 1/4] Commit schedule changes with their checkpoint createSchedule, pauseSchedule, resumeSchedule and deleteSchedule now run through runTransactionalStep, so each commits its effect and its step checkpoint in one transaction under the step name it already used, and a refusal is recorded as the step's error. A create refused because the name is taken therefore replays that refusal even once the holder is gone, rather than creating the schedule on recovery. Called from inside a step, each commits only while the caller still owns its workflow; from a client or the Conductor, it runs in a plain transaction. The DAO methods take the caller's connection instead of opening their own. --- .../java/dev/dbos/transact/DBOSClient.java | 8 +- .../transact/database/SystemDatabase.java | 44 ++++++++-- .../transact/database/dao/SchedulesDAO.java | 37 ++++---- .../execution/DBOSConductorEngine.java | 4 +- .../dbos/transact/execution/DBOSExecutor.java | 32 +------ .../transact/database/SystemDatabaseTest.java | 44 +++++----- .../scheduled/ScheduleCheckpointTest.java | 84 +++++++++++++++++++ 7 files changed, 170 insertions(+), 83 deletions(-) create mode 100644 transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java 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 29ab0c73f..8d53985ed 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -2074,8 +2074,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) { @@ -2113,20 +2120,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) { 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/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 c5dd9a42b..71f74ba04 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1295,13 +1295,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) { @@ -1326,33 +1320,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/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..f02131137 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java @@ -0,0 +1,84 @@ +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.utils.PgContainer; +import dev.dbos.transact.workflow.Workflow; +import dev.dbos.transact.workflow.WorkflowSchedule; + +import org.junit.jupiter.api.AutoClose; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * 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"); + } +} + +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"; + } + } +} From f424e16b6feeef86345728187ac1fc4a3b91dd27 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 10:22:54 -0700 Subject: [PATCH 2/4] Run send, setEvent and stream writes through the transactional step helper send, sendBulk, setEvent, writeStream and closeStream already committed their effect and their checkpoint together, each with its own copy of the check-run-record transaction. They now share runTransactionalStep, and so gain what it adds: a failure is recorded as the step's error, so a send refused because its destination does not exist replays that refusal even once the destination is created; a retry after a lost commit acknowledgement replays the checkpoint; and the transaction runs at READ COMMITTED. A checkpoint with neither output nor error, which is what these wrote, now replays as a null output rather than being rejected, so steps recorded by earlier releases replay unchanged. An empty send still records nothing, as before. Their DAO methods are reduced to the write on the caller's connection. --- .../transact/database/SystemDatabase.java | 85 ++++-- .../database/dao/NotificationsDAO.java | 248 +++++------------- .../transact/database/dao/StreamsDAO.java | 79 +----- .../workflow/InternalStepTransactionTest.java | 33 +++ 4 files changed, 169 insertions(+), 276 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index 8d53985ed..615b86e1f 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1581,11 +1581,17 @@ 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; + } + runTransactionalStep( + caller, + functionName, + conn -> { + NotificationsDAO.sendBulk(ctx, conn, messages, sendToForks, serialization); + return null; + }); } public Object recv( @@ -1617,11 +1623,27 @@ 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); + runTransactionalStep( + caller, + "DBOS.setEvent", + conn -> { + var serialized = + SerializationUtil.serializeValue(message, serialization, ctx.serializer()); + NotificationsDAO.setEvent( + conn, + ctx.schema(), + workflowId, + functionId, + key, + serialized.serializedValue(), + serialized.serialization()); + return null; + }); signal(new SignalKey.Event(workflowId, key)); } @@ -1852,11 +1874,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( @@ -2241,11 +2260,15 @@ 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. + runTransactionalStep( + new StepCaller.FromStep(workflowId, ownerXid), + "DBOS.writeStream", + conn -> { + StreamsDAO.writeStream( + conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); + return null; + }); signal(new SignalKey.Stream(workflowId, key)); } @@ -2256,17 +2279,25 @@ public void writeStreamFromWorkflow( Object value, String serializationFormat, String ownerXid) { - dbRetryIncludingSerializationError( - "writeStreamFromWorkflow", - () -> - StreamsDAO.writeStreamFromWorkflow( - ctx, workflowId, functionId, key, value, serializationFormat, ownerXid)); + runTransactionalStep( + new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), + StreamsDAO.writeStepName(value), + conn -> { + StreamsDAO.writeStream( + conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); + return null; + }); 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)); + runTransactionalStep( + new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), + "DBOS.closeStream", + conn -> { + StreamsDAO.closeStream(conn, ctx.schema(), workflowId, functionId, key); + return null; + }); // 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..7d53ac3b8 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 @@ -6,13 +6,11 @@ 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 +92,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 +111,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,107 +121,68 @@ 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; } } @@ -381,7 +338,11 @@ public static Object recv( } } - 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 +387,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/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/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java index a5f96e6b9..051e7aa3b 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; @@ -180,6 +182,24 @@ 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 aForkThatFailsBeforeCommitLeavesNoFork() throws Exception { var sourceId = "source-%d".formatted(System.nanoTime()); @@ -390,6 +410,8 @@ interface InternalStepService { String forks(String sourceId); + String sends(String destinationId); + String rewinds(String targetId); String startsChild(String childId) throws Exception; @@ -474,6 +496,17 @@ 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 forks(String sourceId) { From 1563f959737c6731aa040719f0d947d2235fd513 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 10:32:36 -0700 Subject: [PATCH 3/4] Run the recv claim and the debouncer's steps through the step helper The recv claim, the debounce bounce and the stranded-debouncer takeover each had their own copy of the check-run-record transaction. They now share the helper, and so gain failure recording, a replay of the checkpoint after a lost commit acknowledgement, and READ COMMITTED. runRecordedStep is the helper for work that says what its checkpoint holds; runTransactionalStep is now the case that serializes the value the work returns. The recv claim records the message as it was stored, and the takeover names the workflow it created as the step's child, as before. recv is split into its wait and its claim. Only the claim is the step's transaction and only the claim is retried, now on connection failures as well as conflicts, and it re-checks the step inside that transaction. A timed-out recv records null through the helper too. --- .../transact/database/SystemDatabase.java | 133 ++++++++++++---- .../database/dao/NotificationsDAO.java | 121 +++++++-------- .../transact/database/dao/WorkflowDAO.java | 144 ++++-------------- .../workflow/InternalStepTransactionTest.java | 28 ++++ 4 files changed, 214 insertions(+), 212 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index 615b86e1f..22b4e6e35 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -1372,17 +1372,34 @@ 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", + conn -> { + String created = + WorkflowDAO.takeOverDebouncerWorkflow( + ctx, conn, debouncerWorkflowId, promised, delayUntilEpochMs, ownerXid); + 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()); } /** @@ -1405,11 +1422,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, @@ -1418,8 +1437,7 @@ public Object debounceDelayedWorkflow( delayUntilEpochMs, args, serializationFormat, - ids, - caller)); + ids)); } /** Returns the given executors' PENDING workflows to their queues; reports which rows moved. */ @@ -1594,6 +1612,14 @@ public void sendBulk( }); } + /** + * Receives a message on {@code topic} as step {@code stepId}, waiting up to {@code timeout}. + * + *

Only the claim is the step's transaction, and only the claim is retried, never the wait + * before it: re-running the wait would register a fresh subscription and restart the caller's + * timeout from zero. A peer that took the message between the wait and the claim leaves the claim + * nothing, and the step records null. + */ public Object recv( String workflowId, int stepId, @@ -1601,18 +1627,30 @@ 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) { + return runTransactionalStep(step, "DBOS.recv", conn -> null); + } + return runRecordedStep( + step, + "DBOS.recv", + conn -> NotificationsDAO.claimMessage(ctx, conn, workflowId, stepId, topic)); } public void setEvent( @@ -1780,6 +1818,37 @@ public void deleteWorkflows( */ @Nullable T runTransactionalStep( @Nullable StepCaller caller, String stepName, SqlTransaction.SqlCall work) { + boolean recorded = caller instanceof StepCaller.FromWorkflow; + return runRecordedStep( + caller, + stepName, + 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) {} + + /** + * As {@link #runTransactionalStep}, for work that says what its checkpoint records: a message + * kept in the serialization it arrived in, or a step naming the child workflow it created. + */ + @Nullable T runRecordedStep( + @Nullable StepCaller caller, String stepName, SqlTransaction.SqlCall> work) { var step = caller instanceof StepCaller.FromWorkflow fromWorkflow ? fromWorkflow : null; long startTime = System.currentTimeMillis(); // What work itself threw on the attempt that ended the retries, as against a failure of the @@ -1817,7 +1886,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) { @@ -1827,8 +1896,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, @@ -1836,10 +1903,10 @@ record Outcome(@Nullable StepResult replayed, @Nullable T output) {} step.workflowId(), step.stepId(), stepName, - serialized.serializedValue(), - null, + result.output(), null, - serialized.serialization()), + result.childWorkflowId(), + result.serialization()), startTime, System.currentTimeMillis(), step.ownerXid()); @@ -1849,7 +1916,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( 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 7d53ac3b8..b4dcd73c5 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,7 +5,6 @@ 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.SystemDatabase; import dev.dbos.transact.database.signal.SignalKey; import dev.dbos.transact.database.signal.SignalMap; @@ -186,7 +185,23 @@ ON CONFLICT (message_uuid) DO NOTHING } } - 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. */ + record TimedOut() implements RecvWait {} + + /** A message is waiting to be claimed. */ + record Ready() 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, @@ -203,23 +218,24 @@ 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( "Running recv, workflowId: {}, stepId: {}, topic: {}", workflowId, stepId, recvTopic); - var startTime = System.currentTimeMillis(); var messageKey = new SignalKey.Message(workflowId, recvTopic); var selectSql = """ @@ -239,8 +255,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(); } } WorkflowDAO.checkWorkflow(conn, ctx.schema(), workflowId, ownerXid); @@ -248,11 +263,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(); } var loopDuration = @@ -261,10 +272,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 @@ -284,58 +302,27 @@ 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); } /** 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 481076104..d1aaf8475 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( @@ -1096,11 +1054,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 +1068,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/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java index 051e7aa3b..7630acb63 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -17,6 +17,7 @@ import java.sql.SQLException; import java.sql.SQLTransientException; +import java.time.Duration; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; @@ -200,6 +201,25 @@ public void aSendToAMissingWorkflowReplaysItsRefusalOnceTheWorkflowExists() thro 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()); + } + @Test public void aForkThatFailsBeforeCommitLeavesNoFork() throws Exception { var sourceId = "source-%d".formatted(System.nanoTime()); @@ -410,6 +430,8 @@ interface InternalStepService { String forks(String sourceId); + String receives(); + String sends(String destinationId); String rewinds(String targetId); @@ -507,6 +529,12 @@ public String sends(String destinationId) { } } + @Override + @Workflow + public String receives() { + return dbos.recv("topic", Duration.ofSeconds(30)).orElse("nothing"); + } + @Override @Workflow public String forks(String sourceId) { From 1854e467730570668a990511681808b3167053c1 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 30 Oct 2026 11:38:21 -0700 Subject: [PATCH 4/4] Record no output for send and stream steps, recv's wait as its start, and a takeover collision as nothing taken over send, sendBulk, setEvent, writeStream and closeStream record neither an output nor a serialization again, as they did before they ran through the step helper, instead of a serialized null tagged with this application's serializer, which other readers of the step would have to decode for a step that returns nothing. runRecordedStep takes the step's start time, and a recv records the start of its wait, so a recv that waited shows the wait in its step's duration again. A stranded-debouncer takeover whose create collides with whatever else holds the key now rolls back to a savepoint taken before the takeover, undoing the cancel with it, and records that it took nothing over. The caller already treated the collision as exactly that; recording it as an error made a replay depend on the serializer restoring the exception's class. The recv javadoc no longer claims a retried wait restarts its timeout, whose deadline is durable, and debounceDelayedWorkflow's javadoc no longer describes a caller it does not take. New tests: a send's row replays with no output, a timed-out recv's step spans its wait, a schedule create that fails before commit leaves neither schedule nor checkpoint, and a colliding takeover inside a workflow records no error and replays as null. --- .../transact/database/SystemDatabase.java | 95 ++++++++++++++----- .../database/dao/NotificationsDAO.java | 13 +-- .../transact/database/dao/WorkflowDAO.java | 7 +- .../workflow/internal/InternalWorkflows.java | 34 +++---- .../scheduled/ScheduleCheckpointTest.java | 37 ++++++++ .../dbos/transact/workflow/DebouncerTest.java | 52 ++++++++++ .../workflow/InternalStepTransactionTest.java | 46 +++++++++ 7 files changed, 228 insertions(+), 56 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java index f12c95b49..eeedf9950 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; @@ -1382,10 +1383,28 @@ public Object takeOverDebouncerWorkflow( return runRecordedStep( stepOf(caller), caller != null ? caller.stepName() : "takeOverDebouncerWorkflow", + System.currentTimeMillis(), conn -> { - String created = - WorkflowDAO.takeOverDebouncerWorkflow( - ctx, conn, debouncerWorkflowId, promised, delayUntilEpochMs, ownerXid); + 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); } @@ -1603,22 +1622,24 @@ public void sendBulk( if (messages.isEmpty()) { return; } - runTransactionalStep( + runRecordedStep( caller, functionName, + System.currentTimeMillis(), conn -> { NotificationsDAO.sendBulk(ctx, conn, messages, sendToForks, serialization); - return null; + return Recorded.nothing(); }); } /** * Receives a message on {@code topic} as step {@code stepId}, waiting up to {@code timeout}. * - *

Only the claim is the step's transaction, and only the claim is retried, never the wait - * before it: re-running the wait would register a fresh subscription and restart the caller's - * timeout from zero. A peer that took the message between the wait and the claim leaves the claim - * nothing, and the step records null. + *

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, @@ -1644,12 +1665,22 @@ public Object recv( return replayed.message(); } var step = new StepCaller.FromWorkflow(workflowId, stepId, ownerXid); - if (waited instanceof NotificationsDAO.RecvWait.TimedOut) { - return runTransactionalStep(step, "DBOS.recv", conn -> null); + 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)); } @@ -1666,9 +1697,10 @@ public void setEvent( asStep ? new StepCaller.FromWorkflow(workflowId, functionId, ownerXid) : new StepCaller.FromStep(workflowId, ownerXid); - runTransactionalStep( + runRecordedStep( caller, "DBOS.setEvent", + System.currentTimeMillis(), conn -> { var serialized = SerializationUtil.serializeValue(message, serialization, ctx.serializer()); @@ -1680,7 +1712,7 @@ public void setEvent( key, serialized.serializedValue(), serialized.serialization()); - return null; + return Recorded.nothing(); }); signal(new SignalKey.Event(workflowId, key)); } @@ -1831,6 +1863,7 @@ public void deleteWorkflows( return runRecordedStep( caller, stepName, + System.currentTimeMillis(), conn -> { T result = work.run(conn); if (!recorded) { @@ -1850,16 +1883,29 @@ public record Recorded( @Nullable T value, @Nullable String output, @Nullable String serialization, - @Nullable String childWorkflowId) {} + @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, or a step naming the child workflow it created. + * 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, SqlTransaction.SqlCall> work) { + @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(); @@ -2357,13 +2403,14 @@ public void writeStreamFromStep( String serializationFormat, String ownerXid) { // Even from inside a step the entry names a step, the one writing it. - runTransactionalStep( + runRecordedStep( new StepCaller.FromStep(workflowId, ownerXid), "DBOS.writeStream", + System.currentTimeMillis(), conn -> { StreamsDAO.writeStream( conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); - return null; + return Recorded.nothing(); }); signal(new SignalKey.Stream(workflowId, key)); } @@ -2375,24 +2422,26 @@ public void writeStreamFromWorkflow( Object value, String serializationFormat, String ownerXid) { - runTransactionalStep( + runRecordedStep( new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), StreamsDAO.writeStepName(value), + System.currentTimeMillis(), conn -> { StreamsDAO.writeStream( conn, ctx.schema(), workflowId, functionId, key, value, serializationFormat); - return null; + return Recorded.nothing(); }); signal(new SignalKey.Stream(workflowId, key)); } public void closeStream(String workflowId, int functionId, String key, String ownerXid) { - runTransactionalStep( + runRecordedStep( new StepCaller.FromWorkflow(workflowId, functionId, ownerXid), "DBOS.closeStream", + System.currentTimeMillis(), conn -> { StreamsDAO.closeStream(conn, ctx.schema(), workflowId, functionId, key); - return null; + 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 b4dcd73c5..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 @@ -190,11 +190,11 @@ 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. */ - record TimedOut() 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. */ - record Ready() implements RecvWait {} + /** A message is waiting to be claimed; the wait began at {@code startedAtEpochMs}. */ + record Ready(long startedAtEpochMs) implements RecvWait {} } /** @@ -236,6 +236,7 @@ public static RecvWait awaitMessage( logger.debug( "Running recv, workflowId: {}, stepId: {}, topic: {}", workflowId, stepId, recvTopic); + var startTime = System.currentTimeMillis(); var messageKey = new SignalKey.Message(workflowId, recvTopic); var selectSql = """ @@ -255,7 +256,7 @@ public static RecvWait awaitMessage( stmt.setString(2, recvTopic); try (var rs = stmt.executeQuery()) { if (rs.next()) { - return new RecvWait.Ready(); + return new RecvWait.Ready(startTime); } } WorkflowDAO.checkWorkflow(conn, ctx.schema(), workflowId, ownerXid); @@ -263,7 +264,7 @@ public static RecvWait awaitMessage( var duration = Duration.between(Instant.now(), timeoutAt); if (duration.isNegative() || duration.isZero()) { - return new RecvWait.TimedOut(); + return new RecvWait.TimedOut(startTime); } var loopDuration = 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 d1aaf8475..653bb68a7 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 @@ -1043,10 +1043,9 @@ public static void checkOwner(Connection conn, String schema, String workflowId, *

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 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/scheduled/ScheduleCheckpointTest.java b/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java index f02131137..249bfc6bb 100644 --- a/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java +++ b/transact/src/test/java/dev/dbos/transact/scheduled/ScheduleCheckpointTest.java @@ -6,13 +6,18 @@ 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 @@ -58,6 +63,38 @@ public void aRefusedCreateReplaysItsRefusalOnceTheNameIsFree() throws Exception 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 { 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 c5199bec4..6cd24a044 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/InternalStepTransactionTest.java @@ -226,6 +226,44 @@ public void aLostCommitAcknowledgementReplaysTheClaimedMessage() throws Exceptio 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()); @@ -485,6 +523,8 @@ interface InternalStepService { String receives(); + String receivesFor(long millis); + String sends(String destinationId); String rewinds(String targetId); @@ -589,6 +629,12 @@ public String sends(String destinationId) { } } + @Override + @Workflow + public String receivesFor(long millis) { + return dbos.recv("topic", Duration.ofMillis(millis)).orElse("nothing"); + } + @Override @Workflow public String receives() {