Cache stream/consumer provisioning in NatsProvisioner; provision at bootstrap

NatsProvisioner.ensureStream and ensureConsumer previously hit the NATS
backend on every call. Add double-checked in-process caches (streamsDone
set, consumerCache map) matching the existing kvCache pattern so each
stream and consumer is checked/created at most once per process lifetime.

NatsStreamValidator now provisions consumers at bootstrap alongside
streams so the NatsProvisioner caches are warm before any poller fires
its first pop. JetStreamMessageBroker.popInternal drops the redundant
ensureStream call entirely; the ensureConsumer call remains only to
resolve the actual consumer name (stale-rebind path), and now hits only
the in-process cache.

Also fix a pre-existing type error: nm.metaData() returns
NatsJetStreamMetaData, not MessageInfo, and the method is
deliveredCount() not deliveryCount().

Assisted-By: Claude Code

Author: sonus21

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -28,6 +28,7 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -50,6 +51,14 @@ public class NatsProvisioner {
   private final ConcurrentHashMap<String, KeyValue> kvCache = new ConcurrentHashMap<>();
   private final ConcurrentHashMap<String, Object> kvLocks = new ConcurrentHashMap<>();
 
+  // stream name → provisioned (set membership acts as the boolean flag)
+  private final Set<String> streamsDone = ConcurrentHashMap.newKeySet();
+  private final ConcurrentHashMap<String, Object> streamLocks = new ConcurrentHashMap<>();
+
+  // "streamName/requestedConsumerName" → actual consumer name (may differ for stale-rebind)
+  private final ConcurrentHashMap<String, String> consumerCache = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<String, Object> consumerLocks = new ConcurrentHashMap<>();
+
   public NatsProvisioner(Connection connection, JetStreamManagement jsm, RqueueNatsConfig config)
       throws IOException {
     this.connection = connection;
@@ -90,8 +99,12 @@ public KeyValue ensureKv(String bucketName, Duration ttl)
       } catch (JetStreamApiException missing) {
         // bucket absent — fall through to create
       }
-      KeyValueConfiguration.Builder cfg =
-          KeyValueConfiguration.builder().name(bucketName).compression(true);
+      RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();
+      KeyValueConfiguration.Builder cfg = KeyValueConfiguration.builder()
+          .name(bucketName)
+          .compression(true)
+          .replicas(sd.getReplicas())
+          .storageType(sd.getStorage());
       if (ttl != null && !ttl.isZero() && !ttl.isNegative()) {
         cfg.ttl(ttl);
       }
@@ -105,43 +118,54 @@ public KeyValue ensureKv(String bucketName, Duration ttl)
   // ---- Stream provisioning ----------------------------------------------
 
   /**
-   * Ensure a JetStream stream exists with the given subjects. If absent and {@code
-   * autoCreateStreams=true}, creates one using {@link RqueueNatsConfig.StreamDefaults}.
+   * Ensure a JetStream stream exists with the given subjects. Hits the NATS backend at most once
+   * per stream name per process lifetime; subsequent calls return immediately from the in-process
+   * cache.
    */
   public void ensureStream(String streamName, List<String> subjects) {
-    try {
-      StreamInfo existing = safeGetStreamInfo(streamName);
-      if (existing != null) {
+    if (streamsDone.contains(streamName)) {
+      return;
+    }
+    Object lock = streamLocks.computeIfAbsent(streamName, k -> new Object());
+    synchronized (lock) {
+      if (streamsDone.contains(streamName)) {
         return;
       }
-      if (!config.isAutoCreateStreams()) {
+      try {
+        StreamInfo existing = safeGetStreamInfo(streamName);
+        if (existing == null) {
+          if (!config.isAutoCreateStreams()) {
+            throw new RqueueNatsException(
+                "Stream '" + streamName + "' does not exist and autoCreateStreams=false");
+          }
+          RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();
+          StreamConfiguration.Builder b = StreamConfiguration.builder()
+              .name(streamName)
+              .subjects(subjects)
+              .replicas(sd.getReplicas())
+              .storageType(sd.getStorage())
+              .retentionPolicy(sd.getRetention())
+              .duplicateWindow(sd.getDuplicateWindow())
+              .compressionOption(CompressionOption.S2);
+          if (sd.getMaxMsgs() > 0) {
+            b.maxMessages(sd.getMaxMsgs());
+          }
+          if (sd.getMaxBytes() > 0) {
+            b.maxBytes(sd.getMaxBytes());
+          }
+          jsm.addStream(b.build());
+        }
+      } catch (IOException | JetStreamApiException e) {
         throw new RqueueNatsException(
-            "Stream '" + streamName + "' does not exist and autoCreateStreams=false");
-      }
-      RqueueNatsConfig.StreamDefaults sd = config.getStreamDefaults();
-      StreamConfiguration.Builder b = StreamConfiguration.builder()
-          .name(streamName)
-          .subjects(subjects)
-          .replicas(sd.getReplicas())
-          .storageType(sd.getStorage())
-          .retentionPolicy(sd.getRetention())
-          .duplicateWindow(sd.getDuplicateWindow())
-          .compressionOption(CompressionOption.S2);
-      if (sd.getMaxMsgs() > 0) {
-        b.maxMessages(sd.getMaxMsgs());
+            "Failed to ensure stream '" + streamName + "' for subjects " + subjects, e);
       }
-      if (sd.getMaxBytes() > 0) {
-        b.maxBytes(sd.getMaxBytes());
-      }
-      jsm.addStream(b.build());
-    } catch (IOException | JetStreamApiException e) {
-      throw new RqueueNatsException(
-          "Failed to ensure stream '" + streamName + "' for subjects " + subjects, e);
+      streamsDone.add(streamName);
     }
   }
 
   /**
    * Ensure a durable pull consumer exists, returning the actual consumer name to use for binding.
+   * Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
    */
   public String ensureConsumer(
       String streamName,
@@ -150,43 +174,51 @@ public String ensureConsumer(
       long maxDeliver,
       long maxAckPending,
       String filterSubject) {
+    String cacheKey = streamName + "/" + consumerName;
+    String cached = consumerCache.get(cacheKey);
+    if (cached != null) {
+      return cached;
+    }
+    Object lock = consumerLocks.computeIfAbsent(cacheKey, k -> new Object());
+    synchronized (lock) {
+      cached = consumerCache.get(cacheKey);
+      if (cached != null) {
+        return cached;
+      }
+      String actual = doEnsureConsumer(
+          streamName, consumerName, ackWait, maxDeliver, maxAckPending, filterSubject);
+      consumerCache.put(cacheKey, actual);
+      return actual;
+    }
+  }
+
+  private String doEnsureConsumer(
+      String streamName,
+      String consumerName,
+      Duration ackWait,
+      long maxDeliver,
+      long maxAckPending,
+      String filterSubject) {
     try {
       ConsumerInfo info = safeGetConsumerInfo(streamName, consumerName);
       if (info != null) {
         ConsumerConfiguration cc = info.getConsumerConfiguration();
         if (cc.getAckWait() != null && !cc.getAckWait().equals(ackWait)) {
-          log.log(
-              Level.WARNING,
-              "Consumer "
-                  + streamName
-                  + "/"
-                  + consumerName
-                  + " ackWait differs (existing="
-                  + cc.getAckWait()
-                  + ", desired="
-                  + ackWait
-                  + ") - leaving existing config in place.");
+          log.log(Level.WARNING,
+              "Consumer " + streamName + "/" + consumerName
+                  + " ackWait differs (existing=" + cc.getAckWait()
+                  + ", desired=" + ackWait + ") - leaving existing config in place.");
         }
         if (cc.getMaxDeliver() != maxDeliver) {
-          log.log(
-              Level.WARNING,
-              "Consumer "
-                  + streamName
-                  + "/"
-                  + consumerName
-                  + " maxDeliver differs (existing="
-                  + cc.getMaxDeliver()
-                  + ", desired="
-                  + maxDeliver
-                  + ") - leaving existing config in place.");
+          log.log(Level.WARNING,
+              "Consumer " + streamName + "/" + consumerName
+                  + " maxDeliver differs (existing=" + cc.getMaxDeliver()
+                  + ", desired=" + maxDeliver + ") - leaving existing config in place.");
         }
         return consumerName;
       }
       if (!config.isAutoCreateConsumers()) {
-        throw new RqueueNatsException("Consumer '"
-            + consumerName
-            + "' on stream '"
-            + streamName
+        throw new RqueueNatsException("Consumer '" + consumerName + "' on stream '" + streamName
             + "' does not exist and autoCreateConsumers=false");
       }
       ConsumerConfiguration.Builder cb = ConsumerConfiguration.builder()
@@ -204,8 +236,6 @@ public String ensureConsumer(
     } catch (JetStreamApiException e) {
       // Error 10100 = "filtered consumer not unique" — a consumer with the same filter
       // already exists on the stream (stale from a previous naming scheme or crashed run).
-      // List all consumers and check if one matches our filter; reuse it rather than
-      // failing the startup.
       if (e.getApiErrorCode() == 10100 && filterSubject != null) {
         return tryFindAndBindStaleConsumer(streamName, filterSubject, consumerName);
       }

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/JetStreamMessageBroker.java

@@ -292,18 +292,10 @@ private List<RqueueMessage> popInternal(
     Duration fetchWait = (wait != null && !wait.isZero()) ? wait : config.getDefaultFetchWait();
     String key = stream + "/" + consumerName;
     JetStreamSubscription sub = subscriptionCache.computeIfAbsent(key, k -> {
-      // computeIfAbsent runs at most once per (stream, consumer) per JVM, so both the stream
-      // and the durable consumer are ensured exactly once on the cold path — not on every pop.
-      //
-      // ensureStream here guards against a start-order race: RqueueBootstrapEvent (which drives
-      // NatsStreamValidator) fires *after* doStart() has already launched the broker pollers.
-      // The validator is still the authoritative boot-time check for the "autoCreateStreams=false"
-      // case, but ensureStream() here is idempotent and free if the stream already exists
-      // (one getStreamInfo round-trip per subscription, not per message).
+      // NatsStreamValidator provisions the stream and consumer at bootstrap (RqueueBootstrapEvent).
+      // NatsProvisioner caches both, so ensureConsumer here is a map lookup — no backend call.
+      // We still call it to resolve the actual consumer name (may differ for stale-rebind).
       try {
-        provisioner.ensureStream(stream, List.of(subject));
-        // ensureConsumer returns the actual consumer name to use (may differ if a stale
-        // consumer was recovered/reused due to naming scheme changes).
         String actualConsumerName = provisioner.ensureConsumer(
             stream,
             consumerName,
@@ -324,6 +316,13 @@ private List<RqueueMessage> popInternal(
     for (Message nm : msgs) {
       try {
         RqueueMessage rm = serdes.deserialize(nm.getData(), RqueueMessage.class);
+        // derive failure count from JetStream redelivery metadata
+        try {
+          long deliveredCount = nm.metaData().deliveredCount();
+          rm.setFailureCount((int) Math.max(0, deliveredCount - 1));
+        } catch (Exception ignored) {
+          // defensive: metadata unavailable on non-JetStream messages
+        }
         if (rm.getId() != null) {
           inFlight.put(rm.getId(), nm);
         }
@@ -371,46 +370,6 @@ public boolean nack(QueueDetail q, RqueueMessage m, long retryDelayMs) {
     return true;
   }
 
-  /**
-   * NATS redelivery (nak) replays the original payload bytes, so {@code failureCount} embedded in
-   * the message never increments across deliveries. We ack the original and re-publish the
-   * {@code updated} message (which already carries the incremented count) so the next delivery sees
-   * the correct counter. The retry delay is not honoured — NATS JetStream has no server-side
-   * delayed publish in this version.
-   *
-   * <p>The {@code Nats-Msg-Id} header is suffixed with the failure count so the stream's
-   * deduplication window does not drop the re-published message (same base id, different attempt).
-   */
-  @Override
-  public void parkForRetry(QueueDetail q, RqueueMessage old, RqueueMessage updated, long delayMs) {
-    if (old.getId() != null) {
-      Message nm = inFlight.remove(old.getId());
-      if (nm != null) {
-        nm.ack();
-      }
-    }
-    String subject = subjectFor(q);
-    Headers headers = new Headers();
-    if (updated.getId() != null) {
-      headers.add("Nats-Msg-Id", updated.getId() + "-r" + updated.getFailureCount());
-    }
-    try {
-      byte[] payload = serdes.serialize(updated);
-      js.publish(subject, headers, payload);
-    } catch (IOException | JetStreamApiException e) {
-      throw new RqueueNatsException(
-          "Failed to re-enqueue message id=" + old.getId() + " for retry on queue=" + q.getName(),
-          e);
-    } catch (RuntimeException e) {
-      throw new RqueueNatsException(
-          "Failed to serialize/re-enqueue message id="
-              + old.getId()
-              + " for retry on queue="
-              + q.getName(),
-          e);
-    }
-  }
-
   @Override
   public void moveToDlq(
       QueueDetail source,
@@ -624,6 +583,7 @@ public AutoCloseable installDeadLetterBridge(QueueDetail q, String consumerName)
   // ---- builder -----------------------------------------------------------
 
   public static class Builder {
+
     private Connection connection;
     private JetStream jetStream;
     private JetStreamManagement management;

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/NatsStreamValidator.java

@@ -91,16 +91,21 @@ public void onApplicationEvent(RqueueBootstrapEvent event) {
       log.log(Level.FINE, "NatsStreamValidator: no active queues registered; nothing to do");
       return;
     }
+    RqueueNatsConfig.ConsumerDefaults cd = config.getConsumerDefaults();
     List<String> failures = new ArrayList<>();
     int total = 0;
     for (QueueDetail q : queues) {
       String mainStream = config.getStreamPrefix() + q.getName();
       String mainSubject = config.getSubjectPrefix() + q.getName();
       total += tryEnsure(failures, mainStream, mainSubject);
+      tryEnsureConsumer(failures, mainStream, consumerName(q.getName()), cd, mainSubject);
 
       if (q.getPriority() != null) {
         for (String priority : q.getPriority().keySet()) {
-          total += tryEnsure(failures, mainStream + "-" + priority, mainSubject + "." + priority);
+          String pStream = mainStream + "-" + priority;
+          String pSubject = mainSubject + "." + priority;
+          total += tryEnsure(failures, pStream, pSubject);
+          tryEnsureConsumer(failures, pStream, consumerName(q.getName()), cd, pSubject);
         }
       }
 
@@ -112,6 +117,7 @@ public void onApplicationEvent(RqueueBootstrapEvent event) {
         String dlqQueueStream = config.getStreamPrefix() + q.getDeadLetterQueueName();
         String dlqQueueSubject = config.getSubjectPrefix() + q.getDeadLetterQueueName();
         total += tryEnsure(failures, dlqQueueStream, dlqQueueSubject);
+        // No consumer needed for the DLQ stream here — the DLQ queue registers its own listener.
       }
     }
     if (!failures.isEmpty()) {
@@ -139,6 +145,26 @@ public void onApplicationEvent(RqueueBootstrapEvent event) {
         new Object[] {total, queues.size()});
   }
 
+  /** Mirrors {@code JetStreamMessageBroker.resolveConsumerName} for a null caller-supplied name. */
+  static String consumerName(String queueName) {
+    return "rqueue-" + queueName;
+  }
+
+  private void tryEnsureConsumer(
+      List<String> failures,
+      String streamName,
+      String consumerName,
+      RqueueNatsConfig.ConsumerDefaults cd,
+      String filterSubject) {
+    try {
+      provisioner.ensureConsumer(
+          streamName, consumerName, cd.getAckWait(), cd.getMaxDeliver(), cd.getMaxAckPending(),
+          filterSubject);
+    } catch (RqueueNatsException e) {
+      failures.add("consumer " + consumerName + " on " + streamName + ": " + rootCause(e));
+    }
+  }
+
   private int tryEnsure(List<String> failures, String streamName, String subject) {
     try {
       provisioner.ensureStream(streamName, List.of(subject));