fix(nats): correct delayed-queue scheduling implementation (ADR-51)

The enqueueWithDelay path was completely broken: it used a made-up
header (Nats-Next-Deliver-Time) that NATS ignores, published to the
work subject instead of a scheduler subject, and the pull consumer
had no filter so it read scheduler entries before the scheduled time.

Correct implementation (NATS >= 2.12, ADR-51):
- Replace HDR_NEXT_DELIVER_TIME with Nats-Schedule, Nats-Schedule-Target,
  and Nats-Rollup: sub — the three headers the server actually requires
- Publish to a per-message scheduler subject (<work>.sched.<msgId>);
  NATS fires the triggered message to the work subject at the due time
- Extend stream subjects to include the <work>.sched.* wildcard pattern
- Set filterSubject on pull consumers so they only receive work-subject
  messages and never see scheduler entries early

Infrastructure fixes that were hiding the bug:
- AbstractJetStreamIT: replace @testcontainers(disabledWithoutDocker=true)
  with assumeTrue(DockerClientFactory.isDockerAvailable()) inside @BeforeAll
  so NATS_RUNNING=1 works even without Docker (annotation was vetoing the
  class before @BeforeAll ran, silently skipping all ITs)
- CI: bump nats-server from v2.10.22 → v2.12.0 so scheduling assumeTrue
  guards stop silently skipping and actually gate the build
- NatsProvisioner: add filterSubject overload to ensureConsumer; add
  allowMessageSchedules flag to stream creation / upgrade path
- AllowMessageSchedulesEnforcementIT: new IT proving server enforces the
  allow_msg_schedules flag with Nats-Schedule (error 10189 without it)

All 306 rqueue-nats tests pass including 3 new scheduling ITs that
verify messages are held until the scheduled time then delivered.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: sonus21

.github/workflows/java-ci.yaml

@@ -365,7 +365,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

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-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -30,7 +30,6 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -53,8 +52,8 @@ public class NatsProvisioner {
   private final ConcurrentHashMap<String, KeyValue> kvCache = new ConcurrentHashMap<>();
   private final ConcurrentHashMap<String, Object> kvLocks = new ConcurrentHashMap<>();
 
-  // stream name → provisioned (set membership acts as the boolean flag)
-  private final Set<String> streamsDone = ConcurrentHashMap.newKeySet();
+  // stream name → schedulingEnabled (true if the stream was created/updated with allowMessageSchedules)
+  private final ConcurrentHashMap<String, Boolean> streamsDone = new ConcurrentHashMap<>();
   private final ConcurrentHashMap<String, Object> streamLocks = new ConcurrentHashMap<>();
 
   // "streamName/requestedConsumerName" → actual consumer name (may differ for stale-rebind)
@@ -63,7 +62,7 @@ public class NatsProvisioner {
 
   /**
    * Minimum NATS server version that supports server-side message scheduling via the
-   * {@code Nats-Next-Deliver-Time} JetStream publish header (ADR-51).
+   * {@code Nats-Schedule} JetStream publish header (ADR-51).
    */
   public static final String SCHEDULING_MIN_VERSION = "2.12.0";
 
@@ -146,12 +145,18 @@ 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, null);
+    ensureStream(streamName, subjects, QueueType.QUEUE, null, false);
   }
 
-  /** See {@link #ensureStream(String, List, QueueType, String)}. */
+  /** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
   public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
-    ensureStream(streamName, subjects, queueType, null);
+    ensureStream(streamName, subjects, queueType, null, false);
+  }
+
+  /** See {@link #ensureStream(String, List, QueueType, String, boolean)}. */
+  public void ensureStream(
+      String streamName, List<String> subjects, QueueType queueType, String description) {
+    ensureStream(streamName, subjects, queueType, description, false);
   }
 
   /**
@@ -168,20 +173,35 @@ public void ensureStream(String streamName, List<String> subjects, QueueType que
    * {@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>{@code allowSchedules} must be {@code true} when the stream will receive messages published
+   * with the {@code Nats-Schedule} header (ADR-51). Only callers that perform delayed
+   * enqueue should pass {@code true}; regular enqueue callers should pass {@code false} (or use the
+   * shorter overloads). Equivalent to the CLI flag {@code --allow-schedules}.
+   *
    * <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
+   * return immediately from the in-process cache. If {@code allowSchedules=true} is later requested
+   * for a stream that was previously created without that flag, the stream is updated in place via
+   * {@link JetStreamManagement#updateStream}. 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, String description) {
-    if (streamsDone.contains(streamName)) {
+      String streamName,
+      List<String> subjects,
+      QueueType queueType,
+      String description,
+      boolean allowSchedules) {
+    // Fast-path: already provisioned with at least as many capabilities as requested.
+    Boolean cached = streamsDone.get(streamName);
+    if (cached != null && (cached || !allowSchedules)) {
       return;
     }
     Object lock = streamLocks.computeIfAbsent(streamName, k -> new Object());
     synchronized (lock) {
-      if (streamsDone.contains(streamName)) {
+      cached = streamsDone.get(streamName);
+      if (cached != null && (cached || !allowSchedules)) {
         return;
       }
+      boolean enableSchedules = allowSchedules && schedulingSupported;
       try {
         StreamInfo existing = safeGetStreamInfo(streamName);
         RetentionPolicy desired =
@@ -199,6 +219,11 @@ public void ensureStream(
               .storageType(sd.getStorage())
               .retentionPolicy(desired)
               .compressionOption(CompressionOption.S2);
+          if (enableSchedules) {
+            // Enable server-side message scheduling (ADR-51 / Nats-Schedule header).
+            // Equivalent to: nats stream add MY_STREAM --allow-schedules
+            b.allowMessageSchedules(true);
+          }
           if (description != null && !description.isEmpty()) {
             b.description(description);
           }
@@ -223,30 +248,63 @@ public void ensureStream(
                     + " — leaving existing config in place.",
                 new Object[] {streamName, actual, desired});
           }
+          // If scheduling was previously not enabled but is now required, update the stream.
+          if (enableSchedules && !existing.getConfiguration().getAllowMsgSchedules()) {
+            StreamConfiguration updated = StreamConfiguration.builder(existing.getConfiguration())
+                .allowMessageSchedules(true)
+                .build();
+            jsm.updateStream(updated);
+            log.log(Level.INFO,
+                "Stream ''{0}'' updated to enable message scheduling (ADR-51).", streamName);
+          }
         }
       } catch (IOException | JetStreamApiException e) {
         throw new RqueueNatsException(
             "Failed to ensure stream '" + streamName + "' for subjects " + subjects, e);
       }
-      streamsDone.add(streamName);
+      streamsDone.put(streamName, enableSchedules);
     }
   }
 
   /**
    * Ensure a durable pull consumer exists, returning the consumer name.
    * Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
    *
-   * <p>No filter subject is set on the consumer: each queue already has its own dedicated stream
-   * with a single subject, so a filter would be redundant. More importantly, omitting the filter
-   * allows multiple independent consumer groups (fan-out) to coexist on the same stream — NATS
-   * rejects two consumers with the same filter subject (error 10100) regardless of retention type.
+   * <p>Overload without a filter subject: used when the stream has only the work subject so the
+   * filter would be redundant, or when multiple independent consumer groups (fan-out) must coexist
+   * on a Limits-retention stream (NATS rejects two consumers with the same filter subject, error
+   * 10100). For WorkQueue streams that also carry scheduler subjects ({@code .sched.*}) a filter
+   * subject MUST be supplied via
+   * {@link #ensureConsumer(String, String, String, Duration, long, long)} so the consumer only
+   * receives work-subject messages and does not accidentally pick up scheduler entries.
    */
   public String ensureConsumer(
       String streamName,
       String consumerName,
       Duration ackWait,
       long maxDeliver,
       long maxAckPending) {
+    return ensureConsumer(streamName, consumerName, null, ackWait, maxDeliver, maxAckPending);
+  }
+
+  /**
+   * Ensure a durable pull consumer exists with an optional subject filter, returning the consumer
+   * name. Hits the NATS backend at most once per (stream, consumer) pair per process lifetime.
+   *
+   * <p>{@code filterSubject} — when non-null, sets the consumer's filter subject so it only
+   * receives messages published to that subject. Required when the stream carries both work subjects
+   * and scheduler subjects ({@code .sched.*}): without a filter the consumer reads scheduler
+   * entries before the scheduled time, delivering the message early. Pass {@code null} for streams
+   * that do not use NATS scheduling, or where fan-out across multiple consumers is needed (Limits
+   * retention).
+   */
+  public String ensureConsumer(
+      String streamName,
+      String consumerName,
+      String filterSubject,
+      Duration ackWait,
+      long maxDeliver,
+      long maxAckPending) {
     String cacheKey = streamName + "/" + consumerName;
     String cached = consumerCache.get(cacheKey);
     if (cached != null) {
@@ -259,7 +317,7 @@ public String ensureConsumer(
         return cached;
       }
       String actual =
-          doEnsureConsumer(streamName, consumerName, ackWait, maxDeliver, maxAckPending);
+          doEnsureConsumer(streamName, consumerName, filterSubject, ackWait, maxDeliver, maxAckPending);
       consumerCache.put(cacheKey, actual);
       return actual;
     }
@@ -268,6 +326,7 @@ public String ensureConsumer(
   private String doEnsureConsumer(
       String streamName,
       String consumerName,
+      String filterSubject,
       Duration ackWait,
       long maxDeliver,
       long maxAckPending) {
@@ -295,16 +354,20 @@ private String doEnsureConsumer(
         throw new RqueueNatsException("Consumer '" + consumerName + "' on stream '" + streamName
             + "' does not exist and autoCreateConsumers=false");
       }
-      jsm.addOrUpdateConsumer(
-          streamName,
-          ConsumerConfiguration.builder()
-              .durable(consumerName)
-              .ackPolicy(AckPolicy.Explicit)
-              .deliverPolicy(DeliverPolicy.All)
-              .ackWait(ackWait)
-              .maxDeliver(maxDeliver)
-              .maxAckPending(maxAckPending)
-              .build());
+      ConsumerConfiguration.Builder ccBuilder = ConsumerConfiguration.builder()
+          .durable(consumerName)
+          .ackPolicy(AckPolicy.Explicit)
+          .deliverPolicy(DeliverPolicy.All)
+          .ackWait(ackWait)
+          .maxDeliver(maxDeliver)
+          .maxAckPending(maxAckPending);
+      if (filterSubject != null && !filterSubject.isEmpty()) {
+        // Filter to the work subject only so that scheduler entries (published to
+        // <workSubject>.sched.*) are not delivered to this consumer before the scheduled time.
+        // The NATS scheduler fires the triggered message to workSubject when the time arrives.
+        ccBuilder.filterSubject(filterSubject);
+      }
+      jsm.addOrUpdateConsumer(streamName, ccBuilder.build());
       return consumerName;
     } catch (JetStreamApiException e) {
       throw new RqueueNatsException(