copilot feedback

Author: devhawk

transact/src/main/java/dev/dbos/transact/database/dao/NotificationsDAO.java

@@ -196,9 +196,9 @@ ON CONFLICT (message_uuid) DO NOTHING
           stmt.executeBatch();
         } catch (SQLException e) {
           if ("23503".equals(e.getSQLState())) {
-            // Find which destination was missing
-            String missingDest = rows.stream().map(InsertRow::destId).findFirst().orElse("unknown");
-            throw new DBOSNonExistentWorkflowException(missingDest);
+            var distinctDests = rows.stream().map(InsertRow::destId).distinct().toList();
+            throw new DBOSNonExistentWorkflowException(
+                distinctDests.size() == 1 ? distinctDests.get(0) : null);
           }
           throw e;
         }

transact/src/main/java/dev/dbos/transact/exceptions/DBOSNonExistentWorkflowException.java

@@ -11,7 +11,10 @@ public class DBOSNonExistentWorkflowException extends RuntimeException {
   private final String workflowId;
 
   public DBOSNonExistentWorkflowException(String workflowId) {
-    super(String.format("Workflow does not exist %s", workflowId));
+    super(
+        workflowId != null
+            ? String.format("Workflow does not exist %s", workflowId)
+            : "One or more destination workflows do not exist");
     this.workflowId = workflowId;
   }
 

transact/src/test/java/dev/dbos/transact/client/ClientTest.java

@@ -14,6 +14,7 @@
 import dev.dbos.transact.exceptions.DBOSNonExistentWorkflowException;
 import dev.dbos.transact.utils.DBUtils;
 import dev.dbos.transact.utils.PgContainer;
+import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.Queue;
 import dev.dbos.transact.workflow.SendMessage;
 import dev.dbos.transact.workflow.WorkflowState;
@@ -386,4 +387,52 @@ public void testClientSendBulk() throws Exception {
     assertEquals("1-msg1", handle1.getResult());
     assertEquals("2-msg2", handle2.getResult());
   }
+
+  @Test
+  public void testClientSendBulkSendToForks() throws Exception {
+    // Parent is a running workflow; child is a fork (ENQUEUED, not executing)
+    var parent = dbos.startWorkflow(() -> service.sendTest(99));
+    var childId = "fork-child-" + UUID.randomUUID();
+    dbos.forkWorkflow(parent.workflowId(), 0, new ForkOptions(childId));
+
+    var sysdb = DBOSTestAccess.getSystemDatabase(dbos);
+    String key = UUID.randomUUID().toString();
+
+    try (var client = pgContainer.dbosClient()) {
+      client.sendBulk(
+          List.of(new SendMessage(parent.workflowId(), "fan-out", "test-topic", key)),
+          DBOSClient.SendOptions.defaults().withSendToForks(true));
+    }
+
+    // Parent receives the message and completes
+    assertEquals("99-fan-out", parent.getResult());
+
+    // Fork also received the notification (verified via DB since it is not executing)
+    assertEquals(1, sysdb.getAllNotifications(childId).size());
+
+    // Re-send with same key is idempotent — {key}::{dest} UUIDs dedup via ON CONFLICT DO NOTHING
+    try (var client = pgContainer.dbosClient()) {
+      client.sendBulk(
+          List.of(new SendMessage(parent.workflowId(), "fan-out", "test-topic", key)),
+          DBOSClient.SendOptions.defaults().withSendToForks(true));
+    }
+    assertEquals(1, sysdb.getAllNotifications(childId).size());
+  }
+
+  @Test
+  public void testClientSendBulkPortableSerialization() throws Exception {
+    var handle1 = dbos.startWorkflow(() -> service.sendTest(1));
+    var handle2 = dbos.startWorkflow(() -> service.sendTest(2));
+
+    try (var client = pgContainer.dbosClient()) {
+      client.sendBulk(
+          List.of(
+              new SendMessage(handle1.workflowId(), "hello", "test-topic", null),
+              new SendMessage(handle2.workflowId(), "world", "test-topic", null)),
+          DBOSClient.SendOptions.portable());
+    }
+
+    assertEquals("1-hello", handle1.getResult());
+    assertEquals("2-world", handle2.getResult());
+  }
 }

transact/src/test/java/dev/dbos/transact/notifications/SendBulkTest.java

@@ -85,7 +85,8 @@ void testSendBulkEmpty() {
 
     // Inside workflow: also a no-op, no step recorded
     String wfId = "empty-wf-" + UUID.randomUUID();
-    service.sendBulkWorkflow(List.of());
+    dbos.startWorkflow(() -> service.sendBulkWorkflow(List.of()), new StartWorkflowOptions(wfId))
+        .getResult();
     assertEquals(0, dbos.listWorkflowSteps(wfId).size());
   }