feat(nats): honour visibilityTimeout/numRetry on JetStream consumer

Map QueueDetail.visibilityTimeout to JetStream ackWait and numRetry to
maxDeliver (numRetry + 1, with Integer.MAX_VALUE as the unlimited
sentinel) when the durable pull consumer is provisioned. Falls back to
RqueueNatsConfig.consumerDefaults when the per-queue value is unset.
Resolution lives in JetStreamMessageBroker.resolveAckWait /
resolveMaxDeliver and is used by both popInternal and the bootstrap
NatsStreamValidator so the consumer is created with the right config
upfront (provisioner only logs a warning on existing-config drift).

Also drops dead, never-set NATS-related fields from QueueDetail
(natsStream, natsSubject, natsDlqStream, natsDlqSubject,
natsMaxDeliverOverride, natsDedupWindow) and their resolved* helpers,
plus the corresponding test. Broker DLQ name helpers now compute
directly from prefix + suffix.

Assisted-By: Claude Code

Author: sonus21

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/QueueDetail.java

@@ -73,18 +73,6 @@ public class QueueDetail extends SerializableBase {
   private String priorityGroup;
   private Set<Class<? extends Throwable>> doNotRetry;
 
-  // ---------------------------------------------------------------------------
-  // NATS / JetStream-related fields. All nullable and additive: when null the
-  // resolved* helpers below derive sensible defaults from queueName /
-  // visibilityTimeout / numRetry. Existing Redis-shaped behavior is unchanged.
-  // ---------------------------------------------------------------------------
-  private final String natsStream;
-  private final String natsSubject;
-  private final String natsDlqStream;
-  private final String natsDlqSubject;
-  private final Duration natsAckWaitOverride;
-  private final Integer natsMaxDeliverOverride;
-  private final Duration natsDedupWindow;
   private final String consumerName;
 
   public boolean isDlqSet() {
@@ -190,77 +178,6 @@ public String resolvedConsumerName() {
     return systemGenerated ? sanitized + "-consumer" : sanitized + "-consumer-primary";
   }
 
-  /**
-   * Resolves the JetStream stream name. When {@link #natsStream} is null the default
-   * derivation {@code "rqueue-" + queueName} is used so existing queue configs keep
-   * working with a NATS broker without explicit overrides.
-   */
-  public String resolvedNatsStream() {
-    return natsStream != null ? natsStream : "rqueue-" + queueName;
-  }
-
-  /**
-   * Resolves the JetStream subject. Falls back to {@code "rqueue." + queueName}.
-   */
-  public String resolvedNatsSubject() {
-    return natsSubject != null ? natsSubject : "rqueue." + queueName;
-  }
-
-  /**
-   * Resolves the JetStream stream name for a specific priority bucket. When {@code priority} is
-   * null or empty falls back to {@link #resolvedNatsStream()}; otherwise appends {@code "-" +
-   * priority} to the resolved base stream name. Used by the NATS broker when a queue declares
-   * per-priority sub-streams.
-   */
-  public String resolvedNatsStreamForPriority(String priority) {
-    if (priority == null || priority.isEmpty()) {
-      return resolvedNatsStream();
-    }
-    return resolvedNatsStream() + "-" + priority;
-  }
-
-  /**
-   * Resolves the JetStream subject for a specific priority bucket. Falls back to
-   * {@link #resolvedNatsSubject()} when {@code priority} is null/empty; otherwise appends
-   * {@code "." + priority}.
-   */
-  public String resolvedNatsSubjectForPriority(String priority) {
-    if (priority == null || priority.isEmpty()) {
-      return resolvedNatsSubject();
-    }
-    return resolvedNatsSubject() + "." + priority;
-  }
-
-  /**
-   * Resolves the dead-letter stream name. Falls back to {@code resolvedNatsStream() + "-dlq"}.
-   */
-  public String resolvedNatsDlqStream() {
-    return natsDlqStream != null ? natsDlqStream : resolvedNatsStream() + "-dlq";
-  }
-
-  /**
-   * Resolves the dead-letter subject. Falls back to {@code resolvedNatsSubject() + ".dlq"}.
-   */
-  public String resolvedNatsDlqSubject() {
-    return natsDlqSubject != null ? natsDlqSubject : resolvedNatsSubject() + ".dlq";
-  }
-
-  /**
-   * Returns the JetStream ack-wait, falling back to the supplied {@code fallback}
-   * (typically the {@link #visibilityDuration()}) when no explicit override is set.
-   */
-  public Duration resolvedAckWait(Duration fallback) {
-    return natsAckWaitOverride != null ? natsAckWaitOverride : fallback;
-  }
-
-  /**
-   * Returns the JetStream max-deliver count, falling back to {@code fallback}
-   * (typically {@code numRetry + 1}) when no explicit override is set.
-   */
-  public int resolvedMaxDeliver(int fallback) {
-    return natsMaxDeliverOverride != null ? natsMaxDeliverOverride : fallback;
-  }
-
   public boolean isDoNotRetryError(Throwable throwable) {
     if (Objects.isNull(throwable)) {
       return false;

rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/QueueDetailNatsFieldsTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024-2026 Sonu Kumar
+ * Copyright (c) 2026 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -65,13 +65,13 @@ public class JetStreamMessageBroker implements MessageBroker, AutoCloseable {
   private static final Capabilities CAPS = new Capabilities(false, false, false, false);
 
   /**
-   * Translation of {@link Duration#ZERO} (the Redis "non-blocking" pop convention used by
-   * {@code RqueueMessagePoller}) into the smallest positive duration JetStream will accept on a
-   * pull fetch. Long enough that messages already buffered for the consumer come back in the same
-   * call, short enough that an empty sub-queue inside a priority group does not stall the poll
-   * cycle.
+   * Lower bound for fetch wait when the caller passes a non-positive duration. JetStream rejects
+   * zero on a pull fetch, so any zero/negative wait is rounded up to this minimum. Callers that
+   * want long-poll semantics should pass the desired wait explicitly (e.g. the listener
+   * container's {@code pollingInterval}); this constant only guards against accidental zero waits
+   * from non-listener callers.
    */
-  private static final Duration NON_BLOCKING_FETCH_WAIT = Duration.ofMillis(50);
+  private static final Duration MIN_FETCH_WAIT = Duration.ofMillis(50);
 
   private final Connection connection;
   private final JetStream js;
@@ -115,7 +115,6 @@ public static Builder builder() {
 
   // ---- subject / stream naming -------------------------------------------
 
-  // TODO: once Phase 1 lands, read additive QueueDetail.getNatsSubject() / getNatsStream() if set.
   private String subjectFor(QueueDetail q) {
     return config.getSubjectPrefix() + q.getName();
   }
@@ -149,15 +148,11 @@ private String streamFor(QueueDetail q, String priority) {
   }
 
   private String dlqStreamFor(QueueDetail q) {
-    return q.getNatsDlqStream() != null
-        ? q.getNatsDlqStream()
-        : streamFor(q) + config.getDlqStreamSuffix();
+    return streamFor(q) + config.getDlqStreamSuffix();
   }
 
   private String dlqSubjectFor(QueueDetail q) {
-    return q.getNatsDlqSubject() != null
-        ? q.getNatsDlqSubject()
-        : subjectFor(q) + config.getDlqSubjectSuffix();
+    return subjectFor(q) + config.getDlqSubjectSuffix();
   }
 
   /** Stream description shown in {@code nats stream info} so operators can map back to rqueue. */
@@ -310,7 +305,13 @@ public Mono<Void> enqueueWithDelayReactive(QueueDetail q, RqueueMessage m, long
   @Override
   public List<RqueueMessage> pop(QueueDetail q, String consumerName, int batch, Duration wait) {
     return popInternal(
-        streamFor(q), subjectFor(q), resolveConsumerName(q.getName(), consumerName), batch, wait);
+        streamFor(q),
+        subjectFor(q),
+        resolveConsumerName(q.getName(), consumerName),
+        batch,
+        wait,
+        resolveAckWait(q, config),
+        resolveMaxDeliver(q, config));
   }
 
   @Override
@@ -321,26 +322,64 @@ public List<RqueueMessage> pop(
         subjectFor(q, priority),
         resolveConsumerName(q.getName(), consumerName),
         batch,
-        wait);
+        wait,
+        resolveAckWait(q, config),
+        resolveMaxDeliver(q, config));
   }
 
   private static String resolveConsumerName(String queueName, String consumerName) {
     return (consumerName != null && !consumerName.isEmpty()) ? consumerName : "rqueue-" + queueName;
   }
 
+  /**
+   * Resolve the JetStream {@code ackWait} for this queue's pull consumer: per-queue
+   * {@link QueueDetail#getVisibilityTimeout()} (when positive), else the global
+   * {@link RqueueNatsConfig.ConsumerDefaults#getAckWait()}. Honouring visibilityTimeout makes the
+   * NATS backend match the contract every other rqueue backend exposes: a message stays invisible
+   * to other consumers for that window and is redelivered if not acked in time.
+   */
+  public static Duration resolveAckWait(QueueDetail q, RqueueNatsConfig config) {
+    long vt = q.getVisibilityTimeout();
+    if (vt > 0) {
+      return Duration.ofMillis(vt);
+    }
+    return config.getConsumerDefaults().getAckWait();
+  }
+
+  /**
+   * Resolve the JetStream {@code maxDeliver} from per-queue {@link QueueDetail#getNumRetry()}
+   * (counted as initial delivery + N retries = numRetry + 1). The {@link Integer#MAX_VALUE}
+   * "retry forever" sentinel maps to JetStream's unlimited value ({@code -1}); non-positive
+   * numRetry falls back to {@link RqueueNatsConfig.ConsumerDefaults#getMaxDeliver()}.
+   */
+  public static long resolveMaxDeliver(QueueDetail q, RqueueNatsConfig config) {
+    int numRetry = q.getNumRetry();
+    if (numRetry == Integer.MAX_VALUE) {
+      return -1L;
+    }
+    if (numRetry > 0) {
+      return numRetry + 1L;
+    }
+    return config.getConsumerDefaults().getMaxDeliver();
+  }
+
   private List<RqueueMessage> popInternal(
-      String stream, String subject, String consumerName, int batch, Duration wait) {
-    // Use default fetch wait if none provided. If zero duration is passed (Redis "non-blocking"
-    // pop convention used by RqueueMessagePoller), translate to a short but positive duration:
-    // JetStream rejects zero, but using the multi-second defaultFetchWait blocks empty pulls long
-    // enough that under Weighted/Strict priority polling the cycle starves real queues — a single
-    // empty sub-queue in a priority group can absorb the whole 30s test budget. Use the smallest
-    // value JetStream still tolerates so empty sub-queues yield back to polling immediately.
+      String stream,
+      String subject,
+      String consumerName,
+      int batch,
+      Duration wait,
+      Duration ackWait,
+      long maxDeliver) {
+    // Honour the caller-supplied wait — this is the listener container's pollingInterval for
+    // RqueueMessagePoller, and lets JetStream long-poll instead of the broker firing a steady
+    // stream of $JS.API.CONSUMER.MSG.NEXT requests. Only fall back when the caller didn't
+    // express a preference; zero/negative waits are rounded up to the JetStream minimum.
     Duration fetchWait;
     if (wait == null) {
       fetchWait = config.getDefaultFetchWait();
-    } else if (wait.isZero()) {
-      fetchWait = NON_BLOCKING_FETCH_WAIT;
+    } else if (wait.isZero() || wait.isNegative()) {
+      fetchWait = MIN_FETCH_WAIT;
     } else {
       fetchWait = wait;
     }
@@ -352,8 +391,8 @@ private List<RqueueMessage> popInternal(
         String actualConsumerName = provisioner.ensureConsumer(
             stream,
             consumerName,
-            config.getConsumerDefaults().getAckWait(),
-            config.getConsumerDefaults().getMaxDeliver(),
+            ackWait,
+            maxDeliver,
             config.getConsumerDefaults().getMaxAckPending());
         PullSubscribeOptions opts = PullSubscribeOptions.bind(stream, actualConsumerName);
         // Consumer has no filter subject; pass null so the NATS client doesn't validate