feat(nats): validate queue name + add stream description on creation

- MessageBroker SPI: new default no-op validateQueueName(String) hook.
- JetStreamMessageBroker overrides it to reject queue names containing
  '.', '*', '>' or whitespace. NATS subjects use '.' as a hierarchy
  separator and stream / consumer names disallow it outright, so a
  silent accept turned into an opaque driver-side rejection at first
  publish; now it fails loudly at registration with a message that
  points users at '-' / '_'.
- BaseMessageSender.registerQueueInternal and
  RqueueMessageListenerContainer.initializeQueueRegistry both call
  validateQueueName, so both the explicit RqueueEndpointManager.registerQueue
  path and the @RqueueListener bootstrap path validate.
- NatsProvisioner.ensureStream gains a description overload; the broker
  passes "rqueue queue: <name>" (and "(priority=<p>)" / "rqueue DLQ for
  queue: <name>" for the priority and DLQ variants) so operators can map
  a stream back to the queue that created it via `nats stream info`.
- NatsStreamValidator passes the same description on the bootstrap path.

Tests: JetStreamMessageBrokerQueueNameValidationTest pins the
character-rejection rules; JetStreamMessageBrokerStreamDescriptionTest
pins the description format on enqueue / onQueueRegistered / priority
enqueue paths; RqueueEndpointManagerImplTest gains a regression covering
broker-validation propagation through registerQueue.

Assisted-By: Claude Code

Author: sonus21

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -204,6 +204,7 @@ protected Object deleteAllMessages(QueueDetail queueDetail) {
 
   protected void registerQueueInternal(String queueName, QueueType type, String... priorities) {
     validateQueue(queueName);
+    messageBroker.validateQueueName(queueName);
     notNull(priorities, "priorities cannot be null");
     Map<String, Integer> priorityMap = new HashMap<>();
     priorityMap.put(DEFAULT_PRIORITY_KEY, 1);

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

@@ -53,6 +53,18 @@ default void enqueue(QueueDetail q, String priority, RqueueMessage m) {
    */
   default void onQueueRegistered(QueueDetail q) {}
 
+  /**
+   * Validate the queue name against backend-specific rules. Called from every queue-registration
+   * path ({@code RqueueEndpointManager.registerQueue} and the {@code @RqueueListener} bootstrap)
+   * before the queue is added to the registry, so an illegal name fails fast with a clear error
+   * instead of surfacing later as an opaque NATS / driver-side rejection.
+   *
+   * <p>Default is a no-op — backends like Redis accept any non-empty name.
+   *
+   * @throws IllegalArgumentException if {@code queueName} is not legal for this backend
+   */
+  default void validateQueueName(String queueName) {}
+
   /**
    * Reactive variant of {@link #enqueue(QueueDetail, RqueueMessage)}. The default falls back to the
    * blocking implementation wrapped in {@code Mono.fromRunnable}; backends with native async

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

@@ -294,9 +294,15 @@ public void stop(Runnable callback) {
   private void initializeQueueRegistry() {
     log.info("Initializing queue registry");
     EndpointRegistry.delete();
+    // Validate every @RqueueListener queue name against backend rules before registering, so
+    // illegal names (e.g. dots when running on NATS) fail loudly at startup instead of surfacing
+    // as opaque driver errors at first publish.
     for (MappingInformation mappingInformation :
         rqueueMessageHandler.getHandlerMethodMap().keySet()) {
       for (String queue : mappingInformation.getQueueNames()) {
+        if (messageBroker != null) {
+          messageBroker.validateQueueName(queue);
+        }
         for (QueueDetail queueDetail : getQueueDetail(queue, mappingInformation)) {
           EndpointRegistry.register(queueDetail);
         }

rqueue-core/src/test/java/com/github/sonus21/rqueue/core/impl/RqueueEndpointManagerImplTest.java

@@ -29,6 +29,7 @@
 import com.github.sonus21.rqueue.core.EndpointRegistry;
 import com.github.sonus21.rqueue.core.RqueueEndpointManager;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
+import com.github.sonus21.rqueue.core.spi.MessageBroker;
 import com.github.sonus21.rqueue.core.spi.redis.RedisMessageBroker;
 import com.github.sonus21.rqueue.dao.RqueueSystemConfigDao;
 import com.github.sonus21.rqueue.listener.RqueueMessageHeaders;
@@ -186,4 +187,29 @@ void pauseQueueWithPriority() {
         .pauseUnpauseQueue(request);
     assertFalse(rqueueEndpointManager.pauseUnpauseQueue(queue, priorities[0], false));
   }
+
+  /**
+   * registerQueue must consult {@link MessageBroker#validateQueueName(String)}; an exception there
+   * must propagate so callers fail fast on illegal names (e.g. dots when running on NATS).
+   */
+  @Test
+  void registerQueue_propagatesBrokerValidationFailure() throws IllegalAccessException {
+    MessageBroker rejectingBroker = new RedisMessageBroker(messageTemplate) {
+      @Override
+      public void validateQueueName(String queueName) {
+        if (queueName.contains(".")) {
+          throw new IllegalArgumentException("dot rejected: " + queueName);
+        }
+      }
+    };
+    RqueueEndpointManager mgr = new RqueueEndpointManagerImpl(
+        messageTemplate, rejectingBroker, messageConverter, messageHeaders);
+    RqueueConfig rqueueConfig = new RqueueConfig(redisConnectionFactory, null, false, 1);
+    FieldUtils.writeField(mgr, "rqueueConfig", rqueueConfig, true);
+
+    IllegalArgumentException ex =
+        assertThrows(IllegalArgumentException.class, () -> mgr.registerQueue("orders.us"));
+    assertTrue(ex.getMessage().contains("orders.us"));
+    assertFalse(mgr.isQueueRegistered("orders.us"));
+  }
 }

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -125,7 +125,12 @@ public KeyValue ensureKv(String bucketName, Duration ttl)
    * use {@link #ensureStream(String, List, QueueType)} instead.
    */
   public void ensureStream(String streamName, List<String> subjects) {
-    ensureStream(streamName, subjects, QueueType.QUEUE);
+    ensureStream(streamName, subjects, QueueType.QUEUE, null);
+  }
+
+  /** See {@link #ensureStream(String, List, QueueType, String)}. */
+  public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
+    ensureStream(streamName, subjects, queueType, null);
   }
 
   /**
@@ -138,11 +143,16 @@ public void ensureStream(String streamName, List<String> subjects) {
    *       independent durable consumer group receives all messages; stream/fan-out semantics.
    * </ul>
    *
+   * <p>{@code description} is forwarded to JetStream as the stream's description (visible via
+   * {@code nats stream info}). Callers should pass the rqueue queue name so operators can map a
+   * stream back to the queue that created it; pass {@code null} to skip.
+   *
    * <p>Hits the NATS backend at most once per stream name per process lifetime; subsequent calls
    * return immediately from the in-process cache. If the stream already exists with a different
    * retention policy, a WARNING is logged and the existing config is left untouched.
    */
-  public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
+  public void ensureStream(
+      String streamName, List<String> subjects, QueueType queueType, String description) {
     if (streamsDone.contains(streamName)) {
       return;
     }
@@ -169,6 +179,9 @@ public void ensureStream(String streamName, List<String> subjects, QueueType que
               .retentionPolicy(desired)
               .duplicateWindow(sd.getDuplicateWindow())
               .compressionOption(CompressionOption.S2);
+          if (description != null && !description.isEmpty()) {
+            b.description(description);
+          }
           if (sd.getMaxMsgs() > 0) {
             b.maxMessages(sd.getMaxMsgs());
           }

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

@@ -14,6 +14,7 @@
 import com.github.sonus21.rqueue.core.RqueueMessage;
 import com.github.sonus21.rqueue.core.spi.Capabilities;
 import com.github.sonus21.rqueue.core.spi.MessageBroker;
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
 import com.github.sonus21.rqueue.nats.RqueueNatsException;
@@ -159,13 +160,29 @@ private String dlqSubjectFor(QueueDetail q) {
         : subjectFor(q) + config.getDlqSubjectSuffix();
   }
 
+  /** Stream description shown in {@code nats stream info} so operators can map back to rqueue. */
+  private static String streamDescription(QueueDetail q) {
+    return "rqueue queue: " + q.getName();
+  }
+
+  /** Stream description for the priority sub-stream. */
+  private static String streamDescription(QueueDetail q, String priority) {
+    return priority == null || priority.isEmpty()
+        ? streamDescription(q)
+        : "rqueue queue: " + q.getName() + " (priority=" + priority + ")";
+  }
+
+  private static String dlqStreamDescription(QueueDetail q) {
+    return "rqueue DLQ for queue: " + q.getName();
+  }
+
   // ---- MessageBroker -----------------------------------------------------
 
   @Override
   public void enqueue(QueueDetail q, RqueueMessage m) {
     String subject = subjectFor(q);
     String stream = streamFor(q);
-    provisioner.ensureStream(stream, List.of(subject), q.getType());
+    provisioner.ensureStream(stream, List.of(subject), q.getType(), streamDescription(q));
     Headers headers = new Headers();
     if (m.getId() != null) {
       headers.add("Nats-Msg-Id", m.getId());
@@ -198,7 +215,8 @@ public void enqueue(QueueDetail q, RqueueMessage m) {
   public void enqueue(QueueDetail q, String priority, RqueueMessage m) {
     String subject = subjectFor(q, priority);
     String stream = streamFor(q, priority);
-    provisioner.ensureStream(stream, List.of(subject), q.getType());
+    provisioner.ensureStream(
+        stream, List.of(subject), q.getType(), streamDescription(q, priority));
     Headers headers = new Headers();
     if (m.getId() != null) {
       headers.add("Nats-Msg-Id", m.getId());
@@ -243,7 +261,7 @@ public Mono<Void> enqueueReactive(QueueDetail q, RqueueMessage m) {
     String subject = subjectFor(q);
     String stream = streamFor(q);
     try {
-      provisioner.ensureStream(stream, List.of(subject), q.getType());
+      provisioner.ensureStream(stream, List.of(subject), q.getType(), streamDescription(q));
     } catch (Exception e) {
       return Mono.error(new RqueueNatsException(
           "Failed to provision stream for reactive enqueue id="
@@ -431,7 +449,8 @@ public void moveToDlq(
       headers.add("Nats-Msg-Id", updated.getId() + "-dlq");
     }
     try {
-      provisioner.ensureStream(dlqStream, List.of(dlqSubject));
+      provisioner.ensureStream(
+          dlqStream, List.of(dlqSubject), QueueType.QUEUE, "rqueue DLQ for queue: " + targetQueue);
       byte[] payload = serdes.serialize(updated);
       js.publish(dlqSubject, headers, payload);
     } catch (IOException | JetStreamApiException e) {
@@ -528,7 +547,36 @@ public void publish(String channel, String payload) {
   public void onQueueRegistered(QueueDetail q) {
     String stream = streamFor(q);
     String subject = subjectFor(q);
-    provisioner.ensureStream(stream, List.of(subject), q.getType());
+    provisioner.ensureStream(stream, List.of(subject), q.getType(), streamDescription(q));
+  }
+
+  /**
+   * NATS subjects use {@code .} as a hierarchy separator and stream / consumer names disallow it
+   * outright. A queue name like {@code "orders.us"} would (a) silently turn the publish subject
+   * into a two-level token tree and (b) make {@code rqueue-js-orders.us} an invalid stream name —
+   * the JetStream API would reject it with an opaque error at first publish. Reject the name at
+   * registration so the failure is loud and local.
+   *
+   * <p>{@code *} and {@code &gt;} are also illegal in subject tokens (they're NATS wildcards), and
+   * whitespace is rejected by the server; check those too while we're here.
+   */
+  @Override
+  public void validateQueueName(String queueName) {
+    if (queueName == null || queueName.isEmpty()) {
+      return;
+    }
+    for (int i = 0; i < queueName.length(); i++) {
+      char c = queueName.charAt(i);
+      if (c == '.' || c == '*' || c == '>' || Character.isWhitespace(c)) {
+        throw new IllegalArgumentException(
+            "Queue name '"
+                + queueName
+                + "' contains illegal character '"
+                + c
+                + "' for the NATS backend. Subject hierarchy ('.'), wildcards ('*', '>') and"
+                + " whitespace are not allowed in queue names — use '-' or '_' instead.");
+      }
+    }
   }
 
   @Override
@@ -578,7 +626,8 @@ public void close() {
   public void provisionDlq(QueueDetail 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)));
+    provisioner.ensureStream(
+        dlqStreamFor(q), List.of(dlqSubjectFor(q)), QueueType.QUEUE, dlqStreamDescription(q));
   }
 
   /**

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

@@ -165,7 +165,8 @@ private void tryEnsureConsumer(
 
   private int tryEnsure(List<String> failures, String streamName, String subject, QueueDetail q) {
     try {
-      provisioner.ensureStream(streamName, List.of(subject), q.getType());
+      provisioner.ensureStream(
+          streamName, List.of(subject), q.getType(), "rqueue queue: " + q.getName());
       return 1;
     } catch (RqueueNatsException e) {
       failures.add(streamName + " (subject " + subject + "): " + rootCause(e));

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

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2024-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
+ */
+package com.github.sonus21.rqueue.nats;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import com.github.sonus21.rqueue.nats.js.JetStreamMessageBroker;
+import com.github.sonus21.rqueue.serdes.RqJacksonSerDes;
+import com.github.sonus21.rqueue.serdes.SerializationUtils;
+import io.nats.client.Connection;
+import io.nats.client.JetStream;
+import io.nats.client.JetStreamManagement;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Pins the dot/wildcard/whitespace rejection on queue names for the NATS backend. NATS subjects
+ * use {@code .} as a hierarchy separator and stream / consumer names disallow it outright, so
+ * silently accepting an illegal name leads to an opaque driver-side rejection at first publish.
+ */
+@NatsUnitTest
+class JetStreamMessageBrokerQueueNameValidationTest {
+
+  private JetStreamMessageBroker newBroker() {
+    return new JetStreamMessageBroker(
+        mock(Connection.class),
+        mock(JetStream.class),
+        mock(JetStreamManagement.class),
+        RqueueNatsConfig.defaults(),
+        new RqJacksonSerDes(SerializationUtils.getObjectMapper()),
+        null);
+  }
+
+  @Test
+  void rejects_queueName_with_dot() {
+    JetStreamMessageBroker broker = newBroker();
+    IllegalArgumentException ex =
+        assertThrows(IllegalArgumentException.class, () -> broker.validateQueueName("orders.us"));
+    assertTrue(ex.getMessage().contains("orders.us"));
+    assertTrue(ex.getMessage().contains("'.'"));
+  }
+
+  @Test
+  void rejects_queueName_with_star_wildcard() {
+    JetStreamMessageBroker broker = newBroker();
+    assertThrows(IllegalArgumentException.class, () -> broker.validateQueueName("foo*bar"));
+  }
+
+  @Test
+  void rejects_queueName_with_gt_wildcard() {
+    JetStreamMessageBroker broker = newBroker();
+    assertThrows(IllegalArgumentException.class, () -> broker.validateQueueName("foo>bar"));
+  }
+
+  @Test
+  void rejects_queueName_with_whitespace() {
+    JetStreamMessageBroker broker = newBroker();
+    assertThrows(IllegalArgumentException.class, () -> broker.validateQueueName("foo bar"));
+    assertThrows(IllegalArgumentException.class, () -> broker.validateQueueName("foo\tbar"));
+  }
+
+  @Test
+  void accepts_legal_queue_names() {
+    JetStreamMessageBroker broker = newBroker();
+    assertDoesNotThrow(() -> broker.validateQueueName("orders"));
+    assertDoesNotThrow(() -> broker.validateQueueName("orders-us"));
+    assertDoesNotThrow(() -> broker.validateQueueName("orders_us"));
+    assertDoesNotThrow(() -> broker.validateQueueName("orders123"));
+    assertDoesNotThrow(() -> broker.validateQueueName("a"));
+  }
+
+  @Test
+  void accepts_null_or_empty_so_that_callers_higher_up_the_stack_handle_those_errors() {
+    JetStreamMessageBroker broker = newBroker();
+    assertDoesNotThrow(() -> broker.validateQueueName(null));
+    assertDoesNotThrow(() -> broker.validateQueueName(""));
+  }
+}