remove temp wf send (#292)

Co-authored-by: Peter Kraft <petereliaskraft@gmail.com>

Author: devhawk

transact/src/main/java/dev/dbos/transact/DBOS.java

@@ -14,8 +14,6 @@
 import dev.dbos.transact.internal.QueueRegistry;
 import dev.dbos.transact.internal.WorkflowRegistry;
 import dev.dbos.transact.migrations.MigrationManager;
-import dev.dbos.transact.tempworkflows.InternalWorkflowsService;
-import dev.dbos.transact.tempworkflows.InternalWorkflowsServiceImpl;
 import dev.dbos.transact.workflow.ForkOptions;
 import dev.dbos.transact.workflow.ListWorkflowsInput;
 import dev.dbos.transact.workflow.Queue;
@@ -92,8 +90,6 @@ public static class Instance {
 
     private DBOSConfig config;
 
-    private InternalWorkflowsService internalWorkflowsService;
-
     private final AtomicReference<DBOSExecutor> dbosExecutor = new AtomicReference<>();
 
     private Instance() {
@@ -186,8 +182,6 @@ void registerQueue(@NonNull Queue queue) {
     }
 
     private void registerInternals() {
-      internalWorkflowsService =
-          registerWorkflows(InternalWorkflowsService.class, new InternalWorkflowsServiceImpl());
       this.registerQueue(new Queue(Constants.DBOS_INTERNAL_QUEUE));
     }
 
@@ -574,7 +568,7 @@ public static <T, E extends Exception> T getResult(@NonNull String workflowId) t
   public static void send(
       @NonNull String destinationId,
       @NonNull Object message,
-      @NonNull String topic,
+      @Nullable String topic,
       @Nullable String idempotencyKey) {
     send(destinationId, message, topic, idempotencyKey, null);
   }
@@ -591,18 +585,11 @@ public static void send(
   public static void send(
       @NonNull String destinationId,
       @NonNull Object message,
-      @NonNull String topic,
+      @Nullable String topic,
       @Nullable String idempotencyKey,
       @Nullable SerializationStrategy serialization) {
     if (serialization == null) serialization = SerializationStrategy.DEFAULT;
-    executor("send")
-        .send(
-            destinationId,
-            message,
-            topic,
-            instance().internalWorkflowsService,
-            idempotencyKey,
-            serialization);
+    executor("send").send(destinationId, message, topic, idempotencyKey, serialization);
   }
 
   /**
@@ -613,7 +600,7 @@ public static void send(
    * @param topic topic to which the message is send
    */
   public static void send(
-      @NonNull String destinationId, @NonNull Object message, @NonNull String topic) {
+      @NonNull String destinationId, @NonNull Object message, @Nullable String topic) {
     DBOS.send(destinationId, message, topic, null, null);
   }
 

transact/src/main/java/dev/dbos/transact/DBOSClient.java

@@ -595,24 +595,13 @@ public void send(
       @NonNull String topic,
       @Nullable String idempotencyKey,
       @Nullable SendOptions options) {
-    if (idempotencyKey == null) {
-      idempotencyKey = UUID.randomUUID().toString();
-    }
-    var workflowId = "%s-%s".formatted(destinationId, idempotencyKey);
 
     String serializationFormat =
         (options != null && options.serialization() != null)
             ? options.serialization().formatName()
             : null;
 
-    var status =
-        WorkflowStatusInternal.builder(workflowId, WorkflowState.SUCCESS)
-            .name("temp_workflow-send-client")
-            .serialization(
-                serializationFormat != null ? serializationFormat : SerializationUtil.NATIVE)
-            .build();
-    systemDatabase.initWorkflowStatus(status, null, false, false);
-    systemDatabase.send(status.workflowId(), 0, destinationId, message, topic, serializationFormat);
+    systemDatabase.sendDirect(destinationId, message, topic, idempotencyKey, serializationFormat);
   }
 
   /**

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

@@ -13,6 +13,7 @@
 import java.sql.SQLException;
 import java.time.Duration;
 import java.util.Objects;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
 import javax.sql.DataSource;
@@ -43,11 +44,12 @@ void speedUpPollingForTest() {
   }
 
   void send(
-      String workflowUuid,
-      int functionId,
-      String destinationUuid,
+      String workflowId,
+      int stepId,
+      String destinationId,
       Object message,
       String topic,
+      String messageId,
       String serialization)
       throws SQLException {
 
@@ -61,52 +63,55 @@ void send(
       try {
         // Check if operation was already executed
         StepResult recordedOutput =
-            StepsDAO.checkStepExecutionTxn(
-                workflowUuid, functionId, functionName, conn, this.schema);
+            StepsDAO.checkStepExecutionTxn(workflowId, stepId, functionName, conn, this.schema);
 
         if (recordedOutput != null) {
           logger.debug(
               "Replaying send, id: {}, destination_uuid: {}, topic: {}",
-              functionId,
-              destinationUuid,
+              stepId,
+              destinationId,
               finalTopic);
           conn.commit();
           return;
         } else {
           logger.debug(
               "Running send, id: {}, destination_uuid: {}, topic: {}",
-              functionId,
-              destinationUuid,
+              stepId,
+              destinationId,
               finalTopic);
         }
 
-        // Serialize the message using the specified format
-        SerializationUtil.SerializedResult serializedMsg =
+        var finalMessageId = (messageId != null) ? messageId : UUID.randomUUID().toString();
+        var serializedMsg =
             SerializationUtil.serializeValue(message, serialization, this.serializer);
 
         // Insert notification with serialization format
         final String sql =
             """
-              INSERT INTO "%s".notifications (destination_uuid, topic, message, serialization) VALUES (?, ?, ?, ?)
+              INSERT INTO "%s".notifications
+                (destination_uuid, topic, message, serialization, message_uuid)
+              VALUES (?, ?, ?, ?, ?)
+              ON CONFLICT (message_uuid) DO NOTHING
             """
                 .formatted(this.schema);
 
         try (PreparedStatement stmt = conn.prepareStatement(sql)) {
-          stmt.setString(1, destinationUuid);
+          stmt.setString(1, destinationId);
           stmt.setString(2, finalTopic);
           stmt.setString(3, serializedMsg.serializedValue());
           stmt.setString(4, serializedMsg.serialization());
+          stmt.setString(5, finalMessageId);
           stmt.executeUpdate();
         } catch (SQLException e) {
           // Foreign key violation
           if ("23503".equals(e.getSQLState())) {
-            throw new DBOSNonExistentWorkflowException(destinationUuid);
+            throw new DBOSNonExistentWorkflowException(destinationId);
           }
           throw e;
         }
 
         // Record operation result
-        var output = new StepResult(workflowUuid, functionId, functionName, null, null, null, null);
+        var output = new StepResult(workflowId, stepId, functionName, null, null, null, null);
         StepsDAO.recordStepResultTxn(
             output, startTime, System.currentTimeMillis(), conn, this.schema);
 
@@ -123,8 +128,40 @@ void send(
     }
   }
 
-  Object recv(
-      String workflowUuid, int functionId, int timeoutFunctionId, String topic, Duration timeout)
+  void sendDirect(
+      String destinationId, Object message, String topic, String messageId, String serialization)
+      throws SQLException {
+    String finalTopic = (topic != null) ? topic : Constants.DBOS_NULL_TOPIC;
+    String finalMessageId = (messageId != null) ? messageId : UUID.randomUUID().toString();
+    var serializedMsg = SerializationUtil.serializeValue(message, serialization, this.serializer);
+
+    final String sql =
+        """
+          INSERT INTO "%s".notifications
+            (destination_uuid, topic, message, message_uuid, serialization)
+          VALUES (?, ?, ?, ?, ?)
+          ON CONFLICT (message_uuid) DO NOTHING
+        """
+            .formatted(this.schema);
+
+    try (var conn = dataSource.getConnection();
+        var stmt = conn.prepareStatement(sql)) {
+      stmt.setString(1, destinationId);
+      stmt.setString(2, finalTopic);
+      stmt.setString(3, serializedMsg.serializedValue());
+      stmt.setString(4, finalMessageId);
+      stmt.setString(5, serializedMsg.serialization());
+      stmt.executeUpdate();
+    } catch (SQLException e) {
+      // Foreign key violation
+      if ("23503".equals(e.getSQLState())) {
+        throw new DBOSNonExistentWorkflowException(destinationId);
+      }
+      throw e;
+    }
+  }
+
+  Object recv(String workflowId, int stepId, int timeoutFunctionId, String topic, Duration timeout)
       throws SQLException {
 
     var startTime = System.currentTimeMillis();
@@ -133,28 +170,26 @@ Object recv(
 
     // First, check for previous executions
     StepResult recordedOutput = null;
-
     try (Connection c = dataSource.getConnection()) {
       recordedOutput =
-          StepsDAO.checkStepExecutionTxn(workflowUuid, functionId, functionName, c, this.schema);
+          StepsDAO.checkStepExecutionTxn(workflowId, stepId, functionName, c, this.schema);
     }
 
     if (recordedOutput != null) {
-      logger.debug("Replaying recv, id: {}, topic: {}", functionId, finalTopic);
+      logger.debug("Replaying recv, id: {}, topic: {}", stepId, finalTopic);
       if (recordedOutput.output() != null) {
         return SerializationUtil.deserializeValue(
             recordedOutput.output(), recordedOutput.serialization(), this.serializer);
       } else {
         throw new RuntimeException("No output recorded in the last recv");
       }
     } else {
-      logger.debug(
-          "Running recv, wfid {}, id: {}, topic: {}", workflowUuid, functionId, finalTopic);
+      logger.debug("Running recv, wfid {}, id: {}, topic: {}", workflowId, stepId, finalTopic);
     }
 
     // Insert a condition to the notifications map
-    String payload = workflowUuid + "::" + finalTopic;
-    NotificationService.LockConditionPair lockPair = new NotificationService.LockConditionPair();
+    String payload = workflowId + "::" + finalTopic;
+    var lockPair = new NotificationService.LockConditionPair();
 
     // Timeout / deadline for the notification
     double actualTimeout = timeout.toMillis();
@@ -166,7 +201,7 @@ Object recv(
       boolean success = notificationService.registerNotificationCondition(payload, lockPair);
       if (!success) {
         // if this happens, the workflow is executing concurrently
-        throw new DBOSWorkflowExecutionConflictException(workflowUuid);
+        throw new DBOSWorkflowExecutionConflictException(workflowId);
       }
 
       while (true) {
@@ -176,12 +211,13 @@ Object recv(
         try (Connection conn = dataSource.getConnection()) {
           final String sql =
               """
-                SELECT topic FROM "%s".notifications WHERE destination_uuid = ? AND topic = ?
+              SELECT topic FROM "%s".notifications
+              WHERE destination_uuid = ? AND topic = ? AND consumed = FALSE
               """
                   .formatted(this.schema);
 
           try (PreparedStatement stmt = conn.prepareStatement(sql)) {
-            stmt.setString(1, workflowUuid);
+            stmt.setString(1, workflowId);
             stmt.setString(2, finalTopic);
             try (ResultSet rs = stmt.executeQuery()) {
               hasExistingNotification = rs.next();
@@ -199,7 +235,7 @@ Object recv(
           actualTimeout =
               StepsDAO.durableSleepDuration(
                       dataSource,
-                      workflowUuid,
+                      workflowId,
                       timeoutFunctionId,
                       timeout,
                       this.schema,
@@ -223,63 +259,61 @@ Object recv(
       notificationService.unregisterNotificationCondition(payload);
     }
 
-    // Transactionally consume and return the message if it's in the database
+    // Transactionally consume and return the oldest unconsumed message, or null if none.
     try (Connection conn = dataSource.getConnection()) {
       conn.setAutoCommit(false);
 
       try {
-        // Find and delete the oldest entry for this workflow+topic
+        // Find and mark consumed the oldest entry for this workflow+topic
         final String sql =
             """
-              WITH oldest_entry AS (
-                  SELECT destination_uuid, topic, message, serialization, created_at_epoch_ms
-                  FROM "%1$s".notifications
-                  WHERE destination_uuid = ? AND topic = ?
-                  ORDER BY created_at_epoch_ms ASC
-                  LIMIT 1
+            UPDATE "%1$s".notifications
+            SET consumed = TRUE
+            WHERE destination_uuid = ?
+              AND topic = ?
+              AND consumed = FALSE
+              AND message_uuid = (
+                SELECT message_uuid FROM "%1$s".notifications
+                WHERE destination_uuid = ?
+                  AND topic = ?
+                  AND consumed = FALSE
+                ORDER BY created_at_epoch_ms ASC
+                LIMIT 1
               )
-              DELETE FROM "%1$s".notifications
-              WHERE destination_uuid = (SELECT destination_uuid FROM oldest_entry)
-                AND topic = (SELECT topic FROM oldest_entry)
-                AND created_at_epoch_ms = (SELECT created_at_epoch_ms FROM oldest_entry)
-              RETURNING message, serialization
+            RETURNING message, serialization
             """
                 .formatted(this.schema);
 
-        Object recvdMessage = null;
+        String serializedMessage = null;
+        String serialization = null;
         try (PreparedStatement stmt = conn.prepareStatement(sql)) {
-          stmt.setString(1, workflowUuid);
+          // JDBC uses positional parameters (?), and each placeholder must be bound explicitly,
+          // so we need to set the same values again for the nested SELECT
+          stmt.setString(1, workflowId);
           stmt.setString(2, finalTopic);
+          stmt.setString(3, workflowId);
+          stmt.setString(4, finalTopic);
 
           try (ResultSet rs = stmt.executeQuery()) {
             if (rs.next()) {
-              String serializedMessage = rs.getString("message");
-              String serialization = rs.getString("serialization");
-              recvdMessage =
-                  SerializationUtil.deserializeValue(
-                      serializedMessage, serialization, this.serializer);
+              serializedMessage = rs.getString("message");
+              serialization = rs.getString("serialization");
             }
           }
         }
 
+        var recvdMessage =
+            SerializationUtil.deserializeValue(serializedMessage, serialization, this.serializer);
+
         // Record operation result
-        Object toSave = recvdMessage;
-        var toSaveSer = SerializationUtil.serializeValue(toSave, null, this.serializer);
         StepResult output =
             new StepResult(
-                    workflowUuid,
-                    functionId,
-                    functionName,
-                    null,
-                    null,
-                    null,
-                    toSaveSer.serialization())
-                .withOutput(toSaveSer.serializedValue());
+                workflowId, stepId, functionName, serializedMessage, null, null, serialization);
         StepsDAO.recordStepResultTxn(
             output, startTime, System.currentTimeMillis(), conn, this.schema);
 
         conn.commit();
-        return toSave;
+        return recvdMessage;
 
       } catch (Exception e) {
         conn.rollback();