in memory cache of dynamic queues for validation

Author: devhawk

transact/src/main/java/dev/dbos/transact/execution/DBOSExecutor.java

@@ -395,7 +395,9 @@ public Optional<RegisteredWorkflow> getRegisteredWorkflow(
   }
 
   public Optional<Queue> findQueue(String queueName) {
-    return findStaticQueue(queueName).or(() -> systemDatabase.findQueue(queueName));
+    return findStaticQueue(queueName)
+        .or(() -> queueService.findDynamicQueue(queueName))
+        .or(() -> systemDatabase.findQueue(queueName));
   }
 
   public Collection<Queue> getStaticQueues() {
@@ -1458,8 +1460,7 @@ private void validateWorkflow(String workflowName, String className, String inst
 
   private void validateQueue(String queueName) {
     if (queueName != null) {
-      findStaticQueue(queueName)
-          .or(() -> systemDatabase.findQueue(queueName))
+      findQueue(queueName)
           .orElseThrow(
               () -> new IllegalStateException("Queue %s is not registered".formatted(queueName)));
     }
@@ -1472,7 +1473,7 @@ private void validateQueue(String queueName, String queuePartitionKey) {
             "DBOS internal queue is not a partitioned queue, but a partition key was provided");
       }
     } else {
-      var queue = findStaticQueue(queueName).or(() -> systemDatabase.findQueue(queueName));
+      var queue = findQueue(queueName);
       if (queue.isPresent()) {
         if (queue.get().partitioningEnabled() && queuePartitionKey == null) {
           throw new IllegalArgumentException(

transact/src/main/java/dev/dbos/transact/execution/QueueService.java

@@ -6,7 +6,9 @@
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -15,6 +17,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -28,6 +31,7 @@ public class QueueService implements AutoCloseable {
   private final AtomicReference<ScheduledExecutorService> execServiceRef = new AtomicReference<>();
   private final AtomicBoolean paused = new AtomicBoolean(false);
   private final Set<String> dbListeningQueues = ConcurrentHashMap.newKeySet();
+  private volatile Map<String, Queue> dynamicQueueMap = Map.of();
 
   private final SystemDatabase systemDatabase;
   private final DBOSExecutor dbosExecutor;
@@ -80,6 +84,10 @@ public boolean isStopped() {
     return this.execServiceRef.get() == null;
   }
 
+  public Optional<Queue> findDynamicQueue(String queueName) {
+    return Optional.ofNullable(dynamicQueueMap.get(queueName));
+  }
+
   private boolean isListening(String queueName) {
     return queueName.equals(Constants.DBOS_INTERNAL_QUEUE)
         || listenQueues.isEmpty()
@@ -102,6 +110,8 @@ private void pollDynamicQueues() {
       if (execServiceRef.get() == null) return;
 
       var dbQueues = systemDatabase.listQueues();
+      dynamicQueueMap =
+          dbQueues.stream().collect(Collectors.toUnmodifiableMap(Queue::name, q -> q));
       if (logger.isDebugEnabled()) {
         logger.debug("pollDynamicQueues found {} queues", dbQueues.size());
         for (var q : dbQueues) {