fix: max retry

Author: sonus21

.github/workflows/java-ci.yaml

@@ -446,6 +446,7 @@ jobs:
 
       - name: Run NATS tests
         env:
+          RQUEUE_TEST_BACKEND: nats
           NATS_RUNNING: "true"
           NATS_URL: nats://127.0.0.1:4222
         run: ./gradlew :rqueue-nats:test :rqueue-spring-boot-starter:test :rqueue-spring:test -DincludeTags=nats

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

@@ -225,18 +225,11 @@ private void handleRetryExceededMessage(JobImpl job, int failureCount, Throwable
     }
   }
 
-  private int getMaxRetryCount(RqueueMessage rqueueMessage, QueueDetail queueDetail) {
-    return rqueueMessage.getRetryCount() == null
-        ? queueDetail.getNumRetry()
-        : rqueueMessage.getRetryCount();
-  }
-
   private void handleFailure(JobImpl job, int failureCount, Throwable throwable) {
     if (job.getQueueDetail().isDoNotRetryError(throwable)) {
       handleRetryExceededMessage(job, failureCount, throwable);
     } else {
-      int maxRetryCount = getMaxRetryCount(job.getRqueueMessage(), job.getQueueDetail());
-      if (failureCount < maxRetryCount) {
+      if (!RetryPolicy.isExhausted(job.getRqueueMessage(), job.getQueueDetail(), failureCount)) {
         long delay = taskExecutionBackoff.nextBackOff(
             job.getMessage(), job.getRqueueMessage(), failureCount, throwable);
         if (delay == TaskExecutionBackOff.STOP) {

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

@@ -0,0 +1,61 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ *
+ */
+
+package com.github.sonus21.rqueue.listener;
+
+import com.github.sonus21.rqueue.config.RqueueConfig;
+import com.github.sonus21.rqueue.core.RqueueMessage;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+final class RetryPolicy {
+
+  static final int UNLIMITED_RETRY_LIMIT = 100_000;
+
+  static int maxRetryCount(RqueueMessage rqueueMessage, QueueDetail queueDetail) {
+    int maxRetryCount = rqueueMessage.getRetryCount() == null
+        ? queueDetail.getNumRetry()
+        : rqueueMessage.getRetryCount();
+    if (maxRetryCount == Integer.MAX_VALUE) {
+      return UNLIMITED_RETRY_LIMIT;
+    }
+    return maxRetryCount;
+  }
+
+  static int remainingRetryCount(
+      RqueueMessage rqueueMessage, QueueDetail queueDetail, int failureCount) {
+    int maxRetryCount = maxRetryCount(rqueueMessage, queueDetail);
+    return Math.max(0, maxRetryCount - failureCount);
+  }
+
+  static int retryCountForPoll(
+      RqueueConfig rqueueConfig,
+      RqueueMessage rqueueMessage,
+      QueueDetail queueDetail,
+      int failureCount) {
+    int remainingRetryCount = remainingRetryCount(rqueueMessage, queueDetail, failureCount);
+    if (rqueueConfig.getRetryPerPoll() == -1) {
+      return remainingRetryCount;
+    }
+    return Math.min(rqueueConfig.getRetryPerPoll(), remainingRetryCount);
+  }
+
+  static boolean isExhausted(
+      RqueueMessage rqueueMessage, QueueDetail queueDetail, int failureCount) {
+    return remainingRetryCount(rqueueMessage, queueDetail, failureCount) == 0;
+  }
+}

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

@@ -111,12 +111,6 @@ private void init() {
     this.failureCount = job.getRqueueMessage().getFailureCount();
   }
 
-  private int getMaxRetryCount() {
-    return Objects.isNull(job.getRqueueMessage().getRetryCount())
-        ? job.getQueueDetail().getNumRetry()
-        : job.getRqueueMessage().getRetryCount();
-  }
-
   private void updateCounter(boolean fail) {
     RqueueMetricsCounter counter = beanProvider.getRqueueMetricsCounter();
     if (Objects.isNull(counter)) {
@@ -179,11 +173,8 @@ private boolean isOldMessage() {
   }
 
   private int getRetryCount() {
-    int maxRetry = getMaxRetryCount();
-    if (beanProvider.getRqueueConfig().getRetryPerPoll() == -1) {
-      return maxRetry;
-    }
-    return Math.min(beanProvider.getRqueueConfig().getRetryPerPoll(), maxRetry);
+    return RetryPolicy.retryCountForPoll(
+        beanProvider.getRqueueConfig(), job.getRqueueMessage(), job.getQueueDetail(), failureCount);
   }
 
   private boolean queueInactive() {
@@ -283,6 +274,7 @@ private void execute() {
   private boolean shouldRetry(long maxProcessingTime, int retryCount, int failureCount) {
     if (retryCount > 0
         && ExecutionStatus.FAILED.equals(status)
+        && !RetryPolicy.isExhausted(rqueueMessage, queueDetail, failureCount)
         && System.currentTimeMillis() < maxProcessingTime) {
       boolean doNoRetry = queueDetail.isDoNotRetryError(error);
       // it should not be retried based on the exception list

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

@@ -0,0 +1,78 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ *
+ */
+
+package com.github.sonus21.rqueue.listener;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+
+import com.github.sonus21.rqueue.CoreUnitTest;
+import com.github.sonus21.rqueue.config.RqueueConfig;
+import com.github.sonus21.rqueue.core.RqueueMessage;
+import com.github.sonus21.rqueue.utils.TestUtils;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+@CoreUnitTest
+class RetryPolicyTest {
+
+  private final QueueDetail queueDetail = TestUtils.createQueueDetail("queue", 2, 900000L, null);
+
+  @Test
+  void retryCountForPollUsesRemainingRetryBudget() {
+    RqueueConfig rqueueConfig = Mockito.mock(RqueueConfig.class);
+    doReturn(100).when(rqueueConfig).getRetryPerPoll();
+    RqueueMessage rqueueMessage = new RqueueMessage();
+
+    assertEquals(1, RetryPolicy.retryCountForPoll(rqueueConfig, rqueueMessage, queueDetail, 1));
+  }
+
+  @Test
+  void retryCountForPollKeepsExplicitMessageRetryCount() {
+    RqueueConfig rqueueConfig = Mockito.mock(RqueueConfig.class);
+    doReturn(-1).when(rqueueConfig).getRetryPerPoll();
+    RqueueMessage rqueueMessage = RqueueMessage.builder().retryCount(1000).build();
+
+    assertEquals(
+        999, RetryPolicy.retryCountForPoll(rqueueConfig, rqueueMessage, queueDetail, 1));
+  }
+
+  @Test
+  void isExhaustedUsesEffectiveRetryCount() {
+    RqueueMessage rqueueMessage = new RqueueMessage();
+
+    assertFalse(RetryPolicy.isExhausted(rqueueMessage, queueDetail, 1));
+    assertTrue(RetryPolicy.isExhausted(rqueueMessage, queueDetail, 2));
+  }
+
+  @Test
+  void retryForeverSentinelUsesFiniteLimit() {
+    QueueDetail retryForeverQueue =
+        TestUtils.createQueueDetail("queue", Integer.MAX_VALUE, 900000L, null);
+    RqueueMessage rqueueMessage = new RqueueMessage();
+
+    assertEquals(RetryPolicy.UNLIMITED_RETRY_LIMIT, RetryPolicy.maxRetryCount(
+        rqueueMessage, retryForeverQueue));
+    assertEquals(
+        1,
+        RetryPolicy.remainingRetryCount(
+            rqueueMessage, retryForeverQueue, RetryPolicy.UNLIMITED_RETRY_LIMIT - 1));
+    assertTrue(RetryPolicy.isExhausted(
+        rqueueMessage, retryForeverQueue, RetryPolicy.UNLIMITED_RETRY_LIMIT));
+  }
+}

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

@@ -228,6 +228,24 @@ void redisDefaultsBrokerAlsoUsesNormalStartQueuePath() throws Exception {
     }
   }
 
+  @Test
+  void startDoesNotStartQueuesAgainWhenContainerIsAlreadyRunning() throws Exception {
+    EndpointRegistry.delete();
+    CountingBroker broker = new CountingBroker(Capabilities.REDIS_DEFAULTS);
+    TrackingContainer container = new TrackingContainer(messageHandler);
+    container.setMessageBroker(broker);
+    container.afterPropertiesSet();
+    try {
+      container.start();
+      container.start();
+
+      assertEquals(1, container.startQueueCalls.get() + container.startGroupCalls.get());
+    } finally {
+      container.stop();
+      container.destroy();
+    }
+  }
+
   @Test
   void pollerForwardsPollingIntervalAsBrokerFetchWait() throws Exception {
     EndpointRegistry.delete();
@@ -288,6 +306,8 @@ private class TrackingContainer extends RqueueMessageListenerContainer {
     final AtomicBoolean startBrokerPollersCalled = new AtomicBoolean();
     final AtomicBoolean startQueueCalled = new AtomicBoolean();
     final AtomicBoolean startGroupCalled = new AtomicBoolean();
+    final AtomicInteger startQueueCalls = new AtomicInteger();
+    final AtomicInteger startGroupCalls = new AtomicInteger();
 
     TrackingContainer(RqueueMessageHandler handler) {
       super(handler, rqueueMessageTemplate);
@@ -297,12 +317,14 @@ private class TrackingContainer extends RqueueMessageListenerContainer {
     @Override
     protected void startQueue(String pollerKey, QueueDetail queueDetail) {
       startQueueCalled.set(true);
+      startQueueCalls.incrementAndGet();
       // Do not actually start the poller; it would need a real broker.
     }
 
     @Override
     protected void startGroup(String groupName, List<QueueDetail> queueDetails) {
       startGroupCalled.set(true);
+      startGroupCalls.incrementAndGet();
     }
   }
 }

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

@@ -101,6 +101,15 @@ void getPollWait_usesConfiguredFetchWait() {
     assertEquals(Duration.ofSeconds(9), f.broker.getPollWait(Duration.ofMillis(137)));
   }
 
+  @Test
+  void getPollWait_fallsBackToPollingIntervalWhenFetchWaitIsUnset() {
+    RqueueNatsConfig cfg = RqueueNatsConfig.defaults().setDefaultFetchWait(null);
+    Fixture f = newFixture(cfg);
+    Duration pollingInterval = Duration.ofMillis(137);
+
+    assertEquals(pollingInterval, f.broker.getPollWait(pollingInterval));
+  }
+
   @Test
   void enqueueWithPriority_appendsPrioritySuffixToSubject() throws Exception {
     Fixture f = newFixture(RqueueNatsConfig.defaults());