Assorted SysDB Improvements (#317)

combination of improvements suggested by claude clode and Oracle's
VSCode extension

Author: devhawk

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

@@ -5,6 +5,7 @@
 import java.sql.Statement;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -25,8 +26,7 @@ public static class LockConditionPair {
   private static final Logger logger = LoggerFactory.getLogger(NotificationService.class);
 
   private final Map<String, LockConditionPair> notificationsMap = new ConcurrentHashMap<>();
-  private volatile boolean running = false;
-  private Thread notificationListenerThread;
+  private final AtomicReference<Thread> notificationListenerThread = new AtomicReference<>(null);
   private final DataSource dataSource;
 
   public NotificationService(DataSource dataSource) {
@@ -46,21 +46,20 @@ public void unregisterNotificationCondition(String key) {
   }
 
   public void start() {
-    if (!running) {
-      running = true;
-      notificationListenerThread = new Thread(this::notificationListener, "NotificationListener");
-      notificationListenerThread.setDaemon(true);
-      notificationListenerThread.start();
+    Thread t = new Thread(this::notificationListener, "NotificationListener");
+    t.setDaemon(true);
+    if (notificationListenerThread.compareAndSet(null, t)) {
+      t.start();
       logger.debug("Notification listener started");
     }
   }
 
   public void stop() {
-    running = false;
-    if (notificationListenerThread != null) {
-      notificationListenerThread.interrupt();
+    Thread t = notificationListenerThread.getAndSet(null);
+    if (t != null) {
+      t.interrupt();
       try {
-        notificationListenerThread.join(5000); // Wait up to 5 seconds
+        t.join(5000); // Wait up to 5 seconds
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
       }
@@ -71,7 +70,7 @@ public void stop() {
   }
 
   private void notificationListener() {
-    while (running) {
+    while (notificationListenerThread.get() == Thread.currentThread()) {
       Connection notificationConnection = null;
 
       try {
@@ -89,11 +88,9 @@ private void notificationListener() {
 
         logger.debug("Listening for PostgreSQL notifications");
 
-        while (running) {
-          // Check for notifications with a timeout
-          PGNotification[] notifications = pgConnection.getNotifications(1000); // 1
-          // second
-          // timeout
+        while (notificationListenerThread.get() == Thread.currentThread()) {
+          // Check for notifications with a one second timeout
+          PGNotification[] notifications = pgConnection.getNotifications(1000);
 
           if (notifications != null) {
             for (PGNotification notification : notifications) {
@@ -102,19 +99,21 @@ private void notificationListener() {
 
               logger.debug("Received notification on channel: {}, payload: {}", channel, payload);
 
-              if ("dbos_notifications_channel".equals(channel)) {
-                handleNotification(payload, "notifications");
-              } else if ("dbos_workflow_events_channel".equals(channel)) {
-                handleNotification(payload, "workflow_events");
-              } else {
-                logger.error("Unknown NOTIFY channel: {}", channel);
-              }
+              if (null == channel) {
+                logger.error("Received notification with null channel. Payload: {}", payload);
+              } else
+                switch (channel) {
+                  case "dbos_notifications_channel" -> handleNotification(payload, "notifications");
+                  case "dbos_workflow_events_channel" ->
+                      handleNotification(payload, "workflow_events");
+                  default -> logger.error("Unknown NOTIFY channel: {}", channel);
+                }
             }
           }
         }
 
       } catch (Exception e) {
-        if (running) {
+        if (notificationListenerThread.get() == Thread.currentThread()) {
           logger.warn("Notification listener error: {}", e.getMessage());
           try {
             Thread.sleep(1000); // Wait before retrying

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

@@ -28,7 +28,7 @@ class NotificationsDAO {
   private final DataSource dataSource;
   private final String schema;
   private final DBOSSerializer serializer;
-  private NotificationService notificationService;
+  private final NotificationService notificationService;
   private long dbPollingIntervalEventMs = 10000;
 
   NotificationsDAO(
@@ -169,7 +169,7 @@ Object recv(String workflowId, int stepId, int timeoutFunctionId, String topic,
     String finalTopic = (topic != null) ? topic : Constants.DBOS_NULL_TOPIC;
 
     // First, check for previous executions
-    StepResult recordedOutput = null;
+    StepResult recordedOutput;
     try (Connection c = dataSource.getConnection()) {
       recordedOutput =
           StepsDAO.checkStepExecutionTxn(workflowId, stepId, functionName, c, this.schema);
@@ -207,7 +207,7 @@ Object recv(String workflowId, int stepId, int timeoutFunctionId, String topic,
       while (true) {
         // Check if the key is already in the database. If not, wait for the
         // notification
-        boolean hasExistingNotification = false;
+        boolean hasExistingNotification;
         try (Connection conn = dataSource.getConnection()) {
           final String sql =
               """
@@ -437,8 +437,7 @@ Object getEvent(
     // Check for previous executions only if it's in a workflow
     if (callerCtx != null) {
 
-      StepResult recordedOutput = null;
-
+      StepResult recordedOutput;
       try (Connection conn = dataSource.getConnection()) {
         recordedOutput =
             StepsDAO.checkStepExecutionTxn(
@@ -506,7 +505,7 @@ Object getEvent(
         if (nowTime > targetTime) break;
 
         // Consult DB - part of timeout may have expired if sleep is durable.
-        if (callerCtx != null & !checkedDBForSleep) {
+        if (callerCtx != null && !checkedDBForSleep) {
           actualTimeout =
               StepsDAO.durableSleepDuration(
                       dataSource,

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

@@ -3,7 +3,11 @@
 import dev.dbos.transact.workflow.Queue;
 import dev.dbos.transact.workflow.WorkflowState;
 
-import java.sql.*;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -259,7 +263,7 @@ THEN EXTRACT(epoch FROM NOW()) * 1000 + workflow_timeout_ms
       }
 
       // Commit only if workflows were dequeued. Avoids WAL bloat and XID advancement.
-      if (updatedWorkflowIds.size() > 0) {
+      if (!updatedWorkflowIds.isEmpty()) {
         connection.commit();
       } else {
         connection.rollback();

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

@@ -308,7 +308,11 @@ static Duration durableSleepDuration(
       Object deserialized =
           SerializationUtil.deserializeValue(
               recordedOutput.output(), recordedOutput.serialization(), serializer);
-      endTime = ((Number) deserialized).longValue();
+      if (deserialized instanceof Long durationLong) {
+        endTime = durationLong;
+      } else {
+        throw new IllegalStateException("Recorded sleep timeout is not a number: " + deserialized);
+      }
     } else {
       logger.debug(
           "Running sleep, workflow {}, id: {}, duration: {}", workflowUuid, functionId, duration);