Nats scheduling fix

.github/workflows/java-ci.yaml

@@ -137,10 +137,25 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y redis-server
           redis-cli --version
+          sudo systemctl stop redis-server
+          redis-server --logfile /tmp/redis.log --daemonize yes
+
+      - name: Log Redis info
+        run: |
+          redis-server --version
+          redis-cli INFO server
 
       - name: Run producer-only tests
         run: ./gradlew test -DincludeTags=producerOnly
 
+      - name: Upload redis-server log
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: producer-redis-server
+          path: /tmp/redis.log
+          if-no-files-found: ignore
+
       - name: Upload JaCoCo exec data
         if: always()
         uses: actions/upload-artifact@v4
@@ -189,10 +204,25 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y redis-server
           redis-cli --version
+          sudo systemctl stop redis-server
+          redis-server --logfile /tmp/redis.log --daemonize yes
+
+      - name: Log Redis info
+        run: |
+          redis-server --version
+          redis-cli INFO server
 
       - name: Run integration tests
         run: ./gradlew test -DincludeTags=integration -DexcludeTags=redisCluster,producerOnly,local
 
+      - name: Upload redis-server log
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: integration-redis-server
+          path: /tmp/redis.log
+          if-no-files-found: ignore
+
       - name: Upload JaCoCo exec data
         if: always()
         uses: actions/upload-artifact@v4
@@ -245,12 +275,12 @@ jobs:
       - name: Setup Redis Cluster
         run: |
           mkdir 9000 9001 9002 9003 9004 9005
-          printf "port 9000 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9000/redis.conf
-          printf "port 9001 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9001/redis.conf
-          printf "port 9002 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9002/redis.conf
-          printf "port 9003 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9003/redis.conf
-          printf "port 9004 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9004/redis.conf
-          printf "port 9005 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes" >> 9005/redis.conf
+          printf "port 9000 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9000.log" >> 9000/redis.conf
+          printf "port 9001 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9001.log" >> 9001/redis.conf
+          printf "port 9002 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9002.log" >> 9002/redis.conf
+          printf "port 9003 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9003.log" >> 9003/redis.conf
+          printf "port 9004 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9004.log" >> 9004/redis.conf
+          printf "port 9005 \ncluster-enabled yes \ncluster-config-file nodes.conf \ncluster-node-timeout 5000 \nappendonly yes\nlogfile /tmp/redis-9005.log" >> 9005/redis.conf
           (cd 9000 && redis-server ./redis.conf) &
           (cd 9001 && redis-server ./redis.conf) &
           (cd 9002 && redis-server ./redis.conf) &
@@ -260,9 +290,23 @@ jobs:
           sleep 30
           yes yes | redis-cli --cluster create 127.0.0.1:9000 127.0.0.1:9001 127.0.0.1:9002 127.0.0.1:9003 127.0.0.1:9004 127.0.0.1:9005 --cluster-replicas 1
 
+      - name: Log Redis cluster info
+        run: |
+          redis-server --version
+          redis-cli -p 9000 INFO server | grep -E 'redis_version|os|tcp_port|uptime_in_seconds'
+          redis-cli -p 9000 CLUSTER INFO
+
       - name: Run Redis cluster tests
         run: ./gradlew test -DincludeTags=redisCluster
 
+      - name: Upload redis-server logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: redis-server-cluster
+          path: /tmp/redis-*.log
+          if-no-files-found: ignore
+
       - name: Upload JaCoCo exec data
         if: always()
         uses: actions/upload-artifact@v4
@@ -313,10 +357,25 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y redis-server
           redis-cli --version
+          sudo systemctl stop redis-server
+          redis-server --logfile /tmp/redis.log --daemonize yes
+
+      - name: Log Redis info
+        run: |
+          redis-server --version
+          redis-cli INFO server
 
       - name: Run reactive integration tests
         run: ./gradlew test -DincludeTags=integration -DexcludeTags=redisCluster,producerOnly,local
 
+      - name: Upload redis-server log
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: reactive-redis-server
+          path: /tmp/redis.log
+          if-no-files-found: ignore
+
       - name: Upload JaCoCo exec data
         if: always()
         uses: actions/upload-artifact@v4
@@ -365,7 +424,7 @@ jobs:
       # instead of pulling a Testcontainers image.
       - name: Install nats-server
         run: |
-          NATS_VERSION=v2.10.22
+          NATS_VERSION=v2.12.0
           curl -sSL "https://github.com/nats-io/nats-server/releases/download/${NATS_VERSION}/nats-server-${NATS_VERSION}-linux-amd64.tar.gz" \
             | tar -xz -C /tmp
           sudo mv "/tmp/nats-server-${NATS_VERSION}-linux-amd64/nats-server" /usr/local/bin/nats-server
@@ -374,14 +433,17 @@ jobs:
       - name: Start nats-server
         run: |
           mkdir -p /tmp/jetstream
-          nohup nats-server -js -sd /tmp/jetstream -p 4222 > /tmp/nats.log 2>&1 &
+          nats-server -js -sd /tmp/jetstream -p 4222 > /tmp/nats.log 2>&1 &
           for i in $(seq 1 20); do
             if (echo > /dev/tcp/127.0.0.1/4222) 2>/dev/null; then
               echo "nats-server ready after ${i}s"; break
             fi
             sleep 1
           done
 
+      - name: Log NATS server info
+        run: nats-server --version
+
       - name: Run NATS tests
         env:
           NATS_RUNNING: "true"
@@ -392,7 +454,7 @@ jobs:
         if: always()
         uses: actions/upload-artifact@v4
         with:
-          name: nats-server-log
+          name: nats-server
           path: /tmp/nats.log
           if-no-files-found: ignore
 

AGENTS.md

@@ -31,7 +31,7 @@ Never remove or increment the base version number. The human decides when the of
 <claude-mem-context>
 # Memory Context
 
-# [rqueue] recent context, 2026-05-09 2:28pm GMT+5:30
+# [rqueue] recent context, 2026-05-09 9:08pm GMT+5:30
 
 No previous sessions found.
 </claude-mem-context>

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

@@ -122,10 +122,15 @@ private void updateCounter(boolean fail) {
     if (Objects.isNull(counter)) {
       return;
     }
+    // Pass the consumer name so the counter can route increments to the per-consumer entry
+    // when multiple @RqueueListener methods share a queue with distinct consumerName overrides;
+    // null/empty (no override) routes to the bare-queue counter unchanged.
+    String queueName = job.getQueueDetail().getName();
+    String consumerName = job.getQueueDetail().getConsumerName();
     if (fail) {
-      counter.updateFailureCount(job.getQueueDetail().getName());
+      counter.updateFailureCount(queueName, consumerName);
     } else {
-      counter.updateExecutionCount(job.getQueueDetail().getName());
+      counter.updateExecutionCount(queueName, consumerName);
     }
   }
 

rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/QueueCounter.java

@@ -29,6 +29,13 @@
 /**
  * Queue counter counts the different types of events related to a queue. Failure and execution
  * count, it supports queue registrations.
+ *
+ * <p><b>Multi-consumer keying.</b> Two {@code @RqueueListener} methods on the same queue with
+ * different {@code consumerName} overrides each produce a distinct {@link QueueDetail} and need
+ * their own counters. The maps are keyed by {@code queueName##consumerName} (or just
+ * {@code queueName} when no override is set) so the second registration does not silently
+ * overwrite the first. Increment lookups follow the same composite key — see
+ * {@link #updateFailureCount(String, String)}.
  */
 public class QueueCounter {
 
@@ -37,40 +44,70 @@ public class QueueCounter {
   private final Map<String, Counter> queueNameToFailureCounter = new HashMap<>();
   private final Map<String, Counter> queueNameToExecutionCounter = new HashMap<>();
 
-  private void updateCounter(Map<String, Counter> map, String queueName) {
-    Counter counter = map.get(queueName);
+  private static String key(String queueName, String consumerName) {
+    return (consumerName == null || consumerName.isEmpty())
+        ? queueName
+        : queueName + "##" + consumerName;
+  }
+
+  private void updateCounter(Map<String, Counter> map, String queueName, String consumerName) {
+    // Try the consumer-specific entry first; fall back to the bare queue-name entry so callers
+    // that don't yet pass a consumer name (older paths, single-consumer queues) still work.
+    Counter counter = map.get(key(queueName, consumerName));
+    if (counter == null && consumerName != null && !consumerName.isEmpty()) {
+      counter = map.get(queueName);
+    }
     if (counter == null) {
       return;
     }
     counter.increment();
   }
 
+  /** Backward-compatible single-arg increment; route to the bare-queue counter only. */
   void updateFailureCount(String queueName) {
-    updateCounter(queueNameToFailureCounter, queueName);
+    updateCounter(queueNameToFailureCounter, queueName, null);
   }
 
+  /** Backward-compatible single-arg increment; route to the bare-queue counter only. */
   void updateExecutionCount(String queueName) {
-    updateCounter(queueNameToExecutionCounter, queueName);
+    updateCounter(queueNameToExecutionCounter, queueName, null);
+  }
+
+  /**
+   * Consumer-aware increment: increments the counter registered for
+   * {@code (queueName, consumerName)}, falling back to the bare {@code queueName} counter when no
+   * consumer-scoped entry exists. Use this from
+   * {@link com.github.sonus21.rqueue.listener.RqueueExecutor} (which has the {@link QueueDetail})
+   * so multi-consumer queues keep accurate per-consumer counts.
+   */
+  void updateFailureCount(String queueName, String consumerName) {
+    updateCounter(queueNameToFailureCounter, queueName, consumerName);
+  }
+
+  /** Consumer-aware execution-count increment. See {@link #updateFailureCount(String, String)}. */
+  void updateExecutionCount(String queueName, String consumerName) {
+    updateCounter(queueNameToExecutionCounter, queueName, consumerName);
   }
 
   void registerQueue(
       MetricsProperties metricsProperties,
       Tags queueTags,
       MeterRegistry registry,
       QueueDetail queueDetail) {
+    String mapKey = key(queueDetail.getName(), queueDetail.getConsumerName());
     if (metricsProperties.countFailure()) {
       Counter.Builder builder = Counter.builder(metricsProperties.getMetricName(FAILURE_COUNT))
           .tags(queueTags.and(QUEUE_KEY, queueDetail.getQueueName()))
           .description("Failure count");
       Counter counter = builder.register(registry);
-      queueNameToFailureCounter.put(queueDetail.getName(), counter);
+      queueNameToFailureCounter.put(mapKey, counter);
     }
     if (metricsProperties.countExecution()) {
       Counter.Builder builder = Counter.builder(metricsProperties.getMetricName(EXECUTION_COUNT))
           .tags(queueTags.and(QUEUE_KEY, queueDetail.getQueueName()))
           .description("Task execution count");
       Counter counter = builder.register(registry);
-      queueNameToExecutionCounter.put(queueDetail.getName(), counter);
+      queueNameToExecutionCounter.put(mapKey, counter);
     }
   }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/RqueueCounter.java

@@ -38,4 +38,14 @@ public void updateFailureCount(String queueName) {
   public void updateExecutionCount(String queueName) {
     queueCounter.updateExecutionCount(queueName);
   }
+
+  @Override
+  public void updateFailureCount(String queueName, String consumerName) {
+    queueCounter.updateFailureCount(queueName, consumerName);
+  }
+
+  @Override
+  public void updateExecutionCount(String queueName, String consumerName) {
+    queueCounter.updateExecutionCount(queueName, consumerName);
+  }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/RqueueMetrics.java

@@ -35,6 +35,14 @@
 public class RqueueMetrics implements RqueueMetricsRegistry {
 
   static final String QUEUE_KEY = "key";
+  /**
+   * Tag added when a {@link QueueDetail} declares a {@code consumerName} override. Without this
+   * tag, two {@code @RqueueListener} methods on the same queue with different consumer names
+   * register gauges with identical (name, tag-set) pairs and Micrometer silently keeps only the
+   * first — losing the second consumer's metrics entirely.
+   */
+  static final String CONSUMER_KEY = "consumer";
+
   private static final String QUEUE_SIZE = "queue.size";
   private static final String SCHEDULED_QUEUE_SIZE = "scheduled.queue.size";
   private static final String PROCESSING_QUEUE_SIZE = "processing.queue.size";
@@ -58,17 +66,31 @@ private void monitor() {
     for (QueueDetail queueDetail : EndpointRegistry.getActiveQueueDetails()) {
       Tags queueTags =
           Tags.concat(metricsProperties.getMetricTags(), "queue", queueDetail.getName());
+      // When a queue carries multiple consumers (multiple @RqueueListener with distinct
+      // consumerName overrides), each gets its own QueueDetail. Without a `consumer` tag the
+      // gauges would share the same (name, tags) and Micrometer would drop all but the first.
+      String consumerName = queueDetail.getConsumerName();
+      boolean hasConsumerOverride = consumerName != null && !consumerName.isEmpty();
+      if (hasConsumerOverride) {
+        queueTags = queueTags.and(CONSUMER_KEY, consumerName);
+      }
       Gauge.builder(
               metricsProperties.getMetricName(QUEUE_SIZE),
               queueDetail,
-              c -> queueMetricsProvider.getPendingMessageCount(queueDetail.getName()))
+              c -> hasConsumerOverride
+                  ? queueMetricsProvider.getPendingMessageCountByConsumer(
+                      queueDetail.getName(), consumerName)
+                  : queueMetricsProvider.getPendingMessageCount(queueDetail.getName()))
           .tags(queueTags.and(QUEUE_KEY, queueDetail.getQueueName()))
           .description("The number of entries in this queue")
           .register(meterRegistry);
       Gauge.builder(
               metricsProperties.getMetricName(PROCESSING_QUEUE_SIZE),
               queueDetail,
-              c -> queueMetricsProvider.getProcessingMessageCount(queueDetail.getName()))
+              c -> hasConsumerOverride
+                  ? queueMetricsProvider.getProcessingMessageCountByConsumer(
+                      queueDetail.getName(), consumerName)
+                  : queueMetricsProvider.getProcessingMessageCount(queueDetail.getName()))
           .tags(queueTags.and(QUEUE_KEY, queueDetail.getProcessingQueueName()))
           .description("The number of entries in the processing queue")
           .register(meterRegistry);

rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/RqueueMetricsCounter.java

@@ -21,4 +21,21 @@ public interface RqueueMetricsCounter {
   void updateFailureCount(String queueName);
 
   void updateExecutionCount(String queueName);
+
+  /**
+   * Consumer-aware failure increment. When a queue carries multiple {@code @RqueueListener}
+   * methods with distinct {@code consumerName} overrides, each consumer has its own counter
+   * registered under {@code (queueName, consumerName)}; calling the bare-queue overload would
+   * route every increment to the same (last-registered) counter and silently lose per-consumer
+   * counts. Defaults to the queue-level path so callers that don't have a consumer name keep
+   * working unchanged.
+   */
+  default void updateFailureCount(String queueName, String consumerName) {
+    updateFailureCount(queueName);
+  }
+
+  /** Consumer-aware execution increment. See {@link #updateFailureCount(String, String)}. */
+  default void updateExecutionCount(String queueName, String consumerName) {
+    updateExecutionCount(queueName);
+  }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/metrics/RqueueQueueMetricsProvider.java

@@ -72,4 +72,25 @@ default long getScheduledMessageCount(String queueName, String priority) {
   default long getProcessingMessageCount(String queueName, String priority) {
     return getProcessingMessageCount(queueName);
   }
+
+  /**
+   * Per-consumer variant of {@link #getPendingMessageCount(String)}. When two
+   * {@code @RqueueListener} methods on the same queue declare different {@code consumerName}
+   * overrides, each gets its own QueueDetail and its own metric registration; backends that can
+   * report per-consumer pending depth (e.g. NATS JetStream Limits/Interest streams or any
+   * fan-out broker) should override this. The default delegates to the queue-level call so
+   * single-consumer queues, and backends without per-consumer state, behave unchanged.
+   *
+   * @param queueName    user-facing queue name
+   * @param consumerName consumer-name override from {@code @RqueueListener(consumerName=...)};
+   *                     {@code null} or empty when no override is set
+   */
+  default long getPendingMessageCountByConsumer(String queueName, String consumerName) {
+    return getPendingMessageCount(queueName);
+  }
+
+  /** Per-consumer variant of {@link #getProcessingMessageCount(String)}. See related javadoc. */
+  default long getProcessingMessageCountByConsumer(String queueName, String consumerName) {
+    return getProcessingMessageCount(queueName);
+  }
 }