25621809822

Committed 10 May 2026 06:27AM UTC coverage: 83.425% (+0.03%) from 83.396%

Build # 25621809822

Build Type

push

github

Committed by

web-flow

Commit Message

Nats scheduling fix (#297)

* ci: compile main sources in coverage_report job

The coverage_report job was producing an effectively empty
jacocoTestReport.xml (3.4KB vs ~1.1MB locally) because no .class files
existed when coverageReportOnly ran — the job checked out source code
and downloaded .exec artifacts, but never compiled. JaCoCo's report
generator skips packages/classes it cannot resolve, so the merged XML
ended up with only <sessioninfo> entries and no <package> elements.

That made coverallsJacoco silently no-op via the
"source file set empty, skipping" branch in CoverallsReporter, so
"Push coverage to Coveralls" reported success without uploading.

Verified by downloading the coverage-report artifact from a recent run
and comparing its XML structure against a local build's report.

Assisted-By: Claude Code

* nats-web: implement pause / soft-delete admin ops and capability-aware Q-detail

Replace the all-stub `NatsRqueueUtilityService` with real impls for the operations
JetStream can model: `pauseUnpauseQueue` persists the `paused` flag on `QueueConfig`
in the queue-config KV bucket and notifies the local listener container so the poller
stops dispatching; `deleteMessage` is a soft delete via `MessageMetadataService`
(stream message persists, dashboard hides via the metadata flag); `getDataType`
reports `STREAM`. `moveMessage`, `enqueueMessage`, and `makeEmpty` deliberately
remain "not supported" — there is no JetStream primitive for those.

Update `RqueueQDetailServiceImpl.getRunningTasks` / `getScheduledTasks` to return
header-only tables when the broker capabilities suppress those sections, instead of
emitting zero rows or 501s on NATS.

20 new unit tests cover the pause/delete paths and lock in the still-unsupported
operations. Updates `nats-task.md` / `nats-task-v2.md` to reflect what landed.

Assisted-By: Claude Code

* nats-web: capability-aware nav / charts and stream-based peek

End-to-end browser-tested the NATS dashboard and shipp... (continued)

Coverage Stats

2628 of 3487 branches covered (75.37%)

Branch coverage included in aggregate %.

128 of 179 new or added lines in 12 files covered. (71.51%)

1 existing line in 1 file now uncovered.

7841 of 9062 relevant lines covered (86.53%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.83

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

/*
 * 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();
  }
}

1	/*
2	* Copyright (c) 2026 Sonu Kumar
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* You may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* https://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and limitations under the License.
14	*
15	*/
16	package com.github.sonus21.rqueue.nats.js;
17
18	import com.github.sonus21.rqueue.config.RqueueConfig;
19	import com.github.sonus21.rqueue.core.EndpointRegistry;
20	import com.github.sonus21.rqueue.core.spi.MessageBroker;
21	import com.github.sonus21.rqueue.listener.QueueDetail;
22	import java.util.ArrayList;
23	import java.util.List;
24	import java.util.logging.Level;
25	import java.util.logging.Logger;
26	import org.springframework.beans.factory.DisposableBean;
27	import org.springframework.beans.factory.SmartInitializingSingleton;
28
29	/**
30	* Bootstrap-time installer for the NATS-native dead-letter advisory bridge. For every active queue
31	* registered in {@link EndpointRegistry}, calls
32	* {@link JetStreamMessageBroker#installDeadLetterBridge(QueueDetail, String)} so that messages
33	* exceeding {@code maxDeliver} on the durable consumer are republished onto the queue's DLQ
34	* stream via the {@code $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES} advisory subject.
35	*
36	* <p>This is the NATS-side equivalent of the Redis backend's {@code RqueueDeadLetterPublisher}:
37	* the rqueue post-processing handler already routes failed deliveries to a configured Rqueue-level
38	* DLQ ({@code @RqueueListener(deadLetterQueue=...)}); the advisory bridge registered here is an
39	* additional, independent path that catches messages whose handler exhausted retries without the
40	* post-processor noticing (e.g. listener restart, container shutdown mid-retry, JetStream-driven
41	* redelivery exhaustion outside the rqueue retry counter).
42	*
43	* <p><b>Lifecycle.</b> Implements {@link SmartInitializingSingleton} so it runs after every
44	* {@code @RqueueListener} bean has registered with {@link EndpointRegistry} and the
45	* {@link com.github.sonus21.rqueue.nats.js.NatsStreamValidator} has provisioned the underlying
46	* streams — but before {@code SmartLifecycle.start()} spawns the message pollers, so the bridge
47	* is in place before the first delivery attempt. Implements {@link DisposableBean} so the
48	* advisory dispatchers are torn down on context shutdown.
49	*
50	* <p><b>Producer-only mode.</b> When {@link RqueueConfig#isProducer()} is true the application
51	* has no listeners and therefore no consumers that could exhaust retries; the registrar exits
52	* early and installs nothing.
53	*
54	* <p><b>Backend gating.</b> Only does its work when the active broker is a
55	* {@link JetStreamMessageBroker}; on Redis or other backends the bean simply no-ops, so it is safe
56	* to wire unconditionally from the NATS auto-config (which is itself gated on
57	* {@code rqueue.backend=nats}).
58	*/
59	public class NatsDeadLetterBridgeRegistrar implements SmartInitializingSingleton, DisposableBean {
60
61	private static final Logger log = Logger.getLogger(NatsDeadLetterBridgeRegistrar.class.getName());	1✔
62
63	private final MessageBroker broker;
64	private final RqueueConfig rqueueConfig;
65	private final List<AutoCloseable> bridges = new ArrayList<>();	1✔
66
67	public NatsDeadLetterBridgeRegistrar(MessageBroker broker, RqueueConfig rqueueConfig) {	1✔
68	this.broker = broker;	1✔
69	this.rqueueConfig = rqueueConfig;	1✔
70	}	1✔
71
72	@Override
73	public void afterSingletonsInstantiated() {
74	if (rqueueConfig != null && rqueueConfig.isProducer()) {	1!
NEW 75	log.log(	×
76	Level.FINE,
77	"NatsDeadLetterBridgeRegistrar: producer-only mode — skipping bridge installation");
NEW 78	return;	×
79	}
80	if (!(broker instanceof JetStreamMessageBroker)) {	1!
81	// Defensive — the bean is wired only by the NATS auto-config, but other backends could
82	// theoretically substitute a different MessageBroker via @Primary.
NEW 83	return;	×
84	}
85	JetStreamMessageBroker nb = (JetStreamMessageBroker) broker;	1✔
86	List<QueueDetail> queues = EndpointRegistry.getActiveQueueDetails();	1✔
87	if (queues.isEmpty()) {	1✔
88	return;	1✔
89	}
90	int installed = 0;	1✔
91	for (QueueDetail q : queues) {	1✔
92	String consumerName = q.resolvedConsumerName();	1✔
93	try {
94	bridges.add(nb.installDeadLetterBridge(q, consumerName));	1✔
95	installed++;	1✔
NEW 96	} catch (RuntimeException e) {	×
97	// Best-effort: a single failure must not abort listener startup. The rqueue-level DLQ
98	// path (PostProcessingHandler.moveToDlq) still works regardless.
NEW 99	log.log(	×
100	Level.WARNING,
NEW 101	"Failed to install dead-letter advisory bridge for queue " + q.getName() + " consumer "	×
NEW 102	+ consumerName + ": " + e.getMessage(),	×
103	e);
104	}	1✔
105	}	1✔
106	log.log(	1✔
107	Level.INFO,
108	"NatsDeadLetterBridgeRegistrar: installed {0} advisory bridge(s) across {1} queue(s)",
109	new Object[] {installed, queues.size()});	1✔
110	}	1✔
111
112	@Override
113	public void destroy() {
114	for (AutoCloseable c : bridges) {	1!
115	try {
NEW 116	c.close();	×
NEW 117	} catch (Exception ignore) {	×
118	// best-effort close; we are shutting down anyway
NEW 119	}	×
NEW 120	}	×
121	bridges.clear();	1✔
122	}	1✔
123	}

sonus21 / rqueue / 25621809822

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous