From e07f89b2b61d2086bab5d366179f76c063c0717f Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 28 Aug 2026 10:59:53 -0700 Subject: [PATCH 1/5] Read an empty error column as no error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Be liberal in what you accept. On a system database shared by several applications the row may have been written by another SDK, and the Go SDK stores a workflow's error as a non-nullable string — so a workflow that succeeded leaves "" behind where this one writes NULL. Java parsed whatever was there as JSON and threw `MismatchedInputException: No content to map due to end-of-input`, on a workflow that never failed. The other SDKs already read the two the same way. TypeScript coerces on the way in (`row.error ? row.error : null`), and Python never reads the column unless the status says ERROR. Empty, not blank: no SDK writes a non-empty run of whitespace here, so a value that is blank without being empty is content to hand back rather than quietly discard. Applied where the column is read — a workflow's and a step's alike, including the transactional step schema — rather than inside the deserializers, which should not have to know how a column came to be written. What this SDK writes is unchanged and stays conservative: an absent error is NULL. --- .../transact/database/SystemDatabase.java | 10 ++ .../dbos/transact/database/dao/StepsDAO.java | 5 +- .../transact/database/dao/WorkflowDAO.java | 4 +- .../dbos/transact/txstep/TxStepSchema.java | 3 +- .../dev/dbos/transact/json/InteropTest.java | 102 ++++++++++++++++++ 5 files changed, 119 insertions(+), 5 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 131dcd48..567776c6 100644 --- a/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java +++ b/transact/src/main/java/dev/dbos/transact/database/SystemDatabase.java @@ -437,6 +437,16 @@ public static Duration toDuration(Long ms) { return ms != null ? Duration.ofMillis(ms) : null; } + /** + * The error column of a workflow or a step, or null when it records no error. + * + *

The Go SDK stores the error as a non-nullable string, so a workflow that succeeded leaves "" + * behind where this one writes NULL. Empty only: no SDK writes a blank non-empty value. + */ + public static String errorOrNull(String error) { + return error != null && error.isEmpty() ? null : error; + } + /** * Initializes the status of a workflow. * diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java index 2791a4ac..a4551add 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java @@ -1,6 +1,7 @@ package dev.dbos.transact.database.dao; import dev.dbos.transact.database.DbContext; +import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.exceptions.*; import dev.dbos.transact.internal.DebugTriggers; import dev.dbos.transact.json.DBOSSerializer; @@ -144,7 +145,7 @@ public static StepResult checkStepResult( try (ResultSet rs = pstmt.executeQuery()) { if (rs.next()) { // Check if any operation output row exists String output = rs.getString("output"); - String error = rs.getString("error"); + String error = SystemDatabase.errorOrNull(rs.getString("error")); String _stepName = rs.getString("function_name"); String serialization = rs.getString("serialization"); result = @@ -222,7 +223,7 @@ static List listWorkflowSteps( int functionId = rs.getInt("function_id"); String functionName = rs.getString("function_name"); String outputData = rs.getString("output"); - String errorData = rs.getString("error"); + String errorData = SystemDatabase.errorOrNull(rs.getString("error")); String childWorkflowId = rs.getString("child_workflow_id"); Long startedAt = rs.getObject("started_at_epoch_ms", Long.class); Long completedAt = rs.getObject("completed_at_epoch_ms", Long.class); 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 894d4a5c..4835a017 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 @@ -1233,7 +1233,7 @@ private static WorkflowStatus resultsToWorkflowStatus( String attributesJson = rs.getString("attributes"); String serializedInput = loadInput ? rs.getString("inputs") : null; String serializedOutput = loadOutput ? rs.getString("output") : null; - String serializedError = loadOutput ? rs.getString("error") : null; + String serializedError = loadOutput ? SystemDatabase.errorOrNull(rs.getString("error")) : null; String serialization = loadInput || loadOutput ? rs.getString("serialization") : null; WorkflowStatus info = new WorkflowStatus( @@ -1327,7 +1327,7 @@ public static Result awaitWorkflowResult( } case ERROR -> { - String error = rs.getString("error"); + String error = SystemDatabase.errorOrNull(rs.getString("error")); Throwable t = SerializationUtil.deserializeError(error, serialization, serializer); return Result.failure(t); } diff --git a/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java b/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java index 0abf5cc7..4d7947de 100644 --- a/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java +++ b/transact/src/main/java/dev/dbos/transact/txstep/TxStepSchema.java @@ -1,5 +1,6 @@ package dev.dbos.transact.txstep; +import dev.dbos.transact.database.SystemDatabase; import dev.dbos.transact.workflow.internal.StepResult; import java.sql.Connection; @@ -66,7 +67,7 @@ public static Optional readResult( stepId, stepName, rs.getString("output"), - rs.getString("error"), + SystemDatabase.errorOrNull(rs.getString("error")), null, rs.getString("serialization"))); } diff --git a/transact/src/test/java/dev/dbos/transact/json/InteropTest.java b/transact/src/test/java/dev/dbos/transact/json/InteropTest.java index 09565c3e..87399dae 100644 --- a/transact/src/test/java/dev/dbos/transact/json/InteropTest.java +++ b/transact/src/test/java/dev/dbos/transact/json/InteropTest.java @@ -229,6 +229,66 @@ private void insertPortableWorkflowRow( } } + /** + * Insert a completed workflow row as another SDK would leave it. + * + *

{@code serialization} and {@code error} are what vary between them: Go stores the error as a + * non-nullable string and so writes "" for a workflow that succeeded, and every SDK writes its + * own native format unless the workflow was declared portable. + */ + private void insertPeerWorkflowRow( + String workflowId, String serialization, String inputs, String output, String error) + throws Exception { + try (Connection conn = dataSource.getConnection()) { + String sql = + """ + INSERT INTO dbos.workflow_status( + workflow_uuid, name, class_name, config_name, + status, inputs, output, error, created_at, serialization, application_name + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """; + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + stmt.setString(2, "echoWorkflow"); + stmt.setString(3, "interop"); + stmt.setString(4, null); + stmt.setString(5, "SUCCESS"); + stmt.setString(6, inputs); + stmt.setString(7, output); + stmt.setString(8, error); + stmt.setLong(9, System.currentTimeMillis()); + stmt.setString(10, serialization); + stmt.setString(11, "interop-peer"); + stmt.executeUpdate(); + } + } + } + + /** Insert a completed step row as another SDK would leave it. */ + private void insertPeerStepRow( + String workflowId, String serialization, String output, String error) throws Exception { + try (Connection conn = dataSource.getConnection()) { + String sql = + """ + INSERT INTO dbos.operation_outputs( + workflow_uuid, function_id, function_name, output, error, serialization, application_name + ) + VALUES (?, ?, ?, ?, ?, ?, ?) + """; + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + stmt.setInt(2, 0); + stmt.setString(3, "echoStep"); + stmt.setString(4, output); + stmt.setString(5, error); + stmt.setString(6, serialization); + stmt.setString(7, "interop-peer"); + stmt.executeUpdate(); + } + } + } + private void insertPortableNotification(String destinationUuid, String topic, String messageJson) throws Exception { try (Connection conn = dataSource.getConnection()) { @@ -422,4 +482,46 @@ public void testInteropNamedArgs() throws Exception { assertEquals(Arrays.asList("a", "b"), storedNamedArgs.get("tags")); } } + + // ============================================================================ + // Test: reading a workflow another application owns + // ============================================================================ + + /** + * A workflow that succeeded, whose error column holds "" rather than NULL. + * + *

That is what the Go SDK leaves behind — it stores the error as a non-nullable string — and + * on a shared system database this runtime is routinely asked about such a row. An empty column + * has to mean the same thing as an absent one; parsing it as JSON does not. + */ + @Test + public void testStatusReadTreatsAnEmptyErrorColumnAsNoError() throws Exception { + String workflowId = "peer-empty-error"; + insertPeerWorkflowRow( + workflowId, "portable_json", "{\"positionalArgs\":[]}", "{\"ok\":true}", ""); + + dbos.launch(); + var status = dbos.getWorkflowStatus(workflowId).orElseThrow(); + + assertEquals(WorkflowState.SUCCESS, status.status()); + assertNull(status.error(), "an empty error column is not an error"); + } + + /** + * A step of a peer's workflow, recorded with an empty error column. + * + *

Steps carry the same column with the same quirk, so they get the same reading. + */ + @Test + public void testStepReadTreatsAnEmptyErrorColumnAsNoError() throws Exception { + String workflowId = "peer-step-empty-error"; + insertPeerWorkflowRow(workflowId, "portable_json", "{\"positionalArgs\":[]}", "{}", ""); + insertPeerStepRow(workflowId, "portable_json", "{\"ok\":true}", ""); + + dbos.launch(); + var steps = dbos.listWorkflowSteps(workflowId); + + assertEquals(1, steps.size()); + assertNull(steps.get(0).error(), "an empty error column is not an error"); + } } From 11897bc27a8aeca00d3b67ef0f16266d0a1f41a0 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 28 Aug 2026 11:00:21 -0700 Subject: [PATCH 2/5] Don't try to deserialize a format we don't recognize MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `DBOS.getWorkflowStatus` threw `IllegalArgumentException: Serialization is not available` on any row written in a format this runtime has no deserializer for. Workflow IDs address the whole system database, so a status read reaches rows another application owns on purpose — and what is wanted from such a row is the metadata: who owns it, what it is, whether it finished. Losing all of that over a payload the caller may not have asked for is the wrong trade. It is not only an interop problem. Two Java applications on one system database, one configured with a custom serializer and one not, hit exactly the same thing: `custom_base64` is as unreadable here as `py_pickle` is. So `SerializationUtil.canDeserialize` asks the question directly — the two built-in formats, plus whatever a configured custom serializer names — and the reads that assemble a record from a row skip the payloads when the answer is no, reporting them as null. A predicate rather than a caught exception: there is no deserializer to be had, and trying and failing is a slower way to learn it. Only where the payload is one field of a record — a workflow's status, a workflow's steps. Everywhere the payload is the answer still throws, because there is nothing else to hand back: getEvent, a workflow's result, a recv'd message, a stream, and a recorded step result on replay. That last one matters most — a step that "returned null" because its output could not be read would corrupt the run it is replaying — so it has a test of its own. Three paths read a record but must not accept a null payload, and check for themselves: * Running a workflow. The arguments are the point, and one invoked with arguments it never had is worse than one marked ERROR, so executeWorkflowById refuses and names the format it would have taken. The error is recorded in this runtime's own format: one we cannot read is one we cannot write, and serializing into it would throw and leave the workflow PENDING forever, which is the hang this refusal exists to prevent. * Exporting one. An export is imported back, so a payload dropped on the way out restores a workflow that never had it. Conductor can ask an application to export a peer's workflow, which is exactly when this bites. * Importing one. Export and import are a single-SDK affair but not a single-configuration one, and import re-serializes every payload, so it needs a serializer for the recorded format as much as export did. It also writes a null payload straight through as NULL, so a lossy batch would import as an emptied workflow rather than an error. Checked for the whole batch before the transaction opens. --- .../dbos/transact/database/dao/StepsDAO.java | 5 +- .../transact/database/dao/WorkflowDAO.java | 53 ++++++- .../dbos/transact/execution/DBOSExecutor.java | 16 ++ .../dbos/transact/json/SerializationUtil.java | 13 ++ .../dev/dbos/transact/json/InteropTest.java | 148 ++++++++++++++++++ .../json/PortableSerializationTest.java | 54 +++++++ 6 files changed, 285 insertions(+), 4 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java b/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java index a4551add..582117ff 100644 --- a/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java +++ b/transact/src/main/java/dev/dbos/transact/database/dao/StepsDAO.java @@ -232,7 +232,10 @@ static List listWorkflowSteps( Object outputVal = null; ErrorResult stepError = null; - if (Objects.requireNonNullElse(loadOutput, true)) { + // As for a workflow's status: the steps are what is wanted, the payloads one field of + // each. See SerializationUtil.canDeserialize. + if (Objects.requireNonNullElse(loadOutput, true) + && SerializationUtil.canDeserialize(serialization, serializer)) { if (outputData != null) { try { outputVal = 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 4835a017..752069f7 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 @@ -25,6 +25,7 @@ import dev.dbos.transact.workflow.GetWorkflowAggregatesInput; import dev.dbos.transact.workflow.ListWorkflowsInput; import dev.dbos.transact.workflow.StepAggregateRow; +import dev.dbos.transact.workflow.StepInfo; import dev.dbos.transact.workflow.WorkflowAggregateRow; import dev.dbos.transact.workflow.WorkflowDelay; import dev.dbos.transact.workflow.WorkflowEvent; @@ -47,6 +48,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -1235,6 +1237,9 @@ private static WorkflowStatus resultsToWorkflowStatus( String serializedOutput = loadOutput ? rs.getString("output") : null; String serializedError = loadOutput ? SystemDatabase.errorOrNull(rs.getString("error")) : null; String serialization = loadInput || loadOutput ? rs.getString("serialization") : null; + // A status read reaches other applications' rows on purpose, and wants their metadata; an + // unreadable payload comes back null rather than failing the read. + boolean readable = SerializationUtil.canDeserialize(serialization, serializer); WorkflowStatus info = new WorkflowStatus( rs.getString("workflow_uuid"), @@ -1247,14 +1252,16 @@ private static WorkflowStatus resultsToWorkflowStatus( (authenticatedRolesJson != null) ? JsonUtility.fromJson(authenticatedRolesJson, new TypeReference>() {}) : null, - loadInput + loadInput && readable ? SerializationUtil.deserializePositionalArgs( serializedInput, serialization, serializer) : null, - loadOutput + loadOutput && readable ? SerializationUtil.deserializeValue(serializedOutput, serialization, serializer) : null, - loadOutput ? ErrorResult.deserialize(serializedError, serialization, serializer) : null, + loadOutput && readable + ? ErrorResult.deserialize(serializedError, serialization, serializer) + : null, rs.getString("executor_id"), SystemDatabase.toInstant(rs.getObject("created_at", Long.class)), SystemDatabase.toInstant(rs.getObject("updated_at", Long.class)), @@ -2246,6 +2253,35 @@ static List listWorkflowStreams(Connection conn, String schema, return streams; } + /** + * Refuse to move a workflow whose payloads this runtime cannot handle. + * + *

Export and import round-trip the payloads through this runtime's serializers, so neither can + * settle for the null a status read reports: a payload dropped on the way through restores a + * workflow that never had it. + */ + private static void requireSerializerFor( + String action, + String workflowId, + String workflowSerialization, + List steps, + DBOSSerializer serializer) { + var formats = new LinkedHashSet(); + if (!SerializationUtil.canDeserialize(workflowSerialization, serializer)) { + formats.add(workflowSerialization); + } + for (var step : steps) { + if (!SerializationUtil.canDeserialize(step.serialization(), serializer)) { + formats.add(step.serialization()); + } + } + if (!formats.isEmpty()) { + throw new IllegalStateException( + "Cannot %s workflow %s: it is serialized as %s, which this application has no serializer for" + .formatted(action, workflowId, String.join(", ", formats))); + } + } + public static List exportWorkflow( DbContext ctx, String workflowId, boolean exportChildren) throws SQLException { @@ -2263,6 +2299,9 @@ public static List exportWorkflow( var steps = StepsDAO.listWorkflowSteps( conn, ctx.schema(), ctx.serializer(), wfid, true, null, null); + if (status != null) { + requireSerializerFor("export", wfid, status.serialization(), steps, ctx.serializer()); + } var events = listWorkflowEvents(conn, ctx.schema(), wfid); var eventHistory = listWorkflowEventHistory(conn, ctx.schema(), wfid); var streams = listWorkflowStreams(conn, ctx.schema(), wfid); @@ -2276,6 +2315,14 @@ public static void importWorkflow(DbContext ctx, List workflow throws SQLException { DBOSSerializer serializer = ctx.serializer(); + // The whole batch, before anything is written: export and import are a single-SDK affair but + // not a single-configuration one, and a payload we cannot re-serialize imports empty. + for (var workflow : workflows) { + var s = workflow.status(); + requireSerializerFor( + "import", s.workflowId(), s.serialization(), workflow.steps(), serializer); + } + var wfSQL = """ INSERT INTO "%s".workflow_status ( 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 714d26ae..03582214 100644 --- a/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java +++ b/transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java @@ -1696,6 +1696,22 @@ public WorkflowHandle executeWorkflowById( throw new DBOSNonExistentWorkflowException(workflowId); } + // Reading the row reports an unreadable payload as null, so running the workflow refuses for + // itself: a workflow invoked with arguments it never had is worse than one marked ERROR. + if (!SerializationUtil.canDeserialize(status.serialization(), serializer)) { + var e = + new IllegalStateException( + "Cannot run workflow %s: its arguments are serialized as %s, which this application has no deserializer for" + .formatted(workflowId, status.serialization())); + logger.error("Unreadable serialization for workflow {}", workflowId, e); + // Recorded in this runtime's own format, not the row's: a format we cannot read is one + // we cannot write either, and serializing the error into it would throw and leave the + // workflow PENDING forever — the hang this refusal exists to prevent. The status is the + // part that has to land. + persistWorkflowError(workflowId, e, null); + throw e; + } + Object[] inputs = status.input(); var wfName = RegisteredWorkflow.fullyQualifiedName( diff --git a/transact/src/main/java/dev/dbos/transact/json/SerializationUtil.java b/transact/src/main/java/dev/dbos/transact/json/SerializationUtil.java index e58cbcd2..890daa81 100644 --- a/transact/src/main/java/dev/dbos/transact/json/SerializationUtil.java +++ b/transact/src/main/java/dev/dbos/transact/json/SerializationUtil.java @@ -26,6 +26,19 @@ public final class SerializationUtil { private SerializationUtil() {} + /** + * Whether this runtime can read the given serialization format. + * + *

The two built-in formats always, and a custom serializer only for the format it names. A + * null format is the native one, written before the column existed. + */ + public static boolean canDeserialize(String serialization, DBOSSerializer customSerializer) { + if (serialization == null || PORTABLE.equals(serialization) || NATIVE.equals(serialization)) { + return true; + } + return customSerializer != null && customSerializer.name().equals(serialization); + } + // ============ Value Serialization ============ /** diff --git a/transact/src/test/java/dev/dbos/transact/json/InteropTest.java b/transact/src/test/java/dev/dbos/transact/json/InteropTest.java index 87399dae..121d5578 100644 --- a/transact/src/test/java/dev/dbos/transact/json/InteropTest.java +++ b/transact/src/test/java/dev/dbos/transact/json/InteropTest.java @@ -4,15 +4,20 @@ import dev.dbos.transact.DBOS; import dev.dbos.transact.DBOSClient; +import dev.dbos.transact.DBOSTestAccess; import dev.dbos.transact.config.DBOSConfig; import dev.dbos.transact.utils.DBUtils; import dev.dbos.transact.utils.PgContainer; +import dev.dbos.transact.workflow.ExportedWorkflow; +import dev.dbos.transact.workflow.ListWorkflowsInput; import dev.dbos.transact.workflow.Queue; import dev.dbos.transact.workflow.SerializationStrategy; +import dev.dbos.transact.workflow.StepInfo; import dev.dbos.transact.workflow.Workflow; import dev.dbos.transact.workflow.WorkflowClassName; import dev.dbos.transact.workflow.WorkflowHandle; import dev.dbos.transact.workflow.WorkflowState; +import dev.dbos.transact.workflow.internal.StepResult; import java.sql.Connection; import java.sql.PreparedStatement; @@ -507,6 +512,52 @@ public void testStatusReadTreatsAnEmptyErrorColumnAsNoError() throws Exception { assertNull(status.error(), "an empty error column is not an error"); } + /** + * A workflow another application ran, in a serialization format this runtime cannot read. + * + *

Workflow IDs address the whole system database, so a status read reaches a peer's rows on + * purpose — and what is wanted there is the metadata: who owns it, what it is, whether it + * finished. A payload in the peer's own format must not take that away, so the fields that cannot + * be deserialized come back null and the read succeeds. Python behaves the same way, in + * safe_deserialize. + */ + @Test + public void testStatusReadOfAPeerRowInAnUnreadableFormat() throws Exception { + String workflowId = "peer-foreign-format"; + insertPeerWorkflowRow(workflowId, "py_pickle", "gASVCgAAAA==", "gASVBAAAAA==", null); + + dbos.launch(); + var status = dbos.getWorkflowStatus(workflowId).orElseThrow(); + + // The metadata is the point, and it is all there. + assertEquals(workflowId, status.workflowId()); + assertEquals("echoWorkflow", status.workflowName()); + assertEquals(WorkflowState.SUCCESS, status.status()); + assertEquals("interop-peer", status.applicationName()); + assertEquals("py_pickle", status.serialization()); + + // The payloads are not, and saying so beats failing the whole read. + assertNull(status.input()); + assertNull(status.output()); + assertNull(status.error()); + } + + /** The same row, through the listing rather than by ID. */ + @Test + public void testListingAPeerRowInAnUnreadableFormat() throws Exception { + String workflowId = "peer-foreign-format-listed"; + insertPeerWorkflowRow(workflowId, "py_pickle", "gASVCgAAAA==", "gASVBAAAAA==", ""); + + dbos.launch(); + var listed = dbos.listWorkflows(new ListWorkflowsInput().withWorkflowIds(workflowId)).get(0); + + assertEquals(workflowId, listed.workflowId()); + assertEquals("interop-peer", listed.applicationName()); + assertNull(listed.input()); + assertNull(listed.output()); + assertNull(listed.error()); + } + /** * A step of a peer's workflow, recorded with an empty error column. * @@ -524,4 +575,101 @@ public void testStepReadTreatsAnEmptyErrorColumnAsNoError() throws Exception { assertEquals(1, steps.size()); assertNull(steps.get(0).error(), "an empty error column is not an error"); } + + /** + * The same gap without another language in it: two Java applications on one system database, one + * configured with a custom serializer and one not. + * + *

`custom_base64` is a format this runtime has no deserializer for, exactly as `py_pickle` is, + * and canDeserialize says so without having to try and fail. + */ + @Test + public void testStatusReadOfARowWrittenByACustomSerializerWeLack() throws Exception { + String workflowId = "peer-custom-serializer"; + insertPeerWorkflowRow(workflowId, "custom_base64", "cG9zaXRpb25hbA==", "b3V0cHV0", null); + + dbos.launch(); + var status = dbos.getWorkflowStatus(workflowId).orElseThrow(); + + assertEquals(WorkflowState.SUCCESS, status.status()); + assertEquals("custom_base64", status.serialization()); + assertNull(status.input()); + assertNull(status.output()); + } + + /** + * Exporting a workflow this runtime cannot read is refused rather than silently emptied. + * + *

A status read reports an unreadable payload as null, which is right when the metadata is + * what was asked for. An export is imported back, so the same null would restore a workflow that + * never had an input. + */ + @Test + public void testExportRefusesAWorkflowItCannotRead() throws Exception { + String workflowId = "peer-export"; + insertPeerWorkflowRow(workflowId, "py_pickle", "gASVCgAAAA==", "gASVBAAAAA==", null); + + dbos.launch(); + var systemDatabase = DBOSTestAccess.getSystemDatabase(dbos); + + var thrown = + assertThrows( + IllegalStateException.class, () -> systemDatabase.exportWorkflow(workflowId, false)); + assertTrue( + thrown.getMessage().contains("py_pickle"), + "The refusal should name the format it could not read, got: " + thrown.getMessage()); + } + + /** + * Importing one is refused too, before anything is written. + * + *

Export and import are a single-SDK affair, but not a single-configuration one: the source + * application may have had a serializer this one does not. Import re-serializes the payloads, so + * it needs that serializer as much as export did — and a payload the export already carried as + * null would otherwise be written as NULL and committed. + */ + @Test + public void testImportRefusesAWorkflowItCannotRead() throws Exception { + String workflowId = "peer-import"; + insertPeerWorkflowRow(workflowId, "portable_json", "{\"positionalArgs\":[]}", "{}", null); + + dbos.launch(); + var systemDatabase = DBOSTestAccess.getSystemDatabase(dbos); + + // Exported readably, then given a step in a format this application has no serializer for — + // which is what a batch arriving from an application configured differently looks like. + var exported = systemDatabase.exportWorkflow(workflowId, false).get(0); + var foreignStep = + new StepInfo( + 0, "echoStep", "cG9zaXRpb25hbA==", null, null, null, null, "custom_base64", null); + var batch = + List.of( + new ExportedWorkflow( + exported.status(), + List.of(foreignStep), + exported.events(), + exported.eventHistory(), + exported.streams())); + + var thrown = + assertThrows(IllegalStateException.class, () -> systemDatabase.importWorkflow(batch)); + assertTrue( + thrown.getMessage().contains("custom_base64"), + "The refusal should name the format, got: " + thrown.getMessage()); + } + + /** + * Replaying a step whose result this application cannot read fails; it does not replay as null. + * + *

The tolerant reads are the ones that assemble a record. A recorded step result is consumed + * to continue a workflow — a step that "returned null" because its output could not be read would + * corrupt the run it is replaying, silently and durably. + */ + @Test + public void testStepResultRefusesToReplayWhatItCannotRead() { + var recorded = + new StepResult("wf", 0, "echoStep", "cG9zaXRpb25hbA==", null, null, "custom_base64"); + + assertThrows(IllegalArgumentException.class, () -> recorded.toResult(null)); + } } diff --git a/transact/src/test/java/dev/dbos/transact/json/PortableSerializationTest.java b/transact/src/test/java/dev/dbos/transact/json/PortableSerializationTest.java index 701b7496..e39788e7 100644 --- a/transact/src/test/java/dev/dbos/transact/json/PortableSerializationTest.java +++ b/transact/src/test/java/dev/dbos/transact/json/PortableSerializationTest.java @@ -1036,6 +1036,60 @@ private WorkflowStatusRow waitForWorkflowTerminal(String workflowId, Duration ti throw new AssertionError("Workflow " + workflowId + " did not reach terminal state in time"); } + /** Insert an enqueued workflow row carrying an arbitrary serialization format. */ + private void insertEnqueuedRowWithSerialization( + String workflowId, String queueName, String inputsJson, String serialization) + throws Exception { + try (Connection conn = dataSource.getConnection()) { + String sql = + """ + INSERT INTO dbos.workflow_status( + workflow_uuid, name, class_name, config_name, + queue_name, status, inputs, created_at, serialization + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """; + try (PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, workflowId); + stmt.setString(2, "recvWorkflow"); + stmt.setString(3, "PortableTestService"); + stmt.setString(4, null); + stmt.setString(5, queueName); + stmt.setString(6, "ENQUEUED"); + stmt.setString(7, inputsJson); + stmt.setLong(8, System.currentTimeMillis()); + stmt.setString(9, serialization); + stmt.executeUpdate(); + } + } + } + + /** + * A workflow whose arguments are in a format this application cannot read is marked ERROR, not + * left in PENDING. + * + *

Reading the row does not fail on such a format any more — a status read reports the + * arguments as null and hands back the metadata — so running the workflow has to refuse for + * itself, and say which format it would have taken. + */ + @Test + public void testUnreadableSerializationMarksTheWorkflowErrored() throws Exception { + Queue testQueue = new Queue("testq"); + dbos.registerQueue(testQueue); + dbos.registerProxy(PortableTestService.class, new PortableTestServiceImpl(dbos)); + dbos.launch(); + + String workflowId = UUID.randomUUID().toString(); + insertEnqueuedRowWithSerialization(workflowId, "testq", "cGlja2xlZA==", "py_pickle"); + + var row = waitForWorkflowTerminal(workflowId, Duration.ofSeconds(30)); + assertEquals(WorkflowState.ERROR.name(), row.status()); + assertNotNull(row.error()); + assertTrue( + row.error().contains("py_pickle"), + "The error should name the format it could not read, got: " + row.error()); + } + /** * Tests that completely invalid (unparseable) JSON in the inputs column results in the workflow * being marked as ERROR rather than being stuck in PENDING forever. From 2a7aeebc16020b935b4d4fbbdb4669051488f00a Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 28 Aug 2026 15:21:57 -0700 Subject: [PATCH 3/5] Restore constructors that omit the application name Adding applicationName to the WorkflowSchedule and Queue records changed their canonical constructors in place, breaking every caller that built one positionally. The DBOS conductor test app was one, and its CI now fails to compile against the published SDK. Add constructors taking the pre-application-name argument lists, which delegate with a null owner -- the same thing those callers meant, since a null owner records the creating application. The overloads delegate positionally, so the test checks every component round-trips; a mis-ordered delegation would otherwise compile cleanly. --- .../dev/dbos/transact/workflow/Queue.java | 23 +++++++ .../transact/workflow/WorkflowSchedule.java | 31 +++++++++ .../workflow/OmittedApplicationNameTest.java | 64 +++++++++++++++++++ 3 files changed, 118 insertions(+) create mode 100644 transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java diff --git a/transact/src/main/java/dev/dbos/transact/workflow/Queue.java b/transact/src/main/java/dev/dbos/transact/workflow/Queue.java index 0b026f7a..ed35adfc 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/Queue.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/Queue.java @@ -45,6 +45,29 @@ public record RateLimit(int limit, Duration period) {} throw new IllegalArgumentException("Queue pollingInterval must be greater than zero"); } + /** + * Constructs a queue with no explicit owning application, which records the registering + * application as the owner. + */ + public Queue( + @NonNull String name, + @Nullable Integer concurrency, + @Nullable Integer workerConcurrency, + boolean priorityEnabled, + boolean partitioningEnabled, + @Nullable RateLimit rateLimit, + @NonNull Duration pollingInterval) { + this( + name, + concurrency, + workerConcurrency, + priorityEnabled, + partitioningEnabled, + rateLimit, + pollingInterval, + null); + } + /** Construct a queue with a given name */ public Queue(@NonNull String name) { this(name, null, null, false, false, null, DEFAULT_POLLING_INTERVAL, null); diff --git a/transact/src/main/java/dev/dbos/transact/workflow/WorkflowSchedule.java b/transact/src/main/java/dev/dbos/transact/workflow/WorkflowSchedule.java index 83977ddf..a57963d0 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/WorkflowSchedule.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/WorkflowSchedule.java @@ -39,6 +39,37 @@ public record WorkflowSchedule( Objects.requireNonNull(status, "status must not be null"); } + /** + * Constructs a schedule with no explicit owning application, which records the creating + * application as the owner. + */ + public WorkflowSchedule( + @Nullable String id, + @NonNull String scheduleName, + @NonNull String workflowName, + @Nullable String className, + @NonNull String cron, + @NonNull ScheduleStatus status, + @Nullable Object context, + @Nullable Instant lastFiredAt, + boolean automaticBackfill, + @Nullable ZoneId cronTimezone, + @Nullable String queueName) { + this( + id, + scheduleName, + workflowName, + className, + cron, + status, + context, + lastFiredAt, + automaticBackfill, + cronTimezone, + queueName, + null); + } + public WorkflowSchedule( @NonNull String scheduleName, @NonNull String workflowName, diff --git a/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java b/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java new file mode 100644 index 00000000..0bb4d7c9 --- /dev/null +++ b/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java @@ -0,0 +1,64 @@ +package dev.dbos.transact.workflow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; + +import org.junit.jupiter.api.Test; + +/** + * Covers the constructors that omit the trailing application name, which exist so callers that + * predate system database sharing keep compiling. They delegate positionally, so each component is + * checked to catch a mis-ordered delegation, which would otherwise compile cleanly. + */ +class OmittedApplicationNameTest { + + @Test + void scheduleConstructedWithoutAnApplicationNameIsUnclaimed() { + var lastFiredAt = Instant.ofEpochSecond(1_700_000_000L); + var schedule = + new WorkflowSchedule( + "schedule-id", + "schedule-name", + "workflow-name", + "class-name", + "* * * * * *", + ScheduleStatus.PAUSED, + "context", + lastFiredAt, + true, + ZoneId.of("America/New_York"), + "queue-name"); + + assertNull(schedule.applicationName()); + assertEquals("schedule-id", schedule.id()); + assertEquals("schedule-name", schedule.scheduleName()); + assertEquals("workflow-name", schedule.workflowName()); + assertEquals("class-name", schedule.className()); + assertEquals("* * * * * *", schedule.cron()); + assertEquals(ScheduleStatus.PAUSED, schedule.status()); + assertEquals("context", schedule.context()); + assertEquals(lastFiredAt, schedule.lastFiredAt()); + assertEquals(true, schedule.automaticBackfill()); + assertEquals(ZoneId.of("America/New_York"), schedule.cronTimezone()); + assertEquals("queue-name", schedule.queueName()); + } + + @Test + void queueConstructedWithoutAnApplicationNameIsUnclaimed() { + var rateLimit = new Queue.RateLimit(5, Duration.ofSeconds(10)); + var queue = new Queue("queue-name", 3, 2, true, true, rateLimit, Duration.ofSeconds(7)); + + assertNull(queue.applicationName()); + assertEquals("queue-name", queue.name()); + assertEquals(3, queue.concurrency()); + assertEquals(2, queue.workerConcurrency()); + assertEquals(true, queue.priorityEnabled()); + assertEquals(true, queue.partitioningEnabled()); + assertEquals(rateLimit, queue.rateLimit()); + assertEquals(Duration.ofSeconds(7), queue.pollingInterval()); + } +} From 68297a2ded49d22449e156babd1ea922c2df2e29 Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 28 Aug 2026 15:31:50 -0700 Subject: [PATCH 4/5] Restore input constructors that omit the application name Adding applicationName to ListWorkflowsInput, GetStepAggregatesInput and GetWorkflowAggregatesInput changed their canonical constructors in place. These are records developers build themselves, so add constructors taking the pre-application-name argument lists. They forward a null filter, which covers the caller's own application plus unclaimed rows -- the same default the no-arg constructors already use. GetWorkflowAggregatesInput gained its two components mid-list rather than appended, so that delegation reorders its arguments and additionally forwards groupByApplicationName as false. The read-only output records that also gained the field are left alone; callers receive those rather than construct them. Tests compare each overload against the canonical constructor, which catches a mis-ordered delegation that would otherwise compile cleanly. --- .../workflow/GetStepAggregatesInput.java | 29 +++ .../workflow/GetWorkflowAggregatesInput.java | 56 ++++++ .../transact/workflow/ListWorkflowsInput.java | 65 +++++++ .../workflow/OmittedApplicationNameTest.java | 176 +++++++++++++++++- 4 files changed, 323 insertions(+), 3 deletions(-) diff --git a/transact/src/main/java/dev/dbos/transact/workflow/GetStepAggregatesInput.java b/transact/src/main/java/dev/dbos/transact/workflow/GetStepAggregatesInput.java index 68c8a633..e298def9 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/GetStepAggregatesInput.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/GetStepAggregatesInput.java @@ -41,6 +41,35 @@ public record GetStepAggregatesInput( } } + /** + * Constructs an input with no application filter, which covers this application's own steps plus + * unclaimed ones. + */ + public GetStepAggregatesInput( + boolean groupByFunctionName, + boolean groupByStatus, + boolean selectCount, + boolean selectMaxDuration, + Duration timeBucketSize, + List status, + List functionName, + List workflowIdPrefix, + Instant completedAfter, + Instant completedBefore) { + this( + groupByFunctionName, + groupByStatus, + selectCount, + selectMaxDuration, + timeBucketSize, + status, + functionName, + workflowIdPrefix, + completedAfter, + completedBefore, + null); + } + public GetStepAggregatesInput() { this(false, false, true, false, null, null, null, null, null, null, null); } diff --git a/transact/src/main/java/dev/dbos/transact/workflow/GetWorkflowAggregatesInput.java b/transact/src/main/java/dev/dbos/transact/workflow/GetWorkflowAggregatesInput.java index 2d47866e..f7b85fd4 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/GetWorkflowAggregatesInput.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/GetWorkflowAggregatesInput.java @@ -65,6 +65,62 @@ public record GetWorkflowAggregatesInput( } /** Constructs a default input with {@code selectCount=true} and no group-by or filter flags. */ + /** + * Constructs an input with no application filter and no grouping by application, which covers + * this application's own workflows plus unclaimed ones. + */ + public GetWorkflowAggregatesInput( + boolean groupByStatus, + boolean groupByName, + boolean groupByQueueName, + boolean groupByExecutorId, + boolean groupByApplicationVersion, + boolean selectCount, + boolean selectMinCreatedAt, + boolean selectMaxQueueWait, + boolean selectMaxTotalLatency, + Duration timeBucketSize, + List workflowName, + List status, + List queueName, + List executorIds, + List applicationVersion, + List workflowIdPrefix, + Instant startTime, + Instant endTime, + Instant completedAfter, + Instant completedBefore, + Instant dequeuedAfter, + Instant dequeuedBefore, + Map attributes) { + this( + groupByStatus, + groupByName, + groupByQueueName, + groupByExecutorId, + groupByApplicationVersion, + false, + selectCount, + selectMinCreatedAt, + selectMaxQueueWait, + selectMaxTotalLatency, + timeBucketSize, + workflowName, + status, + queueName, + executorIds, + applicationVersion, + null, + workflowIdPrefix, + startTime, + endTime, + completedAfter, + completedBefore, + dequeuedAfter, + dequeuedBefore, + attributes); + } + public GetWorkflowAggregatesInput() { this( false, false, false, false, false, false, true, false, false, false, null, null, null, null, diff --git a/transact/src/main/java/dev/dbos/transact/workflow/ListWorkflowsInput.java b/transact/src/main/java/dev/dbos/transact/workflow/ListWorkflowsInput.java index 36c7150c..006134cb 100644 --- a/transact/src/main/java/dev/dbos/transact/workflow/ListWorkflowsInput.java +++ b/transact/src/main/java/dev/dbos/transact/workflow/ListWorkflowsInput.java @@ -54,6 +54,71 @@ public record ListWorkflowsInput( attributes = validateAttributes(attributes); } + /** + * Constructs an input with no application filter, which covers this application's own workflows + * plus unclaimed ones. + */ + public ListWorkflowsInput( + List workflowIds, + List status, + Instant startTime, + Instant endTime, + List workflowName, + String className, + String instanceName, + List applicationVersion, + List authenticatedUser, + Integer limit, + Integer offset, + Boolean sortDesc, + List workflowIdPrefix, + Boolean loadInput, + Boolean loadOutput, + List queueName, + Boolean queuesOnly, + List executorIds, + List forkedFrom, + List parentWorkflowId, + Boolean wasForkedFrom, + Boolean hasParent, + Map attributes, + Instant completedAfter, + Instant completedBefore, + Instant dequeuedAfter, + Instant dequeuedBefore, + List scheduleName) { + this( + workflowIds, + status, + startTime, + endTime, + workflowName, + className, + instanceName, + applicationVersion, + authenticatedUser, + limit, + offset, + sortDesc, + workflowIdPrefix, + loadInput, + loadOutput, + queueName, + queuesOnly, + executorIds, + forkedFrom, + parentWorkflowId, + wasForkedFrom, + hasParent, + attributes, + completedAfter, + completedBefore, + dequeuedAfter, + dequeuedBefore, + scheduleName, + null); + } + public ListWorkflowsInput() { this( null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, diff --git a/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java b/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java index 0bb4d7c9..da8b0e99 100644 --- a/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java +++ b/transact/src/test/java/dev/dbos/transact/workflow/OmittedApplicationNameTest.java @@ -6,13 +6,16 @@ import java.time.Duration; import java.time.Instant; import java.time.ZoneId; +import java.util.List; +import java.util.Map; import org.junit.jupiter.api.Test; /** - * Covers the constructors that omit the trailing application name, which exist so callers that - * predate system database sharing keep compiling. They delegate positionally, so each component is - * checked to catch a mis-ordered delegation, which would otherwise compile cleanly. + * Covers the constructors that omit the application name, which exist so callers that predate + * system database sharing keep compiling. They delegate positionally, so each is checked against + * the canonical constructor to catch a mis-ordered delegation, which would otherwise compile + * cleanly. Distinct values per component are what make a swapped pair visible. */ class OmittedApplicationNameTest { @@ -61,4 +64,171 @@ void queueConstructedWithoutAnApplicationNameIsUnclaimed() { assertEquals(rateLimit, queue.rateLimit()); assertEquals(Duration.ofSeconds(7), queue.pollingInterval()); } + + @Test + void listWorkflowsInputConstructedWithoutAnApplicationNameFiltersOnNone() { + var attributes = Map.of("key", "value"); + var actual = + new ListWorkflowsInput( + List.of("workflow-ids"), + List.of(WorkflowState.SUCCESS), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2), + List.of("workflow-name"), + "class-name", + "instance-name", + List.of("application-version"), + List.of("authenticated-user"), + 10, + 20, + true, + List.of("workflow-id-prefix"), + true, + false, + List.of("queue-name"), + true, + List.of("executor-ids"), + List.of("forked-from"), + List.of("parent-workflow-id"), + false, + true, + attributes, + Instant.ofEpochSecond(3), + Instant.ofEpochSecond(4), + Instant.ofEpochSecond(5), + Instant.ofEpochSecond(6), + List.of("schedule-name")); + + var expected = + new ListWorkflowsInput( + List.of("workflow-ids"), + List.of(WorkflowState.SUCCESS), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2), + List.of("workflow-name"), + "class-name", + "instance-name", + List.of("application-version"), + List.of("authenticated-user"), + 10, + 20, + true, + List.of("workflow-id-prefix"), + true, + false, + List.of("queue-name"), + true, + List.of("executor-ids"), + List.of("forked-from"), + List.of("parent-workflow-id"), + false, + true, + attributes, + Instant.ofEpochSecond(3), + Instant.ofEpochSecond(4), + Instant.ofEpochSecond(5), + Instant.ofEpochSecond(6), + List.of("schedule-name"), + null); + + assertNull(actual.applicationName()); + assertEquals(expected, actual); + } + + @Test + void getStepAggregatesInputConstructedWithoutAnApplicationNameFiltersOnNone() { + var actual = + new GetStepAggregatesInput( + true, + false, + true, + false, + Duration.ofSeconds(30), + List.of("status"), + List.of("function-name"), + List.of("workflow-id-prefix"), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2)); + + var expected = + new GetStepAggregatesInput( + true, + false, + true, + false, + Duration.ofSeconds(30), + List.of("status"), + List.of("function-name"), + List.of("workflow-id-prefix"), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2), + null); + + assertNull(actual.applicationName()); + assertEquals(expected, actual); + } + + // The two components #471 added here landed mid-list rather than appended, so this delegation + // has to reorder its arguments rather than just pass an extra null. + @Test + void getWorkflowAggregatesInputConstructedWithoutAnApplicationNameFiltersOnNone() { + var attributes = Map.of("key", "value"); + var actual = + new GetWorkflowAggregatesInput( + true, + false, + true, + false, + true, + false, + true, + false, + true, + Duration.ofSeconds(30), + List.of("workflow-name"), + List.of("status"), + List.of("queue-name"), + List.of("executor-ids"), + List.of("application-version"), + List.of("workflow-id-prefix"), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2), + Instant.ofEpochSecond(3), + Instant.ofEpochSecond(4), + Instant.ofEpochSecond(5), + Instant.ofEpochSecond(6), + attributes); + + var expected = + new GetWorkflowAggregatesInput( + true, + false, + true, + false, + true, + false, // groupByApplicationName + false, + true, + false, + true, + Duration.ofSeconds(30), + List.of("workflow-name"), + List.of("status"), + List.of("queue-name"), + List.of("executor-ids"), + List.of("application-version"), + null, // applicationName + List.of("workflow-id-prefix"), + Instant.ofEpochSecond(1), + Instant.ofEpochSecond(2), + Instant.ofEpochSecond(3), + Instant.ofEpochSecond(4), + Instant.ofEpochSecond(5), + Instant.ofEpochSecond(6), + attributes); + + assertNull(actual.applicationName()); + assertEquals(false, actual.groupByApplicationName()); + assertEquals(expected, actual); + } } From 3054cc63f3290c57a4726478e6079a29199635ae Mon Sep 17 00:00:00 2001 From: Harry Pierson Date: Fri, 28 Aug 2026 15:52:50 -0700 Subject: [PATCH 5/5] Restore the EnqueueOptions constructor that omits the application name A sweep of every public signature #471 changed turned up one more record developers construct: DBOSClient.EnqueueOptions, which gained a trailing applicationName. Add the constructor taking the pre-application-name argument list, forwarding a null owner, which enqueues for the enqueueing application. That sweep found 30 changed public signatures in all. The rest are either read-only outputs callers receive rather than construct (WorkflowStatus, StepInfo, VersionInfo) or internal plumbing: the conductor wire DTOs, the DAO layer, DbContext, SystemDatabase, DBOSExecutor, and the internal packages. Those are left to change in place. --- .../java/dev/dbos/transact/DBOSClient.java | 43 +++++++++++++ .../transact/client/EnqueueOptionsTest.java | 61 +++++++++++++++++++ 2 files changed, 104 insertions(+) diff --git a/transact/src/main/java/dev/dbos/transact/DBOSClient.java b/transact/src/main/java/dev/dbos/transact/DBOSClient.java index 9b88d5df..57b355be 100644 --- a/transact/src/main/java/dev/dbos/transact/DBOSClient.java +++ b/transact/src/main/java/dev/dbos/transact/DBOSClient.java @@ -371,6 +371,49 @@ public EnqueueOptions(@NonNull String workflowName, @NonNull String queueName) { null); } + /** + * Constructs options with no explicit owning application, which enqueues the workflow for the + * enqueueing application. + */ + public EnqueueOptions( + @NonNull String workflowName, + @Nullable String className, + @Nullable String instanceName, + @NonNull String queueName, + @Nullable String workflowId, + @Nullable String appVersion, + @Nullable Duration timeout, + @Nullable Instant deadline, + @Nullable String deduplicationId, + @Nullable Integer priority, + @Nullable String queuePartitionKey, + @Nullable Duration delay, + @Nullable SerializationStrategy serialization, + @Nullable String authenticatedUser, + @Nullable String assumedRole, + @Nullable List authenticatedRoles, + @Nullable Map attributes) { + this( + workflowName, + className, + instanceName, + queueName, + workflowId, + appVersion, + timeout, + deadline, + deduplicationId, + priority, + queuePartitionKey, + delay, + serialization, + authenticatedUser, + assumedRole, + authenticatedRoles, + attributes, + null); + } + public EnqueueOptions( @NonNull String workflowName, @Nullable String className, @NonNull String queueName) { this( diff --git a/transact/src/test/java/dev/dbos/transact/client/EnqueueOptionsTest.java b/transact/src/test/java/dev/dbos/transact/client/EnqueueOptionsTest.java index dce58b4f..d087cf44 100644 --- a/transact/src/test/java/dev/dbos/transact/client/EnqueueOptionsTest.java +++ b/transact/src/test/java/dev/dbos/transact/client/EnqueueOptionsTest.java @@ -1,10 +1,16 @@ package dev.dbos.transact.client; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import dev.dbos.transact.DBOSClient; +import dev.dbos.transact.workflow.SerializationStrategy; import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; import org.junit.jupiter.api.Test; @@ -50,4 +56,59 @@ public void enqueueOptionsValidation() throws Exception { IllegalArgumentException.class, () -> new DBOSClient.EnqueueOptions("wf-name", "q-name").withDelay(Duration.ofSeconds(-1))); } + + /** + * The constructor omitting the application name exists so callers that predate system database + * sharing keep compiling. It delegates positionally, so it is compared against the canonical + * constructor to catch a mis-ordered delegation, which would otherwise compile cleanly. + */ + @Test + public void constructorWithoutAnApplicationNameEnqueuesForTheEnqueueingApplication() { + var attributes = Map.of("key", "value"); + var roles = List.of("role"); + + var actual = + new DBOSClient.EnqueueOptions( + "workflow-name", + "class-name", + "instance-name", + "queue-name", + "workflow-id", + "app-version", + Duration.ofSeconds(1), + Instant.ofEpochSecond(2), + "deduplication-id", + 3, + "queue-partition-key", + Duration.ofSeconds(4), + SerializationStrategy.PORTABLE, + "authenticated-user", + "assumed-role", + roles, + attributes); + + var expected = + new DBOSClient.EnqueueOptions( + "workflow-name", + "class-name", + "instance-name", + "queue-name", + "workflow-id", + "app-version", + Duration.ofSeconds(1), + Instant.ofEpochSecond(2), + "deduplication-id", + 3, + "queue-partition-key", + Duration.ofSeconds(4), + SerializationStrategy.PORTABLE, + "authenticated-user", + "assumed-role", + roles, + attributes, + null); + + assertNull(actual.applicationName()); + assertEquals(expected, actual); + } }