Fix NATS retry and DLQ routing

Three bugs prevented messages from retrying and reaching the DLQ:

1. Missing parkForRetry override: default nack() replays original payload
   bytes so failureCount embedded in RqueueMessage never accumulated across
   deliveries. Override acks the original and re-publishes the updated message
   (with incremented failureCount). Nats-Msg-Id is suffixed with the failure
   count to bypass the stream deduplication window.

2. Missing moveToDlq override: default re-enqueued to the source queue.
   Override acks the original and publishes to the targetQueue's NATS stream
   and subject (streamPrefix+targetQueue / subjectPrefix+targetQueue), the
   same naming convention used for every other queue.

3. autoCreateDlqStream defaulted to true; changed to false so advisory-bridge
   DLQ streams are opt-in rather than created automatically for every queue.

Assisted-By: Claude Code

Author: sonus21

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/spi/MessageBroker.java

@@ -147,6 +147,38 @@ default boolean extendVisibilityTimeout(QueueDetail q, RqueueMessage m, long del
 
   long size(QueueDetail q);
 
+  /**
+   * Short label for the storage backend shown in the dashboard "Queue Storage Footprint" section
+   * header (e.g. "Redis", "NATS"). Defaults to "Redis".
+   */
+  default String storageKicker() {
+    return "Redis";
+  }
+
+  /**
+   * One-line description for the storage backend shown below the footprint section heading.
+   * Defaults to the Redis description.
+   */
+  default String storageDescription() {
+    return "Underlying Redis structures for the queues visible on this page.";
+  }
+
+  /**
+   * Display name for the primary storage unit backing the given queue's messages (pending,
+   * in-flight, and completed). Returns {@code null} to fall back to the Redis key name.
+   */
+  default String storageDisplayName(QueueDetail q) {
+    return null;
+  }
+
+  /**
+   * Display name for the dead-letter storage unit of the given queue. Returns {@code null} to
+   * fall back to the DLQ key name stored in {@code DeadLetterQueue}.
+   */
+  default String dlqStorageDisplayName(QueueDetail q) {
+    return null;
+  }
+
   AutoCloseable subscribe(String channel, Consumer<String> handler);
 
   void publish(String channel, String payload);

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

@@ -33,7 +33,7 @@ public class RqueueNatsConfig {
 
   private boolean autoCreateStreams = true;
   private boolean autoCreateConsumers = true;
-  private boolean autoCreateDlqStream = true;
+  private boolean autoCreateDlqStream = false;
 
   private StreamDefaults streamDefaults = new StreamDefaults();
   private ConsumerDefaults consumerDefaults = new ConsumerDefaults();

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/dao/NatsRqueueJobDao.java

@@ -13,14 +13,11 @@
 import com.github.sonus21.rqueue.config.NatsBackendCondition;
 import com.github.sonus21.rqueue.dao.RqueueJobDao;
 import com.github.sonus21.rqueue.models.db.RqueueJob;
+import com.github.sonus21.rqueue.nats.internal.NatsProvisioner;
 import com.github.sonus21.rqueue.nats.kv.NatsKvBuckets;
-import io.nats.client.Connection;
 import io.nats.client.JetStreamApiException;
 import io.nats.client.KeyValue;
-import io.nats.client.KeyValueManagement;
-import io.nats.client.api.KeyValueConfiguration;
 import io.nats.client.api.KeyValueEntry;
-import io.nats.client.api.KeyValueStatus;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -31,7 +28,6 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import org.springframework.context.annotation.Conditional;
@@ -56,44 +52,14 @@ public class NatsRqueueJobDao implements RqueueJobDao {
   private static final Logger log = Logger.getLogger(NatsRqueueJobDao.class.getName());
   private static final String BUCKET_NAME = NatsKvBuckets.JOBS;
 
-  private final Connection connection;
-  private final KeyValueManagement kvm;
-  private final AtomicReference<KeyValue> kvRef = new AtomicReference<>();
+  private final NatsProvisioner provisioner;
 
-  public NatsRqueueJobDao(Connection connection) throws IOException {
-    this.connection = connection;
-    this.kvm = connection.keyValueManagement();
+  public NatsRqueueJobDao(NatsProvisioner provisioner) {
+    this.provisioner = provisioner;
   }
 
-  private KeyValue ensureBucket(Duration ttl) throws IOException, JetStreamApiException {
-    KeyValue cached = kvRef.get();
-    if (cached != null) {
-      return cached;
-    }
-    synchronized (this) {
-      cached = kvRef.get();
-      if (cached != null) {
-        return cached;
-      }
-      try {
-        KeyValueStatus status = kvm.getStatus(BUCKET_NAME);
-        if (status != null) {
-          KeyValue kv = connection.keyValue(BUCKET_NAME);
-          kvRef.set(kv);
-          return kv;
-        }
-      } catch (JetStreamApiException missing) {
-        // fall through
-      }
-      KeyValueConfiguration.Builder cfg = KeyValueConfiguration.builder().name(BUCKET_NAME);
-      if (ttl != null && !ttl.isZero() && !ttl.isNegative()) {
-        cfg.ttl(ttl);
-      }
-      kvm.create(cfg.build());
-      KeyValue kv = connection.keyValue(BUCKET_NAME);
-      kvRef.set(kv);
-      return kv;
-    }
+  private KeyValue kv(Duration ttl) throws IOException, JetStreamApiException {
+    return provisioner.ensureKv(BUCKET_NAME, ttl);
   }
 
   @Override
@@ -104,8 +70,7 @@ public void createJob(RqueueJob rqueueJob, Duration expiry) {
   @Override
   public void save(RqueueJob rqueueJob, Duration expiry) {
     try {
-      KeyValue kv = ensureBucket(expiry);
-      kv.put(sanitize(rqueueJob.getId()), serialize(rqueueJob));
+      kv(expiry).put(sanitize(rqueueJob.getId()), serialize(rqueueJob));
     } catch (IOException | JetStreamApiException e) {
       log.log(Level.WARNING, "save job " + rqueueJob.getId() + " failed", e);
     }
@@ -147,8 +112,7 @@ public List<RqueueJob> finByMessageIdIn(List<String> messageIds) {
   @Override
   public void delete(String jobId) {
     try {
-      KeyValue kv = ensureBucket(null);
-      kv.delete(sanitize(jobId));
+      kv(null).delete(sanitize(jobId));
     } catch (IOException | JetStreamApiException e) {
       log.log(Level.WARNING, "delete job " + jobId + " failed", e);
     }
@@ -158,8 +122,7 @@ public void delete(String jobId) {
 
   private RqueueJob loadByKey(String key) {
     try {
-      KeyValue kv = ensureBucket(null);
-      KeyValueEntry entry = kv.get(key);
+      KeyValueEntry entry = kv(null).get(key);
       if (entry == null || entry.getValue() == null) {
         return null;
       }
@@ -172,8 +135,7 @@ private RqueueJob loadByKey(String key) {
 
   private List<RqueueJob> scanForMessageIds(Collection<String> messageIds) {
     try {
-      KeyValue kv = ensureBucket(null);
-      List<String> keys = new ArrayList<>(kv.keys());
+      List<String> keys = new ArrayList<>(kv(null).keys());
       List<RqueueJob> out = new ArrayList<>();
       for (String k : keys) {
         RqueueJob j = loadByKey(k);

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/dao/NatsRqueueQStatsDao.java

@@ -6,46 +6,133 @@
  * 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.nats.dao;
 
 import com.github.sonus21.rqueue.config.NatsBackendCondition;
 import com.github.sonus21.rqueue.dao.RqueueQStatsDao;
 import com.github.sonus21.rqueue.models.db.QueueStatistics;
+import com.github.sonus21.rqueue.nats.internal.NatsProvisioner;
+import com.github.sonus21.rqueue.nats.kv.NatsKvBuckets;
+import io.nats.client.JetStreamApiException;
+import io.nats.client.KeyValue;
+import io.nats.client.api.KeyValueEntry;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Repository;
 
 /**
- * NATS-backend stub for {@link RqueueQStatsDao}. The Redis impl persists per-queue daily
- * aggregates as serialized {@link QueueStatistics} objects driving the dashboard charts; on
- * NATS we no-op writes and return empty reads in v1 so that
- * {@code RqueueJobMetricsAggregatorService} can boot without a missing-bean failure even
- * though the chart panel is empty.
+ * NATS-backed {@link RqueueQStatsDao} using the {@code rqueue-queue-stats} JetStream KV bucket.
+ *
+ * <p>Each {@link QueueStatistics} entry is stored as a single KV record keyed by its
+ * {@link QueueStatistics#getId()} (e.g. {@code __rq::q-stat::job-morgue}), serialized via Java
+ * serialization to match the other NATS DAO implementations. The key is sanitized to the NATS KV
+ * character set before storage.
  *
- * <p>Replace with a NATS-native impl (a dedicated {@code rqueue-queue-stats} KV bucket
- * mirroring the pattern of {@code NatsRqueueSystemConfigDao}) when chart support lands for
- * NATS.
+ * <p>No bucket-level TTL is set: the aggregator service calls
+ * {@link QueueStatistics#pruneStats} before each {@link #save}, so stale per-day entries inside
+ * the object are trimmed to {@code rqueue.web.statistic.history.day} days automatically.
  */
 @Repository
 @Conditional(NatsBackendCondition.class)
+@DependsOn("natsKvBucketValidator")
 public class NatsRqueueQStatsDao implements RqueueQStatsDao {
 
+  private static final Logger log = Logger.getLogger(NatsRqueueQStatsDao.class.getName());
+  private static final String BUCKET_NAME = NatsKvBuckets.QUEUE_STATS;
+
+  private final NatsProvisioner provisioner;
+
+  public NatsRqueueQStatsDao(NatsProvisioner provisioner) {
+    this.provisioner = provisioner;
+  }
+
+  private KeyValue kv() throws IOException, JetStreamApiException {
+    return provisioner.ensureKv(BUCKET_NAME, null);
+  }
+
   @Override
   public QueueStatistics findById(String id) {
-    return null;
+    if (id == null) {
+      return null;
+    }
+    try {
+      KeyValueEntry entry = kv().get(sanitize(id));
+      if (entry == null || entry.getValue() == null) {
+        return null;
+      }
+      return deserialize(entry.getValue());
+    } catch (IOException | JetStreamApiException e) {
+      log.log(Level.WARNING, "findById id=" + id + " failed", e);
+      return null;
+    }
   }
 
   @Override
   public List<QueueStatistics> findAll(Collection<String> ids) {
-    return Collections.emptyList();
+    List<QueueStatistics> out = new ArrayList<>(ids.size());
+    for (String id : ids) {
+      QueueStatistics stat = findById(id);
+      if (stat != null) {
+        out.add(stat);
+      }
+    }
+    return out;
   }
 
   @Override
   public void save(QueueStatistics queueStatistics) {
-    // intentionally no-op until a NATS-native chart store lands
+    if (queueStatistics == null) {
+      throw new IllegalArgumentException("queueStatistics cannot be null");
+    }
+    if (queueStatistics.getId() == null) {
+      throw new IllegalArgumentException("id cannot be null: " + queueStatistics);
+    }
+    try {
+      kv().put(sanitize(queueStatistics.getId()), serialize(queueStatistics));
+    } catch (IOException | JetStreamApiException e) {
+      log.log(Level.WARNING, "save id=" + queueStatistics.getId() + " failed", e);
+    }
+  }
+
+  // ---- helpers ----------------------------------------------------------
+
+  private static byte[] serialize(QueueStatistics stat) throws IOException {
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
+      oos.writeObject(stat);
+    }
+    return baos.toByteArray();
+  }
+
+  private static QueueStatistics deserialize(byte[] bytes) {
+    try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes))) {
+      Object o = ois.readObject();
+      return o instanceof QueueStatistics ? (QueueStatistics) o : null;
+    } catch (IOException | ClassNotFoundException e) {
+      log.log(Level.WARNING, "deserialize QueueStatistics failed", e);
+      return null;
+    }
+  }
+
+  /** KV keys allow {@code [A-Za-z0-9_=.-]} only. */
+  private static String sanitize(String key) {
+    return key == null ? "_" : key.replaceAll("[^A-Za-z0-9_=.-]", "_");
   }
 }