fix: three NATS integration test failures

1. Default retention WorkQueue (was Limits)
   Streams must use WorkQueue retention so that acked messages are removed
   from the stream and broker.size() drains to 0. The peek IT already
   sets Limits explicitly so it is unaffected. Also update
   RqueueNatsProperties so the Spring Boot auto-config matches.

2. provisionDlq always creates the DLQ stream
   autoCreateDlqStream gates automatic bootstrap provisioning, not explicit
   calls. provisionDlq() is opt-in by the caller; guard it with ensureStream
   directly instead of ensureDlqStream, which also checked the flag.

3. Priority stream names use underscore suffix (PriorityUtils convention)
   streamFor/subjectFor(q, priority) used a dash separator (pq-high) but
   the poller's expanded QueueDetail.name uses the PriorityUtils suffix
   (pq_high). Align both to PriorityUtils.getSuffix() so enqueue and pop
   target the same stream. Update NatsStreamValidator's priority loop and
   the unit test assertion accordingly.

Assisted-By: Claude Code

Author: sonus21

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/RqueueNatsConfig.java

@@ -53,7 +53,7 @@ public static RqueueNatsConfig defaults() {
   public static class StreamDefaults {
     private int replicas = 1;
     private StorageType storage = StorageType.File;
-    private RetentionPolicy retention = RetentionPolicy.Limits;
+    private RetentionPolicy retention = RetentionPolicy.WorkQueue;
     private Duration duplicateWindow = Duration.ofMinutes(2);
     private long maxMsgs = -1;
     private long maxBytes = -1;

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

@@ -16,6 +16,7 @@
 import com.github.sonus21.rqueue.core.spi.MessageBroker;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
+import com.github.sonus21.rqueue.utils.PriorityUtils;
 import com.github.sonus21.rqueue.nats.RqueueNatsException;
 import com.github.sonus21.rqueue.nats.internal.NatsProvisioner;
 import com.github.sonus21.rqueue.serdes.RqJacksonSerDes;
@@ -114,26 +115,27 @@ private String streamFor(QueueDetail q) {
   }
 
   /**
-   * Resolve the priority-specific subject. Returns the unsuffixed subject when {@code priority} is
-   * null or empty; otherwise appends {@code "." + priority}. Mirrors the naming used by
-   * {@link QueueDetail#resolvedNatsSubjectForPriority(String)}.
+   * Resolve the priority-specific subject. Uses the same {@code "_priority"} suffix as
+   * {@link com.github.sonus21.rqueue.utils.PriorityUtils#getSuffix(String)} so the subject
+   * matches the expanded {@link QueueDetail#getName()} used by the poller (e.g. {@code "pq_high"}).
    */
   private String subjectFor(QueueDetail q, String priority) {
     if (priority == null || priority.isEmpty()) {
       return subjectFor(q);
     }
-    return subjectFor(q) + "." + priority;
+    return config.getSubjectPrefix() + q.getName() + PriorityUtils.getSuffix(priority);
   }
 
   /**
-   * Resolve the priority-specific stream. Returns the unsuffixed stream when {@code priority} is
-   * null or empty; otherwise appends {@code "-" + priority}.
+   * Resolve the priority-specific stream. Uses the same {@code "_priority"} suffix as
+   * {@link com.github.sonus21.rqueue.utils.PriorityUtils#getSuffix(String)} so the stream name
+   * matches what the poller derives from the expanded {@link QueueDetail#getName()}.
    */
   private String streamFor(QueueDetail q, String priority) {
     if (priority == null || priority.isEmpty()) {
       return streamFor(q);
     }
-    return streamFor(q) + "-" + priority;
+    return config.getStreamPrefix() + q.getName() + PriorityUtils.getSuffix(priority);
   }
 
   private String dlqStreamFor(QueueDetail q) {
@@ -547,10 +549,9 @@ public void close() {
    * {@link #installDeadLetterBridge(QueueDetail, String)}.
    */
   public void provisionDlq(QueueDetail q) {
-    if (!config.isAutoCreateDlqStream()) {
-      return;
-    }
-    provisioner.ensureDlqStream(dlqStreamFor(q), List.of(dlqSubjectFor(q)));
+    // Explicit call — always provision, bypassing the autoCreateDlqStream flag.
+    // That flag gates automatic provisioning at bootstrap; here the caller is explicitly opting in.
+    provisioner.ensureStream(dlqStreamFor(q), List.of(dlqSubjectFor(q)));
   }
 
   /**

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

@@ -18,6 +18,8 @@
 import com.github.sonus21.rqueue.core.EndpointRegistry;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.models.event.RqueueBootstrapEvent;
+import com.github.sonus21.rqueue.utils.Constants;
+import com.github.sonus21.rqueue.utils.PriorityUtils;
 import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
 import com.github.sonus21.rqueue.nats.RqueueNatsException;
 import com.github.sonus21.rqueue.nats.internal.NatsProvisioner;
@@ -102,8 +104,11 @@ public void onApplicationEvent(RqueueBootstrapEvent event) {
 
       if (q.getPriority() != null) {
         for (String priority : q.getPriority().keySet()) {
-          String pStream = mainStream + "-" + priority;
-          String pSubject = mainSubject + "." + priority;
+          if (Constants.DEFAULT_PRIORITY_KEY.equals(priority)) {
+            continue; // DEFAULT entry is the queue itself; already handled above
+          }
+          String pStream = config.getStreamPrefix() + q.getName() + PriorityUtils.getSuffix(priority);
+          String pSubject = config.getSubjectPrefix() + q.getName() + PriorityUtils.getSuffix(priority);
           total += tryEnsure(failures, pStream, pSubject);
           tryEnsureConsumer(failures, pStream, q.resolvedConsumerName(), cd);
         }

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/JetStreamMessageBrokerUnitTest.java

@@ -91,7 +91,7 @@ void enqueueWithPriority_appendsPrioritySuffixToSubject() throws Exception {
         "high",
         RqueueMessage.builder().id("m1").message("hi").build());
     verify(f.js, times(1))
-        .publish(eq("rqueue.js.orders.high"), any(Headers.class), any(byte[].class));
+        .publish(eq("rqueue.js.orders_high"), any(Headers.class), any(byte[].class));
   }
 
   @Test

rqueue-spring-boot-starter/src/main/java/com/github/sonus21/rqueue/spring/boot/RqueueNatsProperties.java

@@ -66,7 +66,7 @@ public static class Connection {
   public static class Stream {
     private int replicas = 1;
     private String storage = "FILE";
-    private String retention = "LIMITS";
+    private String retention = "WORKQUEUE";
     private Duration maxAge = Duration.ofDays(14);
     private long maxBytes = -1;
     private long maxMessages = -1;