refactor(framework): decouple Manager from TronJsonRpcImpl

[0xbigapple]

What does this PR do?

Removes the Manager → TronJsonRpcImpl reverse dependency left by tronprotocol#6732, where core-layer Manager holds the json-rpc filter consumer through a @Lazy injection.

• Adds a standalone FilterCapsuleQueue bean; Manager.postBlockFilter/postLogsFilter produce into it, so Manager no longer references anything under org.tron.core.services.jsonrpc (repo-wide framework/src/main is now @Lazy-free).
• Moves the consumer loop into TronJsonRpcImpl: started by @PostConstruct when isJsonRpcFilterEnabled(), single daemon thread, stopped by close().
• TronJsonRpcImpl constructor becomes (NodeInfoService, Wallet, Manager); setManager() is removed and the direct-construction test sites are migrated.
• close() is rewritten: an AtomicBoolean guard makes it idempotent and doubles as the loop's visibility-safe stop flag; the consumer executor is awaited before logsFilterPool shuts down (the previous order could reject the in-flight submit(...).join() and silently drop events).
• ApplicationImpl.shutdown() now closes the consumer after producers stop and before dbManager.close(), so the consumer structurally never outlives the database; the later Spring-destruction close() is a no-op.
• The instanceof dispatch logs a warning for unknown FilterTriggerCapsule subtypes instead of silently dropping them.

Runtime behavior of the filter API is unchanged: same unbounded queue, same discard-on-shutdown semantics, one consumer shared by the FullNode/solidity/PBFT json-rpc services.

Why are these changes required?

Follow-up agreed in the tronprotocol#6732 review (see discussion). @Lazy tolerates the cycle rather than removing it: FullNode sets allowCircularReferences(false), so no injection style can resolve the cycle — the reverse edge itself has to go, otherwise later json-rpc cleanups keep copying the pattern and core stays coupled to the API layer through a runtime proxy.

This PR has been tested by:

• Unit Tests
• Manual Testing
  • private chain: block production with registered filters; eth_getFilterChanges delivers; SIGTERM log order confirms consumer → logs-filter-pool → dbManager → context close, single shutdown episode, no errors
  • Nile and mainnet lite nodes: 5 restart cycles each with 100 filters created and polled per cycle (~2,500 queries/cycle, zero rpc errors); on mainnet the log path delivered 20k+ real contract log entries through the new queue; filters correctly report filter not found after restart; no RejectedExecutionException, no bean-cycle errors, shutdown order identical in every cycle

Follow up

Extra details

---

Summary by cubic

Decouples core Manager from JSON-RPC TronJsonRpcImpl by inserting a FilterCapsuleQueue between producers and the consumer. This removes the Spring cycle, hardens shutdown (idempotent close and clean exit on interrupt), and keeps filter API behavior unchanged.

• Introduces FilterCapsuleQueue bean with a narrow API (offer, timed poll; stream() is test-only). Manager.postBlockFilter/postLogsFilter offer to it; removes direct references to org.tron.core.services.jsonrpc. Drops unreachable offer-failure handling since the queue is unbounded.
• Moves the filter consumer loop into TronJsonRpcImpl; starts at @PostConstruct only when isJsonRpcFilterEnabled(), runs in a single daemon thread, exits on interrupt, and warns on unknown FilterTriggerCapsule subtypes.
• Makes TronJsonRpcImpl.close() idempotent and stops the consumer before logsFilterPool to avoid dropped events; ApplicationImpl.shutdown() closes TronJsonRpcImpl after producers stop and before dbManager.close(), logging any close failures.
• Adds end-to-end delivery and shutdown-race tests; updates existing tests to the new constructor.

Migration

• Replace direct constructions with new TronJsonRpcImpl(nodeInfoService, wallet, manager) and remove setManager(...) calls.
• If code inspected the filter queue directly, inject FilterCapsuleQueue and use offer(...); for tests, use its stream() accessor.

^{Written for commit d9db070. Summary will update on new commits.}

[coderabbitai]

Important

• 🔍 Trigger review

This repository does not receive automatic reviews because it has fewer than 10 stars.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 8608885c-3d54-4982-b84b-ef6a2422ead4

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[cubic-dev-ai]

3 issues found across 13 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java">

<violation number="1" location="framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java:63">
P3: `consumerThreadStartedOnce` asserts `Thread.getAllStackTraces()` has exactly one thread named "filter". That scan sees every live thread in the whole JVM, not just this test's context. The consumer is a daemon thread named "filter" (ExecutorServiceManager.newSingleThreadExecutor("filter", true)), so if another test class that also enables jsonRpcFilter holds its context open when this method runs (e.g. parallel Surefire, or a not-yet-torn-down daemon thread from a prior context), the count rises above 1 and the assertion fails spuriously. Scope the check to this instance, e.g. offer/await a capsule through the injected queue to verify the consumer loop is running, instead of counting global threads.</violation>
</file>

<file name="framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java">

<violation number="1" location="framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java:17">
P2: The queue is an unbounded LinkedBlockingQueue, so offer() always returns true for a non-null capsule. The producer call-sites in Manager.postBlockFilter/postLogsFilter treat a false return as a full queue and log "Too many filters, block filter lost", but that branch is unreachable, and there is no bound protecting memory if the consumer lags or stops. If loss/dropping on overflow is intended, give the queue a bounded capacity; otherwise the "lost filter" handling in Manager is dead code and the queue can grow without limit.</violation>
</file>

<file name="framework/src/main/java/org/tron/core/db/Manager.java">

<violation number="1" location="framework/src/main/java/org/tron/core/db/Manager.java:255">
P2: Manager.close() previously stopped the json-rpc filter consumer (stopFilterProcessThread). After this change the consumer is stopped only by ApplicationImpl.shutdown() -> TronJsonRpcImpl.close(); direct callers of the public Manager.close() with json-rpc filters enabled no longer terminate the consumer thread or its executor. Keep the leak-safe behavior by documenting/centralizing the shutdown contract, or have Manager.close() delegate the consumer shutdown so no code path leaves the daemon thread running.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

P2: The queue is an unbounded LinkedBlockingQueue, so offer() always returns true for a non-null capsule. The producer call-sites in Manager.postBlockFilter/postLogsFilter treat a false return as a full queue and log "Too many filters, block filter lost", but that branch is unreachable, and there is no bound protecting memory if the consumer lags or stops. If loss/dropping on overflow is intended, give the queue a bounded capacity; otherwise the "lost filter" handling in Manager is dead code and the queue can grow without limit.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At framework/src/main/java/org/tron/common/logsfilter/queue/FilterCapsuleQueue.java, line 17:

<comment>The queue is an unbounded LinkedBlockingQueue, so offer() always returns true for a non-null capsule. The producer call-sites in Manager.postBlockFilter/postLogsFilter treat a false return as a full queue and log "Too many filters, block filter lost", but that branch is unreachable, and there is no bound protecting memory if the consumer lags or stops. If loss/dropping on overflow is intended, give the queue a bounded capacity; otherwise the "lost filter" handling in Manager is dead code and the queue can grow without limit.</comment>

<file context>
@@ -0,0 +1,38 @@
+@Component
+public class FilterCapsuleQueue {
+
+  private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>();
+
+  public boolean offer(FilterTriggerCapsule capsule) {
</file context>

Suggested change

	private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>();
	private final BlockingQueue<FilterTriggerCapsule> queue = new LinkedBlockingQueue<>(10000);

[cubic-dev-ai]

P2: Manager.close() previously stopped the json-rpc filter consumer (stopFilterProcessThread). After this change the consumer is stopped only by ApplicationImpl.shutdown() -> TronJsonRpcImpl.close(); direct callers of the public Manager.close() with json-rpc filters enabled no longer terminate the consumer thread or its executor. Keep the leak-safe behavior by documenting/centralizing the shutdown contract, or have Manager.close() delegate the consumer shutdown so no code path leaves the daemon thread running.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At framework/src/main/java/org/tron/core/db/Manager.java, line 255:

<comment>Manager.close() previously stopped the json-rpc filter consumer (stopFilterProcessThread). After this change the consumer is stopped only by ApplicationImpl.shutdown() -> TronJsonRpcImpl.close(); direct callers of the public Manager.close() with json-rpc filters enabled no longer terminate the consumer thread or its executor. Keep the leak-safe behavior by documenting/centralizing the shutdown contract, or have Manager.close() delegate the consumer shutdown so no code path leaves the daemon thread running.</comment>

<file context>
@@ -253,8 +251,8 @@ public class Manager {
-  private boolean isRunFilterProcessThread = true;
-  private BlockingQueue<FilterTriggerCapsule> filterCapsuleQueue;
+  @Autowired
+  private FilterCapsuleQueue filterCapsuleQueue;
 
   @Getter
</file context>

[cubic-dev-ai]

P3: consumerThreadStartedOnce asserts Thread.getAllStackTraces() has exactly one thread named "filter". That scan sees every live thread in the whole JVM, not just this test's context. The consumer is a daemon thread named "filter" (ExecutorServiceManager.newSingleThreadExecutor("filter", true)), so if another test class that also enables jsonRpcFilter holds its context open when this method runs (e.g. parallel Surefire, or a not-yet-torn-down daemon thread from a prior context), the count rises above 1 and the assertion fails spuriously. Scope the check to this instance, e.g. offer/await a capsule through the injected queue to verify the consumer loop is running, instead of counting global threads.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At framework/src/test/java/org/tron/core/jsonrpc/FilterPipelineDeliveryTest.java, line 63:

<comment>`consumerThreadStartedOnce` asserts `Thread.getAllStackTraces()` has exactly one thread named "filter". That scan sees every live thread in the whole JVM, not just this test's context. The consumer is a daemon thread named "filter" (ExecutorServiceManager.newSingleThreadExecutor("filter", true)), so if another test class that also enables jsonRpcFilter holds its context open when this method runs (e.g. parallel Surefire, or a not-yet-torn-down daemon thread from a prior context), the count rises above 1 and the assertion fails spuriously. Scope the check to this instance, e.g. offer/await a capsule through the injected queue to verify the consumer loop is running, instead of counting global threads.</comment>

<file context>
@@ -0,0 +1,100 @@
+
+  @Test
+  public void consumerThreadStartedOnce() {
+    long count = Thread.getAllStackTraces().keySet().stream()
+        .filter(t -> "filter".equals(t.getName())).count();
+    Assert.assertEquals(1, count);
</file context>