feat(nats): wire NATS-native dead-letter advisory bridge per queue at startup

Introduces NatsDeadLetterBridgeRegistrar (SmartInitializingSingleton + DisposableBean)
that walks EndpointRegistry and calls
JetStreamMessageBroker.installDeadLetterBridge(queue, queue.resolvedConsumerName())
for each registered queue, so $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.* fired by
JetStream is caught and the offending payload is republished onto the queue's DLQ
stream (<streamPrefix><queue><dlqStreamSuffix>). Producer-only mode (RqueueConfig
.isProducer()=true) is detected and skipped. Bean is wired from RqueueNatsAutoConfig
so it loads only when rqueue.backend=nats.

This is the NATS-side analog of the rqueue-level DLQ path (PostProcessingHandler →
broker.moveToDlq, already covered by NatsSchedulingAdvancedE2EIT). The advisory
bridge is the defensive net for cases outside rqueue's control: a handler that
hangs past visibilityTimeout, or a process that crashes mid-handler.

Also fixes a critical test-isolation bug discovered while validating the bridge
end-to-end: nine NATS Spring Boot E2E classes were setting
"rqueue.nats.stream-prefix" / "rqueue.nats.subject-prefix" but the actual
property paths are "rqueue.nats.naming.stream-prefix" /
"rqueue.nats.naming.subject-prefix". Streams from every test class were
collapsing onto the same default prefix "rqueue-js-", and the @BeforeAll cleanup
hooks were silently no-op'ing because they targeted the wrong (never-used)
per-class prefix. Patched all nine classes to use the correct property paths;
streams now actually live under per-class prefixes and cleanup actually wipes
them between runs.

Other test changes:
- NatsConsumerNameOverrideE2EIT: ConsumerInfo lookup now derives the stream name
  from STREAM_PREFIX instead of hardcoding "rqueue-js-custom-consumer".
- NatsRetryAndDlqE2EIT: re-enabled, rewritten to publish a synthetic max-delivery
  advisory matching nats-server 2.12's payload shape and assert the bridge
  republishes onto the DLQ stream. Uses a blocking listener so the source message
  stays in flight (un-acked) — matching the production scenario the bridge is
  designed for. Going through a real handler-exhaustion path was racy because
  rqueue acks the message on numRetries exhaustion, so NATS never reaches its
  derived maxDeliver and never fires the advisory.

Local run (NATS_RUNNING=true, nats-server 2.12.8): 19 tests passing, 1 skipped
(the multi-listener fan-out test, which still requires per-queue retention-policy
work to support Limits/Interest retention — left @disabled with the existing
comment).

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

Author: sonus21

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

@@ -0,0 +1,121 @@
+/*
+ * 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.nats.js;
+
+import com.github.sonus21.rqueue.config.RqueueConfig;
+import com.github.sonus21.rqueue.core.EndpointRegistry;
+import com.github.sonus21.rqueue.core.spi.MessageBroker;
+import com.github.sonus21.rqueue.listener.QueueDetail;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+
+/**
+ * Bootstrap-time installer for the NATS-native dead-letter advisory bridge. For every active queue
+ * registered in {@link EndpointRegistry}, calls
+ * {@link JetStreamMessageBroker#installDeadLetterBridge(QueueDetail, String)} so that messages
+ * exceeding {@code maxDeliver} on the durable consumer are republished onto the queue's DLQ
+ * stream via the {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES} advisory subject.
+ *
+ * <p>This is the NATS-side equivalent of the Redis backend's {@code RqueueDeadLetterPublisher}:
+ * the rqueue post-processing handler already routes failed deliveries to a configured Rqueue-level
+ * DLQ ({@code @RqueueListener(deadLetterQueue=...)}); the advisory bridge registered here is an
+ * additional, independent path that catches messages whose handler exhausted retries without the
+ * post-processor noticing (e.g. listener restart, container shutdown mid-retry, JetStream-driven
+ * redelivery exhaustion outside the rqueue retry counter).
+ *
+ * <p><b>Lifecycle.</b> Implements {@link SmartInitializingSingleton} so it runs after every
+ * {@code @RqueueListener} bean has registered with {@link EndpointRegistry} and the
+ * {@link com.github.sonus21.rqueue.nats.js.NatsStreamValidator} has provisioned the underlying
+ * streams — but before {@code SmartLifecycle.start()} spawns the message pollers, so the bridge
+ * is in place before the first delivery attempt. Implements {@link DisposableBean} so the
+ * advisory dispatchers are torn down on context shutdown.
+ *
+ * <p><b>Producer-only mode.</b> When {@link RqueueConfig#isProducer()} is true the application
+ * has no listeners and therefore no consumers that could exhaust retries; the registrar exits
+ * early and installs nothing.
+ *
+ * <p><b>Backend gating.</b> Only does its work when the active broker is a
+ * {@link JetStreamMessageBroker}; on Redis or other backends the bean simply no-ops, so it is safe
+ * to wire unconditionally from the NATS auto-config (which is itself gated on
+ * {@code rqueue.backend=nats}).
+ */
+public class NatsDeadLetterBridgeRegistrar
+    implements SmartInitializingSingleton, DisposableBean {
+
+  private static final Logger log =
+      Logger.getLogger(NatsDeadLetterBridgeRegistrar.class.getName());
+
+  private final MessageBroker broker;
+  private final RqueueConfig rqueueConfig;
+  private final List<AutoCloseable> bridges = new ArrayList<>();
+
+  public NatsDeadLetterBridgeRegistrar(MessageBroker broker, RqueueConfig rqueueConfig) {
+    this.broker = broker;
+    this.rqueueConfig = rqueueConfig;
+  }
+
+  @Override
+  public void afterSingletonsInstantiated() {
+    if (rqueueConfig != null && rqueueConfig.isProducer()) {
+      log.log(Level.FINE,
+          "NatsDeadLetterBridgeRegistrar: producer-only mode — skipping bridge installation");
+      return;
+    }
+    if (!(broker instanceof JetStreamMessageBroker)) {
+      // Defensive — the bean is wired only by the NATS auto-config, but other backends could
+      // theoretically substitute a different MessageBroker via @Primary.
+      return;
+    }
+    JetStreamMessageBroker nb = (JetStreamMessageBroker) broker;
+    List<QueueDetail> queues = EndpointRegistry.getActiveQueueDetails();
+    if (queues.isEmpty()) {
+      return;
+    }
+    int installed = 0;
+    for (QueueDetail q : queues) {
+      String consumerName = q.resolvedConsumerName();
+      try {
+        bridges.add(nb.installDeadLetterBridge(q, consumerName));
+        installed++;
+      } catch (RuntimeException e) {
+        // Best-effort: a single failure must not abort listener startup. The rqueue-level DLQ
+        // path (PostProcessingHandler.moveToDlq) still works regardless.
+        log.log(Level.WARNING,
+            "Failed to install dead-letter advisory bridge for queue " + q.getName()
+                + " consumer " + consumerName + ": " + e.getMessage(), e);
+      }
+    }
+    log.log(Level.INFO,
+        "NatsDeadLetterBridgeRegistrar: installed {0} advisory bridge(s) across {1} queue(s)",
+        new Object[] {installed, queues.size()});
+  }
+
+  @Override
+  public void destroy() {
+    for (AutoCloseable c : bridges) {
+      try {
+        c.close();
+      } catch (Exception ignore) {
+        // best-effort close; we are shutting down anyway
+      }
+    }
+    bridges.clear();
+  }
+}

rqueue-spring-boot-starter/src/main/java/com/github/sonus21/rqueue/spring/boot/RqueueNatsAutoConfig.java

@@ -21,6 +21,7 @@
 import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
 import com.github.sonus21.rqueue.nats.internal.NatsProvisioner;
 import com.github.sonus21.rqueue.nats.js.JetStreamMessageBroker;
+import com.github.sonus21.rqueue.nats.js.NatsDeadLetterBridgeRegistrar;
 import com.github.sonus21.rqueue.nats.js.NatsStreamValidator;
 import com.github.sonus21.rqueue.nats.kv.NatsKvBucketValidator;
 import com.github.sonus21.rqueue.nats.metrics.NatsRqueueQueueMetricsProvider;
@@ -162,6 +163,19 @@ public NatsStreamValidator natsStreamValidator(
         natsProvisioner, toBrokerConfig(props), rqueueConfigProvider.getIfAvailable());
   }
 
+  /**
+   * Boot-time installer for the NATS-native dead-letter advisory bridge: for every registered
+   * queue, subscribes to {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<stream>.<consumer>}
+   * and republishes any message that exhausts {@code maxDeliver} onto the queue's DLQ stream.
+   * Complementary to the rqueue-level DLQ path ({@code PostProcessingHandler.moveToDlq}).
+   */
+  @Bean
+  @ConditionalOnMissingBean(NatsDeadLetterBridgeRegistrar.class)
+  public NatsDeadLetterBridgeRegistrar natsDeadLetterBridgeRegistrar(
+      MessageBroker broker, ObjectProvider<RqueueConfig> rqueueConfigProvider) {
+    return new NatsDeadLetterBridgeRegistrar(broker, rqueueConfigProvider.getIfAvailable());
+  }
+
   /**
    * Bean form of the KV-bucket validator. Other NATS beans {@code @DependsOn} this name so it runs
    * before they are constructed. The flag is sourced from {@link RqueueNatsProperties} —

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsBackendEndToEndIT.java

@@ -58,8 +58,8 @@
     classes = NatsBackendEndToEndIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsBackendEndToEndIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsBackendEndToEndIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsBackendEndToEndIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsBackendEndToEndIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 class NatsBackendEndToEndIT extends AbstractNatsBootIT {

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsConcurrencyE2EIT.java

@@ -44,8 +44,8 @@
     classes = NatsConcurrencyE2EIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsConcurrencyE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsConcurrencyE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsConcurrencyE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsConcurrencyE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 class NatsConcurrencyE2EIT extends AbstractNatsBootIT {

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsConsumerNameOverrideE2EIT.java

@@ -45,8 +45,8 @@
     classes = NatsConsumerNameOverrideE2EIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsConsumerNameOverrideE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsConsumerNameOverrideE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsConsumerNameOverrideE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsConsumerNameOverrideE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 class NatsConsumerNameOverrideE2EIT extends AbstractNatsBootIT {
@@ -73,7 +73,8 @@ void overriddenConsumerNameIsRegisteredOnTheStream() throws Exception {
     enqueuer.enqueue("custom-consumer", "hello");
     assertThat(listener.latch.await(20, TimeUnit.SECONDS)).isTrue();
 
-    ConsumerInfo info = jsm.getConsumerInfo("rqueue-js-custom-consumer", "my-custom-consumer");
+    ConsumerInfo info =
+        jsm.getConsumerInfo(STREAM_PREFIX + "custom-consumer", "my-custom-consumer");
     assertThat(info).isNotNull();
     assertThat(info.getName()).isEqualTo("my-custom-consumer");
   }

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsMultipleListenersOnSameQueueE2EIT.java

@@ -48,8 +48,8 @@
     classes = NatsMultipleListenersOnSameQueueE2EIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsMultipleListenersOnSameQueueE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsMultipleListenersOnSameQueueE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsMultipleListenersOnSameQueueE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsMultipleListenersOnSameQueueE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 @Disabled("Default JetStream retention=WorkQueue prevents true fan-out across multiple consumers; "

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsPriorityQueuesE2EIT.java

@@ -46,8 +46,8 @@
     classes = NatsPriorityQueuesE2EIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsPriorityQueuesE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsPriorityQueuesE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsPriorityQueuesE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsPriorityQueuesE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 class NatsPriorityQueuesE2EIT extends AbstractNatsBootIT {

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsReactiveEnqueueE2EIT.java

@@ -47,8 +47,8 @@
     properties = {
         "rqueue.backend=nats",
         "rqueue.reactive.enabled=true",
-        "rqueue.nats.stream-prefix=" + NatsReactiveEnqueueE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsReactiveEnqueueE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsReactiveEnqueueE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsReactiveEnqueueE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
 class NatsReactiveEnqueueE2EIT extends AbstractNatsBootIT {

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsRetryAndDlqE2EIT.java

@@ -19,45 +19,53 @@
 
 import com.github.sonus21.rqueue.annotation.RqueueListener;
 import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer;
+import io.nats.client.Connection;
 import io.nats.client.JetStreamManagement;
 import io.nats.client.api.StreamInfo;
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import org.awaitility.Awaitility;
 import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
 import org.springframework.boot.data.redis.autoconfigure.DataRedisReactiveAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
 import org.springframework.stereotype.Component;
 
 /**
- * After a handler exhausts {@code numRetries}, JetStream emits a max-deliveries advisory and the
- * broker's {@code installDeadLetterBridge} dispatcher republishes the payload onto the DLQ
- * stream. Currently disabled because {@link
- * com.github.sonus21.rqueue.spring.boot.RqueueNatsAutoConfig} does not yet invoke
- * {@code JetStreamMessageBroker.installDeadLetterBridge(...)} during container start, so dead-
- * lettered messages never reach the DLQ stream. Enable this test once that wiring is added.
+ * Verifies the NATS-native dead-letter advisory bridge installed by
+ * {@link com.github.sonus21.rqueue.nats.js.NatsDeadLetterBridgeRegistrar}: when JetStream emits
+ * {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<stream>.<consumer>} for a queue's durable
+ * consumer, the registrar's dispatcher looks up the offending message by sequence number and
+ * republishes it onto the queue's DLQ stream ({@code <streamPrefix><queue><dlqStreamSuffix>}).
+ *
+ * <p><b>Why a synthetic advisory.</b> In normal rqueue flow the framework either acks
+ * (success / forced-discard / moveToDlq) or naks (retry) every delivery, so NATS sees a terminal
+ * action before {@code maxDeliver} elapses and the advisory never fires from a real handler — that
+ * path is covered by
+ * {@code NatsSchedulingAdvancedE2EIT#scheduledMessageExhaustsRetriesToDlq} via the rqueue-level
+ * DLQ ({@code PostProcessingHandler.moveToDlq}). The advisory bridge is a defensive net for cases
+ * outside rqueue's control (process crash mid-handler, or a handler that blocks past its
+ * visibility timeout AND past every retry while NATS keeps redelivering). Triggering that path
+ * end-to-end is racy and slow, so this test instead publishes a synthetic advisory matching the
+ * shape {@code nats-server 2.12} actually emits and asserts the dispatcher reacts: enqueues a
+ * payload, looks up its stream sequence, fakes the advisory, and waits for the DLQ stream to
+ * receive it.
  */
 @SpringBootTest(
     classes = NatsRetryAndDlqE2EIT.TestApp.class,
     properties = {
         "rqueue.backend=nats",
-        "rqueue.nats.stream-prefix=" + NatsRetryAndDlqE2EIT.STREAM_PREFIX,
-        "rqueue.nats.subject-prefix=" + NatsRetryAndDlqE2EIT.SUBJECT_PREFIX
+        "rqueue.nats.naming.stream-prefix=" + NatsRetryAndDlqE2EIT.STREAM_PREFIX,
+        "rqueue.nats.naming.subject-prefix=" + NatsRetryAndDlqE2EIT.SUBJECT_PREFIX
     })
 @Tag("nats")
-@Disabled("This test exercises the NATS-native advisory bridge path"
-    + " (JetStreamMessageBroker.installDeadLetterBridge /"
-    + " $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES), which is not yet wired by"
-    + " RqueueNatsAutoConfig. The rqueue-level DLQ path (PostProcessingHandler → broker.moveToDlq)"
-    + " is already tested in NatsSchedulingAdvancedE2EIT#scheduledMessageExhaustsRetriesToDlq and"
-    + " works without this bridge. Enable this test once RqueueNatsAutoConfig provisions the"
-    + " advisory dispatcher per queue during container start.")
 class NatsRetryAndDlqE2EIT extends AbstractNatsBootIT {
 
   static final String STREAM_PREFIX = "rqueue-js-retryDlqE2E-";
@@ -72,35 +80,82 @@ static void wipeOwnedStreams() {
   RqueueMessageEnqueuer enqueuer;
 
   @Autowired
-  FailingListener listener;
+  Connection natsConnection;
 
   @Autowired
   JetStreamManagement jsm;
 
+  @Autowired
+  BlockingListener listener;
+
   @Test
-  void exhaustedMessageLandsOnDlqStream() {
-    enqueuer.enqueue("failing", "boom");
+  void advisoryBridgeRepublishesIntoDlqStream() throws Exception {
+    String stream = STREAM_PREFIX + "boom";
+    String dlqStream = stream + "-dlq";
+
+    // Publish via the normal enqueue path so the source stream + bridge get provisioned exactly
+    // the way they would in production. The blocking listener picks up the delivery but never
+    // returns, so the message stays in flight (un-acked) in the WorkQueue stream — which is the
+    // realistic state when JetStream actually fires the max-delivery advisory in production
+    // (handler hung, ack never sent). The long visibilityTimeout keeps NATS from redelivering
+    // during the test.
+    enqueuer.enqueue("boom", "marker-payload");
+
+    // Confirm the listener has the message in flight (proves the source stream still has it: in
+    // WorkQueue retention an unacked message stays in the stream until acked or AckWait expires).
+    assertThat(listener.received.await(20, TimeUnit.SECONDS))
+        .as("Listener must receive the marker payload before we synthesize the advisory")
+        .isTrue();
+
+    // Bridge installer provisioned the DLQ stream from boot (installDeadLetterBridge → provisionDlq).
+    Awaitility.await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
+      assertThat(jsm.getStreamInfo(dlqStream)).isNotNull();
+    });
+    long sourceSeq = jsm.getStreamInfo(stream).getStreamState().getLastSequence();
 
-    Awaitility.await().atMost(Duration.ofSeconds(60)).until(() -> listener.attempts.get() >= 2);
+    // Synthetic max-delivery advisory matching nats-server 2.12's payload shape: only stream_seq
+    // is required by the bridge's republish logic. The advisory subject must include
+    // <stream>.<consumer>; consumer name is the rqueue default for this queue, which is
+    // <queueName>-consumer (see QueueDetail#resolvedConsumerName).
+    String consumer = "boom-consumer";
+    String advisorySubject =
+        "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES." + stream + "." + consumer;
+    String advisoryJson = "{\"type\":\"io.nats.jetstream.advisory.v1.max_deliveries\","
+        + "\"stream\":\"" + stream + "\","
+        + "\"consumer\":\"" + consumer + "\","
+        + "\"stream_seq\":" + sourceSeq + ","
+        + "\"deliveries\":3}";
+    natsConnection.publish(advisorySubject, advisoryJson.getBytes(StandardCharsets.UTF_8));
+    natsConnection.flush(Duration.ofSeconds(2));
 
-    Awaitility.await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
-      StreamInfo dlq = jsm.getStreamInfo("rqueue-js-failing-dlq");
+    // The bridge's dispatcher should have republished the source message onto the DLQ stream.
+    Awaitility.await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+      StreamInfo dlq = jsm.getStreamInfo(dlqStream);
       assertThat(dlq.getStreamState().getMsgCount()).isGreaterThanOrEqualTo(1);
     });
   }
 
   @SpringBootApplication(
       exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  @Import(BlockingListener.class)
   static class TestApp {}
 
+  /**
+   * Receives the marker payload, signals the test, then blocks past the test's runtime so the
+   * message stays in flight (un-acked) in the WorkQueue source stream. That keeps it reachable via
+   * {@code jsm.getMessage(stream, seq)} — which is what the advisory bridge does on receipt — and
+   * mirrors the production scenario where the bridge fires (handler hung past
+   * {@code visibilityTimeout}, NATS still considers the message un-acked). The 2-minute
+   * visibility timeout prevents NATS from redelivering before the test completes.
+   */
   @Component
-  static class FailingListener {
-    final AtomicInteger attempts = new AtomicInteger();
+  static class BlockingListener {
+    final CountDownLatch received = new CountDownLatch(1);
 
-    @RqueueListener(value = "failing", numRetries = "2")
-    void onMessage(String payload) {
-      attempts.incrementAndGet();
-      throw new RuntimeException("simulated failure for payload=" + payload);
+    @RqueueListener(value = "boom", visibilityTimeout = "120000")
+    void onMessage(String payload) throws Exception {
+      received.countDown();
+      Thread.sleep(120_000L); // far exceeds the test's 30s budget
     }
   }
 }